"""Screen watcher — a local VLM watches the screen so Claude doesn't have to. Ahmed: "watch my gameplay and tell me my mistakes." Sending frames to Claude burns context fast; instead a SMALL local vision model (Qwen2.5-VL via Ollama, running on the same GPU next to the game) analyzes a screenshot every few seconds for free, accumulates observations, and when the watch ends writes ONE summarized report to control/report.txt — which flows to the master Jarvis through the normal worker-report channel, and he relays it in his own words. Standalone process (like hands.py), started/stopped by the engine via control/watch.json: {"on": true, "task": "spot mistakes in my Marvel Rivals gameplay", "minutes": 5, "interval": 4} Env: WATCH_MODEL (default qwen2.5vl:3b — ~3.2GB, fits beside a game on a 12GB card; use qwen2.5vl:7b when not gaming for sharper analysis), OLLAMA_URL (default http://127.0.0.1:11434). """ from __future__ import annotations import base64 import json import os import signal import sys import time from pathlib import Path PROJECT = Path(__file__).resolve().parent.parent CONTROL = PROJECT / "control" MODEL = os.environ.get("WATCH_MODEL", "qwen2.5vl:3b") OLLAMA = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434") FRAME_PROMPT = ( "You are watching {task}. This is one frame of the live screen. In 1-2 " "short sentences, note ONLY what matters for that task (mistakes, " "notable events, state changes). If nothing notable, reply exactly: " "nothing notable.") SUMMARY_PROMPT = ( "You watched a screen over time for this task: {task}. Here are your " "timestamped frame notes:\n\n{notes}\n\nWrite a single concise report " "(3-5 sentences, plain prose, no markdown) with the most useful " "observations and concrete advice. Skip frames that said nothing " "notable.") def _grab_jpeg(max_w: int = 1280) -> bytes: import cv2 import mss import numpy as np with mss.mss() as sct: shot = sct.grab(sct.monitors[0]) img = np.array(shot)[:, :, :3] # BGRA -> BGR if img.shape[1] > max_w: k = max_w / img.shape[1] img = cv2.resize(img, (max_w, int(img.shape[0] * k)), interpolation=cv2.INTER_AREA) ok, buf = cv2.imencode(".jpg", img, [cv2.IMWRITE_JPEG_QUALITY, 82]) if not ok: raise RuntimeError("jpeg encode failed") return buf.tobytes() def _free_vram_gb() -> float: """Free GPU memory right now (0.0 when it can't be read).""" import subprocess try: out = subprocess.run( ["nvidia-smi", "--query-gpu=memory.free", "--format=csv,noheader,nounits"], capture_output=True, text=True, timeout=5) return float(out.stdout.strip().splitlines()[0]) / 1024.0 except Exception: # noqa: BLE001 return 0.0 # Decided ONCE at startup (see main): while a game owns the GPU we force the # vision model fully onto the CPU — a partial GPU load into a full card # causes VRAM thrashing that hitches the game AND everything else. _CPU_ONLY = False def _ollama_chat(prompt: str, image: bytes | None = None, timeout: float = 60.0) -> str: import urllib.request msg: dict = {"role": "user", "content": prompt} if image is not None: msg["images"] = [base64.b64encode(image).decode("ascii")] options: dict = {"num_thread": 4} # never starve the game's CPU either if _CPU_ONLY: options["num_gpu"] = 0 # zero layers on the GPU — no thrash # Resident between frames while a watch is active (every request renews # the lease). On the 16GB Mac the lease is short so the ~3GB VLM unloads # minutes after a watch ends instead of squatting on unified memory. import sys as _sys body = json.dumps({ "model": MODEL, "messages": [msg], "stream": False, "keep_alive": "3m" if _sys.platform == "darwin" else "30m", "options": options, }).encode() req = urllib.request.Request( f"{OLLAMA}/api/chat", data=body, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as r: out = json.loads(r.read()) return (out.get("message") or {}).get("content", "").strip() def _report(line: str) -> None: """Append to control/report.txt — the master relays it to Ahmed.""" CONTROL.mkdir(exist_ok=True) with open(CONTROL / "report.txt", "a", encoding="utf-8") as f: f.write(line.replace("\n", " ").strip() + "\n") def main() -> int: task = os.environ.get("WATCH_TASK", "the screen") minutes = float(os.environ.get("WATCH_MINUTES", "5")) interval = float(os.environ.get("WATCH_INTERVAL", "6")) # live mode: notable events are reported IMMEDIATELY (Jarvis interjects # mid-game: "you're overextending, sir") instead of only at the end. live = os.environ.get("WATCH_LIVE", "0") == "1" live_cooldown = float(os.environ.get("WATCH_LIVE_COOLDOWN", "25")) last_live = -1e9 running = [True] for sig in (signal.SIGTERM, signal.SIGINT): try: signal.signal(sig, lambda *_a: running.__setitem__(0, False)) except Exception: # noqa: BLE001 pass # GAME-SAFE MODE: if the GPU is busy (a game owns it), pin the vision # model to CPU-only, stretch the sampling interval, and drop our own # process priority — watching must never cost frames. global _CPU_ONLY free = _free_vram_gb() if free < 4.5: _CPU_ONLY = True interval = max(interval, 12.0) print(f"watcher: GPU busy ({free:.1f}GB free) — CPU-only vision, " f"interval {interval:.0f}s, low priority", flush=True) try: import psutil psutil.Process().nice(psutil.BELOW_NORMAL_PRIORITY_CLASS) except Exception: # noqa: BLE001 pass # make sure the model is reachable before promising anything try: _ollama_chat("Reply with the word: ready", timeout=120) except Exception as e: # noqa: BLE001 _report(f"Screen watcher couldn't start — the local vision model " f"isn't available ({str(e)[:80]}). Is Ollama running with " f"{MODEL} pulled?") return 1 print(f"watcher: {MODEL} watching '{task}' every {interval:.0f}s " f"for {minutes:.0f}m", flush=True) notes: list[str] = [] t_end = time.monotonic() + minutes * 60 t0 = time.monotonic() while running[0] and time.monotonic() < t_end: try: frame = _grab_jpeg() obs = _ollama_chat(FRAME_PROMPT.format(task=task), frame) stamp = f"{time.monotonic()-t0:4.0f}s" if obs and "nothing notable" not in obs.lower(): notes.append(f"[{stamp}] {obs}") print(f"watcher: [{stamp}] {obs[:100]}", flush=True) # live interjection: hand it to the master NOW (he decides # how to phrase it), throttled so he doesn't nag every frame if live and time.monotonic() - last_live >= live_cooldown: last_live = time.monotonic() _report(f"[live screen watch — interject briefly if " f"worth saying, else stay quiet] {obs}") except Exception as e: # noqa: BLE001 print(f"watcher: frame skipped ({str(e)[:60]})", flush=True) # sleep in small steps so a stop signal lands quickly target = time.monotonic() + interval while running[0] and time.monotonic() < target: time.sleep(0.25) # one summarized report back to the master try: if notes: summary = _ollama_chat(SUMMARY_PROMPT.format( task=task, notes="\n".join(notes[-60:])), timeout=120) _report(f"Screen watch finished ({task}): {summary}") else: _report(f"Screen watch finished ({task}): nothing notable " f"happened while I was watching.") except Exception as e: # noqa: BLE001 _report(f"Screen watch ended but summarizing failed ({str(e)[:60]}); " f"raw notes: " + " | ".join(notes[-5:]) if notes else "no notes.") print("watcher: done", flush=True) return 0 if __name__ == "__main__": sys.exit(main())