Implement hello world dbus and assynchronous enforcing.

antidrift.py

@@ -6,20 +6,35 @@ 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)
+
 
 
 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():
-    FORMAT = '%(levelname)-8s | %(message)s'
-    logging.basicConfig(format=FORMAT, level=logging.DEBUG)
+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:
@@ -37,12 +52,12 @@ def window_is_blocked(config: Config) -> bool:
                 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.")
+        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:
-                logging.debug(f"{window.name[:30]} allowed by {w.name}.")
+                log_debug(f"{window.name[:30]} allowed by {w.name}.")
                 return False
     xwindow.notify(f"{window.name[:30]} not on any whitelist.")
     return True
@@ -62,19 +77,35 @@ def enforce(config: Config):
         xwindow.notify(f"We are gucci again.")
 
 
-def session(config: Config):
-    logging.info(f"Start session with {len(config.whiteblocks)} whiteblocks")
-    while True:
-        enforce(config)
-        time.sleep(config.check_delay)
+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:
-    init_logging()
+    config = Config.load("config.yaml")
+    init_logging(config.log_file)
     logging.info("AntiDrift running")
     check_for_xdotool()
-    config = Config.load("config.yaml")
-    session(config)
+    def _enforce():
+        enforce(config)
+        GLib.timeout_add(config.check_delay, _enforce)
+    logging.info(f"Start session with {len(config.whiteblocks)} whiteblocks")
+    _enforce()
+    AntiDriftService()
+    mainloop = GLib.MainLoop()
+    mainloop.run()
 
 
 if __name__ == "__main__":

config.py

@@ -11,15 +11,12 @@ class Block(BaseModel):
     keywords: List[str]
 
 
-def load(filename: str) -> List[Block]:
-    result = [Block(path=filename, **block) for block in block_list]
-    return result
-
-
 class Config(BaseModel):
     blackblocks: List[Block]
     whiteblocks: List[Block]
-    check_delay: int = 1
+    log_file: Path = Path()
+    config_file: Path = Path()
+    check_delay: int = 1000
     minimize_delay: int = 5
     blackblocks_only: bool = True
 
@@ -32,5 +29,6 @@ class Config(BaseModel):
         with open(config_file, "r") as f:
             config_dict = yaml.safe_load(f)
         config = cls(**config_dict)
+        config.config_file = Path(config_file)
         return config