"""Hermes computer-use events mapped to a small, thread-safe overlay state.""" from __future__ import annotations from dataclasses import dataclass, replace from enum import Enum import threading import time from typing import Any, Mapping class Activity(str, Enum): IDLE = "idle" OBSERVE = "observe" MOVE = "move" CLICK = "click" TYPE = "type" SCROLL = "scroll" DRAG = "drag" WAIT = "wait" ERROR = "error" KILLED = "killed" ACTIVITY_COLORS: dict[Activity, str] = { Activity.IDLE: "#77dba1", Activity.OBSERVE: "#92e6ff", Activity.MOVE: "#77dba1", Activity.CLICK: "#f4fff8", Activity.TYPE: "#c6a7ff", Activity.SCROLL: "#85d7ff", Activity.DRAG: "#ffd58c", Activity.WAIT: "#b8c6c0", Activity.ERROR: "#ff877d", Activity.KILLED: "#ff5d54", } @dataclass(frozen=True, slots=True) class OverlaySnapshot: revision: int = 0 activity: Activity = Activity.IDLE active: bool = False action: str | None = None message: str | None = None pointer_x: float | None = None pointer_y: float | None = None pointer_normalized: bool = False intensity: float = 0.0 color: str = ACTIVITY_COLORS[Activity.IDLE] updated_at: float = 0.0 expires_at: float | None = None killed: bool = False remote_active: bool = False remote_node_name: str | None = None camera_active: bool = False camera_device_name: str | None = None camera_session_id: str | None = None camera_operation_id: str | None = None overtake_active: bool = False def as_dict(self) -> dict[str, object]: return { "revision": self.revision, "activity": self.activity.value, "active": self.active, "action": self.action, "message": self.message, "pointer": ( { "x": self.pointer_x, "y": self.pointer_y, "normalized": self.pointer_normalized, } if self.pointer_x is not None and self.pointer_y is not None else None ), "intensity": self.intensity, "color": self.color, "updated_at": self.updated_at, "expires_at": self.expires_at, "killed": self.killed, "remote": { "active": self.remote_active, "node_name": self.remote_node_name, }, "camera": { "active": self.camera_active, "device_name": self.camera_device_name, "session_id": self.camera_session_id, "operation_id": self.camera_operation_id, }, "overtake": { "active": self.overtake_active, "label": "Nelloz uses your Computer" if self.overtake_active else None, "cancel_key": "Esc" if self.overtake_active else None, }, } @dataclass(frozen=True, slots=True) class ActivityUpdate: activity: Activity active: bool action: str | None = None message: str | None = None pointer_x: float | None = None pointer_y: float | None = None pointer_normalized: bool = False intensity: float = 0.65 ttl_kind: str = "activity" class HermesActivityMapper: """Accept common Hermes/tool-bus spellings without coupling to one transport.""" _COMPUTER_TOOLS = { "computer", "computer-use", "computer_use", "desktop", "mouse", "keyboard", "screen", } _START_WORDS = {"start", "started", "begin", "began", "queued", "running"} _END_WORDS = { "end", "ended", "finish", "finished", "complete", "completed", "done", "result", "returned", "stop", "stopped", } _ERROR_WORDS = {"error", "failed", "failure", "cancelled", "canceled", "denied"} def map(self, event: Mapping[str, Any]) -> ActivityUpdate | None: event_type = self._event_type(event) data = event.get("data") if isinstance(event.get("data"), Mapping) else {} payload = dict(data) payload.update({key: value for key, value in event.items() if key != "data"}) if not self._is_computer_event(event_type, payload): return None lifecycle = self._last_word(event_type) message = self._string(payload.get("message") or payload.get("detail")) x, y, normalized = self._pointer(payload) if lifecycle in self._ERROR_WORDS: return ActivityUpdate( activity=Activity.ERROR, active=True, action="error", message=message or "Computer use failed", pointer_x=x, pointer_y=y, pointer_normalized=normalized, intensity=0.9, ttl_kind="error", ) if lifecycle in self._END_WORDS: return ActivityUpdate( activity=Activity.IDLE, active=False, action="complete", message=message, intensity=0.0, ttl_kind="none", ) action = self._action(payload) activity = self._activity_for_action(action) if lifecycle in self._START_WORDS and not action: activity = Activity.OBSERVE action = "observe" return ActivityUpdate( activity=activity, active=True, action=action or activity.value, message=message, pointer_x=x, pointer_y=y, pointer_normalized=normalized, intensity=self._intensity(activity), ) @staticmethod def _event_type(event: Mapping[str, Any]) -> str: raw = event.get("type") or event.get("event") or event.get("event_type") or "" return str(raw).strip().lower().replace("_", "-") @staticmethod def _last_word(event_type: str) -> str: return event_type.replace("/", ".").replace(":", ".").split(".")[-1].split("-")[-1] def _is_computer_event(self, event_type: str, payload: Mapping[str, Any]) -> bool: if any(marker in event_type for marker in ("computer", "desktop", "mouse", "keyboard")): return True tool = payload.get("tool") or payload.get("tool_name") or payload.get("toolName") if isinstance(tool, Mapping): tool = tool.get("name") or tool.get("type") tool_name = str(tool or "").strip().lower() return tool_name in self._COMPUTER_TOOLS or tool_name.startswith(("computer_", "computer-")) @staticmethod def _string(value: object) -> str | None: return str(value)[:240] if value not in (None, "") else None @staticmethod def _action(payload: Mapping[str, Any]) -> str | None: arguments = payload.get("arguments") or payload.get("input") if not isinstance(arguments, Mapping): arguments = {} raw: object = ( payload.get("action") or payload.get("operation") or payload.get("command") or arguments.get("action") or arguments.get("operation") ) if isinstance(raw, Mapping): raw = raw.get("type") or raw.get("name") or raw.get("action") if raw is None: return None return str(raw).strip().lower().replace("-", "_").replace(" ", "_") @staticmethod def _activity_for_action(action: str | None) -> Activity: if not action: return Activity.OBSERVE if any(word in action for word in ("screenshot", "observe", "inspect", "look")): return Activity.OBSERVE if any(word in action for word in ("type", "key", "hotkey", "write")): return Activity.TYPE if any(word in action for word in ("double_click", "right_click", "click", "mouse_press")): return Activity.CLICK if "scroll" in action or "wheel" in action: return Activity.SCROLL if "drag" in action or "drop" in action: return Activity.DRAG if any(word in action for word in ("move", "hover", "point")): return Activity.MOVE if any(word in action for word in ("wait", "sleep", "pause")): return Activity.WAIT return Activity.OBSERVE @staticmethod def _intensity(activity: Activity) -> float: return { Activity.OBSERVE: 0.46, Activity.MOVE: 0.58, Activity.CLICK: 1.0, Activity.TYPE: 0.78, Activity.SCROLL: 0.68, Activity.DRAG: 0.9, Activity.WAIT: 0.34, }.get(activity, 0.6) @classmethod def _pointer(cls, payload: Mapping[str, Any]) -> tuple[float | None, float | None, bool]: action = payload.get("action") if isinstance(payload.get("action"), Mapping) else {} arguments = payload.get("arguments") or payload.get("input") if not isinstance(arguments, Mapping): arguments = {} position = ( payload.get("pointer") or payload.get("position") or payload.get("coordinate") or action.get("position") or action.get("coordinate") or arguments.get("position") or arguments.get("coordinate") ) x: object = payload.get("x", action.get("x", arguments.get("x"))) y: object = payload.get("y", action.get("y", arguments.get("y"))) if isinstance(position, Mapping): x = position.get("x", x) y = position.get("y", y) elif isinstance(position, (list, tuple)) and len(position) >= 2: x, y = position[0], position[1] if isinstance(x, bool) or isinstance(y, bool): return None, None, False try: x_value, y_value = float(x), float(y) except (TypeError, ValueError): return None, None, False coordinate_space = str( payload.get("coordinate_space") or action.get("coordinate_space") or arguments.get("coordinate_space") or "" ).lower() normalized = coordinate_space in {"normalized", "relative", "unit"} if coordinate_space in {"normalized-1000", "normalized_1000", "qwen-1000", "qwen_1000"}: x_value /= 1000.0 y_value /= 1000.0 normalized = True if not coordinate_space and 0.0 <= x_value <= 1.0 and 0.0 <= y_value <= 1.0: normalized = True return x_value, y_value, normalized class OverlayStateStore: """Atomic state projection consumed by the Tk overlay thread.""" def __init__( self, *, activity_ttl_seconds: float = 1.4, error_ttl_seconds: float = 2.0, mapper: HermesActivityMapper | None = None, ) -> None: self._activity_ttl = activity_ttl_seconds self._error_ttl = error_ttl_seconds self._mapper = mapper or HermesActivityMapper() self._lock = threading.RLock() self._snapshot = OverlaySnapshot(updated_at=time.time()) def apply(self, event: Mapping[str, Any], *, now: float | None = None) -> bool: update = self._mapper.map(event) if update is None: return False timestamp = time.time() if now is None else now with self._lock: if self._snapshot.killed: return False current = self._snapshot ttl = self._error_ttl if update.ttl_kind == "error" else self._activity_ttl expires_at = timestamp + ttl if update.active and update.ttl_kind != "none" else None self._snapshot = OverlaySnapshot( revision=current.revision + 1, activity=update.activity, active=update.active, action=update.action, message=update.message, pointer_x=update.pointer_x, pointer_y=update.pointer_y, pointer_normalized=update.pointer_normalized, intensity=update.intensity, color=ACTIVITY_COLORS[update.activity], updated_at=timestamp, expires_at=expires_at, remote_active=current.remote_active, remote_node_name=current.remote_node_name, camera_active=current.camera_active, camera_device_name=current.camera_device_name, camera_session_id=current.camera_session_id, camera_operation_id=current.camera_operation_id, overtake_active=( update.active and update.activity is not Activity.ERROR and update.ttl_kind != "none" ), ) return True def snapshot(self, *, now: float | None = None) -> OverlaySnapshot: timestamp = time.time() if now is None else now with self._lock: current = self._snapshot if current.active and current.expires_at is not None and timestamp >= current.expires_at: current = replace( current, revision=current.revision + 1, activity=Activity.IDLE, active=False, action=None, message=None, intensity=0.0, color=ACTIVITY_COLORS[Activity.IDLE], updated_at=timestamp, expires_at=None, ) self._snapshot = current return current def set_overtake(self, active: bool) -> OverlaySnapshot: """Keep the full-screen takeover indicator alive between tool events.""" timestamp = time.time() with self._lock: if self._snapshot.killed: return self._snapshot self._snapshot = replace( self._snapshot, revision=self._snapshot.revision + 1, overtake_active=bool(active), updated_at=timestamp, ) return self._snapshot def set_remote_activity( self, active: bool, *, node_name: str | None = None, ) -> OverlaySnapshot: timestamp = time.time() with self._lock: if self._snapshot.killed: return self._snapshot self._snapshot = replace( self._snapshot, revision=self._snapshot.revision + 1, remote_active=active, remote_node_name=" ".join((node_name or "Remote node").split())[:80] if active else None, updated_at=timestamp, ) return self._snapshot def set_camera_activity( self, active: bool, *, device_name: str | None = None, session_id: str | None = None, operation_id: str | None = None, ) -> OverlaySnapshot: timestamp = time.time() with self._lock: if self._snapshot.killed: return self._snapshot self._snapshot = replace( self._snapshot, revision=self._snapshot.revision + 1, camera_active=active, camera_device_name=" ".join((device_name or "Camera").split())[:80] if active else None, camera_session_id=session_id[:160] if active and session_id else None, camera_operation_id=operation_id[:160] if active and operation_id else None, updated_at=timestamp, ) return self._snapshot def kill(self, reason: str = "Emergency stop") -> OverlaySnapshot: timestamp = time.time() with self._lock: if not self._snapshot.killed: self._snapshot = OverlaySnapshot( revision=self._snapshot.revision + 1, activity=Activity.KILLED, active=False, action="kill", message=reason[:240], intensity=0.0, color=ACTIVITY_COLORS[Activity.KILLED], updated_at=timestamp, killed=True, overtake_active=False, ) return self._snapshot class HermesEventConsumer: """Small synchronous consumer suitable for HTTP, SSE or in-process event buses.""" def __init__(self, state: OverlayStateStore) -> None: self.state = state def consume(self, event: Mapping[str, Any], *, now: float | None = None) -> bool: if not isinstance(event, Mapping): raise TypeError("event must be an object") return self.state.apply(event, now=now)