From 2f0e8bab25928bdb35e06f2d32cd07664bb002d6 Mon Sep 17 00:00:00 2001 From: Ahnaf Shahriar Date: Sun, 23 Aug 2026 01:10:10 -0400 Subject: [PATCH 1/5] Sim(backend="rust"): the Rust engine behind the same API One Sim, two in-process engines. backend="rust" hosts the simantic_rust extension (pyrite crates/simantic-rust) through a small backend adapter; the Renode path moves behind the same adapter unchanged. expect(), record paging and symbol lookup are shared Python, so a script runs on either engine. Rust-side pieces the .NET engine used to provide: .replx rendering (_replx.py, defaults + arithmetic), ELF .symtab lookup (_elf.py), and mcu= resolution via ~/.sim_cache / the backend (same as sim --mcu). What the Rust engine lacks raises NotSupported naming simantic-core#183, never a silent no-op. The engine wheel is fetched on first use from the pyrite product manifest (engine-rust-) into ~/.simantic/engine-rust/; simantic install engine-rust fetches it up front. Bump to 0.3.0. --- README.md | 29 +++- pyproject.toml | 2 +- src/simantic/__init__.py | 10 +- src/simantic/_cli.py | 8 +- src/simantic/_elf.py | 39 +++++ src/simantic/_replx.py | 108 ++++++++++++++ src/simantic/_rust.py | 130 +++++++++++++++++ src/simantic/engine.py | 44 ++++++ src/simantic/install.py | 70 ++++++++- src/simantic/session.py | 281 +++++++++++++++++++++++-------------- tests/test_install.py | 35 +++++ tests/test_rust_backend.py | 150 ++++++++++++++++++++ 12 files changed, 786 insertions(+), 120 deletions(-) create mode 100644 src/simantic/_elf.py create mode 100644 src/simantic/_replx.py create mode 100644 src/simantic/_rust.py create mode 100644 tests/test_rust_backend.py diff --git a/README.md b/README.md index 689a573..5e1819b 100644 --- a/README.md +++ b/README.md @@ -6,9 +6,9 @@ time by exact amounts, inject UART/GPIO/CAN/radio, read memory and RTOS state, and run as many simulations in parallel as you have cores. pytest is one way to use it, not a requirement. -> **Alpha — not stable.** Version 0.2.x. The API, the CLI surface, and the +> **Alpha — not stable.** Version 0.3.x. The API, the CLI surface, and the > report schema may change without a deprecation period, and any release may -> break the previous one. Pin an exact version (`simantic==0.2.0`) if you +> break the previous one. Pin an exact version (`simantic==0.3.0`) if you > depend on it. Not recommended for production pipelines yet. ```bash @@ -16,9 +16,10 @@ pip install simantic ``` That is the whole setup for Python. The first `Sim(...)` fetches the -simulation engine (Simantic.Core plus a private .NET runtime — nothing else -to install) into `~/.simantic/engine//`, checksum-verified against -the public release manifest. `simantic install` fetches it up front, along +simulation engine it needs — the Renode engine (Simantic.Core plus a +private .NET runtime) into `~/.simantic/engine//`, or the Rust +engine into `~/.simantic/engine-rust//` — checksum-verified +against the public release manifest. Nothing else to install. `simantic install` fetches it up front, along with the `sim` binary if you also want the command-line tool. A Simantic account (`simantic auth`) is needed for one thing: resolving MCU @@ -65,6 +66,24 @@ functions. See One-shot runs ("run 5 s, give me the transcript") are `run_firmware(...)`. +## Pick your engine + +`Sim` runs on either of two engines, both hosted in your process, selected +per simulation: + +```python +Sim(elf="fw.elf", mcu="STM32F401RE", uart="usart2") # Renode engine (default) +Sim(elf="fw.elf", mcu="STM32F401RE", uart="usart2", backend="rust") # Simantic's Rust engine +``` + +The script is the same; only the engine changes. The Rust engine is a +single small extension module (fetched on first use, like the Renode +engine), runs one machine, and is considerably faster. What it does not do +yet — multi-machine scenarios, network services, scripted peers, CAN/radio +injection, RTOS thread views — raises `simantic.NotSupported` naming the +gap rather than silently doing nothing. The capability table both engines +are ticked against is [simantic-core#183](https://github.com/simantic-dev/simantic-core/issues/183). + ## Using it from pytest (optional) `Sim` needs no plugin — construct it inside any test. If you also keep diff --git a/pyproject.toml b/pyproject.toml index a7076f4..456f041 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "simantic" -version = "0.2.1" +version = "0.3.0" description = "Python SDK and pytest plugin for the Simantic circuit and firmware simulators" # PyYAML reads sim-fixtures test.yaml manifests; pythonnet hosts the # simulation engine (Simantic.Core, .NET) in-process for `simantic.Sim`. diff --git a/src/simantic/__init__.py b/src/simantic/__init__.py index 99d6c72..60184d3 100644 --- a/src/simantic/__init__.py +++ b/src/simantic/__init__.py @@ -25,10 +25,11 @@ ) from .mcu import ServerNotConfigured, SimError, SimRun, sim_binary from .mcu import run as run_firmware -from .engine import EngineNotFound, engine_dir -from .session import ExpectTimeout, Match, Sim +from .engine import EngineNotFound, engine_dir, rust_engine_dir +from .session import BACKENDS, ExpectTimeout, Match, Sim +from ._rust import NotSupported -__version__ = "0.2.1" +__version__ = "0.3.0" __all__ = [ # firmware @@ -46,7 +47,10 @@ "Match", "ExpectTimeout", "EngineNotFound", + "NotSupported", + "BACKENDS", "engine_dir", + "rust_engine_dir", # shared "BinaryNotFound", ] diff --git a/src/simantic/_cli.py b/src/simantic/_cli.py index b982832..3baebfb 100644 --- a/src/simantic/_cli.py +++ b/src/simantic/_cli.py @@ -39,12 +39,14 @@ def _auth(args) -> int: def _install(args) -> int: - names = args.binary or [name for name, _ in BINARIES] + [install.ENGINE_KEY] + names = args.binary or [name for name, _ in BINARIES] + [install.ENGINE_KEY, install.RUST_ENGINE_KEY] failures = 0 for name in names: try: if name == install.ENGINE_KEY: path = install.install_engine(force=args.force, channel=args.channel) + elif name == install.RUST_ENGINE_KEY: + path = install.install_rust_engine(force=args.force, channel=args.channel) else: path = install.install(name, force=args.force, channel=args.channel) print(f"{name}: {path}") @@ -72,6 +74,8 @@ def _status(args) -> int: print(f" {name}: not found (run `simantic install {name}`)") engine = install.installed_engine() print(f" engine: {engine if engine else 'not found (fetched on first use, or `simantic install engine`)'}") + rust = install.installed_rust_engine() + print(f" engine-rust: {rust if rust else 'not found (fetched on first use of backend=\"rust\", or `simantic install engine-rust`)'}") print(telemetry.describe()) return 0 @@ -98,7 +102,7 @@ def main(argv: list[str] | None = None) -> int: p_auth.set_defaults(func=_auth) p_install = sub.add_parser("install", help="download simulator binaries") - p_install.add_argument("binary", nargs="*", help="defaults to all known binaries and the engine") + p_install.add_argument("binary", nargs="*", help="defaults to all known binaries and both engines") p_install.add_argument( "--force", action="store_true", help="re-download even if already present" ) diff --git a/src/simantic/_elf.py b/src/simantic/_elf.py new file mode 100644 index 0000000..98213ad --- /dev/null +++ b/src/simantic/_elf.py @@ -0,0 +1,39 @@ +"""Symbol addresses from a 32-bit little-endian ELF, with the standard library. + +The Renode backend resolves symbols inside the engine. The Rust engine does +not carry a symbol table, so `Sim.symbol()` on that backend reads `.symtab` +here — the same answer, from the same file. +""" + +from __future__ import annotations + +import struct + +SHT_SYMTAB = 2 +STT_FUNC = 2 + + +def symbols(elf: bytes) -> dict[str, int]: + if elf[:4] != b"\x7fELF" or elf[4] != 1 or elf[5] != 1: + raise ValueError("only 32-bit little-endian ELF images are supported") + (shoff,) = struct.unpack_from(" str: + """Fold a numeric expression; anything else passes through untouched.""" + expr = expr.strip() + if not _ARITHMETIC.fullmatch(expr) or re.fullmatch(r"[0-9.]+", expr): + return expr + tree = ast.parse(expr, mode="eval") + + def fold(node): + if isinstance(node, ast.Expression): + return fold(node.body) + if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)): + return node.value + if isinstance(node, ast.BinOp) and isinstance(node.op, (ast.Add, ast.Sub, ast.Mult, ast.Div)): + a, b = fold(node.left), fold(node.right) + return {ast.Add: a + b, ast.Sub: a - b, ast.Mult: a * b, ast.Div: a / b}[type(node.op)] + if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub): + return -fold(node.operand) + raise ValueError(f"unsupported expression in platform template: {expr!r}") + + value = fold(tree) + return str(int(value)) if float(value).is_integer() else str(value) + + +def render(text: str) -> str: + """`.replx` → `.repl`: placeholders take their defaults.""" + return _PLACEHOLDER.sub(lambda m: _evaluate(m.group(1).split(":")[-1]), text) + + +def cache_dir() -> Path: + return Path(os.environ.get("HOME", "")) / ".sim_cache" + + +def model_replx(mcu: str, *, use_cache: bool = True) -> str: + """The `.replx` text for a model name, like `sim --mcu`.""" + if os.environ.get(MCU_LIB_ENV): + return platform_path(mcu, None, Path.cwd()).read_text() + cached = cache_dir() / f"{mcu.lower()}.json" + if use_cache and cached.exists(): + replx = json.loads(cached.read_text()).get("replx") + if replx: + return replx + try: + credentials = auth.load() + except auth.NotAuthenticated as exc: + raise SimError(f"mcu={mcu!r} needs credentials to fetch the model: {exc}") from None + request = urllib.request.Request( + f"{MCU_DETAILS_URL}?model={urllib.parse.quote(mcu)}", + headers={"Authorization": f"Bearer {credentials.api_key}"}, + ) + try: + with urllib.request.urlopen(request, timeout=30) as response: + details = json.loads(response.read()) + except urllib.error.HTTPError as exc: + raise SimError(f"mcu={mcu!r} is not a supported model (HTTP {exc.code})") from None + except urllib.error.URLError as exc: + raise SimError(f"cannot reach the model backend: {exc.reason}") from None + replx = details.get("replx") + if not replx: + raise SimError(f"the backend returned no platform for mcu={mcu!r}") + cached.parent.mkdir(parents=True, exist_ok=True) + cached.write_text(json.dumps({"model": mcu, "replx": replx, "deprecated": bool(details.get("deprecated"))})) + return replx + + +def platform_text(*, repl: Path | None, mcu: str | None, overlay: Path | None) -> str: + """Rendered `.repl` text for one machine.""" + if repl is not None: + text = repl.read_text() + else: + assert mcu is not None + text = model_replx(mcu) + if overlay is not None: + # The platform grammar has no comment syntax; strip note lines first. + body = "\n".join(l for l in overlay.read_text().splitlines() if not l.lstrip().startswith(("#", "//"))) + text = text.rstrip() + "\n\n" + body.strip() + "\n" + return render(text) diff --git a/src/simantic/_rust.py b/src/simantic/_rust.py new file mode 100644 index 0000000..327f848 --- /dev/null +++ b/src/simantic/_rust.py @@ -0,0 +1,130 @@ +"""The Rust backend of `Sim`: `simantic_rust.Session` hosted in this process. + +Same vocabulary as the Renode backend, one machine at a time. What the Rust +engine does not do yet raises `NotSupported` rather than silently doing +nothing — the gap list is simantic-core#183. +""" + +from __future__ import annotations + +import re +import time +from pathlib import Path + +from . import _elf, _replx +from .engine import load_rust +from .mcu import SimError + +#: Virtual time advanced between UART checks while waiting in expect(). +SLICE_SECONDS = 0.001 + + +class NotSupported(SimError): + """The Rust backend has no implementation of this yet (simantic-core#183).""" + + +class RustBackend: + def __init__(self, machines: list[dict], *, base: Path, media, services, quantum, + trace_symbols, trace_interrupts, engine_dir): + if len(machines) != 1: + raise NotSupported("backend='rust' runs one machine; multi-machine scenarios need backend='renode'") + if media or services: + raise NotSupported("backend='rust' has no media or network services yet (simantic-core#183)") + if trace_symbols or trace_interrupts: + raise NotSupported("backend='rust' has no symbol/interrupt tracing yet (simantic-core#183)") + m = machines[0] + self.machines = [m["name"]] + self._elf = (base / m["elf"]).read_bytes() + repl = base / m["repl"] if m.get("repl") else None + overlay = base / m["overlay"] if m.get("overlay") else None + text = _replx.platform_text(repl=repl, mcu=m.get("mcu"), overlay=overlay) + engine = load_rust(engine_dir) + try: + self._s = engine.Session(text, self._elf) + except Exception as exc: + raise SimError(str(exc)) from None + self._symbols: dict[str, int] | None = None + self._records: dict[str, list[dict]] = {k: [] for k in ("uart", "frames", "logs", "interrupts", "symbol_trace")} + self._records["logs"] = [{"t": 0.0, "level": "Warning", "source": "platform", "message": w} + for w in self._s.warnings()] + + # -- stimulus --------------------------------------------------------- + + def send(self, data: bytes, uart: str, machine: str | None) -> None: + self._s.send_uart(uart, bytes(data)) + + def inject_gpio(self, peripheral: str, pin: int, state: bool, machine: str | None) -> None: + self._s.inject_gpio(peripheral, int(pin), bool(state)) + + def inject_can(self, *_a, **_k) -> None: + raise NotSupported("backend='rust' has no CAN injection yet (simantic-core#183)") + + def inject_radio(self, *_a, **_k) -> None: + raise NotSupported("backend='rust' has no radio injection yet (simantic-core#183)") + + # -- time ------------------------------------------------------------- + + def run_for(self, seconds: float) -> float: + self._advance(seconds) + return self.time + + @property + def time(self) -> float: + return self._s.time() + + def expect(self, pattern: str, uart: str, machine: str | None, timeout: float) -> tuple[bool, str, float]: + """Run in slices until `pattern` shows up on `uart`; `timeout` is wall-clock.""" + rx = re.compile(pattern) + deadline = time.monotonic() + timeout + text = "" + while True: + for rec in self._advance(SLICE_SECONDS): + if rec["label"] == uart: + text += rec["text"] + if rx.search(text): + return True, text, self.time + if time.monotonic() > deadline: + return False, text, self.time + + def _advance(self, seconds: float) -> list[dict]: + self._s.run_for(float(seconds)) + fresh: list[dict] = [] + for t, label, byte in self._s.take_uart(): + ch = chr(byte) + if fresh and fresh[-1]["label"] == label: + fresh[-1]["text"] += ch + else: + fresh.append({"t": t, "machine": self.machines[0], "label": label, "text": ch}) + self._records["uart"].extend(fresh) + return fresh + + # -- observation ------------------------------------------------------ + + def records(self, kind: str, cursor: int, limit: int) -> tuple[list[dict], int, bool]: + items = self._records[kind] + page = items[cursor : cursor + limit] + nxt = cursor + len(page) + return page, nxt, nxt < len(items) + + def read_memory(self, address: int, count: int, machine: str | None) -> bytes: + try: + return bytes(self._s.read_memory(int(address), int(count))) + except Exception as exc: + raise SimError(str(exc)) from None + + def symbol(self, name: str, machine: str | None) -> int: + if self._symbols is None: + self._symbols = _elf.symbols(self._elf) + try: + return self._symbols[name] + except KeyError: + raise SimError(f"no symbol {name!r} in the ELF") from None + + def threads(self, machine: str | None): + raise NotSupported("backend='rust' has no RTOS thread view through Sim yet (simantic-core#183)") + + def heap(self, machine: str | None): + raise NotSupported("backend='rust' has no heap report through Sim yet (simantic-core#183)") + + def close(self) -> None: + self._s = None diff --git a/src/simantic/engine.py b/src/simantic/engine.py index abc6770..59d2b18 100644 --- a/src/simantic/engine.py +++ b/src/simantic/engine.py @@ -22,6 +22,7 @@ from .mcu import sim_binary ENV_DIR = "SIMANTIC_ENGINE_DIR" +RUST_ENV_DIR = "SIMANTIC_RUST_ENGINE_DIR" class EngineNotFound(RuntimeError): @@ -89,3 +90,46 @@ def load(explicit: str | os.PathLike[str] | None = None): import Simantic.Core.Emulation.Session as session_ns # type: ignore[import-not-found] return session_ns + + +def rust_engine_dir(explicit: str | os.PathLike[str] | None = None, *, fetch: bool = True) -> Path: + """The directory holding the `simantic_rust` extension module. + + Order: an explicit path, $SIMANTIC_RUST_ENGINE_DIR, the managed install + under ~/.simantic/engine-rust; else fetched from the public release. + """ + candidates = [Path(p) for p in (explicit, os.environ.get(RUST_ENV_DIR)) if p] + for d in candidates: + if install.is_rust_engine(d): + return d + managed = install.installed_rust_engine() + if managed is not None: + return managed + if fetch: + try: + return install.install_rust_engine() + except install.InstallError as exc: + raise EngineNotFound(f"could not fetch the Rust engine: {exc}") from None + raise EngineNotFound( + f"simantic_rust not found. Run `simantic install engine-rust`, or set ${RUST_ENV_DIR}." + ) + + +@cache +def load_rust(explicit: str | os.PathLike[str] | None = None): + """Import the Rust engine. A `simantic_rust` already importable (a + development `maturin develop`, or the wheel installed directly) wins.""" + try: + import simantic_rust # type: ignore[import-not-found] + + return simantic_rust + except ImportError: + pass + d = rust_engine_dir(explicit) + if str(d) not in sys.path: + sys.path.insert(0, str(d)) + try: + import simantic_rust # type: ignore[import-not-found] + except ImportError as exc: + raise EngineNotFound(f"{d} does not contain a loadable simantic_rust: {exc}") from None + return simantic_rust diff --git a/src/simantic/install.py b/src/simantic/install.py index 19d6f05..40be34f 100644 --- a/src/simantic/install.py +++ b/src/simantic/install.py @@ -138,7 +138,10 @@ def resolve( binary: str, *, rid: str | None = None, channel: str | None = None ) -> Artifact: """The artifact this machine should download.""" - manifest = fetch_manifest(binary, channel=channel) + return _resolve(fetch_manifest(binary, channel=channel), binary, rid) + + +def _resolve(manifest: dict, binary: str, rid: str | None) -> Artifact: version = manifest.get("version") artifacts = manifest.get("artifacts") if not version or not isinstance(artifacts, dict): @@ -282,7 +285,65 @@ def install_engine(*, force: bool = False, channel: str | None = None) -> Path: payload = download(artifact) if not payload.startswith(b"PK\x03\x04"): raise InstallError("engine artifact is not a zip archive") - incoming = target.with_name(f".{artifact.version}.incoming") + _unpack_zip(payload, target) + return target + + +# -- the Rust engine: the simantic_rust extension module ---------------------- + +RUST_ENGINE_KEY = "engine-rust" +#: The Rust engine is published under its own product manifest. +RUST_ENGINE_PRODUCT = "pyrite" + + +def rust_engine_root() -> Path: + """Where Rust engine releases live: ~/.simantic/engine-rust//.""" + return simantic_home() / "engine-rust" + + +def is_rust_engine(d: Path) -> bool: + return d.is_dir() and any(p.name.startswith("simantic_rust.") for p in d.iterdir()) + + +def installed_rust_engine() -> Path | None: + root = rust_engine_root() + if not root.exists(): + return None + candidates = [d for d in root.iterdir() if is_rust_engine(d)] + return max(candidates, key=lambda d: _version_key(d.name)) if candidates else None + + +def fetch_rust_manifest(*, channel: str | None = None, timeout: float = 30) -> dict: + channel = channel or default_channel() + url = f"{releases_url()}/{RUST_ENGINE_PRODUCT}/{channel}.json" + try: + with urllib.request.urlopen(urllib.request.Request(url), timeout=timeout) as response: + return json.loads(response.read()) + except urllib.error.HTTPError as exc: + raise InstallError(f"no published Rust engine (HTTP {exc.code} from {url})") from None + except urllib.error.URLError as exc: + raise InstallError(f"cannot reach the release server: {exc.reason}") from None + except json.JSONDecodeError as exc: + raise InstallError(f"release manifest is not valid JSON: {exc}") from None + + +def install_rust_engine(*, force: bool = False, channel: str | None = None) -> Path: + """Download the Rust engine wheel for this machine and unpack it into + rust_engine_root()/. A wheel is a zip; the module inside is abi3, + so one wheel per platform serves every supported Python.""" + artifact = _resolve(fetch_rust_manifest(channel=channel), RUST_ENGINE_KEY, f"{RUST_ENGINE_KEY}-{current_rid()}") + target = rust_engine_root() / artifact.version + if is_rust_engine(target) and not force: + return target + payload = download(artifact) + if not payload.startswith(b"PK\x03\x04"): + raise InstallError("Rust engine artifact is not a wheel") + _unpack_zip(payload, target) + return target + + +def _unpack_zip(payload: bytes, target: Path) -> None: + incoming = target.with_name(f".{target.name}.incoming") if incoming.exists(): shutil.rmtree(incoming) incoming.mkdir(parents=True) @@ -292,15 +353,14 @@ def install_engine(*, force: bool = False, channel: str | None = None) -> Path: # Refuse anything that would land outside the target. dest = (incoming / name).resolve() if not str(dest).startswith(str(incoming.resolve())): - raise InstallError(f"engine archive has an unsafe path: {name}") + raise InstallError(f"archive has an unsafe path: {name}") archive.extractall(incoming) if target.exists(): shutil.rmtree(target) os.replace(incoming, target) except (OSError, zipfile.BadZipFile) as exc: shutil.rmtree(incoming, ignore_errors=True) - raise InstallError(f"cannot unpack the engine into {target}: {exc}") from None - return target + raise InstallError(f"cannot unpack into {target}: {exc}") from None def installed_version(binary: str) -> str | None: diff --git a/src/simantic/session.py b/src/simantic/session.py index b1c7f07..d358e7f 100644 --- a/src/simantic/session.py +++ b/src/simantic/session.py @@ -32,8 +32,12 @@ keys. (`$SIMANTIC_MCU_LIB` switches `mcu=` to a local model library for model development.) -The engine is `Simantic.Core`, hosted in-process (see `engine.py`); this -class adds vocabulary, not semantics. +`backend=` picks the engine, both hosted in-process: `"renode"` (the +default; `Simantic.Core`, see `engine.py`) or `"rust"` (`simantic_rust`, the +pure-Rust engine — one machine, faster, and missing some capabilities that +raise `NotSupported` rather than silently no-op; simantic-core#183 is the +table). A script written against one runs unchanged on the other wherever +both tick. This class adds vocabulary, not semantics. """ from __future__ import annotations @@ -49,6 +53,8 @@ class adds vocabulary, not semantics. from .fixtures import MCU_LIB_ENV, platform_path from .mcu import SimError +BACKENDS = ("renode", "rust") + class ExpectTimeout(AssertionError): """expect() did not match; carries the text collected while waiting.""" @@ -139,7 +145,11 @@ def __init__( show_logs: bool = False, cwd: str | os.PathLike[str] | None = None, engine_dir: str | os.PathLike[str] | None = None, + backend: str = "renode", ): + if backend not in BACKENDS: + raise ValueError(f"backend must be one of {BACKENDS}, got {backend!r}") + self.backend = backend self.machine = machine self.uart = uart self._cursors = {"uart": 0, "frames": 0, "logs": 0, "interrupts": 0, "symbol_trace": 0} @@ -160,68 +170,32 @@ def __init__( elif elf is None or (repl is None) == (mcu is None): raise ValueError("give elf= and exactly one of repl= or mcu= (or scenario=)") - ns = load(engine_dir) - spec = ns.SessionSpec() - spec.TraceInterrupts = trace_interrupts - spec.ShowBackendLogs = show_logs - for s in trace_symbols: - spec.TraceSymbols.Add(s) - if scenario is not None: - self._fill_scenario(spec, scenario) + machines = [dict(name=name, **m) for name, m in scenario["machines"].items()] + for m in machines: + if "elf" not in m or ("repl" in m) == ("mcu" in m): + raise ValueError(f"machine {m['name']!r} needs elf and exactly one of repl/mcu") else: - self._add_machine(spec, "machine", repl, mcu, overlay, elf) - - telemetry.record("sdk.session") - try: - self._session = ns.Session.Start(spec) - except Exception as exc: # .NET exceptions surface as Python exceptions - raise SimError(f"could not start the simulation: {exc}") from None - self.machines: list[str] = list(self._session.Machines) - - # -- platform / scenario preparation ----------------------------------- - - def _add_machine(self, spec, name: str, repl, mcu, overlay, elf) -> None: - """Platform file → AddMachine; model name → the local model library when - $SIMANTIC_MCU_LIB is set (development), else the engine's own resolver - (~/.sim_cache, then the backend with stored credentials — like `sim --mcu`).""" - elf_path = str(self._base / elf) - if repl is not None: - if overlay is not None: - raise ValueError("overlay= applies to mcu=, not repl=") - spec.AddMachine(name, str(self._base / repl), elf_path) - return - if os.environ.get(MCU_LIB_ENV): - platform = platform_path(mcu, self._base / overlay if overlay else None, self._work) - spec.AddMachine(name, str(platform), elf_path) - return - fragment = (self._base / overlay).read_text() if overlay else None - spec.AddModel(name, mcu, elf_path, fragment) - - def _fill_scenario(self, spec, scenario: dict[str, Any]) -> None: - machines = scenario.get("machines") or {} - if not machines: - raise ValueError("scenario needs at least one machine") - for name, m in machines.items(): - if "elf" not in m or ("repl" in m) == ("mcu" in m): - raise ValueError(f"machine {name!r} needs elf and exactly one of repl/mcu") - self._add_machine(spec, name, m.get("repl"), m.get("mcu"), m.get("overlay"), m["elf"]) - for med in scenario.get("media") or []: - sm = spec.AddMedium(med["type"], list(med.get("connect") or [])) - sm.Strict = bool(med.get("strict", False)) - if med.get("hostBridge"): - sm.HostBridge = med["hostBridge"] - for svc in scenario.get("networkServices") or []: - spec.AddService(svc["name"], svc["host"], int(svc.get("port", 0)), - svc.get("type", "Antmicro.Renode.Peripherals.Network.EchoService"), - self._service_args(svc.get("args", ""))) - if scenario.get("quantum") is not None: - spec.QuantumSeconds = float(scenario["quantum"]) - - def _service_args(self, args: str) -> str: - # A script path is the common case; make it absolute against cwd=. - p = self._base / args - return str(p) if args and p.exists() else args + machines = [{"name": "machine", "elf": elf, "repl": repl, "mcu": mcu, "overlay": overlay}] + scenario = scenario or {} + + telemetry.record(f"sdk.session.{backend}") + if backend == "rust": + from ._rust import RustBackend + + self._b = RustBackend( + machines, base=self._base, media=scenario.get("media"), + services=scenario.get("networkServices"), quantum=scenario.get("quantum"), + trace_symbols=trace_symbols, trace_interrupts=trace_interrupts, engine_dir=engine_dir, + ) + else: + self._b = _RenodeBackend( + machines, base=self._base, work=self._work, media=scenario.get("media"), + services=scenario.get("networkServices"), quantum=scenario.get("quantum"), + trace_symbols=trace_symbols, trace_interrupts=trace_interrupts, + show_logs=show_logs, engine_dir=engine_dir, + ) + self.machines: list[str] = list(self._b.machines) # -- stimulus ----------------------------------------------------------- @@ -232,33 +206,33 @@ def send(self, text: str, line_ending: str = "\r", uart: str | None = None, self.send_bytes((text + line_ending).encode("latin-1"), uart, machine) def send_bytes(self, data: bytes, uart: str | None = None, machine: str | None = None) -> None: - self._session.Send(bytes(data), uart or self.uart, machine or self.machine) + self._b.send(bytes(data), uart or self.uart, machine or self.machine) def inject_gpio(self, peripheral: str, pin: int, state: bool, machine: str | None = None) -> None: """Drive an external GPIO input line (a button press/release).""" - self._session.InjectGpio(peripheral, pin, state, machine or self.machine) + self._b.inject_gpio(peripheral, pin, state, machine or self.machine) def inject_can(self, peripheral: str, can_id: int, data: bytes, *, extended: bool = False, remote: bool = False, fd: bool = False, brs: bool = False, machine: str | None = None) -> None: """Put a CAN frame on the bus as seen by `peripheral`.""" - self._session.InjectCan(peripheral, can_id, bytes(data), extended, remote, fd, brs, - machine or self.machine) + self._b.inject_can(peripheral, can_id, bytes(data), extended, remote, fd, brs, + machine or self.machine) def inject_radio(self, peripheral: str, frame: bytes, machine: str | None = None) -> None: """Deliver a raw radio frame to a radio peripheral.""" - self._session.InjectRadio(peripheral, bytes(frame), machine or self.machine) + self._b.inject_radio(peripheral, bytes(frame), machine or self.machine) # -- time control ------------------------------------------------------- def run_for(self, virtual_seconds: float) -> float: """Advance exactly this much virtual time, then hold. Returns elapsed virtual time.""" - return self._await(self._session.RunForAsync(float(virtual_seconds))) + return self._b.run_for(float(virtual_seconds)) @property def time(self) -> float: """Elapsed virtual time in seconds.""" - return self._session.VirtualTime + return self._b.time def expect(self, pattern: str, timeout: float = 30, uart: str | None = None, machine: str | None = None) -> Match: @@ -278,11 +252,11 @@ def expect(self, pattern: str, timeout: float = 30, uart: str | None = None, self._consume(m.end()) return Match(m.group(0), t) - r = self._await(self._session.ExpectAsync(pattern, uart or self.uart, machine or self.machine, float(timeout))) - if not r.Matched: - raise ExpectTimeout(pattern, text + r.Text, r.VirtualSeconds) - live = rx.search(r.Text) - matched_text = live.group(0) if live else r.Text + matched, live_text, at = self._b.expect(pattern, uart or self.uart, machine or self.machine, float(timeout)) + if not matched: + raise ExpectTimeout(pattern, text + live_text, at) + live = rx.search(live_text) + matched_text = live.group(0) if live else live_text # Consume the stream through the live match and no further, so lines # printed in the overshoot stay buffered for the next expect. self._drain_pending() @@ -294,7 +268,7 @@ def expect(self, pattern: str, timeout: float = 30, uart: str | None = None, idx = text.rfind(matched_text) if idx >= 0: self._consume(idx + len(matched_text)) - return Match(matched_text, r.VirtualSeconds) + return Match(matched_text, at) # -- observation (never advances time) ---------------------------------- @@ -306,53 +280,53 @@ def read_uart(self, from_start: bool = False) -> str: def uart_records(self, from_start: bool = False) -> list[dict]: """Timestamped UART records: {t, machine, label, text}.""" - return self._records("uart", self._session.ReadUart, _uart, from_start) + return self._records("uart", from_start) def frames(self, from_start: bool = False) -> list[dict]: """Captured bus frames (CAN/SPI/I2C/BLE/Ethernet) since the last call.""" - return self._records("frames", self._session.ReadFrames, _frame, from_start) + return self._records("frames", from_start) def logs(self, from_start: bool = False) -> list[dict]: """Simulator-side logs — unhandled registers, model warnings.""" - return self._records("logs", self._session.ReadLogs, _log, from_start) + return self._records("logs", from_start) def interrupts(self, from_start: bool = False) -> list[dict]: """Interrupt entry/exit records (needs trace_interrupts=True).""" - return self._records("interrupts", self._session.ReadInterrupts, _interrupt, from_start) + return self._records("interrupts", from_start) def symbol_trace(self, from_start: bool = False) -> list[dict]: """Hits on trace_symbols= with their argument registers (non-halting).""" - return self._records("symbol_trace", self._session.ReadSymbolTrace, _symbol_trace, from_start) + return self._records("symbol_trace", from_start) def read_memory(self, address: int | str, count: int = 4, machine: str | None = None) -> bytes: """Read bytes from the system bus; `address` is an int or a symbol name.""" if isinstance(address, str): address = self.symbol(address, machine) - return _bytes(self._session.ReadMemory(int(address), int(count), machine or self.machine)) + return self._b.read_memory(int(address), int(count), machine or self.machine) def read_u32(self, address: int | str, machine: str | None = None) -> int: return int.from_bytes(self.read_memory(address, 4, machine), "little") def symbol(self, name: str, machine: str | None = None) -> int: """Address of an ELF symbol.""" - return int(self._session.ResolveSymbol(name, machine or self.machine)) + return self._b.symbol(name, machine or self.machine) def threads(self, machine: str | None = None) -> dict | None: """RTOS thread snapshot, e.g. {"rtos": "Zephyr", "threads": [{"name", "state", "priority", ...}], "truncated": False}; None when no RTOS is recognised.""" - return _as_dict(self._session.Threads(machine or self.machine)) + return self._b.threads(machine or self.machine) def heap(self, machine: str | None = None) -> dict | None: """Heap report, e.g. {"arenaStart", "arenaSizeBytes", "usedBytes", "freeBytes", "largestFreeBlockBytes", "fragmentationRatio", ...}; None when not recognised.""" - return _as_dict(self._session.Heap(machine or self.machine)) + return self._b.heap(machine or self.machine) # -- lifecycle ---------------------------------------------------------- def close(self) -> None: - if getattr(self, "_session", None) is not None: - self._session.Dispose() - self._session = None + if getattr(self, "_b", None) is not None: + self._b.close() + self._b = None def __enter__(self) -> "Sim": return self @@ -362,26 +336,13 @@ def __exit__(self, *_exc: Any) -> None: # -- internals ---------------------------------------------------------- - @staticmethod - def _await(task): - """Wait for an engine task while releasing the GIL: scripted peers run - Python on the emulation thread and need it while the clock is running.""" - import time - - while not task.IsCompleted: - time.sleep(0.0005) - if task.IsFaulted: - raise SimError(str(task.Exception.GetBaseException().Message)) - return task.Result - - def _records(self, key: str, reader, convert, from_start: bool) -> list[dict]: + def _records(self, key: str, from_start: bool) -> list[dict]: cursor = 0 if from_start else self._cursors[key] out: list[dict] = [] while True: - page = reader(cursor, 2000) - out.extend(convert(r) for r in page.Records) - cursor = page.Next - if not page.Truncated: + page, cursor, truncated = self._b.records(key, cursor, 2000) + out.extend(page) + if not truncated: break self._cursors[key] = cursor return out @@ -407,3 +368,115 @@ def _consume(self, offset: int) -> None: else: self._pending[0] = (t, s[offset:]) offset = 0 + + +class _RenodeBackend: + """`Simantic.Core.Emulation.Session` via pythonnet — the default engine.""" + + _READERS = {"uart": ("ReadUart", _uart), "frames": ("ReadFrames", _frame), "logs": ("ReadLogs", _log), + "interrupts": ("ReadInterrupts", _interrupt), "symbol_trace": ("ReadSymbolTrace", _symbol_trace)} + + def __init__(self, machines: list[dict], *, base: Path, work: Path, media, services, quantum, + trace_symbols, trace_interrupts, show_logs, engine_dir): + self._base, self._work = base, work + ns = load(engine_dir) + spec = ns.SessionSpec() + spec.TraceInterrupts = trace_interrupts + spec.ShowBackendLogs = show_logs + for sym in trace_symbols: + spec.TraceSymbols.Add(sym) + for m in machines: + self._add_machine(spec, m["name"], m.get("repl"), m.get("mcu"), m.get("overlay"), m["elf"]) + for med in media or []: + sm = spec.AddMedium(med["type"], list(med.get("connect") or [])) + sm.Strict = bool(med.get("strict", False)) + if med.get("hostBridge"): + sm.HostBridge = med["hostBridge"] + for svc in services or []: + spec.AddService(svc["name"], svc["host"], int(svc.get("port", 0)), + svc.get("type", "Antmicro.Renode.Peripherals.Network.EchoService"), + self._service_args(svc.get("args", ""))) + if quantum is not None: + spec.QuantumSeconds = float(quantum) + try: + self._session = ns.Session.Start(spec) + except Exception as exc: # .NET exceptions surface as Python exceptions + raise SimError(f"could not start the simulation: {exc}") from None + self.machines = list(self._session.Machines) + + def _add_machine(self, spec, name: str, repl, mcu, overlay, elf) -> None: + """Platform file → AddMachine; model name → the local model library when + $SIMANTIC_MCU_LIB is set (development), else the engine's own resolver + (~/.sim_cache, then the backend with stored credentials — like `sim --mcu`).""" + elf_path = str(self._base / elf) + if repl is not None: + if overlay is not None: + raise ValueError("overlay= applies to mcu=, not repl=") + spec.AddMachine(name, str(self._base / repl), elf_path) + return + if os.environ.get(MCU_LIB_ENV): + platform = platform_path(mcu, self._base / overlay if overlay else None, self._work) + spec.AddMachine(name, str(platform), elf_path) + return + fragment = (self._base / overlay).read_text() if overlay else None + spec.AddModel(name, mcu, elf_path, fragment) + + def _service_args(self, args: str) -> str: + # A script path is the common case; make it absolute against cwd=. + p = self._base / args + return str(p) if args and p.exists() else args + + def send(self, data, uart, machine): + self._session.Send(data, uart, machine) + + def inject_gpio(self, peripheral, pin, state, machine): + self._session.InjectGpio(peripheral, pin, state, machine) + + def inject_can(self, peripheral, can_id, data, extended, remote, fd, brs, machine): + self._session.InjectCan(peripheral, can_id, data, extended, remote, fd, brs, machine) + + def inject_radio(self, peripheral, frame, machine): + self._session.InjectRadio(peripheral, frame, machine) + + def run_for(self, seconds: float) -> float: + return self._await(self._session.RunForAsync(seconds)) + + @property + def time(self) -> float: + return self._session.VirtualTime + + def expect(self, pattern, uart, machine, timeout): + r = self._await(self._session.ExpectAsync(pattern, uart, machine, timeout)) + return bool(r.Matched), r.Text, r.VirtualSeconds + + def records(self, kind, cursor, limit): + reader_name, convert = self._READERS[kind] + page = getattr(self._session, reader_name)(cursor, limit) + return [convert(r) for r in page.Records], page.Next, bool(page.Truncated) + + def read_memory(self, address, count, machine) -> bytes: + return _bytes(self._session.ReadMemory(address, count, machine)) + + def symbol(self, name, machine) -> int: + return int(self._session.ResolveSymbol(name, machine)) + + def threads(self, machine): + return _as_dict(self._session.Threads(machine)) + + def heap(self, machine): + return _as_dict(self._session.Heap(machine)) + + def close(self) -> None: + self._session.Dispose() + + @staticmethod + def _await(task): + """Wait for an engine task while releasing the GIL: scripted peers run + Python on the emulation thread and need it while the clock is running.""" + import time + + while not task.IsCompleted: + time.sleep(0.0005) + if task.IsFaulted: + raise SimError(str(task.Exception.GetBaseException().Message)) + return task.Result diff --git a/tests/test_install.py b/tests/test_install.py index 782d356..5ef9f05 100644 --- a/tests/test_install.py +++ b/tests/test_install.py @@ -399,3 +399,38 @@ def test_engine_dir_explains_when_no_release_is_reachable(home, monkeypatch): monkeypatch.setattr(install, "install_engine", lambda **kw: (_ for _ in ()).throw(install.InstallError("offline"))) with pytest.raises(engine.EngineNotFound, match="could not fetch the engine: offline"): engine.engine_dir() + + +# -- the Rust engine ---------------------------------------------------------- + +RUST_MANIFEST = { + "version": "0.3.0", + "artifacts": { + "engine-rust-osx-arm64": {"url": "https://releases.example/r.whl", "sha256": None}, + "engine-rust-linux-x64": {"url": "https://releases.example/r.whl", "sha256": None}, + }, +} + + +def test_install_rust_engine_unpacks_the_wheel_under_a_version_dir(home, monkeypatch): + monkeypatch.setattr(install, "fetch_rust_manifest", lambda channel=None: RUST_MANIFEST) + monkeypatch.setattr(install, "download", lambda artifact, timeout=300: engine_zip( + {"simantic_rust.abi3.so": b"\x7fELF", "simantic_rust-0.3.0.dist-info/METADATA": b""})) + monkeypatch.setattr(install, "current_rid", lambda: "osx-arm64") + target = install.install_rust_engine() + assert target == install.rust_engine_root() / "0.3.0" + assert install.is_rust_engine(target) + assert install.installed_rust_engine() == target + + +def test_rust_manifest_is_its_own_product(monkeypatch): + seen = [] + + def capture(request, **k): + seen.append(request.full_url) + raise install.urllib.error.URLError("stop here") + + monkeypatch.setattr(install.urllib.request, "urlopen", capture) + with pytest.raises(install.InstallError): + install.fetch_rust_manifest() + assert seen[0].endswith("/pyrite/latest.json") diff --git a/tests/test_rust_backend.py b/tests/test_rust_backend.py new file mode 100644 index 0000000..603bd7d --- /dev/null +++ b/tests/test_rust_backend.py @@ -0,0 +1,150 @@ +"""The Rust backend through Sim, against a fake engine module. No Rust build +required: what is tested is the Python half — platform rendering, symbol +lookup, the expect loop, record paging, and that unsupported calls say so.""" +import struct +import sys +import types + +import pytest + +from simantic import NotSupported, Sim, _elf, _replx +from simantic.engine import load_rust + + +# -- platform rendering ------------------------------------------------------- + +def test_render_takes_defaults_and_folds_arithmetic(): + text = "cpu: CPU.CortexM\n PerformanceInMips: {{RCC.AHBFreq_Value:84000000 / 1000000 * 1.25}}\n" \ + "nvic: X\n systickFrequency: {{RCC.AHBFreq_Value:84000000}}\n cpuType: \"{{cpu:cortex-m4f}}\"" + out = _replx.render(text) + assert "PerformanceInMips: 105\n" in out + assert "systickFrequency: 84000000\n" in out + assert 'cpuType: "cortex-m4f"' in out + + +def test_render_refuses_code_in_templates(): + assert _replx.render("x: {{a:abc}}") == "x: abc" + assert _replx.render("x: {{a:1+2}}") == "x: 3" + + +# -- ELF symbols -------------------------------------------------------------- + +def _tiny_elf(symbols: dict[str, tuple[int, int]]) -> bytes: + """A minimal ELF32 LE with just .strtab and .symtab; symbols = {name: (value, info)}.""" + strtab = b"\0" + entries = [] + for name, (value, info) in symbols.items(): + idx = len(strtab) + strtab += name.encode() + b"\0" + entries.append(struct.pack(" Date: Sun, 23 Aug 2026 01:17:35 -0400 Subject: [PATCH 2/5] Fix status line for Python 3.11: no backslash inside an f-string expression --- src/simantic/_cli.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/simantic/_cli.py b/src/simantic/_cli.py index 3baebfb..4866b64 100644 --- a/src/simantic/_cli.py +++ b/src/simantic/_cli.py @@ -75,7 +75,8 @@ def _status(args) -> int: engine = install.installed_engine() print(f" engine: {engine if engine else 'not found (fetched on first use, or `simantic install engine`)'}") rust = install.installed_rust_engine() - print(f" engine-rust: {rust if rust else 'not found (fetched on first use of backend=\"rust\", or `simantic install engine-rust`)'}") + missing = 'not found (fetched on first use of backend="rust", or `simantic install engine-rust`)' + print(f" engine-rust: {rust if rust else missing}") print(telemetry.describe()) return 0 From 631e3e9c41c8632180d9803b950ff4dee51c9d1a Mon Sep 17 00:00:00 2001 From: Ahnaf Shahriar Date: Sun, 23 Aug 2026 01:46:54 -0400 Subject: [PATCH 3/5] Review pass against pyrenode3: batch UART, version gate, chained errors From docs/competitors/pyrenode3.md's checklist for our own Python path: - rule 3 (batch across the boundary): the Rust UART path crossed one Python object per byte. It now takes runs of bytes. Measured on the decode side alone, 1 MB of UART is 42.0 ms per-byte against 0.1 ms as runs -- 295x, before the pyo3 object construction it also removes. - their 4.7 (no version contract): check_engine_version refuses an engine older than the Session API this package calls, naming the fix. An unversioned development directory still passes. - their 4.8 (leaky error translation): engine exceptions are chained, so __cause__ keeps the original rather than only our summary. --- src/simantic/_rust.py | 16 +++++++--------- src/simantic/engine.py | 36 ++++++++++++++++++++++++++++++++++++ src/simantic/session.py | 5 ++++- tests/test_rust_backend.py | 27 ++++++++++++++++++++++++++- 4 files changed, 73 insertions(+), 11 deletions(-) diff --git a/src/simantic/_rust.py b/src/simantic/_rust.py index 327f848..0282702 100644 --- a/src/simantic/_rust.py +++ b/src/simantic/_rust.py @@ -42,7 +42,7 @@ def __init__(self, machines: list[dict], *, base: Path, media, services, quantum try: self._s = engine.Session(text, self._elf) except Exception as exc: - raise SimError(str(exc)) from None + raise SimError(str(exc)) from exc self._symbols: dict[str, int] | None = None self._records: dict[str, list[dict]] = {k: [] for k in ("uart", "frames", "logs", "interrupts", "symbol_trace")} self._records["logs"] = [{"t": 0.0, "level": "Warning", "source": "platform", "message": w} @@ -88,13 +88,11 @@ def expect(self, pattern: str, uart: str, machine: str | None, timeout: float) - def _advance(self, seconds: float) -> list[dict]: self._s.run_for(float(seconds)) - fresh: list[dict] = [] - for t, label, byte in self._s.take_uart(): - ch = chr(byte) - if fresh and fresh[-1]["label"] == label: - fresh[-1]["text"] += ch - else: - fresh.append({"t": t, "machine": self.machines[0], "label": label, "text": ch}) + # The engine hands back runs of bytes, not one entry per byte: the + # per-object boundary cost is what dominates a chatty UART. + fresh = [{"t": t, "machine": self.machines[0], "label": label, + "text": bytes(data).decode("latin-1")} + for t, label, data in self._s.take_uart()] self._records["uart"].extend(fresh) return fresh @@ -110,7 +108,7 @@ def read_memory(self, address: int, count: int, machine: str | None) -> bytes: try: return bytes(self._s.read_memory(int(address), int(count))) except Exception as exc: - raise SimError(str(exc)) from None + raise SimError(str(exc)) from exc def symbol(self, name: str, machine: str | None) -> int: if self._symbols is None: diff --git a/src/simantic/engine.py b/src/simantic/engine.py index 59d2b18..41521b1 100644 --- a/src/simantic/engine.py +++ b/src/simantic/engine.py @@ -24,6 +24,41 @@ ENV_DIR = "SIMANTIC_ENGINE_DIR" RUST_ENV_DIR = "SIMANTIC_RUST_ENGINE_DIR" +#: Oldest engine this package can drive. The Session API it calls landed in +#: sim 0.5.4; an older engine fails with a missing-member error deep inside +#: pythonnet, so it is checked here where the message can say what to do. +MIN_ENGINE = (0, 5, 4) + + +class EngineTooOld(RuntimeError): + """The installed engine predates the API this package calls.""" + + +def _version_of(d: Path) -> tuple | None: + """The engine's version, from the directory name a managed install uses.""" + parts = [] + for piece in d.name.split("-")[0].split("."): + if not piece.isdigit(): + return None + parts.append(int(piece)) + return tuple(parts) if len(parts) >= 3 else None + + +def check_engine_version(d: Path) -> None: + """Refuse an engine older than MIN_ENGINE; unknown versions pass. + + A development publish directory has no version in its name — that path is + the developer's own problem, and blocking it would break local work. + """ + found = _version_of(d) + if found is not None and found < MIN_ENGINE: + want = ".".join(str(p) for p in MIN_ENGINE) + have = ".".join(str(p) for p in found) + raise EngineTooOld( + f"engine {have} at {d} is older than {want}, which this package needs. " + f"Run `simantic install engine --force` to fetch the current one." + ) + class EngineNotFound(RuntimeError): """The engine assemblies could not be located or loaded.""" @@ -71,6 +106,7 @@ def engine_dir(explicit: str | os.PathLike[str] | None = None, *, fetch: bool = def load(explicit: str | os.PathLike[str] | None = None): """Host the .NET runtime and import Simantic.Core. Returns the Session namespace.""" d = engine_dir(explicit) + check_engine_version(d) try: from pythonnet import load as load_runtime except ImportError as exc: # pragma: no cover - dependency declared in pyproject diff --git a/src/simantic/session.py b/src/simantic/session.py index d358e7f..1df1d5d 100644 --- a/src/simantic/session.py +++ b/src/simantic/session.py @@ -401,7 +401,10 @@ def __init__(self, machines: list[dict], *, base: Path, work: Path, media, servi try: self._session = ns.Session.Start(spec) except Exception as exc: # .NET exceptions surface as Python exceptions - raise SimError(f"could not start the simulation: {exc}") from None + # Chained, not swallowed: the engine's own exception stays + # reachable as __cause__ so a traceback shows what actually failed + # rather than only this wrapper's summary. + raise SimError(f"could not start the simulation: {exc}") from exc self.machines = list(self._session.Machines) def _add_machine(self, spec, name: str, repl, mcu, overlay, elf) -> None: diff --git a/tests/test_rust_backend.py b/tests/test_rust_backend.py index 603bd7d..f89d17e 100644 --- a/tests/test_rust_backend.py +++ b/tests/test_rust_backend.py @@ -82,7 +82,7 @@ def take_uart(self): out, keep = [], [] for at, data in self._script: if at <= self.t: - out.extend((at, "usart2", b) for b in data) + out.append((at, "usart2", data)) else: keep.append((at, data)) self._script = keep @@ -148,3 +148,28 @@ def test_unsupported_calls_say_so(fake_engine): def test_backend_name_is_validated(): with pytest.raises(ValueError, match="backend"): Sim(elf="fw.elf", repl="a.repl", backend="qemu") + + +# -- the contracts pyrenode3 lacks (docs/competitors/pyrenode3.md §4.7/§4.8) -- + +def test_an_engine_older_than_the_api_is_refused(tmp_path): + from simantic.engine import EngineTooOld, check_engine_version + + check_engine_version(tmp_path / "0.5.4") # exactly the minimum + check_engine_version(tmp_path / "0.6.0") + check_engine_version(tmp_path / "dev-publish") # unversioned: developer's own + with pytest.raises(EngineTooOld, match="0.5.3"): + check_engine_version(tmp_path / "0.5.3") + + +def test_engine_failures_keep_their_cause(fake_engine, monkeypatch): + repl, elf = fake_engine + + def boom(repl_text, elf_bytes): + raise RuntimeError("repl parse error at line 5") + + monkeypatch.setattr(sys.modules["simantic_rust"], "Session", boom) + with pytest.raises(Exception) as exc: + Sim(elf=elf, repl=repl, backend="rust") + assert "repl parse error at line 5" in str(exc.value) + assert isinstance(exc.value.__cause__, RuntimeError) From 27e73ebef868b05c62bcda8589e9b5306efe1c53 Mon Sep 17 00:00:00 2001 From: Ahnaf Shahriar Date: Sun, 23 Aug 2026 02:00:24 -0400 Subject: [PATCH 4/5] Pin the batching contract with tests 705 ns per .NET->CPython crossing measured on this machine, so per-byte traffic is what turns a display frame or a flash write into seconds of pure overhead. Two tests pin what must stay batched: a 10,000-byte burst crosses as runs (<= 4 objects, not 10,000), and observation is pulled in bounded pages. --- tests/test_rust_backend.py | 43 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/tests/test_rust_backend.py b/tests/test_rust_backend.py index f89d17e..db45d4a 100644 --- a/tests/test_rust_backend.py +++ b/tests/test_rust_backend.py @@ -173,3 +173,46 @@ def boom(repl_text, elf_bytes): Sim(elf=elf, repl=repl, backend="rust") assert "repl parse error at line 5" in str(exc.value) assert isinstance(exc.value.__cause__, RuntimeError) + + +# -- the batching contract (docs/competitors/pyrenode3.md §7 rule 3) ---------- + +class CountingSession(FakeSession): + """Reports how many objects crossed the boundary, against bytes delivered.""" + + def __init__(self, repl_text, elf): + super().__init__(repl_text, elf) + self.handed_over = 0 + self._script = [(0.001, b"x" * 10_000)] + + def take_uart(self): + runs = super().take_uart() + self.handed_over += len(runs) + return runs + + +def test_a_burst_crosses_the_boundary_as_runs_not_per_byte(fake_engine, monkeypatch): + """10,000 bytes must not cost 10,000 Python objects. + + At the measured 705 ns per .NET->CPython crossing, per-byte traffic is what + turns a display frame or a flash write into seconds of pure overhead. The + engine hands back runs; this pins that so a future change cannot quietly + regress to one object per byte. + """ + repl, elf = fake_engine + monkeypatch.setattr(sys.modules["simantic_rust"], "Session", CountingSession) + with Sim(elf=elf, repl=repl, uart="usart2", backend="rust") as sim: + sim.run_for(0.002) + text = sim.read_uart(from_start=True) + session = CountingSession.instances[-1] + assert len(text) == 10_000 + assert session.handed_over <= 4, f"{session.handed_over} objects for 10,000 bytes" + + +def test_records_are_paged_not_returned_whole(fake_engine): + """Observation is pulled in bounded pages, so a long run cannot hand the + caller one unbounded list built object by object.""" + repl, elf = fake_engine + with Sim(elf=elf, repl=repl, uart="usart2", backend="rust") as sim: + page, cursor, truncated = sim._b.records("uart", 0, 1) + assert len(page) <= 1 and cursor <= 1 and isinstance(truncated, bool) From f0876184df8ca7761d6201c951de0bf0d81922f7 Mon Sep 17 00:00:00 2001 From: Ahnaf Shahriar Date: Sun, 23 Aug 2026 22:39:48 -0400 Subject: [PATCH 5/5] Add the in-process pytest surface The plugin so far ran every fixture through the `sim` binary: one subprocess, and with it ~3 s of engine start-up, per fixture. That is the right shape for a manifest, which is one coarse pass/fail, but it is the wrong shape for hand-written tests -- many assertions against one running machine. Adds a `sim` fixture that drives the engine in-process, so start-up is paid once per worker rather than once per test, and every machine a test makes is closed when it ends. `--sim-backend=renode|rust|both` picks the engine; `both` runs each test on each and names the engine in the test id. A capability the chosen backend lacks becomes a skip with its reason rather than a failure, so one suite can target both engines and report honestly what each covers. On failure the UART transcript is attached to the report -- it is the useful evidence and it is gone once the session closes. The docstrings carry one measured discipline: on the Renode backend every hand-off costs ~400-800 us, because resuming rendezvouses with the time-source dispatcher threads (MasterTimeSource.cs:127,202, SlaveTimeSource.cs:278). Reading is free; the pause/resume is not. So `expect()` -- one crossing -- is the default verb, and a per-millisecond poll loop runs the same test ~10x slower. On the Rust backend the same hand-off is ~1 us. Converting the test.yaml collector to the in-process path is deliberately left out: manifests do not name a UART, and running the full timeout window rather than exiting at the last match would widen where expect_absent applies. Both need settling against the fixture suite before that path changes. --- README.md | 28 ++++++++ src/simantic/pytest_plugin.py | 132 ++++++++++++++++++++++++++++++++++ tests/test_pytest_surface.py | 60 ++++++++++++++++ 3 files changed, 220 insertions(+) create mode 100644 tests/test_pytest_surface.py diff --git a/README.md b/README.md index 5e1819b..b463e1a 100644 --- a/README.md +++ b/README.md @@ -105,6 +105,34 @@ Tests that cannot run in the current environment skip rather than fail — a missing binary or an unconfigured server. A red run means a simulation ran and disagreed with its expectations. +### The `sim` fixture + +For hand-written tests — many assertions against one running machine — take the +`sim` fixture. It drives the engine **in-process**, so engine start-up is paid +once per worker rather than once per test, and it closes every machine it made +when the test ends. + +```python +def test_timer_irq_fires(sim): + s = sim(elf="fw.elf", mcu="STM32F401RE", uart="usart2") + s.expect("fired=1", timeout=8) + s.expect("RESULT: PASS", timeout=8) +``` + +`--sim-backend=renode|rust|both` picks the engine; `both` runs each test on each +and names the engine in the test id. Anything the chosen backend cannot do +skips with the reason rather than failing, so one suite can target both and +report honestly what each covers. When a test fails, the UART transcript is +attached to the report. + +**Budget your check-ins on the Renode backend.** Every hand-off between Python +and the engine costs ~400–800 µs there, because resuming rendezvouses with +Renode's time-source dispatcher threads — reading is free, it is the +pause/resume that is not. Prefer `expect()`, which crosses once, over a poll +loop that crosses per millisecond: the same test written the chatty way runs +about 10× slower. On the Rust backend the same hand-off is ~1 µs and you can +poll freely. + ## Library The one-shot runner: diff --git a/src/simantic/pytest_plugin.py b/src/simantic/pytest_plugin.py index 447c830..d0a18f7 100644 --- a/src/simantic/pytest_plugin.py +++ b/src/simantic/pytest_plugin.py @@ -154,3 +154,135 @@ def test_boot(firmware): except BinaryNotFound as exc: pytest.skip(str(exc)) return run_firmware + + +# --- the in-process test surface ------------------------------------------- +# +# The fixtures above run a whole manifest through the `sim` binary: one +# subprocess, and with it ~3 s of engine start-up, per fixture. That is the +# right shape for a manifest, which is one coarse pass/fail. +# +# Hand-written tests are the other shape: many assertions against one running +# machine. For those, `sim` below drives the engine *in-process*, so start-up is +# paid once per worker instead of once per test. +# +# One discipline this surface exists to encode (measured; see +# docs/competitors/simantic-py-review-vs-pyrenode3.md §1b): on the Renode +# backend every hand-off between Python and the engine costs ~400-800 us, +# because resuming rendezvouses with Renode's time-source dispatcher threads. +# Reading is free -- it is the pause/resume that is not. So prefer `expect()`, +# which crosses once, over a poll loop that crosses per millisecond. On the +# Rust backend the same hand-off is ~1 us and the discipline does not apply. + +BACKEND_OPTION = "--sim-backend" + + +def pytest_addoption(parser): + group = parser.getgroup("simantic") + group.addoption( + BACKEND_OPTION, + default="renode", + choices=["renode", "rust", "both"], + help="engine the `sim` fixture drives; 'both' runs each test on each.", + ) + + +def pytest_generate_tests(metafunc): + """`both` becomes one test item per backend, so failures name the engine.""" + if "sim_backend" not in metafunc.fixturenames: + return + choice = metafunc.config.getoption(BACKEND_OPTION) + backends = ["renode", "rust"] if choice == "both" else [choice] + metafunc.parametrize("sim_backend", backends, scope="session") + + +@pytest.fixture(scope="session") +def sim_backend(request): + """The engine under test. Parametrized by --sim-backend=both.""" + return request.config.getoption(BACKEND_OPTION) + + +@pytest.fixture(scope="session") +def _sim_engine(sim_backend): + """Load the engine once per worker, before any test is timed. + + Without this the first test in a process absorbs the whole start-up cost + and reads as mysteriously slow; with it, start-up is attributed to the + session where it belongs. Also turns a missing engine into one clear skip + rather than a failure per test. + """ + from . import engine + + try: + engine.load_rust() if sim_backend == "rust" else engine.load() + except engine.EngineNotFound as exc: + pytest.skip(f"no {sim_backend} engine: {exc}") + return sim_backend + + +@pytest.fixture +def sim(request, sim_backend, _sim_engine): + """Factory for an in-process simulation, closed when the test ends. + + def test_timer_fires(sim): + s = sim(elf="fw.elf", mcu="STM32F401RE", uart="usart2") + s.expect("RESULT: PASS", timeout=8) + + Anything the chosen backend cannot do skips rather than fails, so one suite + can run on both engines and report honestly what each covers. On failure the + UART transcript is attached to the report -- what the firmware printed is + almost always the useful evidence, and it is gone once the session closes. + """ + from .session import Sim + from ._rust import NotSupported + + made = [] + + def make(**kwargs): + kwargs.setdefault("backend", sim_backend) + try: + s = Sim(**kwargs) + except NotSupported as exc: + pytest.skip(str(exc)) + made.append(s) + return s + + yield make + + failed = getattr(request.node, "_sim_failed", False) + for s in made: + if failed: + try: + request.node.add_report_section( + "call", f"UART ({s.backend})", s.read_uart(from_start=True) + ) + except Exception: + pass + s.close() + + +@pytest.hookimpl(hookwrapper=True) +def pytest_runtest_call(item): + """A capability the chosen backend lacks is a skip, not a failure. + + Scoped to tests that take the `sim` fixture, so this never reinterprets an + unrelated error. It is what lets one suite run on both engines and report + what each actually covers instead of a wall of red on the narrower one. + """ + if "sim" not in getattr(item, "fixturenames", ()): + yield + return + from ._rust import NotSupported + + outcome = yield + exc = outcome.excinfo + if exc is not None and issubclass(exc[0], NotSupported): + outcome.force_exception(pytest.skip.Exception(str(exc[1]))) + + +@pytest.hookimpl(hookwrapper=True) +def pytest_runtest_makereport(item, call): + """Let the `sim` fixture's teardown know whether the test failed.""" + report = (yield).get_result() + if report.when == "call" and report.failed: + item._sim_failed = True diff --git a/tests/test_pytest_surface.py b/tests/test_pytest_surface.py new file mode 100644 index 0000000..dab90cc --- /dev/null +++ b/tests/test_pytest_surface.py @@ -0,0 +1,60 @@ +"""The pytest surface, exercised against a real engine and real firmware. + +These are the tests a user would write. They run on whichever engine +--sim-backend names (`both` runs each twice), so they double as the evidence +that one suite can target both. +""" +import os +import pytest + +FIX = os.environ.get( + "SIMANTIC_FIXTURES", + "/Users/anoof/dev/simantic/sim-fixtures/tests/stm/nucleo-f401re", +) +MCU, UART = "STM32F401RE", "usart2" + +pytestmark = pytest.mark.skipif( + not os.path.isdir(FIX), reason=f"no fixture tree at {FIX}" +) + + +def f401(sim, name): + return sim(elf=f"{FIX}/{name}/nucleo_f401re.elf", mcu=MCU, uart=UART) + + +def test_expect_reaches_the_end(sim): + """The idiomatic shape: one crossing per assertion.""" + s = f401(sim, "mips-profile") + s.expect("BENCH alu", timeout=30) + s.expect("SUGGEST PerformanceInMips", timeout=30) + assert s.time > 0 + + +def test_transcript_is_readable_after_the_run(sim): + s = f401(sim, "mips-profile") + s.expect("SUGGEST PerformanceInMips", timeout=30) + text = s.read_uart(from_start=True) + assert "BENCH udiv" in text and len(text) > 500 + + +def test_memory_and_time_advance(sim): + s = f401(sim, "mips-profile") + s.run_for(0.005) + first = s.time + s.read_u32(0x20000000) # RAM is readable + s.run_for(0.005) + assert s.time > first + + +def test_backend_is_the_one_requested(sim, sim_backend): + s = f401(sim, "mips-profile") + assert s.backend == sim_backend + + +def test_unsupported_capability_skips_not_fails(sim): + """A multi-machine scenario is Renode-only; on rust this must skip.""" + s = sim(scenario={"machines": { + "a": {"elf": f"{FIX}/mips-profile/nucleo_f401re.elf", "mcu": MCU}, + "b": {"elf": f"{FIX}/mips-profile/nucleo_f401re.elf", "mcu": MCU}, + }}, uart=UART) + assert len(s.machines) == 2