diff --git a/antidrift.py b/antidrift.py index df2761b..5b49838 100644 --- a/antidrift.py +++ b/antidrift.py @@ -62,7 +62,7 @@ def main() -> None: return check_for_xdotool() - if client.antidrift_is_running(): + if client.antidrift_daemon_is_running(): init_logging(config.log_file, "[blue]client[/blue]") client.client_mode(config) else: diff --git a/antidrift/client.py b/antidrift/client.py index bfcf7cb..09ac793 100644 --- a/antidrift/client.py +++ b/antidrift/client.py @@ -1,22 +1,45 @@ from antidrift.config import Config -from antidrift.daemon import DaemonStatus +from typing import Optional, List +from rich import print +import argparse +import sys +import logging import dbus import dbus.service -def antidrift_is_running() -> bool: - """ Check if AntiDrift is running via the DBUS """ +def get_dbus_interface() -> Optional[dbus.Interface]: bus = dbus.SessionBus() try: bus_object = bus.get_object("com.antidrift", "/com/antidrift") interface = dbus.Interface(bus_object, "com.antidrift") + return interface except dbus.exceptions.DBusException: + return None + + +def antidrift_daemon_is_running() -> bool: + """ Check if AntiDrift is running via the DBUS """ + interface = get_dbus_interface() + if interface is None: return False reply = interface.status() - if reply == DaemonStatus.RUNNING.value: + if reply: + logging.info(reply) return True return False def client_mode(_config: Config): - print("client_mode") + parser = argparse.ArgumentParser(description='AntiDrift CLI.') + parser.add_argument('--start', metavar='whiteblock', nargs='+', + help='start session with whiteblocks') + parser.add_argument('--stop', action='store_true', + help='stop session') + args = parser.parse_args() + interface = get_dbus_interface() + if args.start: + reply = interface.start(args.start) + elif args.stop: + reply = interface.stop() + print(reply) diff --git a/antidrift/config.py b/antidrift/config.py index 514e2da..a2b90b3 100644 --- a/antidrift/config.py +++ b/antidrift/config.py @@ -13,11 +13,12 @@ class Block(BaseModel): class Config(BaseModel): blackblocks: List[Block] whiteblocks: List[Block] + active_blackblocks: List[Block] = [] + active_whiteblocks: List[Block] = [] log_file: Path = Path() config_file: Path = Path() check_delay: int = 1000 minimize_delay: int = 5 - blackblocks_only: bool = True class Config: extra = 'forbid' diff --git a/antidrift/daemon.py b/antidrift/daemon.py index fdad309..5bd209c 100644 --- a/antidrift/daemon.py +++ b/antidrift/daemon.py @@ -2,10 +2,12 @@ import logging import time import os import sys +import argparse import antidrift.xwindow as xwindow from antidrift.config import Config from gi.repository import GLib, Gio from enum import Enum +from typing import List import dbus import dbus.service @@ -14,11 +16,6 @@ IFACE = "com.antidrift" OPATH = "/com/antidrift" -class DaemonStatus(Enum): - RUNNING = "running" - ERROR = "error" - - def reload_callback(m, f, o, event): # Without this check, multiple 'ok's will be printed for each file change logging.warning("[bold green]Restart.[/bold green]") @@ -32,11 +29,47 @@ class AntiDriftDaemon(dbus.service.Object): bus_name = dbus.service.BusName(BUS_NAME, bus=bus) dbus.service.Object.__init__(self, bus_name, OPATH) self.config = config + self.config.active_whiteblocks = [] + self.config.active_blackblocks = self.config.blackblocks @dbus.service.method(dbus_interface=IFACE, in_signature="", out_signature="s") def status(self) -> str: - return DaemonStatus.RUNNING.value + return "active" + + + @dbus.service.method(dbus_interface=IFACE, + in_signature="as", out_signature="s") + def start(self, whiteblocks: List[str]) -> str: + self.config.active_whiteblocks = [] + all_whiteblocks = {wb.name: wb for wb in self.config.whiteblocks} + success_wbs = [] + fail_wbs = [] + for wb_name in whiteblocks: + if wb_name in all_whiteblocks: + self.config.active_whiteblocks.append(all_whiteblocks[wb_name]) + success_wbs.append(wb_name) + else: + fail_wbs.append(wb_name) + if success_wbs: + r = f"Start whiteblocks [blue]{', '.join(success_wbs)}[/blue]." + logging.info(r) + else: + r = "No whiteblocks started." + if fail_wbs: + m = f"No blackblocks [red]{', '.join(fail_wbs)}[/red]." + logging.warning(m) + return r + + @dbus.service.method(dbus_interface=IFACE, + in_signature="", out_signature="s") + def stop(self) -> str: + self.config.active_whiteblocks = [] + self.config.active_blackblocks = self.config.blackblocks + m = '[red]Stop[/red] all whitelists. Blacklist only mode.' + logging.info(m) + return m + def run(self): def _enforce(): @@ -52,7 +85,7 @@ class AntiDriftDaemon(dbus.service.Object): monitor.connect("changed", reload_callback) monitors.append(monitor) - logging.info("AntiDriftDaemon run.") + logging.info("Start AntiDriftDaemon.") _enforce() mainloop = GLib.MainLoop() mainloop.run() @@ -60,8 +93,8 @@ class AntiDriftDaemon(dbus.service.Object): def window_is_blocked(config: Config, silent: bool = False) -> bool: # These should be selectable in the future (not all at the same time) - blackblocks = config.blackblocks - whiteblocks = config.whiteblocks + blackblocks = config.active_blackblocks + whiteblocks = config.active_whiteblocks window = xwindow.XWindow() if not window.keywords: @@ -74,9 +107,9 @@ def window_is_blocked(config: Config, silent: bool = False) -> bool: logging.warning(f"[red]{window.name[:50]}[/red] " f"blocked by [red]{b.name}[/red].") return True - if config.blackblocks_only: + if not whiteblocks: if not silent: - logging.debug("All non-blacklisted windows are allowed.") + logging.debug("All non-blackblock windows are allowed.") return False for w in whiteblocks: for k in w.keywords: @@ -86,7 +119,7 @@ def window_is_blocked(config: Config, silent: bool = False) -> bool: f"allowed by [blue]{w.name}[/blue].") return False if not silent: - xwindow.notify(f"{window.name[:30]} not on any whitelist.") + xwindow.notify(f"[red]{window.name[:30]}[/red] not on any whiteblock.") return True