AI Flow

"""The capture recipe: the same picture for every engine, with stage markers.

open:  the board on the desktop's KiCad through the bridge, foreground OpenGL on a test box we own
       (the etiquette pushes a new window back once, so the restore is repeated), maximised, zoom to fit.
start: ab's native monitor recorder; stop: stops it and pulls the MP4 into the run directory.
mark:  writes a labelled timestamp into the run manifest; the split-screen cut aligns engines on them.
"""
import json
import subprocess
import time


def bridge(ai_thread, target, verb, args, timeout=300):
    cmd = ["adom-bridge", "--ai-thread", ai_thread, "--target", target, verb, json.dumps(args)]
    out = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout).stdout
    try:
        return json.loads(out)
    except Exception:
        return {"status": "parse_error", "raw": out[:400]}


def open_board(ai_thread, target, remote_board, reason):
    r = bridge(ai_thread, target, "kicad_open_board", {"filePath": remote_board, "foreground": True, "foregroundReason": reason})
    hwnd = r.get("hwnd")
    if not hwnd:
        return None, r
    bridge(ai_thread, target, "desktop_ui_window", {"hwnd": hwnd, "action": "maximize", "reason": "boardflow capture: fill the monitor"})
    time.sleep(40)
    bridge(ai_thread, target, "desktop_bring_to_front", {"hwnd": hwnd, "reason": "boardflow capture: keep the PCB editor on screen on the test box"})
    time.sleep(2)
    bridge(ai_thread, target, "kicad_send_key", {"hwnd": hwnd, "key": "ctrl+home"})
    time.sleep(3)
    return hwnd, r


def record_start(ai_thread, target, reason, max_ms=1_800_000):
    r = bridge(ai_thread, target, "desktop_record_start", {"monitor": 0, "audio": False, "fps": 30, "reason": reason, "maxDurationMs": max_ms})
    return r.get("recordingId") or r.get("id"), r


def record_stop(ai_thread, target, recording_id, save_to):
    r = bridge(ai_thread, target, "desktop_record_stop", {"recordingId": recording_id})
    p = r.get("filePath") or r.get("path")
    if not p and r.get("output"):
        try:
            p = json.loads(r["output"]).get("data", {}).get("path")
        except Exception:
            p = None
    p = (p or "").replace("\\", "/")
    pulled = bridge(ai_thread, target, "pull_file", {"filePaths": [p], "saveTo": save_to}, timeout=900) if p else {}
    return p, r, pulled