bootloader/test/pbwindow.py

#!/usr/bin/env python3
"""The activation window as a behavioral duration gate.

The loader's window is a counted poll loop whose per-poll cost is hand-counted
in the source (`link::poll_cycles`) — but the loop compiles in consumer
context, so only the running image can prove the count. This test installs a
real application beside the loader (the host tool's own `plan_flash` supplies
the reset-vector surgery), starts the simulator with the line idle, and reads
the cycle of the first transmit activity: nothing talks until the window
closes and the application banners, so that cycle *is* the window, give or
take a banner lead measured in microseconds. Asserted at ±2 % — one
mis-counted cycle per poll shifts a window by 10 % and more.

Fixed-baud loaders declare their window in seconds (--seconds, the build's
TIMEOUT). The autobaud loader's window is its calibration poll budget
(--autobaud-polls); the seconds it amounts to are budget × 10 / f_cpu, the
measured cost of the calibrate() wait loop this gate pins.
"""
import argparse
import importlib.util
import pathlib
import select
import sys
import time

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
from pbsim import Device

# The calibrate() budget loop's cycles per poll in the built image — what the
# README's window arithmetic rests on, verified here. A measured fact, not a
# design constant: the wait's exit branches land where the compiler's block
# layout puts them, and the bounded-calibration rework moved the loop from
# ten cycles to nine.
AUTOBAUD_POLL_CYCLES = 9


def load_tool(path):
    spec = importlib.util.spec_from_file_location("pureboot", path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def compose_flash(pb, loader_bytes, app_bytes, mcu, base, page):
    """The flash image a completed programming session leaves: application
    (with the tinies' vector surgery), loader at base — built through the
    host tool's own planner so the surgery is the shipped one, not a copy."""
    flash_size = base + pb.SLOT
    patch = not mcu.startswith("atmega") or mcu.startswith("atmega48")
    word_flash = flash_size > 0x10000
    wire_base = base // 2 if word_flash else base
    flags = (1 if patch else 0) | (2 if word_flash else 0)
    raw = bytes((ord("P"), ord("B"), 5, 0, 0, 0, page & 0xFF,
                 wire_base & 0xFF, wire_base >> 8, 0, 0, flags))
    info = pb.Info(raw)

    flash = bytearray(b"\xff" * flash_size)
    for address, content in pb.plan_flash(app_bytes, info).items():
        flash[address:address + len(content)] = content
    flash[base:base + len(loader_bytes)] = loader_bytes
    return bytes(flash)


def first_tx_cycle(device, deadline):
    """The PB_WINDOW_TX report, or None. The runner prints it once."""
    stream = device.proc.stdout
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return None
        ready, _, _ = select.select([stream], [], [], remaining)
        if not ready:
            return None
        line = stream.readline()
        if not line:
            return None
        if line.startswith("PB_WINDOW_TX"):
            return int(line.split()[1])


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--device", required=True)
    parser.add_argument("--loader", required=True)
    parser.add_argument("--mcu", required=True)
    parser.add_argument("--hz", type=int, required=True)
    parser.add_argument("--base", required=True)
    parser.add_argument("--page", type=int, required=True)
    parser.add_argument("--baud", type=int, required=True)
    parser.add_argument("--app", required=True)
    parser.add_argument("--tool", required=True)
    parser.add_argument("--workdir", required=True)
    parser.add_argument("--link", default=None)
    parser.add_argument("--seconds", type=float, default=None)
    parser.add_argument("--autobaud-polls", type=int, default=None)
    args = parser.parse_args()
    if (args.seconds is None) == (args.autobaud_polls is None):
        parser.error("exactly one of --seconds / --autobaud-polls")

    pb = load_tool(args.tool)
    base = int(args.base, 0)
    expected = (args.seconds if args.seconds is not None
                else args.autobaud_polls * AUTOBAUD_POLL_CYCLES / args.hz)

    work = pathlib.Path(args.workdir)
    work.mkdir(parents=True, exist_ok=True)
    # Every loader target objcopies its slot content beside the ELF (.bin).
    loader_bytes = pathlib.Path(args.loader + ".bin").read_bytes()
    app_bytes = pathlib.Path(args.app).read_bytes()
    flash_file = work / "window-flash.bin"
    flash_file.write_bytes(compose_flash(pb, loader_bytes, app_bytes, args.mcu, base, args.page))

    device = Device(args.device, args.loader, args.mcu, str(args.hz), args.base, args.page,
                    args.baud, str(work / "window-dump.bin"), resume=str(flash_file),
                    link=args.link, window=True)
    try:
        # Simulation speed is machine-dependent; a few hundred thousand
        # cycles per wall second is the pessimistic floor.
        budget = max(60.0, expected * args.hz / 300000)
        cycle = first_tx_cycle(device, time.monotonic() + budget)
    finally:
        device.stop()

    if cycle is None:
        print(f"  [FAIL] no transmit activity within {budget:.0f} s wall "
              f"(expected a {expected:.2f} s window)")
        return 1
    measured = cycle / args.hz
    error = (measured - expected) / expected
    ok = abs(error) <= 0.02
    print(f"  [{'PASS' if ok else 'FAIL'}] window {measured:.3f} s vs declared "
          f"{expected:.3f} s ({error:+.1%}, gate ±2%)")
    return 0 if ok else 1


if __name__ == "__main__":
    raise SystemExit(main())