diff --git a/antidrift.py b/antidrift.py index 10d794d..d989b85 100644 --- a/antidrift.py +++ b/antidrift.py @@ -4,27 +4,20 @@ import logging import shutil import sys -import time -import re from pathlib import Path -from typing import List -from gi.repository import GLib, GObject -import dbus -import dbus.service -from dbus.mainloop.glib import DBusGMainLoop -from antidrift.config import Config, Block +import antidrift.client as client +from antidrift.config import Config 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") - if not r: + """ Check if xdotool is in path and exit if not """ + 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(): - client_mode(config) + if client.antidrift_is_running(): + client.client_mode(config) else: - a = AntiDriftDaemon(config) - a.run() + add = AntiDriftDaemon(config) + add.run() -DBusGMainLoop(set_as_default=True) if __name__ == "__main__": main() diff --git a/antidrift/__init__.py b/antidrift/__init__.py index e69de29..20c0351 100644 --- a/antidrift/__init__.py +++ b/antidrift/__init__.py @@ -0,0 +1 @@ +""" AntiDrift module init """ diff --git a/antidrift/client.py b/antidrift/client.py new file mode 100644 index 0000000..b86f8d2 --- /dev/null +++ b/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") diff --git a/antidrift/config.py b/antidrift/config.py index ee7adb1..514e2da 100644 --- a/antidrift/config.py +++ b/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 - diff --git a/antidrift/daemon.py b/antidrift/daemon.py index 14eb056..13da75b 100644 --- a/antidrift/daemon.py +++ b/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, Block -from typing import List -from functools import lru_cache -from gi.repository import GLib, GObject +from antidrift.config import Config +from gi.repository import GLib +from enum import Enum 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): - logging.info("status requested") - return "running" + def status(self) -> str: + return DaemonStatus.RUNNING.value 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.") diff --git a/antidrift/xwindow.py b/antidrift/xwindow.py index 91cb8d9..ae3fa7d 100644 --- a/antidrift/xwindow.py +++ b/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) -