#!/usr/bin/env python3 """Demo: capture the Eternal Lands window, analyze it, and log results as JSONL. Usage: python3 main.py [--title "Eternal Lands"] [--fps 5] [--save-every 0] [--log analysis.jsonl] python3 main.py --replay tests/images/harvesting [--log analysis.jsonl] """ import argparse import json import os import time import analyze from capture import WindowCapture, WindowNotFound class WindowTracker: """Cache last-known window positions and re-locate them when they move. Cached positions are re-verified cheaply every frame. The first failed verify after a stable period triggers an immediate full-frame rescan; afterwards rescans back off to SCAN_PERIOD so a closed or unfindable window cannot pin the loop to full-frame OCR forever. """ SCAN_PERIOD = 30.0 # seconds between recovery rescans while a window is missing def __init__(self): self.pos = {"manufacturing": None, "inventory": None} self._last_scan = {"manufacturing": 0.0, "inventory": 0.0} self._verified = {"manufacturing": -1, "inventory": -1} self._frame = 0 def update(self, image): """Return currently verified window positions (None when unverified).""" verified = {"manufacturing": None, "inventory": None} for window in ("manufacturing", "inventory"): cached = self.pos[window] if cached is not None and analyze.verify_window(image, cached[0], cached[1], window): verified[window] = cached self._verified[window] = self._frame continue just_verified = cached is not None and self._verified[window] == self._frame - 1 if just_verified or (cached is None and self._frame == 0) or ( time.monotonic() - self._last_scan[window] >= self.SCAN_PERIOD ): self._last_scan[window] = time.monotonic() pos = analyze.find_windows(image)[window] if pos is not None: self.pos[window] = pos verified[window] = pos self._verified[window] = self._frame self._frame += 1 return verified def analyze_and_print(image, frame, prev, logf, tracker): global _last_t info = analyze.summarize(image) harvest = analyze.is_harvesting(image) bars = analyze.read_bar_values(image) windows = tracker.update(image) fps = 1.0 / max(1e-6, time.monotonic() - _last_t) _last_t = time.monotonic() mfg = windows["manufacturing"] message = analyze.read_manufacturing_message(image, mfg[0], mfg[1]) if mfg else None inv = windows["inventory"] record = { "time": _now(), "frame": frame, "is_harvesting": harvest["is_harvesting"], "harvest_pixel": harvest["color"], "harvest_luma": harvest["luma"], "bars": bars, "manufacturing_window": [mfg[0], mfg[1]] if mfg else None, "manufacturing_message": message, "inventory_window": [inv[0], inv[1]] if inv else None, } if logf is not None: logf.write(json.dumps(record) + "\n") logf.flush() small = image.convert("RGB").resize(analyze.ANALYSIS_SIZE) delta = None if prev is None else analyze.difference(prev, small) top = ", ".join(f"{c['hex']} {c['fraction'] * 100:.0f}%" for c in info["dominant"][:3]) status = "harvesting" if harvest["is_harvesting"] else "not harvesting" frame_str = f"{frame:05d}" if isinstance(frame, int) else frame msg = f"mfg={mfg} msg={message['category'] if message else None}" if mfg else "mfg=none" print( f"[{frame_str}] {status} luma={harvest['luma']} " f"size={info['size'][0]}x{info['size'][1]} " f"mean={info['mean_rgb']} top=[{top}] bars={bars} " f"delta={delta} loop_fps={fps:.1f} {msg}", flush=True, ) return small def run_replay(args): """Process PNGs from a directory instead of a live window.""" logf = open(args.log, "a") if args.log else None prev = None tracker = WindowTracker() try: for name in sorted(os.listdir(args.replay)): if not name.lower().endswith(".png"): continue path = os.path.join(args.replay, name) image = analyze.load_image(path) if image is None: print(f"[skipped] unreadable image: {path}", flush=True) continue prev = analyze_and_print(image, name, prev, logf, tracker) finally: if logf: logf.close() def run_live(args): cap = WindowCapture(args.title) os.makedirs(args.snapshots, exist_ok=True) interval = 1.0 / args.fps print(f"Capturing window matching {args.title!r} at {args.fps:.1f} fps (Ctrl-C to stop)", flush=True) logf = open(args.log, "a") if args.log else None prev = None tracker = WindowTracker() frame = 0 try: while True: started = time.monotonic() try: image = cap.capture() except WindowNotFound as exc: print(f"[{_now()}] {exc}; retrying...", flush=True) time.sleep(1.0) continue prev = analyze_and_print(image, frame, prev, logf, tracker) if args.save_every and frame % args.save_every == 0: path = os.path.join(args.snapshots, f"frame_{frame:05d}.png") image.save(path) frame += 1 elapsed = time.monotonic() - started time.sleep(max(0.0, interval - elapsed)) finally: if logf: logf.close() def main(): parser = argparse.ArgumentParser(description="Capture and analyze a specific window on X11") parser.add_argument( "--title", default="Eternal Lands", help="window title to match (substring, case-insensitive)", ) parser.add_argument("--fps", type=float, default=5.0, help="target capture rate") parser.add_argument("--snapshots", default="snapshots", help="directory for PNG snapshots") parser.add_argument( "--save-every", type=int, default=0, metavar="N", help="save a snapshot every N frames (0 = off)", ) parser.add_argument( "--log", default="analysis.jsonl", metavar="PATH", help="append one JSON record per frame to PATH (empty string disables)", ) parser.add_argument( "--replay", metavar="DIR", help="process existing PNGs from DIR instead of capturing a window", ) args = parser.parse_args() if args.replay: run_replay(args) else: run_live(args) _last_t = time.monotonic() def _now(): return time.strftime("%H:%M:%S") if __name__ == "__main__": try: main() except KeyboardInterrupt: print("\nstopped")