me0nline
a969c91c28
Previously, fallback window detection only checked user32.GetForegroundWindow(). If the user had YouTube Shorts open in Chrome on a second monitor or behind the terminal, it was ignored. Now check_for_distraction(): 1. Checks the active foreground window first. 2. If not a distraction, scans all visible top-level desktop windows via find_window_by_match(). 3. Immediately wakes the cat from sleeping or sitting if any distraction window is discovered.
245 lines
9.7 KiB
Python
245 lines
9.7 KiB
Python
"""
|
|
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
|