Clean up code base.

antidrift.py

@@ -4,27 +4,20 @@
 import logging
 import shutil
 import sys
-import time
-import re
 from pathlib import Path
-from typing import List
+import antidrift.client as client
-from gi.repository import GLib, GObject
+from antidrift.config import Config
-import dbus
-import dbus.service
-from dbus.mainloop.glib import DBusGMainLoop
-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):
+        def filter(self, record) -> bool:
             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',
@@ -34,41 +27,25 @@ def init_logging(log_file: Path):
     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")
+    """ Check if xdotool is in path and exit if not """
-    if not r:
+    result = shutil.which("xdotool")
+    if not result:
         logging.critical("Please install xdotool")
         sys.exit(1)
 
 
 def main() -> None:
+    """ Main routine that dispatches to client or daemon """
     check_for_xdotool()
     config = Config.load("config.yaml")
     init_logging(config.log_file)
-    if antidrift_is_running():
+    if client.antidrift_is_running():
-        client_mode(config)
+        client.client_mode(config)
     else:
-        a = AntiDriftDaemon(config)
+        add = AntiDriftDaemon(config)
-        a.run()
+        add.run()
 
 
-DBusGMainLoop(set_as_default=True)
 if __name__ == "__main__":
     main()

antidrift/__init__.py

@@ -0,0 +1 @@
+""" AntiDrift module init """

antidrift/client.py

@@ -0,0 +1,24 @@
+from antidrift.config import Config
+from antidrift.daemon import DaemonStatus
+import dbus
+import dbus.service
+from dbus.mainloop.glib import DBusGMainLoop
+DBusGMainLoop(set_as_default=True)
+
+
+def antidrift_is_running() -> bool:
+    """ Check if AntiDrift is running via the DBUS """
+    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 == DaemonStatus.RUNNING.value:
+        return True
+    return False
+
+
+def client_mode(_config: Config):
+    print("client_mode")

antidrift/config.py

@@ -1,4 +1,3 @@
-import json
 import os
 import yaml
 from pathlib import Path
@@ -31,4 +30,3 @@ class Config(BaseModel):
         config = cls(**config_dict)
         config.config_file = Path(config_file)
         return config
-

antidrift/daemon.py

@@ -1,23 +1,22 @@
 import logging
-import shutil
-import sys
 import time
-import re
 import antidrift.xwindow as xwindow
-from pathlib import Path
+from antidrift.config import Config
-from antidrift.config import Config, Block
+from gi.repository import GLib
-from typing import List
+from enum import Enum
-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 DaemonStatus(Enum):
+    RUNNING = "running"
+    ERROR = "error"
+
+
 class AntiDriftDaemon(dbus.service.Object):
     def __init__(self, config: Config):
         bus = dbus.SessionBus()
@@ -28,16 +27,16 @@ class AntiDriftDaemon(dbus.service.Object):
 
     @dbus.service.method(dbus_interface=IFACE,
                          in_signature="", out_signature="s")
-    def status(self):
+    def status(self) -> str:
-        logging.info("status requested")
+        return DaemonStatus.RUNNING.value
-        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")
+
+        logging.info("AntiDriftDaemon run")
         _enforce()
         mainloop = GLib.MainLoop()
         mainloop.run()
@@ -58,7 +57,7 @@ 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.")
+        logging.debug("All non-blacklisted windows are allowed.")
         return False
     for w in whiteblocks:
         for k in w.keywords:
@@ -72,7 +71,8 @@ def window_is_blocked(config: Config) -> bool:
 def enforce(config: Config):
     if not window_is_blocked(config):
         return
-    xwindow.notify(f"AntiDrift will minimize the window in {config.minimize_delay} seconds.")
+    delay = config.minimize_delay
+    xwindow.notify(f"AntiDrift will minimize in {delay} seconds.")
     time.sleep(config.minimize_delay)
     window = xwindow.XWindow()
     if window_is_blocked(config):
@@ -80,5 +80,4 @@ def enforce(config: Config):
         xwindow.notify(f"Minimize {window.name[:30]}.")
         window.minimize()
     else:
-        xwindow.notify(f"We are gucci again.")
+        xwindow.notify("We are gucci again.")
-

antidrift/xwindow.py

@@ -15,7 +15,7 @@ class XWindow:
             self.name = self._run(["getwindowname", self.window])
             self.cls = self._run(["getwindowclassname", self.window])
             self.pid = self._run(["getwindowpid", self.window])
-        self.keywords = list(re.findall("\w+", self.name.lower()))
+        self.keywords = list(re.findall(r"\w+", self.name.lower()))
 
     def _run(self, cmd) -> str:
         cmd = ["xdotool"] + cmd
@@ -44,4 +44,3 @@ def notify(message: str) -> None:
     else:
         cmd = ["runuser", "-m", "-u", user, "notify-send", message]
     subprocess.run(cmd, env=env)
-