Peer examples, writing-tests guide, ship examples + py.typed

docs/writing-tests.md

@@ -0,0 +1,95 @@
+# Writing tests against scripted peers
+
+You have firmware and a scenario with scripted peers (a GPS on a UART, an IMU on
+SPI, a barometer on I2C). This guide shows how to turn that into a pytest suite
+that asserts on what the peers saw and what the firmware did with it. The worked
+sample is [`examples/tests/test_peers.py`](../examples/tests/test_peers.py),
+which drives the peers in [`examples/peers/`](../examples/peers/). It is a
+trimmed copy of a suite written for a flight controller: five tests here, eight
+in the original, one run of 12 virtual seconds (about 30 s of wall time there).
+
+Peer and `scenario.yaml` reference (plain paths, on the Simantic site):
+/docs/reference/peers, /docs/reference/scenario-yaml. The session methods used
+below are in [session-api.md](session-api.md).
+
+## Shape of a suite
+
+1. **One simulation per module.** Booting and running firmware is the expensive
+   part; asserting is free. Run once in a `scope="module"` fixture, collect
+   everything you will want (decoded output, bus frames), and let each test
+   look at a different slice.
+2. **Assert on the wire, not on the firmware's own claims.** Count frames per
+   peer (`Sim.frames()`, each record has `label` = the part name, `t`, `data`,
+   `miso`, ...) and decode the firmware's real output (here MAVLink read with
+   `Sim.read_uart_bytes()`, which is byte-accurate; `read_uart()` is not).
+3. **Expected values live in the peer's constants.** The GPS peer reports a 3D
+   fix at a fixed position, the IMU peer tilts the board by `PITCH_TRIM_DEG`.
+   The test asserts the firmware reports what the peer put on the wire, and its
+   comment names the constant. Change a peer, update one number.
+4. **A known firmware bug is a strict `xfail`.** It documents the bug with a
+   reason, keeps the suite green, and turns red the day someone fixes the bug
+   (XPASS strict), prompting you to delete the marker.
+
+## The shared run
+
+```python
+@pytest.fixture(scope="module")
+def run():
+    scenario = yaml.safe_load((PEERS / "scenario.yaml").read_text())
+    scenario["machines"]["dut"]["elf"] = str(Path(elf).resolve())
+    with Sim(scenario=scenario, machine="dut", uart=MAVLINK_UART, cwd=PEERS) as s:
+        s.run_for(RUN_SECONDS)                                   # virtual seconds
+        msgs = list(mavlink_v2(s.read_uart_bytes(from_start=True)))
+        bus = Counter(f["label"] for f in s.frames(from_start=True))
+    return msgs, bus
+```
+
+- `Sim(scenario=...)` takes the same dict shape as `sim --scenario`'s YAML;
+  `machine=` picks the machine whose UART `uart=` names. `cwd=` is the base for
+  relative `elf:` and `script:` paths.
+- Peers (`parts:`) need the default `renode` engine. The pytest plugin's
+  `--sim-backend=renode|rust|both` option chooses the engine for the `sim`
+  fixture, and `simantic.run_scenario` (a one-shot Rust-engine run) rejects
+  `parts:` with `NotSupported`. That is why the sample builds `Sim` itself.
+- The `sim` fixture is function-scoped (a factory that closes each machine when
+  its test ends). For a module-wide run, as here, construct `Sim` in your own
+  module-scoped fixture, as above, and close it with the `with` block.
+- The sample skips when `FIRMWARE` is unset: the firmware is yours to supply.
+
+## The assertions
+
+```python
+def test_gps_fix_reaches_mavlink(run):
+    msgs, _ = run
+    assert last(msgs, GPS_RAW_INT)[28] >= 3        # gps_peer reports a 3D fix
+
+def test_baro_is_read_over_i2c(run):
+    _, bus = run
+    assert bus["baro"] > 1000                      # frames the baro peer saw
+
+@pytest.mark.xfail(strict=True, reason="known firmware bug: ...")
+def test_baro_reaches_scaled_pressure(run):
+    ...
+```
+
+Frames are timestamped in virtual time, so a rate is a subtraction between
+consecutive frames and an ordering is a comparison (see the README's "Test what
+went over the wire"). Prefer `expect()` over polling loops on the Renode engine,
+where each Python/engine hand-off costs a few hundred microseconds.
+
+## Running it
+
+```
+pip install simantic pytest
+FIRMWARE=/path/to/firmware.elf pytest examples/tests
+```
+
+If a test fails, read the peer's own log first: peers report with `ctx.info` /
+`ctx.warning`; pass `show_logs=True` to `Sim`, or read `s.logs()`, to see
+the simulator's log.
+
+## Things this sample does not do
+
+- It does not check MAVLink CRCs (the original does, with a small helper).
+- Peers are plant-free constants: a fixed fix, a stationary board. Closing a
+  loop against a physics model is a different job.

examples/README.md

@@ -0,0 +1,16 @@
+# Examples
+
+| File | What it shows |
+| --- | --- |
+| `step_and_peek.py` | Plain script, no pytest: boot a board, `expect()` a line, read memory, step time. `python step_and_peek.py fw.elf STM32F401RE` |
+| `parallel_sweep.py` | Run the same firmware N times in parallel and collect when each reaches a line. |
+| `peers/gps_peer.py` | u-blox UBX GPS on a UART: ACKs CFG-VALSET, streams NAV-PVT at the rate the firmware programs. |
+| `peers/icm42688_peer.py` | TDK ICM-42688-P IMU on SPI: registers, FIFO, ODR from the firmware's config. |
+| `peers/icp20100_peer.py` | TDK ICP-20100 barometer on I2C: register file and FIFO. |
+| `peers/scenario.yaml` | Wires the three peers to a machine. Needs your own `firmware.elf`; see the header. |
+| `tests/test_peers.py` | pytest suite over those peers: one shared run per module, strict `xfail`. See `docs/writing-tests.md`. |
+
+No firmware is shipped. The peers hold their expected values as constants at
+the top of each file; the tests read those same numbers.
+
+Peer and `scenario.yaml` reference: /docs/reference/peers, /docs/reference/scenario-yaml

examples/peers/gps_peer.py

@@ -0,0 +1,93 @@
+# u-blox M9/M10-class UBX GPS as a scripted UART peer.
+# Protocol: u-blox UBX framing (sync B5 62, class, id, length LE16, payload,
+# 8-bit Fletcher checksum over class..payload). Known-good vector:
+#   CFG-VALSET 06 8A 09 00 00 01 00 00 07 00 91 20 01 -> ckA=53 ckB=48
+# Behaviour: ACK every CFG-VALSET, read the measurement rate and nav ratio the
+# firmware programs (CFG-RATE-MEAS/NAV), then stream a static 3D-fix NAV-PVT at
+# exactly that period. Only the modern CFG-VALSET path is modelled.
+import struct
+
+SYNC = b"\xb5\x62"
+CLASS_NAV, CLASS_ACK, CLASS_CFG = 0x01, 0x05, 0x06
+ID_NAV_PVT, ID_ACK_ACK, ID_CFG_VALSET = 0x07, 0x01, 0x8A
+KEY_RATE_MEAS, KEY_RATE_NAV, KEY_MSGOUT_NAV_PVT_UART1 = 0x30210001, 0x30210002, 0x20910007
+DEFAULT_RATE_MEAS_MS, DEFAULT_RATE_NAV = 1000, 1  # u-blox power-on defaults
+
+LAT_DEG, LON_DEG, ALT_M, NUM_SV = 43.4723, -80.5449, 334.0, 11  # static fix
+
+
+def checksum(body: bytes) -> bytes:
+    a = b = 0
+    for x in body:
+        a = (a + x) & 0xFF
+        b = (b + a) & 0xFF
+    return bytes((a, b))
+
+
+def frame(cls: int, mid: int, payload: bytes = b"") -> bytes:
+    body = bytes((cls, mid)) + struct.pack("<H", len(payload)) + payload
+    return SYNC + body + checksum(body)
+
+
+def nav_pvt(itow_ms: int) -> bytes:
+    p = bytearray(92)
+    struct.pack_into("<I", p, 0, itow_ms & 0xFFFFFFFF)
+    struct.pack_into("<HBBBBB", p, 4, 2026, 9, 15, 12, 0, 0)
+    p[11], p[20], p[21], p[23] = 0x07, 3, 0x01, NUM_SV  # valid, fixType 3D, gnssFixOK, numSV
+    struct.pack_into("<iiii", p, 24, round(LON_DEG * 1e7), round(LAT_DEG * 1e7),
+                     round(ALT_M * 1000), round(ALT_M * 1000))  # lon lat height hMSL
+    return frame(CLASS_NAV, ID_NAV_PVT, bytes(p))
+
+
+class Peer:
+    def __init__(self, ctx):
+        self.ctx, self.rx = ctx, bytearray()
+        self.itow_ms = 0
+        self.rate_meas = self.rate_nav = None
+        self.pvt_on = self.streaming = False
+
+    def on_byte(self, b):
+        self.rx.append(b)
+        while True:
+            while len(self.rx) >= 2 and self.rx[:2] != SYNC:
+                del self.rx[0]
+            if len(self.rx) < 8:
+                return
+            total = 8 + (self.rx[4] | self.rx[5] << 8)
+            if len(self.rx) < total:
+                return
+            raw = bytes(self.rx[:total])
+            del self.rx[:total]
+            self._frame(raw)
+
+    def _frame(self, raw):
+        body = raw[2:-2]
+        if checksum(body) != raw[-2:]:
+            self.ctx.warning(f"bad checksum, dropping {raw.hex()}")
+            return
+        if body[:2] == bytes((CLASS_CFG, ID_CFG_VALSET)):
+            # payload: version layers reserved(2) key(4) value(<=4); body[4:] is the payload
+            key = struct.unpack_from("<I", body, 8)[0]
+            val = int.from_bytes(body[12:], "little")
+            self.ctx.send(frame(CLASS_ACK, ID_ACK_ACK, bytes((CLASS_CFG, ID_CFG_VALSET))))
+            self.ctx.info(f"CFG-VALSET key=0x{key:08x} value={val} -> ACK-ACK")
+            if key == KEY_MSGOUT_NAV_PVT_UART1:
+                self.pvt_on = val == 1
+            elif key == KEY_RATE_MEAS:
+                self.rate_meas = val
+            elif key == KEY_RATE_NAV:
+                self.rate_nav = val
+            self._maybe_stream()
+
+    def _maybe_stream(self):
+        if self.streaming or not self.pvt_on:
+            return
+        self.streaming = True
+        meas = self.rate_meas or DEFAULT_RATE_MEAS_MS
+        nav = self.rate_nav or DEFAULT_RATE_NAV
+        self.ctx.info(f"NAV-PVT every {meas * nav} ms (as programmed by the firmware)")
+        self.ctx.schedule_periodic(meas * nav * 1000)
+
+    def on_timer(self):
+        self.itow_ms += self.rate_meas or DEFAULT_RATE_MEAS_MS
+        self.ctx.send(nav_pvt(self.itow_ms))

examples/peers/icm42688_peer.py

@@ -0,0 +1,131 @@
+# TDK ICM-42688-P IMU as a scripted SPI peer. scenario.yaml lists one
+# `spi-device` part per chip select with this script.
+# Framing: the multiplexer calls on_finish when the part's CS goes high.
+# Beat 0 is the register byte (bit7 = read), beats 1.. are data.
+#
+# Timing: once PWR_MGMT0 turns gyro and accel on, the part produces one
+# accel+gyro sample every 1/ODR, ODR from the GYRO_CONFIG0 code the firmware
+# writes (this peer assumes ACCEL_CONFIG0 carries the same code). In stream-to-FIFO
+# mode each sample becomes a 16-byte packet; FIFO_COUNT (in records, the
+# firmware's INTF_CONFIG0 setting) and FIFO_DATA follow those ticks, and each
+# packet's 16-bit timestamp is the sample time in 1 us units.
+# Source: TDK ICM-42688-P datasheet DS-000347 rev 1.2.
+import math
+import struct
+
+# The board sits stationary, nose up PITCH_TRIM_DEG. Axis signs are the
+# firmware's convention: adjust them until your attitude estimate reads as
+# expected (with the author's firmware: roll 0, pitch -10 deg).
+PITCH_TRIM_DEG = 10.0
+
+ODR_HZ = {1: 32000, 2: 16000, 3: 8000, 4: 4000, 5: 2000, 6: 1000, 7: 200, 8: 100,
+          9: 50, 10: 25, 11: 12.5, 15: 500}  # GYRO_ODR code: DS-000347 5.6 (reset code 6 = 1 kHz)
+STARTUP_US = 30_000     # gyro start-up, enable -> drive ready, typ: DS-000347 Table 1 (accel is 10 ms, Table 2)
+FIFO_PACKETS = (2048 - 16) // 16  # 2 KB FIFO less one packet reserved: DS-000347 6.3
+# MODELED SIMPLIFICATION: a full stream-mode FIFO keeps the newest packets
+# (DS-000347 rev 1.2 does not state which end is dropped).
+
+BANK_SEL, WHO_AM_I, DEVICE_CONFIG, GYRO_DATA_X1 = 0x76, 0x75, 0x11, 0x25
+FIFO_CONFIG, FIFO_COUNTH, FIFO_DATA, SIGNAL_PATH_RESET = 0x16, 0x2E, 0x30, 0x4B
+PWR_MGMT0, GYRO_CONFIG0, ACCEL_CONFIG0 = 0x4E, 0x4F, 0x50
+
+
+class Slave:
+    def __init__(self, ctx):
+        self.ctx = ctx
+        self.beat = 0
+        self.reg = 0
+        self.read = False
+        self.out = b""
+        self._reset()
+
+    def _reset(self):
+        self.bank = 0
+        self.regs = {GYRO_CONFIG0: 0x06, ACCEL_CONFIG0: 0x06}  # bank-0 reset values: DS-000347 section 14
+        self.t_on = None   # time of the first sample; None while sensors are off
+        self.next = 0      # index of the next sample the FIFO hands out
+
+    def _period_us(self):
+        return 1e6 / ODR_HZ.get(self.regs[GYRO_CONFIG0] & 0x0F, 1000)
+
+    def _produced(self):
+        if self.t_on is None or self.ctx.now_us < self.t_on:
+            return 0
+        return int((self.ctx.now_us - self.t_on) // self._period_us()) + 1
+
+    def _fifo_count(self):
+        if not self.regs.get(FIFO_CONFIG, 0) & 0xC0:   # bypass mode: FIFO unused
+            return 0
+        n = self._produced()
+        self.next = max(self.next, n - FIFO_PACKETS)
+        return n - self.next
+
+    def _gyro_raw(self):
+        lsb_dps = 16.4 * (1 << ((self.regs[GYRO_CONFIG0] >> 5) & 7))  # GYRO_FS_SEL, 16.4 LSB/dps at +-2000: DS-000347 Table 1, 5.6
+        return tuple(int(g * lsb_dps) for g in (0.0, 0.0, 0.0))  # stationary
+
+    def _packet(self):
+        if self._fifo_count() == 0:
+            return bytes([0x80]) + bytes(15)   # HEADER_MSG: FIFO empty (DS-000347 6.2)
+        ts = int(self.t_on + self.next * self._period_us())
+        self.next += 1
+        lsb_g = 2048 << ((self.regs[ACCEL_CONFIG0] >> 5) & 3)   # ACCEL_FS_SEL: +-16/8/4/2 g (DS-000347 5.6)
+        theta = math.radians(PITCH_TRIM_DEG)
+        ax, ay, az = math.sin(theta), 0.0, math.cos(theta)  # gravity in the body frame, g
+        accel = (int(lsb_g * ax), int(lsb_g * ay), int(lsb_g * az))
+        gyro = self._gyro_raw()
+        # Packet 3: header accel|gyro|ODR timestamp (0x68), accel, gyro, temp, timestamp (DS-000347 6.1)
+        return bytes([0x68]) + struct.pack(">6h", *accel, *gyro) + b"\x00" + struct.pack(">H", ts & 0xFFFF)
+
+    def on_transmit(self, b):
+        i = self.beat
+        self.beat += 1
+        if i == 0:
+            self.read = bool(b & 0x80)
+            self.reg = b & 0x7F
+            if self.read:
+                self.out = self._read(self.reg)
+            return 0x00
+        if self.read:
+            if self.reg == FIFO_DATA and self.bank == 0:   # packets pop as they are clocked out
+                if (i - 1) % 16 == 0:
+                    self.out = self._packet()
+                return self.out[(i - 1) % 16]
+            return self.out[i - 1] if i - 1 < len(self.out) else 0x00
+        self._write(self.reg + i - 1, b)
+        return 0x00
+
+    def _write(self, reg, v):
+        if reg == BANK_SEL:
+            self.bank = v & 7
+            return
+        if self.bank != 0:
+            return
+        if reg == DEVICE_CONFIG and v & 0x01:          # soft reset
+            self._reset()
+            return
+        if reg == SIGNAL_PATH_RESET and v & 0x02:      # FIFO_FLUSH
+            self.next = self._produced()
+        elif reg == PWR_MGMT0:                         # GYRO_MODE 11 = LN, ACCEL_MODE 1x = LP/LN
+            on = (v >> 2) & 3 == 3 and v & 3 >= 2
+            if not on:
+                self.t_on = None
+            elif self.t_on is None:
+                self.t_on, self.next = self.ctx.now_us + STARTUP_US, 0
+        self.regs[reg] = v
+
+    def _read(self, reg):
+        if reg == BANK_SEL:
+            return bytes([self.bank])
+        if reg == WHO_AM_I:
+            return b"\x47"
+        if reg == GYRO_DATA_X1:
+            return struct.pack(">hhh", *self._gyro_raw())
+        if reg == FIFO_COUNTH:
+            return struct.pack(">H", self._fifo_count())   # records, big endian (INTF_CONFIG0 = 0xF0)
+        if reg == FIFO_DATA:
+            return b""
+        return bytes([self.regs.get(reg, 0x00)]) * 2
+
+    def on_finish(self):
+        self.beat = 0