harness.py

Part of You’re the OS.

"""Scripted-input harness: drives the port headlessly and captures seven stages.

Stages:
    01_boot.png            the title screen, before the machine is booted
    02_first_process.png   one process visible
    03_queue_filling.png   ~20 processes spawned, filling slots
    04_cpu_loaded.png      click-toggled several processes onto CPUs
    05_pages_swapped.png   triggered page swaps via clicks
    06_io_event.png        I/O queue lit, event delivered
    07_late_game.png       the shutdown notice, after ten unattended processes starved

Input goes through ``InputSimulator``, so every click and key travels the same
path a real mouse or keyboard would: polled state, scene-tree events, and UI
events all move in lockstep.
"""

from __future__ import annotations

import sys
from collections.abc import Callable
from pathlib import Path

_PORT_DIR = Path(__file__).parent
if str(_PORT_DIR) not in sys.path:
    sys.path.insert(0, str(_PORT_DIR))

from main import OsRoot  # noqa: E402
from nodes.stage import (  # noqa: E402
    DESIGN_HEIGHT,
    DESIGN_WIDTH,
    PAGE_GAP,
    PAGE_H,
    PAGE_W,
    PROCESS_AREA_X,
    PROCESS_AREA_Y,
    PROCESS_GAP,
    PROCESS_SIZE,
    RAM_AREA_X,
    RAM_Y,
)

from simvx.core.input.controls import Key  # noqa: E402
from simvx.core.testing import InputSimulator  # noqa: E402
from simvx.graphics import App, save_png  # noqa: E402

OUT = _PORT_DIR / "screenshots"


def _idle_slot_centre(idx: int, cols: int) -> tuple[float, float]:
    """Centre of idle process slot *idx*, from the stage's own layout constants."""
    col, row = idx % cols, idx // cols
    return (
        PROCESS_AREA_X + col * (PROCESS_SIZE + PROCESS_GAP) + PROCESS_SIZE * 0.5,
        PROCESS_AREA_Y + row * (PROCESS_SIZE + PROCESS_GAP) + PROCESS_SIZE * 0.5,
    )


def _ram_page_centre(idx: int, cols: int) -> tuple[float, float]:
    """Centre of RAM page slot *idx*, from the stage's own layout constants."""
    col, row = idx % cols, idx // cols
    return (
        RAM_AREA_X + col * (PAGE_W + PAGE_GAP) + PAGE_W * 0.5,
        RAM_Y + row * (PAGE_H + PAGE_GAP) + PAGE_H * 0.5,
    )


def run_harness() -> None:
    OUT.mkdir(exist_ok=True)
    app = App(width=DESIGN_WIDTH, height=DESIGN_HEIGHT, title="You're the OS (Harness)", visible=False)
    root = OsRoot()
    sim = InputSimulator(root)

    def click_at(pos: tuple[float, float]) -> Callable[[], None]:
        return lambda: sim.click(pos)

    from nodes import state as gs

    captures: dict[int, str] = {}
    actions: dict[int, Callable[[], None]] = {}

    # Stage 1: the title screen, before anything is clicked.
    captures[1] = "01_boot.png"

    # Boot the machine, then let the startup burst spawn its first process.
    actions[4] = lambda: sim.click((DESIGN_WIDTH * 0.5, DESIGN_HEIGHT * 0.5))
    captures[60] = "02_first_process.png"

    # Stage 3: most slots filled by frame ~600 (10 s of spawns).
    captures[600] = "03_queue_filling.png"

    # Stage 4: click four idle processes onto the four CPUs.
    for i in range(4):
        actions[610 + i * 8] = click_at(_idle_slot_centre(i, gs.NUM_PROCESS_SLOT_COLS))
    captures[660] = "04_cpu_loaded.png"

    # Stage 5: click the first few RAM pages to request swaps to disk.
    for i in range(3):
        actions[700 + i * 6] = click_at(_ram_page_centre(i, gs.PAGES_PER_ROW))
    captures[800] = "05_pages_swapped.png"

    # Stage 6: by now an I/O event has queued; deliver it with Space.
    actions[1200] = lambda: sim.press_key(Key.SPACE)
    actions[1205] = lambda: sim.release_key(Key.SPACE)
    captures[1300] = "06_io_event.png"

    # Stage 7: only four processes were ever scheduled, so the rest starve and
    # the machine reaches its shutdown notice.
    captures[1800] = "07_late_game.png"

    total_frames = 1850

    def on_frame(idx: int, t: float) -> None:
        action = actions.get(idx)
        if action is not None:
            action()

    frames = app.run_headless(
        root,
        frames=total_frames,
        on_frame=on_frame,
        capture_fn=lambda idx: idx in captures,
    )

    for frame_index, image in zip(sorted(captures), frames, strict=False):
        out = OUT / captures[frame_index]
        save_png(image, out)
        print(f"saved {out}")
    print(f"{len(captures)} screenshots written")


if __name__ == "__main__":
    run_harness()