auto_el/analyze/main.py

#!/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")