Pick your engine, and drive it from pytest in-process

README.md

@@ -6,19 +6,20 @@ 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
 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/<version>/`, 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/<version>/`, or the Rust
+engine into `~/.simantic/engine-rust/<version>/` — 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
@@ -86,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:

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`.

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",
 ]

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,9 @@ 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()
+    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
 
@@ -98,7 +103,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"
     )

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("<I", elf, 0x20)
+    shentsize, shnum = struct.unpack_from("<HH", elf, 0x2E)
+    sections = [struct.unpack_from("<IIIIIIIIII", elf, shoff + i * shentsize) for i in range(shnum)]
+
+    out: dict[str, int] = {}
+    for sh in sections:
+        _, sh_type, _, _, offset, size, link, _, _, entsize = sh
+        if sh_type != SHT_SYMTAB or entsize == 0:
+            continue
+        strtab_off, strtab_size = sections[link][4], sections[link][5]
+        strtab = elf[strtab_off : strtab_off + strtab_size]
+        for i in range(size // entsize):
+            name_idx, value, _, info = struct.unpack_from("<IIIB", elf, offset + i * entsize)
+            if name_idx == 0:
+                continue
+            end = strtab.index(b"\0", name_idx)
+            name = strtab[name_idx:end].decode("utf-8", "replace")
+            if info & 0xF == STT_FUNC:
+                value &= ~1  # Thumb bit is a call-site convention, not the address
+            out.setdefault(name, value)
+    return out

src/simantic/_replx.py

@@ -0,0 +1,108 @@
+"""Platform text for the Rust backend.
+
+The Renode backend hands `.replx` templates to Simantic.Core, which renders
+them. The Rust engine parses plain `.repl`, so the same rendering happens
+here: every `{{a:b:default}}` placeholder becomes its default, and a default
+that is an arithmetic expression (`84000000 / 1000000 * 1.25`) is evaluated.
+Model names resolve the way `sim --mcu` does — `~/.sim_cache`, else the
+backend with stored credentials, then cached — or from a local model library
+when $SIMANTIC_MCU_LIB is set.
+"""
+
+from __future__ import annotations
+
+import ast
+import json
+import os
+import re
+import urllib.error
+import urllib.parse
+import urllib.request
+from pathlib import Path
+
+from . import auth
+from .fixtures import MCU_LIB_ENV, platform_path
+from .mcu import SimError
+
+MCU_DETAILS_URL = "https://drjdhqfvrttolueolzif.supabase.co/functions/v1/get-mcu-details"
+
+_PLACEHOLDER = re.compile(r"\{\{([^}]*)\}\}")
+_ARITHMETIC = re.compile(r"[0-9. */+()-]+")
+
+
+def _evaluate(expr: str) -> 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)