145 lines
4.4 KiB
Python
145 lines
4.4 KiB
Python
#!/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
|
|
|
|
|
|
def analyze_and_print(image, frame, prev, logf):
|
|
global _last_t
|
|
info = analyze.summarize(image)
|
|
harvest = analyze.is_harvesting(image)
|
|
bars = analyze.read_bar_values(image)
|
|
fps = 1.0 / max(1e-6, time.monotonic() - _last_t)
|
|
_last_t = time.monotonic()
|
|
|
|
record = {
|
|
"time": _now(),
|
|
"frame": frame,
|
|
"is_harvesting": harvest["is_harvesting"],
|
|
"harvest_pixel": harvest["color"],
|
|
"harvest_luma": harvest["luma"],
|
|
"bars": bars,
|
|
}
|
|
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
|
|
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}",
|
|
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
|
|
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)
|
|
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
|
|
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)
|
|
|
|
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")
|