Log antidrift status to activity watch (requires aw-client and aw-core packages.

antidrift/antidrift_watcher.py

@@ -0,0 +1,30 @@
+import logging
+from datetime import datetime, timedelta, timezone
+from dataclasses import dataclass
+
+from aw_core.models import Event
+from aw_client import ActivityWatchClient
+
+
+@dataclass
+class Settings:
+    timeout: int
+    poll_time: int
+
+
+class AntidriftWatcher:
+    def __init__(self):
+        self.settings = Settings(timeout=180, poll_time=5)
+        self.client = ActivityWatchClient("aw-watcher-antidrift")
+        self.bucketname = "{}_{}".format(
+            self.client.client_name, self.client.client_hostname
+        )
+        logging.info("aw-watcher-antidrift started")
+        eventtype = "antidrift_status"
+        self.client.create_bucket(self.bucketname, eventtype, queued=False)
+
+    def ping(self, status: str):
+        data = {"status": status}
+        now = datetime.now(timezone.utc)
+        event = Event(timestamp=now, data=data)
+        inserted_event = self.client.heartbeat(self.bucketname, event, 2.0)

antidrift/daemon.py

@@ -5,6 +5,7 @@ import antidrift.xwindow as xwindow
 from antidrift.config import Config
 from gi.repository import GLib, Gio
 from typing import List
+from antidrift.antidrift_watcher import AntidriftWatcher
 import dbus
 import dbus.service
 
@@ -31,6 +32,7 @@ class AntiDriftDaemon(dbus.service.Object):
         self.config.active_blackblocks = self.config.blackblocks
         self.enforce_count = 0
         self.enforce_value = int(config.enforce_delay_ms / config.polling_cycle_ms)
+        self.antidrift_watcher = AntidriftWatcher()
 
     @dbus.service.method(dbus_interface=IFACE,
                          in_signature="as", out_signature="s")
@@ -110,15 +112,21 @@ class AntiDriftDaemon(dbus.service.Object):
             xwindow.notify(f"Minimize {window.name[:30]}.")
             window.minimize()
             self.enforce_count = 0
+            self.antidrift_watcher.ping("blocked")
         elif self.enforce_count > 0 and window_is_blocked(config, True):
             self.enforce_count += 1
+            self.antidrift_watcher.ping("blocked")
         elif self.enforce_count == 0 and window_is_blocked(config):
             self.enforce_count += 1
             delay = int(config.enforce_delay_ms / 1000)
             xwindow.notify(f"AntiDrift will minimize in {delay}s.")
+            self.antidrift_watcher.ping("blocked")
         elif self.enforce_count > 0:
             xwindow.notify("We are gucci again.")
             self.enforce_count = 0
+            self.antidrift_watcher.ping("okay")
+        else:
+            self.antidrift_watcher.ping("okay")
 
 
 def window_is_blocked(config: Config, silent: bool = False) -> bool: