diff --git a/src/simantic/_replx.py b/src/simantic/_replx.py deleted file mode 100644 index da59b41..0000000 --- a/src/simantic/_replx.py +++ /dev/null @@ -1,125 +0,0 @@ -"""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. -""" - -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 .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`.""" - cached = cache_dir() / f"{mcu.lower()}.json" - if cached.resolve().parent != cache_dir().resolve(): - raise SimError(f"invalid model name {mcu!r}") - 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: - # The first wall a new user hits, so it says what to do rather than - # what went wrong. The path to ~/.sim_id is a detail they did not ask - # about and cannot act on. - raise SimError( - f"{mcu} needs an account to download its model.\n" - f"\n" - f" simantic auth sign in (opens a browser)\n" - f" simantic status check who you are signed in as" - ) 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: - if exc.code in (401, 403): - # Distinct from "no such model": the name is fine, this account - # cannot have it. Saying "not supported" here sends people off - # hunting for a typo that is not there. - raise SimError( - f"your account does not have access to {mcu}.\n" - f"\n" - f" simantic status check who you are signed in as\n" - f"\n" - f"If that is the wrong account, `simantic auth` again. " - f"Otherwise ask us to enable it." - ) from None - raise SimError( - f"{mcu} is not a model we publish (HTTP {exc.code}). " - f"`simantic status` lists what you can run." - ) 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(*, mcu: str, overlay: Path | None) -> str: - """Rendered `.repl` text for one machine.""" - 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 index ee5fc15..436e4b3 100644 --- a/src/simantic/_rust.py +++ b/src/simantic/_rust.py @@ -14,7 +14,7 @@ from dataclasses import dataclass, field from pathlib import Path -from . import _elf, _replx +from . import _elf from .engine import load_rust from .mcu import SimError @@ -56,6 +56,11 @@ def output(self) -> str: return render_uart(self.uart_records, multi=True) +def _arm(elf: bytes) -> bool: + """Whether the ELF's `e_machine` is ARM (r0-r3 arguments) rather than RISC-V (a0-a3).""" + return int.from_bytes(elf[18:20], "little") == 40 + + def run_scenario(scenario: dict, timeout: float, *, base: str | os.PathLike[str] | None = None, engine_dir=None, trace_symbols: list[str] = (), trace_memory: list[str] = (), trace_interrupts: bool = False) -> ScenarioRun: @@ -81,14 +86,13 @@ def run_scenario(scenario: dict, timeout: float, *, base: str | os.PathLike[str] raise NotSupported("backend='rust' cannot build parts= yet; use backend='renode'") if m.get("gdbPort"): raise NotSupported("backend='rust' does not serve gdbPort through simantic; use `sim --scenario` with gdbPort, or backend='renode'") - overlay = base / m["overlay"] if m.get("overlay") else None - if overlay: - declared |= {f"machine '{name}': {entry}" for entry in re.findall(r"^(\w+)\s*:", overlay.read_text(), re.M)} - machines.append((name, _replx.platform_text(mcu=m["mcu"], overlay=overlay), (base / m["elf"]).read_bytes())) + board = (base / m["overlay"]).read_text() if m.get("overlay") else "" + declared |= {f"machine '{name}': {entry}" for entry in re.findall(r"^(\w+)\s*:", board, re.M)} + machines.append((name, m["mcu"], board, (base / m["elf"]).read_bytes())) # Per machine: its symbols, and whether its registers are r0-r3 or a0-a3. symbols = {name: _elf.symbols((base / (m.get("symbolsElfPath") or m["elf"])).read_bytes()) for name, m in scenario["machines"].items()} if trace_symbols or trace_memory else {} - cortex_m = {name: "CPU.CortexM" in text for name, text, _elf_bytes in machines} + cortex_m = {name: _arm(elf) for name, _mcu, _board, elf in machines} def resolve(where: str) -> list[tuple[str, int]]: if where[:2].lower() == "0x": @@ -174,8 +178,7 @@ def __init__(self, machines: list[dict], *, base: Path, media, services, quantum self._elf = (base / m["elf"]).read_bytes() # Symbols come from the side ELF when the image that boots is stripped. self._symbols_elf = (base / m["symbolsElfPath"]).read_bytes() if m.get("symbolsElfPath") else self._elf - overlay = base / m["overlay"] if m.get("overlay") else None - text = _replx.platform_text(mcu=m["mcu"], overlay=overlay) + board = (base / m["overlay"]).read_text() if m.get("overlay") else "" engine = load_rust(engine_dir) # Only what was asked for is passed, so an engine that predates an # option still runs every script that does not use it. @@ -185,21 +188,21 @@ def __init__(self, machines: list[dict], *, base: Path, media, services, quantum if itm: options["itm"] = True try: - self._s = engine.Session(text, self._elf, **options) + self._s = engine.Session(m["mcu"], self._elf, board=board, **options) except TypeError as exc: raise NotSupported(f"{'/'.join(options)} {_NEWER_ENGINE}") from exc except Exception as exc: raise SimError(str(exc)) from exc # An older engine does not report what it left out. skipped = getattr(self._s, "skipped", list)() - declared = set(re.findall(r"^(\w+)\s*:", overlay.read_text(), re.M)) if overlay else set() + declared = set(re.findall(r"^(\w+)\s*:", board, re.M)) lost = [s for s in skipped if s.split(" ")[0] in declared] if lost: raise NotSupported(f"backend='rust' has no model for {', '.join(lost)}; use backend='renode'") self._symbols: dict[str, int] | None = None self._trace_interrupts, self._interrupts_seen = trace_interrupts, 0 self._itm = itm - self._cortex_m = "CPU.CortexM" in text + self._cortex_m = _arm(self._elf) self._names: dict[int, str] | None = None self._records: dict[str, list[dict]] = {k: [] for k in ( "uart", "frames", "logs", "interrupts", "symbol_trace", "memory_trace", "itm")} diff --git a/tests/test_fixtures.py b/tests/test_fixtures.py index 2757f8c..6c94bf6 100644 --- a/tests/test_fixtures.py +++ b/tests/test_fixtures.py @@ -209,11 +209,3 @@ def test_sim_binary_flags_make_a_manifest_unsupported(tmp_path): load_manifest(tmp_path / "test.yaml") (tmp_path / "test.yaml").write_text("mcu: X\nelf: fw.elf\nsim_args: ['--ascii']\n") assert load_manifest(tmp_path / "test.yaml").mcu == "X" - - -def test_model_name_cannot_escape_the_cache_dir(tmp_path, monkeypatch): - from simantic import _replx - from simantic.session import SimError - monkeypatch.setenv("HOME", str(tmp_path)) - with pytest.raises(SimError, match="invalid model name"): - _replx.model_replx("../../evil") diff --git a/tests/test_model_auth_error.py b/tests/test_model_auth_error.py deleted file mode 100644 index 973fde8..0000000 --- a/tests/test_model_auth_error.py +++ /dev/null @@ -1,18 +0,0 @@ -import pytest - - -def test_unauthenticated_model_error_says_what_to_do(tmp_path, monkeypatch): - """The first wall a new user hits: actions, not a file path.""" - from simantic import _replx - from simantic.mcu import SimError - - monkeypatch.setenv("HOME", str(tmp_path)) - monkeypatch.setenv("SIMANTIC_HOME", str(tmp_path)) - with pytest.raises(SimError) as caught: - _replx.model_replx("ESP32-C3") - - message = str(caught.value) - assert "ESP32-C3 needs an account" in message - assert "simantic auth" in message - # The path to ~/.sim_id is a detail the reader cannot act on. - assert ".sim_id" not in message diff --git a/tests/test_rust_backend.py b/tests/test_rust_backend.py index 47dba14..c20a74c 100644 --- a/tests/test_rust_backend.py +++ b/tests/test_rust_backend.py @@ -1,32 +1,16 @@ """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.""" +required: what is tested is the Python half — 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, SimError, _elf, _replx +from simantic import NotSupported, Sim, SimError, _elf 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: @@ -63,8 +47,8 @@ class FakeSession: """Prints 'boot\\n' then 'RESULT: PASS\\n' on usart2, 1 ms apart, from t=2 ms.""" instances = [] - def __init__(self, repl_text, elf, symbols_elf=None, itm=False): - self.repl_text, self.elf, self.t = repl_text, elf, 0.0 + def __init__(self, mcu, elf, board="", symbols_elf=None, itm=False): + self.mcu, self.board, self.elf, self.t = mcu, board, elf, 0.0 self.symbols_elf, self.itm_on, self.pcs, self.watches = symbols_elf, itm, [], [] self.sent, self.gpio, self.can, self.attached, self.adc = [], [], [], [], [] self._script = [(0.002, b"boot\n"), (0.003, b"RESULT: PASS\n")] @@ -149,7 +133,6 @@ def fake_engine(monkeypatch, tmp_path): load_rust.cache_clear() FakeSession.instances.clear() mcu = "FAKE-MCU" - monkeypatch.setattr(_replx, "model_replx", lambda _mcu: "cpu: CPU.CortexM\n freq: {{F:84000000}}\n") elf = tmp_path / "fw.elf" elf.write_bytes(_tiny_elf({"main": (0x08000495, 0x12), "counter": (0x20000004, 0x11)})) yield mcu, elf @@ -160,7 +143,7 @@ def test_expect_records_and_symbols_on_rust(fake_engine): mcu, elf = fake_engine with Sim(elf=elf, mcu=mcu, uart="usart2", backend="rust") as sim: assert sim.backend == "rust" and sim.machines == ["machine"] - assert "freq: 84000000" in FakeSession.instances[0].repl_text + assert FakeSession.instances[0].mcu == "FAKE-MCU" m = sim.expect("boot") assert m.virtual_seconds == pytest.approx(0.002, abs=1e-9) m = sim.expect(r"RESULT: (PASS|FAIL)") @@ -213,13 +196,13 @@ def test_an_engine_older_than_the_api_is_refused(tmp_path): def test_engine_failures_keep_their_cause(fake_engine, monkeypatch): mcu, elf = fake_engine - def boom(repl_text, elf_bytes): - raise RuntimeError("repl parse error at line 5") + def boom(mcu, elf_bytes, board=""): + raise RuntimeError("board parse error at line 5") monkeypatch.setattr(sys.modules["simantic_rust"], "Session", boom) with pytest.raises(Exception) as exc: Sim(elf=elf, mcu=mcu, backend="rust") - assert "repl parse error at line 5" in str(exc.value) + assert "board parse error at line 5" in str(exc.value) assert isinstance(exc.value.__cause__, RuntimeError) @@ -243,8 +226,8 @@ def test_parts_and_unbuilt_overlay_entries_are_not_supported(fake_engine, monkey class CountingSession(FakeSession): """Reports how many objects crossed the boundary, against bytes delivered.""" - def __init__(self, repl_text, elf): - super().__init__(repl_text, elf) + def __init__(self, mcu, elf, board=""): + super().__init__(mcu, elf, board) self.handed_over = 0 self._script = [(0.001, b"x" * 10_000)] @@ -321,8 +304,8 @@ def test_an_engine_without_the_debug_calls_says_to_upgrade(fake_engine, monkeypa mcu, elf = fake_engine class Old(FakeSession): - def __init__(self, repl_text, elf): - super().__init__(repl_text, elf) + def __init__(self, mcu, elf, board=""): + super().__init__(mcu, elf, board) trace_pc = property() monkeypatch.setattr(sys.modules["simantic_rust"], "Session", Old) @@ -362,8 +345,7 @@ def test_a_scenario_runs_in_one_call_on_rust(fake_engine, monkeypatch, tmp_path) run = run_scenario(scenario, 2.5, base=tmp_path) records, logs = run.uart_records, run.warnings machines, media, nets, timeout, quantum = calls[0] - assert [(n, e) for n, _text, e in machines] == [("nodea", elf.read_bytes()), ("nodeb", elf.read_bytes())] - assert "freq: 84000000" in machines[0][1] + assert machines == [("nodea", mcu, "", elf.read_bytes()), ("nodeb", mcu, "", elf.read_bytes())] assert media == [("can", [("nodea", "can1"), ("nodeb", "can1")])] assert nets == [[("nodea", "gpioa", 3), ("nodeb", "gpiob", 4)]] assert (timeout, quantum) == (3, 0.0001)