""" ActivityWatch Client and Window Tracker for Catser. Integrates with the ActivityWatch backend (https://activitywatch.net/) running locally at http://localhost:5600/api/0. Features: - Automatic discovery of 'aw-watcher-window_*' buckets. - Polling of active window events: executable name, window title, and timestamp. - Evaluation against distraction/drift content rules (Shorts, Reels, TikTok, etc.). - Seamless fallback to native Win32 window APIs when ActivityWatch is offline, ensuring Catser remains fully functional under all conditions. """ import time import json import logging import urllib.request import urllib.error from dataclasses import dataclass from typing import Optional, Dict, Any, List from .config import Config from .window_manager import WindowManager, WindowInfo logger = logging.getLogger("catser.aw") @dataclass class DistractionEvent: """Represents a detected distraction target window.""" app: str title: str matched_rule: str window_info: WindowInfo timestamp: float = 0.0 def __post_init__(self): if self.timestamp == 0.0: self.timestamp = time.time() import threading class ActivityWatchClient: """ Client interface for ActivityWatch API with automatic native Win32 fallback. Runs all HTTP requests in a background thread so that network latency or timeouts when ActivityWatch is offline NEVER block the 60 FPS animation loop. """ def __init__(self, config: Config): self.config = config # Normalise localhost to 127.0.0.1 to prevent Windows IPv6 resolution timeouts url = config.aw_url.rstrip("/") if "localhost" in url: url = url.replace("localhost", "127.0.0.1") self.base_url = url self.api_url = f"{self.base_url}/api/0" self.bucket_id: Optional[str] = None self.is_connected: bool = False self.connect_retry_delay: float = 5.0 # Retry connecting every 5s self._latest_event: Optional[Dict[str, Any]] = None self._lock = threading.Lock() self._running = True self._worker_thread = threading.Thread( target=self._background_worker, name="ActivityWatchWorker", daemon=True, ) self._worker_thread.start() def _background_worker(self) -> None: """ Runs in the background, continuously attempting connection and polling ActivityWatch without ever touching or blocking the main GUI thread. """ while self._running: try: if not self.is_connected or not self.bucket_id: self._attempt_connect() if self.is_connected and self.bucket_id: ev = self._fetch_latest_event() with self._lock: self._latest_event = ev else: with self._lock: self._latest_event = None except Exception as e: logger.debug(f"AW worker error: {e}") self.is_connected = False with self._lock: self._latest_event = None sleep_time = self.config.poll_interval_sec if self.is_connected else self.connect_retry_delay time.sleep(sleep_time) def _attempt_connect(self) -> bool: """Pings the ActivityWatch /api/0/info endpoint with a short timeout.""" try: req = urllib.request.Request(f"{self.api_url}/info", headers={"User-Agent": "Catser"}) with urllib.request.urlopen(req, timeout=0.8) as resp: if resp.status == 200: data = json.loads(resp.read().decode("utf-8")) logger.info(f"Connected to ActivityWatch server v{data.get('version', 'unknown')}") self.is_connected = True self._discover_window_bucket() return True except Exception: if self.is_connected: logger.warning("ActivityWatch disconnected. Falling back to native tracking.") self.is_connected = False return False return False def _discover_window_bucket(self) -> Optional[str]: """Discovers the active 'aw-watcher-window_*' bucket.""" try: req = urllib.request.Request(f"{self.api_url}/buckets", headers={"User-Agent": "Catser"}) with urllib.request.urlopen(req, timeout=0.8) as resp: if resp.status == 200: buckets = json.loads(resp.read().decode("utf-8")) for b_id in buckets.keys(): if b_id.startswith("aw-watcher-window"): self.bucket_id = b_id logger.info(f"Using ActivityWatch window bucket: {self.bucket_id}") return self.bucket_id except Exception as e: logger.debug(f"Error querying ActivityWatch buckets: {e}") return None def _fetch_latest_event(self) -> Optional[Dict[str, Any]]: """Fetches the latest event from the active window bucket.""" if not self.bucket_id: return None try: url = f"{self.api_url}/buckets/{self.bucket_id}/events?limit=1" req = urllib.request.Request(url, headers={"User-Agent": "Catser"}) with urllib.request.urlopen(req, timeout=0.8) as resp: if resp.status == 200: events = json.loads(resp.read().decode("utf-8")) if events and isinstance(events, list) and len(events) > 0: ev_data = events[0].get("data", {}) return { "app": ev_data.get("app", ""), "title": ev_data.get("title", ""), "timestamp": events[0].get("timestamp"), } except Exception as e: logger.debug(f"ActivityWatch query error: {e}") self.is_connected = False return None def check_connection(self) -> bool: """Returns connection status immediately without blocking.""" return self.is_connected def get_current_window_event(self) -> Optional[Dict[str, Any]]: """ Returns the latest window event from the background cache. Completely non-blocking (0ms latency). """ with self._lock: return self._latest_event def stop(self) -> None: """Signals the background worker to exit.""" self._running = False def check_for_distraction(self) -> Optional[DistractionEvent]: """ Checks current active window against distraction rules. Uses ActivityWatch if available, otherwise falls back to native Win32. """ app_name = "" window_title = "" active_window_info: Optional[WindowInfo] = None # 1. Try ActivityWatch aw_event = self.get_current_window_event() if aw_event: app_name = (aw_event.get("app") or "").lower() window_title = aw_event.get("title") or "" # Locate corresponding Win32 HWND for screen coordinates and close button active_window_info = WindowManager.find_window_by_match( title_predicate=lambda t: window_title.lower() in t.lower() or t.lower() in window_title.lower(), process_predicate=lambda p: app_name in p.lower(), ) # 2. Fallback to native Win32 if ActivityWatch is unavailable or HWND not found if not active_window_info and self.config.auto_fallback_to_win32: active_window_info = WindowManager.get_foreground_window_info() if active_window_info: app_name = active_window_info.process_name.lower() window_title = active_window_info.title if not active_window_info or not window_title: return None title_lower = window_title.lower() # Check for title keyword matches (e.g. "shorts", "reels", "tiktok") for kw in self.config.distraction_keywords: if kw.lower() in title_lower: return DistractionEvent( app=app_name, title=window_title, matched_rule=f"keyword '{kw}'", window_info=active_window_info, ) # Check for application process matches for target_app in self.config.distraction_apps: if target_app.lower() in app_name: return DistractionEvent( app=app_name, title=window_title, matched_rule=f"app '{target_app}'", window_info=active_window_info, ) return None