From 0f8315fca2fb31ba01cc278fb811a00e30e7d8af Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Thu, 23 Jun 2022 17:11:23 -0400 Subject: [PATCH] Continue with client daemon implmentaton. --- antidrift.py | 127 ++++++++++------------------- antidrift/__init__.py | 0 config.py => antidrift/config.py | 0 antidrift/daemon.py | 84 +++++++++++++++++++ xwindow.py => antidrift/xwindow.py | 0 5 files changed, 128 insertions(+), 83 deletions(-) create mode 100644 antidrift/__init__.py rename config.py => antidrift/config.py (100%) create mode 100644 antidrift/daemon.py rename xwindow.py => antidrift/xwindow.py (100%) diff --git a/antidrift.py b/antidrift.py index 413f7dd..10d794d 100644 --- a/antidrift.py +++ b/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") - check_for_xdotool() - 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 antidrift_is_running(): + client_mode(config) + else: + a = AntiDriftDaemon(config) + a.run() +DBusGMainLoop(set_as_default=True) if __name__ == "__main__": main() - diff --git a/antidrift/__init__.py b/antidrift/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/config.py b/antidrift/config.py similarity index 100% rename from config.py rename to antidrift/config.py diff --git a/antidrift/daemon.py b/antidrift/daemon.py new file mode 100644 index 0000000..14eb056 --- /dev/null +++ b/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.") + diff --git a/xwindow.py b/antidrift/xwindow.py similarity index 100% rename from xwindow.py rename to antidrift/xwindow.py