Continue with client daemon implmentaton.

antidrift.py

@@ -1,113 +1,74 @@
 #!/usr/bin/env python3
+""" AntiDrift """
 
 import logging
 import shutil
 import sys
 import time
 import re
-import xwindow
 from pathlib import Path
-from config import Config, Block
 from typing import List
-from functools import lru_cache
 from gi.repository import GLib, GObject
 import dbus
 import dbus.service
 from dbus.mainloop.glib import DBusGMainLoop
-DBusGMainLoop(set_as_default=True)
+from antidrift.config import Config, Block
+from antidrift.daemon import AntiDriftDaemon
 
 
+def init_logging(log_file: Path):
+    class DuplicateFilter(logging.Filter):
+        def filter(self, record):
+            current_log = (record.module, record.levelno, record.msg)
+            if current_log != getattr(self, "last_log", None):
+                self.last_log = current_log
+                return True
+            return False
+
+    logging.basicConfig(filename=log_file,
+                        format='%(asctime)s | %(levelname)-8s | %(message)s',
+                        datefmt='%H:%M:%S',
+                        encoding='utf-8',
+                        level=logging.DEBUG)
+    logger = logging.getLogger()
+    logger.addFilter(DuplicateFilter())
+
+
+def antidrift_is_running() -> bool:
+    bus = dbus.SessionBus()
+    try:
+        bus_object = bus.get_object("com.antidrift", "/com/antidrift")
+        interface = dbus.Interface(bus_object, "com.antidrift")
+    except dbus.exceptions.DBusException:
+        return False
+    reply = interface.status()
+    if reply == "running":
+        return True
+    return False
+
+
+def client_mode(config: Config):
+    print("client_mode")
+
 
 def check_for_xdotool():
     r = shutil.which("xdotool")
     if not r:
-        # TODO: kill X here?
         logging.critical("Please install xdotool")
         sys.exit(1)
 
 
-def init_logging(log_file: Path):
-    format = '%(levelname)-8s | %(message)s'
-    logging.basicConfig(filename=log_file, format=format,
-                        encoding='utf-8', level=logging.DEBUG)
-
-
-@lru_cache(1)
-def log_debug(message: str):
-    logging.info(message)
-
-
-def window_is_blocked(config: Config) -> bool:
-    # These should be selectable in the future (not all at the same time)
-    blackblocks = config.blackblocks
-    whiteblocks = config.whiteblocks
-
-    window = xwindow.XWindow()
-    if not window.keywords:
-        return False
-    for b in blackblocks:
-        for k in b.keywords:
-            if k in window.keywords:
-                xwindow.notify(f"{window.name[:30]} blocked by {b.name}.")
-                logging.warning(f"{window.name[:30]} blocked by {b.name}.")
-                return True
-    if config.blackblocks_only:
-        log_debug(f"All non-blacklisted windows are allowed.")
-        return False
-    for w in whiteblocks:
-        for k in w.keywords:
-            if k in window.keywords:
-                log_debug(f"{window.name[:30]} allowed by {w.name}.")
-                return False
-    xwindow.notify(f"{window.name[:30]} not on any whitelist.")
-    return True
-
-
-def enforce(config: Config):
-    if not window_is_blocked(config):
-        return
-    xwindow.notify(f"AntiDrift will minimize the window in {config.minimize_delay} seconds.")
-    time.sleep(config.minimize_delay)
-    window = xwindow.XWindow()
-    if window_is_blocked(config):
-        window = xwindow.XWindow()
-        xwindow.notify(f"Minimize {window.name[:30]}.")
-        window.minimize()
-    else:
-        xwindow.notify(f"We are gucci again.")
-
-
-OPATH = "/com/antidrift"
-IFACE = "com.antidrift"
-BUS_NAME = "com.antidrift"
-class AntiDriftService(dbus.service.Object):
-    def __init__(self):
-        bus = dbus.SessionBus()
-        bus.request_name(BUS_NAME)
-        bus_name = dbus.service.BusName(BUS_NAME, bus=bus)
-        dbus.service.Object.__init__(self, bus_name, OPATH)
-
-    @dbus.service.method(dbus_interface=IFACE,
-                         in_signature="", out_signature="")
-    def hello(self):
-        print("hello, world")
-
-
 def main() -> None:
+    check_for_xdotool()
     config = Config.load("config.yaml")
     init_logging(config.log_file)
-    logging.info("AntiDrift running")
+    if antidrift_is_running():
-    check_for_xdotool()
+        client_mode(config)
-    def _enforce():
+    else:
-        enforce(config)
+        a = AntiDriftDaemon(config)
-        GLib.timeout_add(config.check_delay, _enforce)
+        a.run()
-    logging.info(f"Start session with {len(config.whiteblocks)} whiteblocks")
-    _enforce()
-    AntiDriftService()
-    mainloop = GLib.MainLoop()
-    mainloop.run()
 
 
+DBusGMainLoop(set_as_default=True)
 if __name__ == "__main__":
     main()
-

antidrift/daemon.py

@@ -0,0 +1,84 @@
+import logging
+import shutil
+import sys
+import time
+import re
+import antidrift.xwindow as xwindow
+from pathlib import Path
+from antidrift.config import Config, Block
+from typing import List
+from functools import lru_cache
+from gi.repository import GLib, GObject
+import dbus
+import dbus.service
+from dbus.mainloop.glib import DBusGMainLoop
+
+BUS_NAME = "com.antidrift"
+IFACE = "com.antidrift"
+OPATH = "/com/antidrift"
+
+
+class AntiDriftDaemon(dbus.service.Object):
+    def __init__(self, config: Config):
+        bus = dbus.SessionBus()
+        bus.request_name(BUS_NAME)
+        bus_name = dbus.service.BusName(BUS_NAME, bus=bus)
+        dbus.service.Object.__init__(self, bus_name, OPATH)
+        self.config = config
+
+    @dbus.service.method(dbus_interface=IFACE,
+                         in_signature="", out_signature="s")
+    def status(self):
+        logging.info("status requested")
+        return "running"
+
+    def run(self):
+        logging.info("AntiDriftDaemon run")
+        def _enforce():
+            enforce(self.config)
+            GLib.timeout_add(self.config.check_delay, _enforce)
+        logging.info(f"Start session with {len(self.config.whiteblocks)} whiteblocks")
+        _enforce()
+        mainloop = GLib.MainLoop()
+        mainloop.run()
+
+
+def window_is_blocked(config: Config) -> bool:
+    # These should be selectable in the future (not all at the same time)
+    blackblocks = config.blackblocks
+    whiteblocks = config.whiteblocks
+
+    window = xwindow.XWindow()
+    if not window.keywords:
+        return False
+    for b in blackblocks:
+        for k in b.keywords:
+            if k in window.keywords:
+                xwindow.notify(f"{window.name[:30]} blocked by {b.name}.")
+                logging.warning(f"{window.name[:30]} blocked by {b.name}.")
+                return True
+    if config.blackblocks_only:
+        logging.debug(f"All non-blacklisted windows are allowed.")
+        return False
+    for w in whiteblocks:
+        for k in w.keywords:
+            if k in window.keywords:
+                logging.debug(f"{window.name[:30]} allowed by {w.name}.")
+                return False
+    xwindow.notify(f"{window.name[:30]} not on any whitelist.")
+    return True
+
+
+def enforce(config: Config):
+    if not window_is_blocked(config):
+        return
+    xwindow.notify(f"AntiDrift will minimize the window in {config.minimize_delay} seconds.")
+    time.sleep(config.minimize_delay)
+    window = xwindow.XWindow()
+    if window_is_blocked(config):
+        window = xwindow.XWindow()
+        xwindow.notify(f"Minimize {window.name[:30]}.")
+        window.minimize()
+    else:
+        xwindow.notify(f"We are gucci again.")
+