harness.py

Part of You’re the OS.

  1"""Scripted-input harness: drives the port headlessly and captures seven stages.
  2
  3Stages:
  4    01_boot.png            the title screen, before the machine is booted
  5    02_first_process.png   one process visible
  6    03_queue_filling.png   ~20 processes spawned, filling slots
  7    04_cpu_loaded.png      click-toggled several processes onto CPUs
  8    05_pages_swapped.png   triggered page swaps via clicks
  9    06_io_event.png        I/O queue lit, event delivered
 10    07_late_game.png       the shutdown notice, after ten unattended processes starved
 11
 12Input goes through ``InputSimulator``, so every click and key travels the same
 13path a real mouse or keyboard would: polled state, scene-tree events, and UI
 14events all move in lockstep.
 15"""
 16
 17from __future__ import annotations
 18
 19import sys
 20from collections.abc import Callable
 21from pathlib import Path
 22
 23_PORT_DIR = Path(__file__).parent
 24if str(_PORT_DIR) not in sys.path:
 25    sys.path.insert(0, str(_PORT_DIR))
 26
 27from main import OsRoot  # noqa: E402
 28from nodes.stage import (  # noqa: E402
 29    DESIGN_HEIGHT,
 30    DESIGN_WIDTH,
 31    PAGE_GAP,
 32    PAGE_H,
 33    PAGE_W,
 34    PROCESS_AREA_X,
 35    PROCESS_AREA_Y,
 36    PROCESS_GAP,
 37    PROCESS_SIZE,
 38    RAM_AREA_X,
 39    RAM_Y,
 40)
 41
 42from simvx.core.input.enums import Key  # noqa: E402
 43from simvx.core.testing import InputSimulator  # noqa: E402
 44from simvx.graphics import App, save_png  # noqa: E402
 45
 46OUT = _PORT_DIR / "screenshots"
 47
 48
 49def _idle_slot_centre(idx: int, cols: int) -> tuple[float, float]:
 50    """Centre of idle process slot *idx*, from the stage's own layout constants."""
 51    col, row = idx % cols, idx // cols
 52    return (
 53        PROCESS_AREA_X + col * (PROCESS_SIZE + PROCESS_GAP) + PROCESS_SIZE * 0.5,
 54        PROCESS_AREA_Y + row * (PROCESS_SIZE + PROCESS_GAP) + PROCESS_SIZE * 0.5,
 55    )
 56
 57
 58def _ram_page_centre(idx: int, cols: int) -> tuple[float, float]:
 59    """Centre of RAM page slot *idx*, from the stage's own layout constants."""
 60    col, row = idx % cols, idx // cols
 61    return (
 62        RAM_AREA_X + col * (PAGE_W + PAGE_GAP) + PAGE_W * 0.5,
 63        RAM_Y + row * (PAGE_H + PAGE_GAP) + PAGE_H * 0.5,
 64    )
 65
 66
 67def run_harness() -> None:
 68    OUT.mkdir(exist_ok=True)
 69    app = App(width=DESIGN_WIDTH, height=DESIGN_HEIGHT, title="You're the OS (Harness)", visible=False)
 70    root = OsRoot()
 71    sim = InputSimulator()
 72
 73    from nodes import state as gs
 74
 75    captures: dict[int, str] = {}
 76    actions: dict[int, Callable[[], None]] = {}
 77
 78    # Stage 1: the title screen, before anything is clicked.
 79    captures[1] = "01_boot.png"
 80
 81    # Boot the machine, then let the startup burst spawn its first process.
 82    actions[4] = lambda: sim.click((DESIGN_WIDTH * 0.5, DESIGN_HEIGHT * 0.5))
 83    captures[60] = "02_first_process.png"
 84
 85    # Stage 3: most slots filled by frame ~600 (10 s of spawns).
 86    captures[600] = "03_queue_filling.png"
 87
 88    # Stage 4: click four idle processes onto the four CPUs.
 89    for i in range(4):
 90        actions[610 + i * 8] = _click_action(sim, _idle_slot_centre(i, gs.NUM_PROCESS_SLOT_COLS))
 91    captures[660] = "04_cpu_loaded.png"
 92
 93    # Stage 5: click the first few RAM pages to request swaps to disk.
 94    for i in range(3):
 95        actions[700 + i * 6] = _click_action(sim, _ram_page_centre(i, gs.PAGES_PER_ROW))
 96    captures[800] = "05_pages_swapped.png"
 97
 98    # Stage 6: by now an I/O event has queued; deliver it with Space.
 99    actions[1200] = lambda: sim.press_key(Key.SPACE)
100    actions[1205] = lambda: sim.release_key(Key.SPACE)
101    captures[1300] = "06_io_event.png"
102
103    # Stage 7: only four processes were ever scheduled, so the rest starve and
104    # the machine reaches its shutdown notice.
105    captures[1800] = "07_late_game.png"
106
107    total_frames = 1850
108
109    def on_frame(idx: int, t: float) -> None:
110        action = actions.get(idx)
111        if action is not None:
112            action()
113
114    frames = app.run_headless(
115        root,
116        frames=total_frames,
117        on_frame=on_frame,
118        capture_fn=lambda idx: idx in captures,
119    )
120
121    for frame_index, image in zip(sorted(captures), frames, strict=False):
122        out = OUT / captures[frame_index]
123        save_png(image, out)
124        print(f"saved {out}")
125    print(f"{len(captures)} screenshots written")
126
127
128def _click_action(sim: InputSimulator, pos: tuple[float, float]) -> Callable[[], None]:
129    return lambda: sim.click(pos)
130
131
132if __name__ == "__main__":
133    run_harness()