""" 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 _match_event(self, app_name: str, window_title: str, win_info: WindowInfo) -> Optional[DistractionEvent]: """Evaluates a window title and process name against distraction rules.""" title_lower = window_title.lower() app_lower = app_name.lower() 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=win_info, ) for target_app in self.config.distraction_apps: if target_app.lower() in app_lower: return DistractionEvent( app=app_name, title=window_title, matched_rule=f"app '{target_app}'", window_info=win_info, ) return None def check_for_distraction(self) -> Optional[DistractionEvent]: """ Checks current active window and visible background windows against distraction rules. Uses ActivityWatch if available, otherwise falls back to native Win32. """ # 1. Try ActivityWatch (non-blocking read of cached event) 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 "" 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(), ) if active_window_info: matched = self._match_event(app_name, window_title, active_window_info) if matched: return matched if not self.config.auto_fallback_to_win32: return None # 2. Native Win32: Check foreground window first fg_info = WindowManager.get_foreground_window_info() if fg_info and fg_info.title: matched = self._match_event(fg_info.process_name, fg_info.title, fg_info) if matched: return matched # 3. Native Win32: If foreground window is not a distraction, check any visible desktop window! # This catches distraction windows open in the background, on a 2nd monitor, or when user clicks away def title_matches(t: str) -> bool: tl = t.lower() return any(kw.lower() in tl for kw in self.config.distraction_keywords) def app_matches(p: str) -> bool: pl = p.lower() return any(a.lower() in pl for a in self.config.distraction_apps) matched_window = WindowManager.find_window_by_match( title_predicate=title_matches if self.config.distraction_keywords else None, process_predicate=app_matches if self.config.distraction_apps else None, ) if matched_window and matched_window.title: return self._match_event(matched_window.process_name, matched_window.title, matched_window) return None