diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/value_demos/_common.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/value_demos/_common.py | 196 |
1 files changed, 196 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/value_demos/_common.py b/debug/risc-v-gsp-rm-vm/harness/value_demos/_common.py new file mode 100644 index 0000000..f70ef1f --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/value_demos/_common.py @@ -0,0 +1,196 @@ +#!/usr/bin/env python3 +"""Shared plumbing for the green-eye value-validation demos. + +Every demo maps one *real* Open-IOV scenario (cited from the handoffs) onto a +runnable demonstration on the bundled demo firmware / native RV64 simulator, +prints the question, the green-eye commands, the answer, the evidence artifact +and the measured wall-clock, and emits one machine-readable ``RESULT_JSON`` +line so ``run_all.py`` can aggregate. + +Stdlib only; no hardware; runs from the tool root +(``open-iov/debug/risc-v-gsp-rm-vm``). +""" +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import sys +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.dirname(os.path.dirname(HERE)) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +ARTIFACT_ROOT = os.path.join(HERE, "artifacts") +DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") +SCHEMA_JSON = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "gsp_ctrl.schema.json") +SDK_ELF = os.path.join(TOOL_ROOT, "sdk", "build", "injected_demo.elf") + +CTRL_MAGIC = 0x47535031 +CTRL_RPC_COUNT = 0x10 +CTRL_LAST_RPC = 0x18 +CTRL_MAILBOX = 0x20 +CTRL_ARG0 = 0x28 +CTRL_ARG1 = 0x30 +CTRL_ARG2 = 0x38 +CTRL_RESULT = 0x40 +CTRL_BUF = 0x48 +CTRL_SIZE = 0x190 +PING_XOR = 0xBEEF +INJECT_CMD_BASE = 16 +SERVICE_SLOT = 6 + +#: Per-demo wall-clock ceiling the runner enforces (subprocess timeout). +DEMO_TIMEOUT_S = 45.0 +#: Runner ceiling so the whole suite stays well under ~3 min. +SUITE_BUDGET_S = 180.0 + + +class DemoError(RuntimeError): + """A demo precondition failed in a way the caller should report, not mask.""" + + +def qemu_available() -> bool: + return bool(shutil.which("qemu-system-riscv64")) + + +def demo_available() -> bool: + return os.path.isfile(DEMO_ELF) + + +def choose_backend(preferred: str = "native") -> str: + """Environment override wins; QEMU is used only when preferred and present.""" + env = os.environ.get("VALUE_DEMO_BACKEND") + if env: + return env + if preferred == "qemu": + return "qemu" if qemu_available() else "native" + return "native" + + +def sha256_file(path: str) -> str: + digest = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +class Stopwatch: + def __init__(self) -> None: + self.t0 = time.monotonic() + self.elapsed = 0.0 + + def stop(self) -> float: + self.elapsed = time.monotonic() - self.t0 + return self.elapsed + + +def artifact_dir(name: str) -> str: + path = os.path.join(ARTIFACT_ROOT, name) + os.makedirs(path, exist_ok=True) + return path + + +def write_json(path: str, payload) -> str: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + json.dump(payload, fh, indent=2, sort_keys=True) + fh.write("\n") + return path + + +def write_text(path: str, text: str) -> str: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + fh.write(text) + return path + + +def load_schema(ge): + with open(SCHEMA_JSON, "r", encoding="utf-8") as fh: + return ge.schema_from_json(json.load(fh)) + + +def decode_ctrl(ge, data: bytes, schema=None) -> dict: + schema = schema or load_schema(ge) + out = schema.decode(data, base=0) + if isinstance(out.get("buf"), (bytes, bytearray)): + out["buf"] = bytes(out["buf"]).split(b"\x00", 1)[0].decode("utf-8", "replace") + return out + + +def short_hex(value) -> str: + if value is None: + return "None" + if isinstance(value, str): + return value + return f"{value:#x}" if isinstance(value, int) else str(value) + + +def render_banner(demo: str, question: str, scenario: str, baseline: str, commands) -> None: + print(f"=== {demo} ===") + print(f"real scenario : {scenario}") + print(f"baseline : {baseline}") + print(f"question : {question}") + print("commands :") + for command in commands: + print(f" - {command}") + + +def finish(result: dict, artifact: str, elapsed: float) -> dict: + """Persist the per-demo result and print the human + machine summary.""" + result["elapsed_s"] = round(elapsed, 4) + result["artifact"] = os.path.relpath(artifact, TOOL_ROOT) if artifact else None + result.setdefault("success", True) + if artifact: + write_json(os.path.join(artifact, "result.json"), result) + print(f"answer : {result.get('answer')}") + print(f"evidence : {result.get('artifact')}") + print(f"analog time : {elapsed:.3f} s ({result.get('backend')} backend)") + if result.get("real_baseline"): + print(f"real baseline : {result['real_baseline']}") + if result.get("projected"): + print(f"projected : {result['projected']}") + if result.get("blocked_on_device_model"): + print("blocked : real case blocked on H2 device model; workflow shown on analog") + if not result.get("success", True): + print(f"status : FAILED - {result.get('failure', 'unknown')}") + print("RESULT_JSON: " + json.dumps(result, sort_keys=True)) + sys.stdout.flush() + return result + + +def timed_demo(fn, artifact: str) -> dict: + watch = Stopwatch() + try: + result = fn() + except Exception as exc: # report, never hide + import traceback + + result = { + "success": False, + "failure": f"{type(exc).__name__}: {exc}", + "traceback": traceback.format_exc()[-2000:], + } + return finish(result, artifact, watch.stop()) + + +def scenario( + *, + ids, + baseline, + projection, + blocked=False, + device=None, +) -> dict: + return { + "citation_ids": ids, + "static_baseline": baseline, + "projection": projection, + "blocked_on_device_model": blocked, + "device_model_needed": device, + } |