Merge pull request #81 from Project-Babble/feat/desktop-notifications

(feat) Add Linux desktop notifications

BabbleApp/babbleapp.py

@@ -21,23 +21,25 @@
 import queue
 import requests
 import threading
-from ctypes import windll, c_int
+import asyncio
+from ctypes import c_int
 from babble_model_loader import *
 from camera_widget import CameraWidget
 from config import BabbleConfig
-from tab import CamInfo, Tab
+from tab import Tab
 from osc import VRChatOSCReceiver, VRChatOSC
 from general_settings_widget import SettingsWidget
 from algo_settings_widget import AlgoSettingsWidget
 from calib_settings_widget import CalibSettingsWidget
+from notification_manager import NotificationManager
 from utils.misc_utils import EnsurePath, is_nt, bg_color_highlight, bg_color_clear
 from lang_manager import LocaleStringManager as lang
 
 winmm = None
 
 if is_nt:
-    from winotify import Notification
     try:
+        from ctypes import windll
         winmm = windll.winmm
     except OSError:
         print(f'\033[91m[{lang._instance.get_string("log.error")}] {lang._instance.get_string("error.winmm")}.\033[0m')
@@ -57,7 +59,7 @@
 ALGO_SETTINGS_RADIO_NAME = "-ALGOSETTINGSRADIO-"
 CALIB_SETTINGS_RADIO_NAME = "-CALIBSETTINGSRADIO-"
 
-page_url = "https://github.com/SummerSigh/ProjectBabble/releases/latest"
+page_url = "https://github.com/Project-Babble/ProjectBabble/releases/latest"
 appversion = "Babble v2.0.7"
 
 def timerResolution(toggle):
@@ -70,69 +72,53 @@ def timerResolution(toggle):
         else:
             winmm.timeEndPeriod(1)
 
-def main():
-    EnsurePath()
-
-    # Get Configuration
-    config: BabbleConfig = BabbleConfig.load()
-
-    # Init locale manager
-    lang("Locale", config.settings.gui_language)
-
-    config.save()
-
-    cancellation_event = threading.Event()
-    ROSC = False
-    # Check to see if we can connect to our video source first. If not, bring up camera finding
-    # dialog.
-
+async def check_for_updates(config, notification_manager):
     if config.settings.gui_update_check:
         try:
             response = requests.get(
-                "https://api.github.com/repos/SummerSigh/ProjectBabble/releases/latest"
+                "https://api.github.com/repos/Project-Babble/ProjectBabble/releases/latest"
             )
             latestversion = response.json()["name"]
-            if (
-                appversion == latestversion
-            ):  # If what we scraped and hardcoded versions are same, assume we are up to date.
+
+            if appversion == latestversion:
                 print(
                     f'\033[92m[{lang._instance.get_string("log.info")}] {lang._instance.get_string("babble.latestVersion")}! [{latestversion}]\033[0m'
                 )
             else:
                 print(
                     f'\033[93m[{lang._instance.get_string("log.info")}] {lang._instance.get_string("babble.needUpdateOne")} [{appversion}] {lang._instance.get_string("babble.needUpdateTwo")} [{latestversion}] {lang._instance.get_string("babble.needUpdateThree")}.\033[0m'
                 )
-                try:
-                    if is_nt:
-                        cwd = os.getcwd()
-                        icon = cwd + "\Images\logo.ico"
-                        toast = Notification(
-                            app_id=lang._instance.get_string("babble.name"),
-                            title=lang._instance.get_string("babble.updatePresent"),
-                            msg=f'{lang._instance.get_string("babble.updateTo")} {latestversion}',
-                            icon=r"{}".format(icon),
-                        )
-                        toast.add_actions(
-                            label=lang._instance.get_string("babble.downloadPage"),
-                            launch="https://github.com/SummerSigh/ProjectBabble/releases/latest",
-                        )
-                        toast.show()
-                except Exception as e:
-                    print(
-                        f'[{lang._instance.get_string("log.info")}] {lang._instance.get_string("babble.noToast")}'
-                    )
-        except:
+                await notification_manager.show_notification(appversion, latestversion, page_url)
+        except Exception as e:
             print(
-                f'[{lang._instance.get_string("log.info")}] {lang._instance.get_string("babble.noInternet")}.'
+                f'[{lang._instance.get_string("log.info")}] {lang._instance.get_string("babble.noInternet")}. Error: {e}'
             )
+
+async def async_main():
+    EnsurePath()
+
+    # Get Configuration
+    config: BabbleConfig = BabbleConfig.load()
+
+    # Init locale manager
+    lang("Locale", config.settings.gui_language)
+
+    config.save()
+
+    notification_manager = NotificationManager()
+    await notification_manager.initialize()
+
+    # Run the update check
+    await check_for_updates(config, notification_manager)
+
+    cancellation_event = threading.Event()
+    ROSC = False
+
     timerResolution(True)
-    # Check to see if we have an ROI. If not, bring up ROI finder GUI.
 
-    # Spawn worker threads
     osc_queue: queue.Queue[tuple[bool, int, int]] = queue.Queue(maxsize=10)
     osc = VRChatOSC(cancellation_event, osc_queue, config)
     osc_thread = threading.Thread(target=osc.run)
-    # start worker threads
     osc_thread.start()
 
     cams = [
@@ -233,7 +219,6 @@ def main():
 
     tint = 33
     fs = False
-    # GUI Render loop
     while True:
         # First off, check for any events from the GUI
         event, values = window.read(timeout=tint)
@@ -342,7 +327,12 @@ def main():
         for setting in settings:
             if setting.started():
                 setting.render(window, event, values)
-
+
+        # Does adding this help?
+        # await asyncio.sleep(0)
+
+def main():
+    asyncio.run(async_main())
 
 if __name__ == "__main__":
     main()

BabbleApp/notification_manager.py

@@ -0,0 +1,53 @@
+import os
+import webbrowser
+import signal
+import asyncio
+from pathlib import Path
+from desktop_notifier import DesktopNotifier, Urgency, Button, Icon, DEFAULT_SOUND
+from lang_manager import LocaleStringManager as lang
+from utils.misc_utils import  is_nt
+
+class NotificationManager:
+    def __init__(self):
+        self.notifier = DesktopNotifier(app_name="Babble App")
+        self.loop = None
+        self.stop_event = None
+
+    async def show_notification(self, appversion, latestversion, page_url):
+        logo = Icon(
+            path=Path(os.path.join(os.getcwd(), "Images", "logo.ico"))
+        )
+
+        notification_message = (
+            f'{lang._instance.get_string("babble.needUpdateOne")} '
+            f'{appversion} '
+            f'{lang._instance.get_string("babble.needUpdateTwo")} '
+            f'{latestversion} '
+            f'{lang._instance.get_string("babble.needUpdateThree")}!'
+        )
+
+        await self.notifier.send(
+            title=lang._instance.get_string("babble.updatePresent"),
+            message=notification_message,
+            urgency=Urgency.Normal,
+            buttons=[
+                Button(
+                    title=lang._instance.get_string("babble.downloadPage"),
+                    on_pressed=lambda: webbrowser.open(page_url)
+                )
+            ],
+            icon=logo,
+            sound=DEFAULT_SOUND,
+            on_dismissed=lambda: self.stop_event.set()
+        )
+
+        await self.stop_event.wait()
+
+    async def initialize(self):
+        self.stop_event = asyncio.Event()
+        self.loop = asyncio.get_running_loop()
+
+        # Add non nt signal handlers
+        if not is_nt:
+            self.loop.add_signal_handler(signal.SIGINT, self.stop_event.set)
+            self.loop.add_signal_handler(signal.SIGTERM, self.stop_event.set)

BabbleApp/requirements.txt

@@ -9,7 +9,7 @@ torchvision==0.19.1
 pydantic==1.10.18
 pyserial==3.5
 colorama==0.4.6
-winotify==1.1.0
+desktop-notifier==6.0.0
 comtypes==1.4.8
 pygrabber==0.2
 poetry