From 1d8d82c9711914203063eb8454e567b41206be2a Mon Sep 17 00:00:00 2001 From: Ahnaf Shahriar Date: Tue, 6 Oct 2026 03:00:16 -0700 Subject: [PATCH 1/2] Rust backend: inject_can Calls the engine's Session.inject_can; an older engine without it raises NotSupported. Docs table updated; radio injection still unsupported. Co-Authored-By: Claude Opus 5.5 --- docs/session-api.md | 3 ++- src/simantic/_rust.py | 8 ++++++-- tests/test_rust_backend.py | 9 +++++++-- 3 files changed, 15 insertions(+), 5 deletions(-) diff --git a/docs/session-api.md b/docs/session-api.md index 4c08e5a..ac4e18e 100644 --- a/docs/session-api.md +++ b/docs/session-api.md @@ -105,7 +105,8 @@ and `uart=` are the defaults. | `threads()` / `heap()` | yes | yes, `Sim` only | | `logs()` | yes | no: always empty | | `inject_gpio()` | yes | yes | -| `inject_can()` / `inject_radio()` | yes | no: raises `NotSupported` | +| `inject_can()` | yes | yes | +| `inject_radio()` | yes | no: raises `NotSupported` | | `parts=` and `networkServices` | yes | no: raises `NotSupported` (a manifest is skipped) | | Several machines joined by `media` | yes | `run_scenario` and `test.yaml` manifests; `Sim` drives one machine | diff --git a/src/simantic/_rust.py b/src/simantic/_rust.py index 88042f8..3a3dfd5 100644 --- a/src/simantic/_rust.py +++ b/src/simantic/_rust.py @@ -223,8 +223,12 @@ def send(self, data: bytes, uart: str, machine: str | None) -> None: 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") + def inject_can(self, peripheral: str, can_id: int, data: bytes, extended: bool, remote: bool, + fd: bool, brs: bool, machine: str | None) -> None: + try: + self._s.inject_can(peripheral, int(can_id), bytes(data), extended, remote, fd, brs) + except AttributeError: + raise NotSupported("CAN injection needs a newer Rust engine") from None def inject_radio(self, *_a, **_k) -> None: raise NotSupported("backend='rust' has no radio injection yet") diff --git a/tests/test_rust_backend.py b/tests/test_rust_backend.py index 8979f1c..c8d751f 100644 --- a/tests/test_rust_backend.py +++ b/tests/test_rust_backend.py @@ -66,7 +66,7 @@ class FakeSession: def __init__(self, repl_text, elf, symbols_elf=None, itm=False): self.repl_text, self.elf, self.t = repl_text, elf, 0.0 self.symbols_elf, self.itm_on, self.pcs, self.watches = symbols_elf, itm, [], [] - self.sent, self.gpio = [], [] + self.sent, self.gpio, self.can = [], [], [] self._script = [(0.002, b"boot\n"), (0.003, b"RESULT: PASS\n")] FakeSession.instances.append(self) @@ -92,6 +92,9 @@ def take_uart(self): def send_uart(self, uart, data): self.sent.append((uart, bytes(data))) + def inject_can(self, p, can_id, data, extended, remote, fd, brs): + self.can.append((p, can_id, data, extended, remote, fd, brs)) + def inject_gpio(self, p, pin, level): self.gpio.append((p, pin, level)) @@ -160,15 +163,17 @@ def test_expect_records_and_symbols_on_rust(fake_engine): assert sim.logs()[0]["message"].startswith("spi1") sim.send("hi", uart="usart2") sim.inject_gpio("gpioc", 13, True) + sim.inject_can("can1", 0x123, b"\x01", extended=True) assert FakeSession.instances[0].sent == [("usart2", b"hi\r")] assert FakeSession.instances[0].gpio == [("gpioc", 13, True)] + assert FakeSession.instances[0].can == [("can1", 0x123, b"\x01", True, False, False, False)] def test_unsupported_calls_say_so(fake_engine): mcu, elf = fake_engine with Sim(elf=elf, mcu=mcu, uart="usart2", backend="rust") as sim: with pytest.raises(NotSupported): - sim.inject_can("can1", 0x123, b"\x01") + sim.inject_radio("radio", b"\x01") with pytest.raises(NotSupported, match="one machine"): Sim(scenario={"machines": {"a": {"elf": str(elf), "mcu": mcu}, "b": {"elf": str(elf), "mcu": mcu}}}, backend="rust") From 97691aa0e5e05db3f59b0a4f7352fa77ea11e419 Mon Sep 17 00:00:00 2001 From: Ahnaf Shahriar Date: Tue, 6 Oct 2026 12:02:56 -0700 Subject: [PATCH 2/2] Rust backend: inject_adc, attach_i2c, attach_spi Calls the engine's set_adc / attach_i2c / attach_spi; an older engine raises NotSupported. Docs table updated. Co-Authored-By: Claude Opus 5.5 --- docs/session-api.md | 5 +++++ src/simantic/_rust.py | 22 ++++++++++++++++++++++ src/simantic/session.py | 33 +++++++++++++++++++++++++++++++++ tests/test_rust_backend.py | 16 +++++++++++++++- 4 files changed, 75 insertions(+), 1 deletion(-) diff --git a/docs/session-api.md b/docs/session-api.md index ac4e18e..72b8b79 100644 --- a/docs/session-api.md +++ b/docs/session-api.md @@ -66,6 +66,9 @@ introspection can resolve — the image that actually runs is still `elf=`. | `send(text)` / `send_bytes(b)` | no — delivered when time next advances | | | `inject_gpio(peripheral, pin, state)` | no | | | `inject_can(peripheral, id, data)` / `inject_radio(peripheral, frame)` | no | | +| `inject_adc(peripheral, channel, volts)` | no | applies from the next conversion | +| `attach_i2c(peripheral, address, device)` | no | `device.write(bytes)`, `device.read(n) -> bytes`, optional `device.stop()` | +| `attach_spi(peripheral, device)` | no | `device(mosi) -> miso`, called per byte; replaces the platform's device | `expect` keeps a pexpect-style stream: output that arrived in a previous call's overshoot is matched first, so a burst of lines can be expected one by @@ -106,6 +109,8 @@ and `uart=` are the defaults. | `logs()` | yes | no: always empty | | `inject_gpio()` | yes | yes | | `inject_can()` | yes | yes | +| `inject_adc()` | no: raises `NotSupported` (use the overlay's `channelMillivolts`) | yes: STM32 F0/F4/G4, RP2040, ESP32-C3/P4 | +| `attach_i2c()` / `attach_spi()` | no: raises `NotSupported` | yes: STM32 I2C (both IP versions) and SPI (all three) | | `inject_radio()` | yes | no: raises `NotSupported` | | `parts=` and `networkServices` | yes | no: raises `NotSupported` (a manifest is skipped) | | Several machines joined by `media` | yes | `run_scenario` and `test.yaml` manifests; `Sim` drives one machine | diff --git a/src/simantic/_rust.py b/src/simantic/_rust.py index 3a3dfd5..6fe0136 100644 --- a/src/simantic/_rust.py +++ b/src/simantic/_rust.py @@ -230,6 +230,28 @@ def inject_can(self, peripheral: str, can_id: int, data: bytes, extended: bool, except AttributeError: raise NotSupported("CAN injection needs a newer Rust engine") from None + def inject_adc(self, peripheral: str, channel: int, volts: float) -> None: + try: + self._s.set_adc(peripheral, channel, volts) + except AttributeError: + raise NotSupported("ADC injection needs a newer Rust engine") from None + except ValueError as e: + raise SimError(str(e)) from None + + def attach_i2c(self, peripheral: str, address: int, device) -> None: + self._attach("attach_i2c", "I2C devices", peripheral, address, device) + + def attach_spi(self, peripheral: str, device) -> None: + self._attach("attach_spi", "SPI devices", peripheral, device) + + def _attach(self, method: str, what: str, *args) -> None: + try: + getattr(self._s, method)(*args) + except AttributeError: + raise NotSupported(f"{what} from Python need a newer Rust engine") from None + except ValueError as e: + raise SimError(str(e)) from None + def inject_radio(self, *_a, **_k) -> None: raise NotSupported("backend='rust' has no radio injection yet") diff --git a/src/simantic/session.py b/src/simantic/session.py index 9310fb0..553763e 100644 --- a/src/simantic/session.py +++ b/src/simantic/session.py @@ -275,6 +275,27 @@ def inject_can(self, peripheral: str, can_id: int, data: bytes, *, extended: boo self._b.inject_can(peripheral, can_id, bytes(data), extended, remote, fd, brs, machine or self.machine) + def inject_adc(self, peripheral: str, channel: int, volts: float) -> None: + """Hold an ADC input channel at `volts` (applies from the next sample).""" + self._b.inject_adc(peripheral, int(channel), float(volts)) + + def attach_i2c(self, peripheral: str, address: int, device: Any) -> None: + """Make a Python object the I2C device at 7-bit `address` on `peripheral`. + + `device` needs `write(data: bytes)` and `read(count: int) -> bytes`, + and may have `stop()`. An exception it raises comes out of the next + `run_for()` / `expect()`. + """ + self._b.attach_i2c(peripheral, int(address), device) + + def attach_spi(self, peripheral: str, device: Any) -> None: + """Make a Python callable `device(mosi: int) -> int` the SPI device on `peripheral`. + + Called once per byte exchanged; it returns the byte put on MISO. A + controller holds one device, so this replaces any the platform attached. + """ + self._b.attach_spi(peripheral, device) + def inject_radio(self, peripheral: str, frame: bytes, machine: str | None = None) -> None: """Deliver a raw radio frame to a radio peripheral.""" self._b.inject_radio(peripheral, bytes(frame), machine or self.machine) @@ -541,6 +562,18 @@ def inject_can(self, peripheral, can_id, data, extended, remote, fd, brs, machin def inject_radio(self, peripheral, frame, machine): self._session.InjectRadio(peripheral, frame, machine) + def inject_adc(self, peripheral, channel, volts): + from ._rust import NotSupported + raise NotSupported("backend='renode' has no ADC injection; put channelMillivolts in the board overlay") + + def attach_i2c(self, peripheral, address, device): + from ._rust import NotSupported + raise NotSupported("backend='renode' takes I2C devices from the platform, not Python objects") + + def attach_spi(self, peripheral, device): + from ._rust import NotSupported + raise NotSupported("backend='renode' takes SPI devices from the platform, not Python objects") + def run_for(self, seconds: float) -> float: return self._await(self._session.RunForAsync(seconds)) diff --git a/tests/test_rust_backend.py b/tests/test_rust_backend.py index c8d751f..1c5538b 100644 --- a/tests/test_rust_backend.py +++ b/tests/test_rust_backend.py @@ -66,7 +66,7 @@ class FakeSession: def __init__(self, repl_text, elf, symbols_elf=None, itm=False): self.repl_text, self.elf, self.t = repl_text, elf, 0.0 self.symbols_elf, self.itm_on, self.pcs, self.watches = symbols_elf, itm, [], [] - self.sent, self.gpio, self.can = [], [], [] + self.sent, self.gpio, self.can, self.attached, self.adc = [], [], [], [], [] self._script = [(0.002, b"boot\n"), (0.003, b"RESULT: PASS\n")] FakeSession.instances.append(self) @@ -92,6 +92,15 @@ def take_uart(self): def send_uart(self, uart, data): self.sent.append((uart, bytes(data))) + def attach_i2c(self, p, address, device): + self.attached.append(("i2c", p, address, device)) + + def attach_spi(self, p, device): + self.attached.append(("spi", p, device)) + + def set_adc(self, p, channel, volts): + self.adc.append((p, channel, volts)) + def inject_can(self, p, can_id, data, extended, remote, fd, brs): self.can.append((p, can_id, data, extended, remote, fd, brs)) @@ -164,8 +173,13 @@ def test_expect_records_and_symbols_on_rust(fake_engine): sim.send("hi", uart="usart2") sim.inject_gpio("gpioc", 13, True) sim.inject_can("can1", 0x123, b"\x01", extended=True) + sim.inject_adc("adc1", 3, 1.0) + sim.attach_i2c("i2c1", 0x48, "dev") + sim.attach_spi("spi1", "dev") assert FakeSession.instances[0].sent == [("usart2", b"hi\r")] assert FakeSession.instances[0].gpio == [("gpioc", 13, True)] + assert FakeSession.instances[0].adc == [("adc1", 3, 1.0)] + assert FakeSession.instances[0].attached == [("i2c", "i2c1", 0x48, "dev"), ("spi", "spi1", "dev")] assert FakeSession.instances[0].can == [("can1", 0x123, b"\x01", True, False, False, False)]