Get basic signals working

Author: Eeems

liboxide/_apps.py

@@ -39,6 +39,23 @@ class Application(APIObject):
         ("unregister", None),
         ("setEnvironment", None, dict[str, object]),
     ]
+    signals = [
+        ("launched",),
+        ("paused",),
+        ("resumed",),
+        ("unregistered",),
+        ("exited", int),
+        ("permissionsChanged", list[str]),
+        ("displayNameChanged", str),
+        ("onPauseChanged", str),
+        ("onResumeChanged", str),
+        ("onStopChanged", str),
+        ("autoStartChanged", bool),
+        ("iconChanged", str),
+        ("environmentChanged", list[str]),
+        ("workingDirectoryChanged", str),
+        ("directoriesChanged", str),
+    ]
 
 
 @static_init
@@ -62,6 +79,14 @@ class AppsAPI(API):
         ("getApplicationPath", str, str),
         ("previousApplication", Application),
     ]
+    signals = [
+        ("applicationRegistered", Application),
+        ("applicationLaunched", Application),
+        ("applicationUnregistered", Application),
+        ("applicationPaused", Application),
+        ("applicationResumed", Application),
+        ("applicationExited", Application, int),
+    ]
 
     @classmethod
     def getApplication(cls, name):

liboxide/_base.py

@@ -1,18 +1,33 @@
-from ._util import _registerProperty
+from ._util import _listenMethod
 from ._util import _registerMethod
+from ._util import _registerProperty
+
+
+class SignalHandler:
+    def __init__(self, host):
+        self.__host = host
+
+    def __getattr__(self, name):
+        for _name, *signature in self.__host.signals:
+            if name == _name:
+                return _listenMethod(self.__host, name, *signature)
+
+        raise NotImplementedError(name)
 
 
 class API:
     rotArgs = []
     properties = []
     methods = []
+    signals = []
 
     @staticmethod
     def _raise(e):
         raise e
 
     @classmethod
     def static_init(cls):
+        cls.on = SignalHandler(cls)
         cls.rotArgs = [cls.__name__.lower()[:-3]]
         for name in cls.properties:
             _registerProperty(cls, name)
@@ -26,8 +41,10 @@ class APIObject:
     folder = None
     properties = []
     methods = []
+    signals = []
 
     def __init__(self, path):
+        self.on = SignalHandler(self)
         if self.__class__.folder is None:
             self.__class__.folder = self.__class__.__name__.lower()
 

liboxide/_errors.py

@@ -4,3 +4,11 @@ class TooManyArguments(Exception):
 
 class TooFewArguments(Exception):
     pass
+
+
+class ObjectNotFound(Exception):
+    pass
+
+
+class RotException(Exception):
+    pass

liboxide/_notification.py

@@ -19,6 +19,12 @@ class Notification(APIObject):
         ("remove", None),
         ("click", None),
     ]
+    signals = [
+        ("changed", dict[str, str]),
+        ("removed",),
+        ("displayed",),
+        ("clicked",),
+    ]
 
 
 @static_init
@@ -33,6 +39,11 @@ class NotificationAPI(API):
         ("take", Notification, bool, str),
         ("get", Notification, str),
     ]
+    signals = [
+        ("notificationAdded", Notification),
+        ("notificationRemoved", Notification),
+        ("notificationChanged", Notification),
+    ]
 
     @classproperty
     def notifications(cls):

liboxide/_power.py

@@ -11,3 +11,12 @@ class PowerAPI(API):
         "batteryTemperature",
         "chargerState",
     ]
+    signals = [
+        ("stateChanged", int),
+        ("batteryStateChanged", int),
+        ("batteryLevelChanged", int),
+        ("batteryTemperatureChanged", int),
+        ("batteryWarning",),
+        ("batteryAlert",),
+        ("chargerWarning",),
+    ]

liboxide/_screen.py

@@ -14,6 +14,10 @@ class Screenshot(APIObject):
     methods = [
         ("remove", None),
     ]
+    signals = [
+        ("modified",),
+        ("removed",),
+    ]
 
 
 @static_init
@@ -26,6 +30,11 @@ class ScreenAPI(API):
         ("drawFullScreenImage", Screenshot, str),
         ("screenshot", Screenshot),
     ]
+    signals = [
+        ("screenshotAdded", Screenshot),
+        ("screenshotRemoved", Screenshot),
+        ("screenshotModified", Screenshot),
+    ]
 
     @classproperty
     def screenshots(cls):

liboxide/_system.py

@@ -19,3 +19,14 @@ class SystemAPI(API):
         ("inhibitPowerOff", None),
         ("uninhibitPowerOff", None),
     ]
+    signals = [
+        ("leftAction",),
+        ("homeAction",),
+        ("rightAction",),
+        ("powerAction",),
+        ("sleepInhibitedChanged", bool),
+        ("powerOffInhibitedChanged", bool),
+        ("autoSleepChanged", bool),
+        ("deviceSuspending",),
+        ("deviceResuming",),
+    ]

liboxide/_util.py

@@ -4,6 +4,8 @@
 
 from ._errors import TooFewArguments
 from ._errors import TooManyArguments
+from ._errors import ObjectNotFound
+from ._errors import RotException
 
 
 def static_init(cls):
@@ -24,9 +26,16 @@ def __contains__(self, obj):
 
 
 def rot(*args):
-    stdout = subprocess.check_output(["/opt/bin/rot"] + list(args))
-    if stdout:
-        return json.loads(stdout)
+    try:
+        stdout = subprocess.check_output(["/opt/bin/rot"] + list(args))
+        if stdout:
+            return json.loads(stdout)
+    except subprocess.CalledProcessError as e:
+        stdout = e.output.decode("utf-8")
+        if "org.freedesktop.DBus.Error.UnknownObject" in stdout:
+            raise ObjectNotFound(stdout) from e
+
+        raise RotException(stdout) from e
 
     return None
 
@@ -44,6 +53,21 @@ def fn(self):
         setattr(obj.__class__, name, property(fn))
 
 
+def _convert(_type, val):
+    if _type is None:
+        return None
+
+    from ._base import APIObject
+
+    if issubclass(_type, APIObject):
+        return _type(val)
+
+    if _type in (str, bool, int):
+        return val
+
+    raise NotImplementedError(_type)
+
+
 def _createMethod(cls, name, retType, *signature):
     def fn(cls, *args):
         if len(args) > len(signature):
@@ -68,23 +92,11 @@ def fn(cls, *args):
             elif issubclass(argType, dict):
                 qtype = "QVariant"
             else:
-                raise NotImplementedError()
+                raise NotImplementedError(argType)
 
             arguments.append(f"{qtype}:{json.dumps(args[0])}")
 
-        res = rot(*cls.rotArgs, "call", name, *arguments)
-        if retType is None:
-            return None
-
-        from ._base import APIObject
-
-        if issubclass(APIObject, retType):
-            return retType(res)
-
-        if retType in (str, bool, int):
-            return res
-
-        raise NotImplementedError()
+        return _convert(retType, rot(*cls.rotArgs, "call", name, *arguments))
 
     return fn
 
@@ -105,3 +117,18 @@ def _registerMethod(obj, name, retType, *signature):
             name,
             _createMethod(obj, name, retType, *signature),
         )
+
+
+def _listenMethod(obj, name, *signature):
+    if len(signature) > 1:
+        raise NotImplementedError()
+
+    # TODO cache these instead of creating them every time
+    def fn():
+        res = rot(*obj.rotArgs, "--once", "listen", name)
+        if signature:
+            return _convert(signature[0], res)
+
+        return res
+
+    return fn

liboxide/_wifi.py

@@ -19,6 +19,13 @@ class Network(APIObject):
         ("connect", None),
         ("remove", None),
     ]
+    signals = [
+        ("stateChanged", bool),
+        ("propertiesChanged", dict[str, str]),
+        ("connected",),
+        ("disconnected",),
+        ("removed",),
+    ]
 
     @property
     def bSSs(self):
@@ -42,6 +49,10 @@ class BSS(APIObject):
     methods = [
         ("connect", Network),
     ]
+    signals = [
+        ("removed",),
+        ("propertiesChanged", dict[str, str]),
+    ]
 
     @property
     def network(self):
@@ -78,6 +89,17 @@ class WifiAPI(API):
         ("removeBlob", None, str),
         ("getBlob", str, str),
     ]
+    signals = [
+        ("stateChanged", int),
+        ("linkChanged", int),
+        ("networkAdded", Network),
+        ("networkRemoved", Network),
+        ("networkConnected", Network),
+        ("disconnected",),
+        ("bssFound", BSS),
+        ("bssRemoved", BSS),
+        ("scanningChanged", bool),
+    ]
 
     @classproperty
     def bSSs(cls):

pyproject.toml

@@ -2,7 +2,7 @@
 name = "liboxide"
 description = "Python wrapper around the oxide command line tools."
 readme = "README.md"
-version = "0.0.1"
+version = "0.1.0"
 requires-python = ">= 3.9"
 authors = [
     {name = "Nathaniel van Diepen", email = "[email protected]"},

test.py

@@ -18,3 +18,6 @@
 print(f"Lockscreen App: {app.displayName}")
 autoLock = liboxide.settings.get("autoLock")
 print(f"Automatic lock: {autoLock} minutes")
+liboxide.wifi.scan(False)
+bss = liboxide.wifi.on.bssFound()
+print(f"Found BSS for SSID: {bss.ssid}")