Rust backend: inject_can

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
@@ -105,7 +108,10 @@ 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_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 |
 

src/simantic/_rust.py

@@ -223,8 +223,34 @@ 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_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")

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))
 

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.attached, self.adc = [], [], [], [], []
         self._script = [(0.002, b"boot\n"), (0.003, b"RESULT: PASS\n")]
         FakeSession.instances.append(self)
 
@@ -92,6 +92,18 @@ 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))
+
     def inject_gpio(self, p, pin, level):
         self.gpio.append((p, pin, level))
 
@@ -160,15 +172,22 @@ 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)
+        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)]
 
 
 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")