me0nline
4d1d7cc655
Synchronous urllib calls in the 60 FPS main loop caused a 3-second network connect timeout every 5 seconds whenever ActivityWatch was offline (due to Windows IPv6/IPv4 localhost resolution). This completely froze the main thread, stalled the Windows message pump, and triggered Windows Hung-App detection (hourglass / wait cursor). ActivityWatchClient now polls in a dedicated daemon worker thread: - All socket connections and HTTP timeouts occur in the background thread. - get_current_window_event() reads cached state in 0ms without blocking. - Fallback to native Win32 window APIs is immediate with zero latency.
226 lines
8.7 KiB
Python
226 lines
8.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 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
|