diff --git a/packages/embeddedci-mcp/CHANGELOG.md b/packages/embeddedci-mcp/CHANGELOG.md index 99f807c..a42aa42 100644 --- a/packages/embeddedci-mcp/CHANGELOG.md +++ b/packages/embeddedci-mcp/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog — `embeddedci-mcp` +## 2.13.14 + +- Pin directions are on by default: `connect` and `set_wiring` send the wiring profile's DUT + directions to the pod (`la_dir`, firmware 3.10.4+; older firmware is skipped). `wiring` shows + each channel's `dut_dir` and both report `directions`; `status` warns when a driven pin kept + them from going out. Profiles take `dut_dirs` to override a role's or signal's default. +- Needs embeddedci 2.13.14. + ## 2.13.13 - I2C master tools (gateware v55+): `i2c_open`, `i2c_transfer` (a NACK returns `ack: false` and diff --git a/packages/embeddedci-mcp/README.md b/packages/embeddedci-mcp/README.md index 592d55c..16d3696 100644 --- a/packages/embeddedci-mcp/README.md +++ b/packages/embeddedci-mcp/README.md @@ -192,7 +192,7 @@ description, ranges and enums, so an agent sees more than this table. | Tool | Purpose | Parameters | | --- | --- | --- | | `wiring` | Which DUT signal is on which LA channel, plus the power rail, UART baud and SWD target. | none | -| `set_wiring` | Replace the wiring profile for this connection, or store it on embeddedci.com. Also sends the profile's pin directions to the pod (`la_dir`; `directions` in the result says whether it worked). | **`profile`**, `save=false` | +| `set_wiring` | Replace the wiring profile for this connection, or store it on embeddedci.com. Also sends the profile's pin directions to the pod (`la_dir`, firmware 3.10.4+; `directions` in the result says whether it worked). | **`profile`**, `save=false` | ### Power @@ -301,7 +301,7 @@ version the SDK pins (0.1.0, bundled) unless `BENCHPOD_SENSOR_LIBRARY`, the CLI' | `gpio_wait` | Wait until an LA channel reads a level (`reached: false` on timeout). | **`la`**, `level=1`, `timeout=5` | | `gpio_pulse` | FPGA-timed pulses on an LA channel: `width` 4 us to 65.535 ms, `count` up to 65535. | **`la`**, **`width`**, `count=1` | | `gpio_release` | Return channels to high-Z, watched by captures. | `la` | -| `la_dir` | Which way each pin's DUT signal goes (`dut_input`, `dut_output`, `bidir`, `unknown`); on a DUT output the pod refuses push-pull drivers. `from_wiring` sends the wiring profile's directions; no arguments lists every pin. | `la`, `dir`, `from_wiring=false` | +| `la_dir` | Which way each pin's DUT signal goes (`dut_input`, `dut_output`, `bidir`, `unknown`); on a DUT output the pod refuses push-pull drivers. `connect` and `set_wiring` already send the wiring profile's; `from_wiring` sends them again; no arguments lists every pin. | `la`, `dir`, `from_wiring=false` | ### Analog @@ -458,6 +458,14 @@ capture tools' summaries, `la_timing` and `decode_la` instead. registered, server unreachable, another account) the defaults apply and nothing fails; over the LAN or USB it gives up after about 3 seconds. The STM32 pod's USB text console cannot report its registration, so over USB the defaults apply. `BENCHPOD_WIRING_SOURCE=local` skips the lookup. +- **Pin directions on by default.** Every wired role and signal has a DUT direction, shown per + channel in `wiring` (`dut_dir`): `uart_rx` and `spi_miso` are DUT outputs; `uart_tx`, the other + SPI pins and SWCLK DUT inputs; I2C and SWDIO bidirectional; a signal the pod watches (`input`) a + DUT output, one it drives (`output`) a DUT input. The profile's `dut_dirs` overrides any of them. + On firmware 3.10.4+ `connect` and `set_wiring` send them to the pod (`la_dir`), which then + refuses a push-pull driver on a DUT output pin, so the pod never drives against the DUT. Older + firmware is skipped quietly. `wiring` and `set_wiring` report `directions`, and `status` warns + when a driven pin kept them from going out. - **Agent-sized captures.** `capture_adc` returns calibrated statistics, the dominant frequency and a min/max envelope; `capture_la` returns per-channel levels, edges and frequencies. The last captures stay in the session, so `decode_la`, `replay(from_last_capture=true)` and diff --git a/packages/embeddedci-mcp/pyproject.toml b/packages/embeddedci-mcp/pyproject.toml index d0228e1..51380af 100644 --- a/packages/embeddedci-mcp/pyproject.toml +++ b/packages/embeddedci-mcp/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "embeddedci-mcp" -version = "2.13.13" +version = "2.13.14" description = "MCP server that lets AI agents drive an EmbeddedCI BenchPod hardware-in-the-loop bench" readme = "README.md" requires-python = ">=3.10" @@ -26,7 +26,7 @@ classifiers = [ dependencies = [ # cloud: the embeddedci: destination; discovery: connect("discover") over mDNS; # analysis: dominant frequency in capture summaries; yaml: sensor definitions an agent writes. - "embeddedci[cloud,discovery,analysis,yaml]>=2.13.13,<3", + "embeddedci[cloud,discovery,analysis,yaml]>=2.13.14,<3", "mcp>=1.12,<2", "anyio>=4", "pydantic>=2.7", diff --git a/packages/embeddedci-mcp/server.json b/packages/embeddedci-mcp/server.json index 99dda54..b03c849 100644 --- a/packages/embeddedci-mcp/server.json +++ b/packages/embeddedci-mcp/server.json @@ -3,7 +3,7 @@ "name": "io.github.embeddedci-com/embeddedci-mcp", "title": "EmbeddedCI BenchPod", "description": "Drive a BenchPod hardware-in-the-loop bench: power, flash, UART, I2C, CAN, analog and logic.", - "version": "2.13.13", + "version": "2.13.14", "websiteUrl": "https://www.embeddedci.com/docs/benchpod-mcp", "repository": { "url": "https://github.com/embeddedci-com/embeddedci-python", @@ -15,7 +15,7 @@ "registryType": "pypi", "registryBaseUrl": "https://pypi.org", "identifier": "embeddedci-mcp", - "version": "2.13.13", + "version": "2.13.14", "runtimeHint": "uvx", "transport": { "type": "stdio" diff --git a/packages/embeddedci-mcp/src/embeddedci_mcp/guide.py b/packages/embeddedci-mcp/src/embeddedci_mcp/guide.py index 1cb1770..2246937 100644 --- a/packages/embeddedci-mcp/src/embeddedci_mcp/guide.py +++ b/packages/embeddedci-mcp/src/embeddedci_mcp/guide.py @@ -21,7 +21,7 @@ Typical flows: - Flash and boot: flash(file) then power_cycle_and_capture(until_regex) — with a wiring profile that is all; otherwise pass swclk, swdio, target, nreset and rx, tx. - SPI flash (a 25-series chip on four LA pins, wiring spi_sclk/spi_mosi/spi_miso/spi_cs): spi_flash_info to see it answers, then spi_flash_program(file, addr); hold_reset=true keeps the DUT off the bus. spi_transfer talks to any other SPI device. -- I2C master (gateware v55+): set_pull([sda, scl], true), i2c_open, then i2c_transfer(addr, write_hex, read) or i2c_scan; i2c_close. capture_la(send={...}) runs one I2C, SPI or gpio transaction right after the capture arms and triggers on it. la_dir marks DUT outputs so the pod never drives against them (set_wiring sends the profile's directions). +- I2C master (gateware v55+): set_pull([sda, scl], true), i2c_open, then i2c_transfer(addr, write_hex, read) or i2c_scan; i2c_close. capture_la(send={...}) runs one I2C, SPI or gpio transaction right after the capture arms and triggers on it. la_dir marks DUT outputs so the pod never drives against them; connect and set_wiring send the wiring profile's directions (firmware 3.10.4+), so this is on by default. - Interactive console: uart_open, power_on, uart_read(until_regex), uart_write, uart_close. - Emulate sensors (I2C, SPI, UART, analog; firmware with the 'sensors' capability): the pod plays a sensor from a definition. sensor_library lists ready ones (bme280, bmp280, sht4x, mpu6050, gps_nmea, ...) and sensor_show gives one's YAML and params. For I2C, set_pull([sda, scl], true) on LA1-LA6 (LA7/LA8 pull DOWN) first. sensor_start(definition=, pins={sda, scl} | {tx} | ...), then sensor_set(instance, values) changes readings, sensor_stimulus_load + sensor_stimulus_start play values over time, sensor_fault injects a fault, sensor_status shows bus activity, sensor_capture waits for the DUT's next transaction on an I2C sensor's bus and decodes it, sensor_stop ends it. A part the library lacks: read sensor_format, write a definition from the datasheet, check it with sensor_validate, then sensor_start it. A GPS module is gps_nmea on pins={tx: }. - Analog: dac_output (DC), generate (sine/square/sawtooth) and replay drive the DAC and route its path; adc_read gives one calibrated value, capture_adc a waveform summary; dac_stop ends any DAC output. diff --git a/packages/embeddedci-mcp/src/embeddedci_mcp/models.py b/packages/embeddedci-mcp/src/embeddedci_mcp/models.py index 116200a..3ee94f2 100644 --- a/packages/embeddedci-mcp/src/embeddedci_mcp/models.py +++ b/packages/embeddedci-mcp/src/embeddedci_mcp/models.py @@ -150,6 +150,10 @@ class WiringPin(BaseModel): wired_to: Optional[str] = Field(None, description=( "The role (uart_rx, i2c_sda, swd_swclk, …) or signal name on this channel; null = unused.")) pull: Optional[str] = Field(None, description="Fixed bias resistor, e.g. '4.7k up'; null on LA9-LA14.") + dut_dir: Optional[str] = Field(None, description=( + "Which way the DUT's signal on this channel goes (dut_input, dut_output or bidir): the " + "role's or signal's default, or the profile's dut_dirs. The pod gets it with la_dir and " + "refuses push-pull drivers on dut_output pins. Null when nothing is wired.")) class WiringResult(BaseModel): @@ -173,7 +177,8 @@ class WiringResult(BaseModel): profile: Dict[str, Any] = Field(default_factory=dict, description="The profile as its JSON object.") saved: bool = Field(False, description="set_wiring stored this profile on embeddedci.com.") directions: Optional[str] = Field(None, description=( - "set_wiring: 'sent' when the pod now has the profile's pin directions (la_dir), else why not.")) + "'sent' when the pod has the profile's pin directions (la_dir: sent on connect and by " + "set_wiring), else 'not sent: ' (older firmware, or a DUT output pin driven right now).")) # -- LA pin ownership + GPIO ------------------------------------------------------- diff --git a/packages/embeddedci-mcp/src/embeddedci_mcp/server.py b/packages/embeddedci-mcp/src/embeddedci_mcp/server.py index b7874e9..0a2f9d7 100644 --- a/packages/embeddedci-mcp/src/embeddedci_mcp/server.py +++ b/packages/embeddedci-mcp/src/embeddedci_mcp/server.py @@ -348,6 +348,10 @@ def _status() -> m.StatusResult: wiring_source: Optional[str] = pod.wiring.source except (BenchPodError, ValueError, OSError): wiring_source = None + dirs = getattr(pod, "pin_directions_status", "") + if isinstance(dirs, str) and dirs.startswith("not sent") and "firmware" not in dirs: + warnings.append(f"the wiring's pin directions were {dirs} (call la_dir(from_wiring=true) " + "once that pin is released)") if la is None: warnings.append("LA I/O voltage not selected — call set_la_voltage (1.8 or 3.3, matching " "the DUT) before flash, UART, LA capture, pulls or sensor emulation") @@ -424,7 +428,13 @@ async def wiring() -> m.WiringResult: plus warnings about risky wiring. Every tool whose channel, baud, rail or SWD arguments are omitted takes them from this profile, and channel arguments accept these names. """ - return await _call(lambda: wiring_summary(SESSION.require().wiring)) + def op() -> m.WiringResult: + pod = SESSION.require() + result = wiring_summary(pod.wiring) + result.directions = pod.pin_directions_status + return result + + return await _call(op) @mcp.tool(annotations=_ann("Set the wiring profile", idempotent=True, cloud=True)) @@ -449,23 +459,13 @@ def op() -> m.WiringResult: if save: pod.save_wiring() result = wiring_summary(pod.wiring, saved=save) - result.directions = _push_directions(pod) + # Assigning the profile sent its pin directions (la_dir) on firmware that has it. + result.directions = pod.pin_directions_status return result return await _call(op) -def _push_directions(pod: Any) -> str: - """Send the profile's pin directions (la_dir); 'sent', or why not (older firmware, a driven pin).""" - try: - pod.apply_pin_directions() - return "sent" - except _errors.UnsupportedFeatureError: - return "not sent: this pod's firmware has no la_dir" - except BenchPodError as exc: - return f"not sent: {_error_text(exc)}" - - # -- power ---------------------------------------------------------------------------- def _switch_power(efuse: Optional[int], on: bool, delay: Optional[float]) -> m.PowerResult: @@ -1349,12 +1349,13 @@ async def la_dir( dir: Annotated[Optional[Literal["dut_input", "dut_output", "bidir", "unknown"]], Field(description=( "Which way the DUT signal goes. On dut_output the pod refuses push-pull drivers."))] = None, from_wiring: Annotated[bool, Field(description=( - "Send the wiring profile's directions instead (uart_rx, spi_miso and input signals are " - "DUT outputs)."))] = False, + "Send the wiring profile's directions again (each role's and signal's default, or the " + "profile's dut_dirs)."))] = False, ) -> m.GpioPinsResult: """Tell the pod which way each pin's DUT signal goes, so it never drives against a DUT output. - The pod forgets this on a reboot; set_wiring sends the profile's directions. + connect and set_wiring already send the wiring profile's directions (firmware 3.10.4+); + use this for a pin the profile doesn't name. The pod forgets them on a reboot. """ def op() -> m.GpioPinsResult: pod = SESSION.require() diff --git a/packages/embeddedci-mcp/src/embeddedci_mcp/summaries.py b/packages/embeddedci-mcp/src/embeddedci_mcp/summaries.py index 439c755..b30972a 100644 --- a/packages/embeddedci-mcp/src/embeddedci_mcp/summaries.py +++ b/packages/embeddedci-mcp/src/embeddedci_mcp/summaries.py @@ -120,11 +120,13 @@ def _bias(la: int) -> Optional[str]: def wiring_summary(wiring: Wiring, *, saved: bool = False) -> m.WiringResult: """The effective profile, its 14-row pin table and anything risky about it.""" pins = wiring.pins() + dirs = wiring.pin_dut_dirs() return m.WiringResult( source=wiring.source, version=wiring.version, la_voltage=wiring.la_voltage, efuse=wiring.efuse, uart_baud=wiring.uart_baud, i2c_address=wiring.i2c_address, swd_nreset=wiring.swd_nreset, swd_target=wiring.swd_target, - pins=[m.WiringPin(la=la, wired_to=pins.get(la), pull=_bias(la)) for la in range(1, 15)], + pins=[m.WiringPin(la=la, wired_to=pins.get(la), pull=_bias(la), dut_dir=dirs.get(la)) + for la in range(1, 15)], signals=[m.WiringSignal(name=s.name, la=s.la, direction=s.direction, active_low=s.active_low, description=s.description) for s in wiring.signals], diff --git a/packages/embeddedci-mcp/tests/conftest.py b/packages/embeddedci-mcp/tests/conftest.py index c6a5996..c1f993d 100644 --- a/packages/embeddedci-mcp/tests/conftest.py +++ b/packages/embeddedci-mcp/tests/conftest.py @@ -117,7 +117,7 @@ def status(self) -> Any: caps = list(self.caps) if self.image is not None: caps.append("dac_control_loop" if self.image == 0 else "dac_deep_replay") - return {"version": "2.0.0", "board": "stm32h563", "adc_bits": 16, "adc_fullscale_mv": 4096, + return {"version": getattr(self, "version", "2.0.0"), "board": "stm32h563", "adc_bits": 16, "adc_fullscale_mv": 4096, "caps": caps, "gateware": 55} def ping(self) -> Any: diff --git a/packages/embeddedci-mcp/tests/test_bus_master_tools.py b/packages/embeddedci-mcp/tests/test_bus_master_tools.py index a071969..3a7b1ca 100644 --- a/packages/embeddedci-mcp/tests/test_bus_master_tools.py +++ b/packages/embeddedci-mcp/tests/test_bus_master_tools.py @@ -59,9 +59,27 @@ def test_la_dir_sets_lists_and_follows_the_wiring(connected): def test_set_wiring_sends_the_directions(connected): - r = call("set_wiring", profile={"uart_rx": 7, "uart_tx": 8}) + from embeddedci_mcp.session import SESSION + + connected.version = "3.10.4" + SESSION.require().refresh_capabilities() + r = call("set_wiring", profile={"uart_rx": 7, "uart_tx": 8, + "signals": [{"name": "READY", "la": 9}], + "dut_dirs": {"READY": "bidir"}}) assert r["directions"] == "sent" assert connected.dirs[7] == "dut_output" and connected.dirs[8] == "dut_input" + assert connected.dirs[9] == "bidir" and connected.dirs[5] == "unknown" + pins = {p["la"]: p for p in call("wiring")["pins"]} + assert pins[7]["dut_dir"] == "dut_output" and pins[9]["dut_dir"] == "bidir" + assert pins[3]["dut_dir"] is None + + +def test_older_firmware_skips_the_directions(connected): + r = call("set_wiring", profile={"uart_rx": 7, "uart_tx": 8}) + assert r["directions"] == "not sent: this pod's firmware (2.0.0) has no la_dir (3.10.4+)" + assert 7 not in connected.dirs + # Older firmware is not a problem to warn about. + assert not any("pin directions" in w for w in call("status")["warnings"]) def test_spi_modes_1_and_2(connected): diff --git a/packages/embeddedci-mcp/tests/tools_surface.json b/packages/embeddedci-mcp/tests/tools_surface.json index 75a17d9..1b08b8c 100644 --- a/packages/embeddedci-mcp/tests/tools_surface.json +++ b/packages/embeddedci-mcp/tests/tools_surface.json @@ -2551,7 +2551,7 @@ }, "from_wiring": { "default": false, - "description": "Send the wiring profile's directions instead (uart_rx, spi_miso and input signals are DUT outputs).", + "description": "Send the wiring profile's directions again (each role's and signal's default, or the profile's dut_dirs).", "title": "From Wiring", "type": "boolean" }, diff --git a/packages/embeddedci/CHANGELOG.md b/packages/embeddedci/CHANGELOG.md index 3891af8..b680c24 100644 --- a/packages/embeddedci/CHANGELOG.md +++ b/packages/embeddedci/CHANGELOG.md @@ -1,5 +1,19 @@ # Changelog — `embeddedci` +## 2.13.14 + +- Pin directions are on by default. On a pod with firmware 3.10.4+ the SDK sends the wiring + profile's directions with `la_dir` when it connects and whenever `bp.wiring` changes, and marks + the pins the profile doesn't name `unknown`; older firmware is skipped quietly. The connection + never fails over it: `bp.pin_directions_status` is `"sent"` or `"not sent: "`. +- Every wired role and signal has a DUT direction. New defaults: `swd_swclk` is a DUT input, + `swd_swdio` is `bidir` (SWD pins were left out before). The profile's new `dut_dirs` key + overrides any role or signal (`{"uart_rx": "bidir", "READY": "dut_input"}`). Old profiles stay + valid. `Wiring.dut_dir(name)`, `Wiring.pin_dut_dirs()`; `describe()` shows the direction. +- `apply_pin_directions()` sends DUT outputs last, so a driven pin refusing its mark leaves the + other directions in place. +- `Capabilities.la_dir`: the firmware takes pin directions (3.10.4+). + ## 2.13.13 - The pod as a bus master (gateware v55+). `bp.open_i2c(sda, scl, *, hz=100_000, pullups="pod")` diff --git a/packages/embeddedci/README.md b/packages/embeddedci/README.md index e09a465..59192fb 100644 --- a/packages/embeddedci/README.md +++ b/packages/embeddedci/README.md @@ -511,7 +511,7 @@ with bp.open_uart() as uart: # rx=3, tx=4, 115200 baud from th bp.flash(file="app.elf") # swclk/swdio/nreset/target from the profile bp.sensors.start("bmp280") # sda/scl from the profile la = bp.capture_la(100_000, trigger=Trigger("READY")) -print(wiring.describe()) # a pin table with each channel's bias resistor +print(wiring.describe()) # a pin table: bias resistor and DUT direction per channel ``` | Key | Default | Meaning | @@ -523,6 +523,7 @@ print(wiring.describe()) # a pin table with each channel' | `swd_swclk`, `swd_swdio`, `swd_nreset`, `swd_target` | `11`, `12`, `False`, `""` | flashing | | `spi_sclk`, `spi_mosi`, `spi_miso`, `spi_cs` | not wired | decoder channels | | `signals` | `[]` | named signals: `Signal(name, la, direction="input"/"output"/"open_drain"/"bidir", active_low=False, description="")` | +| `dut_dirs` | `{}` | the DUT direction of a role or signal where its default is wrong: `{"uart_rx": "bidir", "READY": "dut_input"}` (role key or signal name to `dut_input`, `dut_output` or `bidir`); see [pin directions](#gpio-on-the-la-pins) | A role set to `None` is not wired. Construction validates the whole profile and raises `ValueError` listing every problem — including two roles or signals on one LA channel; `wiring.warnings()` lists @@ -922,15 +923,27 @@ try: except DirectionConflictError: # use "open_drain", or fix the wiring bp.gpio(9, "open_drain") -bp.apply_pin_directions() # the wiring profile's directions, in one go +print(bp.pin_directions_status) # "sent": the wiring's directions went out on connect print(bp.pin_directions()) # {5: 'dut_output', 4: 'dut_input', 1: 'bidir', ...} +bp.apply_pin_directions() # send them again (raises on a conflict) print(bp.la_dir()) # every pin's LaPinState, dir included ``` -`pin_directions()` reads them from the wiring profile: `uart_rx` and `spi_miso` are DUT outputs, -`uart_tx`, `spi_sclk`, `spi_mosi` and `spi_cs` DUT inputs, the I2C pins `bidir`; a signal with -direction `input` (the pod watches it) is a DUT output, `output` a DUT input, `open_drain` and -`bidir` are `bidir`. The pod forgets the directions on a reboot, so send them after connecting. +**The wiring profile's directions are on by default.** On a pod with firmware 3.10.4 or later (`bp.capabilities.la_dir`) +the SDK sends them with `la_dir` when it connects and whenever `bp.wiring` changes, and marks the pins +the profile doesn't name `unknown`. On older firmware it skips this quietly. +`bp.pin_directions_status` says how it went: `"sent"`, or `"not sent: "` (older firmware, or +a pin the profile calls a DUT output that is driven right now; the other directions still go out). +The connection never fails over it. The pod forgets the directions on a reboot; the next connect +sends them again. + +Every wired role and signal has a direction. The defaults: `uart_rx` and `spi_miso` are DUT +outputs (the DUT drives them); `uart_tx`, `spi_sclk`, `spi_mosi`, `spi_cs` and `swd_swclk` DUT +inputs; the I2C pins and `swd_swdio` `bidir`; a signal with direction `input` (the pod watches it) +is a DUT output, `output` a DUT input, `open_drain` and `bidir` are `bidir`. The profile's +`dut_dirs` overrides any of them (`wiring.dut_dir(name)` is one role's or signal's direction, +`wiring.pin_dut_dirs()` the per-channel map). Note the point of view: `la_dir` and `dut_dirs` +describe the DUT's signal, while a signal's `direction` is seen from the pod. `LaPinState.dir` is a pin's direction (`None` when unknown), and `LaPinState.warn` holds warning bits that `LaPinState.warnings` spells out: a pull-up under a push-pull output, and a pin driven high or pulled up while the target power is off (it powers an unpowered DUT through its I/O pins). @@ -1944,7 +1957,7 @@ lease_wait=600.0, lease_ttl=120, wiring=None)`. A context manager; `close()` is | I2C master | `open_i2c(sda=None, scl=None, *, hz=100_000, pullups="pod")` (see [I2C master](#i2c-master)) | | UART | `capture_uart(...)`, `power_cycle_and_capture(...)`, `open_uart(...)` | | Bias resistors | `enable_pullup(*las)`, `disable_pullup(*las)`, `enable_pulldown(*las)`, `disable_pulldown(*las)` (LA7/LA8), `set_pull(la, enabled)`, `pull_state(la)`, `enabled_pulls()` | -| GPIO and pins | `gpio(la, mode="output", *, level=None)`, `gpio_pins(las, mode="output", *, level=None)`, `configure_gpio(las, mode="output", *, level=None)`, `set_gpio(la, level)`, `read_gpio(la)`, `wait_for_level(la, level, *, timeout, poll=0.005)` (`False` when `timeout` passes first), `pin_levels()`, `la_pins()`, `la_dir(la=None, dir=None)`, `pin_directions(wiring=None)`, `apply_pin_directions(wiring=None)`, `release_gpio(*las)`, `la_step(la, *, steps, delay, dir_la=None, direction=0)` | +| GPIO and pins | `gpio(la, mode="output", *, level=None)`, `gpio_pins(las, mode="output", *, level=None)`, `configure_gpio(las, mode="output", *, level=None)`, `set_gpio(la, level)`, `read_gpio(la)`, `wait_for_level(la, level, *, timeout, poll=0.005)` (`False` when `timeout` passes first), `pin_levels()`, `la_pins()`, `la_dir(la=None, dir=None)`, `pin_directions(wiring=None)`, `apply_pin_directions(wiring=None)`, `pin_directions_status`, `release_gpio(*las)`, `la_step(la, *, steps, delay, dir_la=None, direction=0)` | | Emulated sensors | `sensors` (a `SensorEmulation`: `start(definition, *, instance=None, pins=None, address=None, baud=None, values=None)`, `define()`, `validate()`, `set()`, `fault()`, `clear_fault()`, `load_stimulus()`, `start_stimulus()`, `stop_stimulus()`, `stop()`, `status()`, `regs()`, `capture()`, `i2c_transactions()`, `library()`); see [Emulated sensors](#emulated-sensors) | | Analog | `analog_path(path, *, range=None)`, `adc_read(source="ext", *, range=None)`, `adc_range(range=None)`, `adc_range_test()`, `dac_output(path, *, volts=None)`, `current_out(current)`, `current_out_range()`, `calibrate(source="current_in")`, `calibration()`, `clear_calibration(part=None)`, `calibrate_dac(path, points)`, `calibrate_adc_gain(source, *, range="normal")`, `DMM_SET_POINTS` | | Captures | `capture_adc(...)`, `capture_la(..., send=None)`, `capture_correlated(...)`, `decode(source, protocol="i2c", ...)` | @@ -2023,7 +2036,7 @@ few built from other shapes); `to_dict()` goes the other way where a type is sen | `MotorEmulator` (class) | `open_motor_emulator` | `configure()`, `boards()`, `probe()`, `read()`, `read_signed(reg)`, `write()`, `status()`, `trips()`, `clear_trips()`, `arm()`, `set_control()`, `set_pwm()`, `set_duties(a, b, c)`, `set_shape(shape)`, `set_protection()`, `set_battery()`, `battery_state()`, `set_pv_calibration(offset, gain)`, `sample()`, `time()`, `start_log()`, `stop_log()`, `read_log()`, `eeprom_read(addr, length)`, `eeprom_write(addr, data)`, `close()` | | `BatteryModel` | `MotorEmulator.set_battery` | `capacity_ah`, `ocv`, `r0_ohm`, `r1_ohm`, `tau_s`, `soc`, `v_min`, `v_max`, `invert_current` | | `EmulatorCalibration` | `open_motor_emulator(calibration=...)` | `codes_per_amp`, `codes_per_volt`, `offsets`, `sample_rate_hz`, `amps()`, `volts()`, `current_code()`, `bus_code()`, `resistance_code()` | -| `Wiring` | `wiring`, `save_wiring` | see [Wiring profile](#wiring-profile); also `defaults()`, `load(path)`, `coerce(value)` (a `Wiring`, dict or file path), `from_dict(data, *, strict=True)` (`strict=False` keeps unknown keys in `extra`), `to_dict()`, `with_changes(**changes)`, `assignments()`, `pins()`, `la(name)`, `signal(name)`, `describe()`, `warnings()`, `la_voltage`, `i2c_address` | +| `Wiring` | `wiring`, `save_wiring` | see [Wiring profile](#wiring-profile); also `defaults()`, `load(path)`, `coerce(value)` (a `Wiring`, dict or file path), `from_dict(data, *, strict=True)` (`strict=False` keeps unknown keys in `extra`), `to_dict()`, `with_changes(**changes)`, `assignments()`, `pins()`, `la(name)`, `signal(name)`, `dut_dir(name)`, `pin_dut_dirs()`, `describe()`, `warnings()`, `la_voltage`, `i2c_address`, `dut_dirs` | | `Signal` | `Wiring.signals` | `name`, `la`, `direction`, `active_low`, `description`, `to_dict()` | | `Waveform` | `WaveformLibrary` | `id`, `name`, `kind`, `sample_count`, `sample_rate_hz`, `bits`, `full_scale_v`, `recording_size_bytes`, `created_at`, `dac_path`, `segments`, `unit`, `is_recording` | | `WaveformLibrary` (class) | `waveforms` | `list()`, `get()`, `find()`, `save_recording()`, `save_waveform()`, `save_segments()`, `rename()`, `delete()`, `preview()`, `download_recording()`, `samples_b64(waveform_id)` | diff --git a/packages/embeddedci/pyproject.toml b/packages/embeddedci/pyproject.toml index cabc198..03bb293 100644 --- a/packages/embeddedci/pyproject.toml +++ b/packages/embeddedci/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "embeddedci" -version = "2.13.13" +version = "2.13.14" description = "BenchPod SDK and pytest plugin for EmbeddedCI hardware-in-the-loop testing" readme = "README.md" requires-python = ">=3.10" diff --git a/packages/embeddedci/src/embeddedci/benchpod/capabilities.py b/packages/embeddedci/src/embeddedci/benchpod/capabilities.py index 863bba8..0ddff47 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/capabilities.py +++ b/packages/embeddedci/src/embeddedci/benchpod/capabilities.py @@ -21,8 +21,9 @@ from __future__ import annotations +import re from dataclasses import dataclass, field -from typing import Any, Dict, Mapping, Optional +from typing import Any, Dict, Mapping, Optional, Tuple # Built-in nominal front-end calibration per board — the affine fit ``volts = a + b*count`` # for boards whose analog front-end is inverting (so the naive count/full-scale model would @@ -85,6 +86,14 @@ def _as_bool(m: Mapping[str, Any], *keys: str) -> Optional[bool]: #: server as ``cap.``), command_handler.c handle_status / cloud_client.c. #: The first gateware with the I2C master, SPI modes 1 and 2 and send-and-capture. BUS_MASTER_GATEWARE = 55 +#: The first firmware with ``la_dir`` (pin directions). +LA_DIR_FIRMWARE = (3, 10, 4) + + +def _version_tuple(version: str) -> Optional[Tuple[int, int, int]]: + """``"3.10.4"`` / ``"v3.10.4-dev"`` as ``(3, 10, 4)``; None when it is not a version.""" + m = re.match(r"^v?(\d+)\.(\d+)\.(\d+)", (version or "").strip()) + return (int(m.group(1)), int(m.group(2)), int(m.group(3))) if m else None _STATUS_FLAGS = ("ota_sig", "sig_policy_cmd", "lan_policy_cmd", "tunnel_max_tier", "lease_state", "cloud_ca", "cloud_proxy", "blob_slots", "can_fd") @@ -499,6 +508,13 @@ def boot_warning(self) -> Optional[str]: # -- scaling ------------------------------------------------------------ + @property + def la_dir(self) -> bool: + """The pod takes pin directions (``la_dir``, firmware 3.10.4+). False when the firmware + version is unknown.""" + v = _version_tuple(self.firmware_version) + return v is not None and v >= LA_DIR_FIRMWARE + @property def bus_master(self) -> bool: """The I2C master, SPI modes 1 and 2 and send-and-capture (gateware v55+).""" diff --git a/packages/embeddedci/src/embeddedci/benchpod/client.py b/packages/embeddedci/src/embeddedci/benchpod/client.py index 327d95e..f184c3e 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/client.py +++ b/packages/embeddedci/src/embeddedci/benchpod/client.py @@ -152,13 +152,6 @@ _log = logging.getLogger("embeddedci.benchpod") -#: The DUT direction of each wiring role (:meth:`BenchPod.pin_directions`). -_ROLE_DIRS = {"uart_rx": "dut_output", "uart_tx": "dut_input", "i2c_sda": "bidir", "i2c_scl": "bidir", - "spi_sclk": "dut_input", "spi_mosi": "dut_input", "spi_miso": "dut_output", - "spi_cs": "dut_input"} -#: A wiring signal's direction (seen from the pod) as the DUT's (:meth:`BenchPod.pin_directions`). -_SIGNAL_DIRS = {"input": "dut_output", "output": "dut_input", "open_drain": "bidir", "bidir": "bidir"} - #: Capability flags only the analog board announces: the digital-only board lacks them whatever its #: firmware (cloud_caps_hw.c). _ANALOG_FLAGS = frozenset({"scope", "dac_replay", "dac_deep_replay", "dac_control_loop", "dac_cotrig", @@ -306,6 +299,9 @@ def __init__( self._waveforms: Optional["WaveformLibrary"] = None self._sensors: Optional["SensorEmulation"] = None self._device_name = "" + #: What happened to the wiring profile's pin directions the last time the SDK sent them on + #: its own (on connect, on a new :attr:`wiring`): ``"sent"`` or ``"not sent: "``. + self.pin_directions_status = "not sent: not connected yet" #: Hardware controls below the named-path API (raw mux/relay/DAC-code access). Not covered #: by the stability guarantee. self.lowlevel = LowLevel(self) @@ -347,6 +343,7 @@ def __init__( raise ValueError(f"{LA_VOLTAGE_ENV} must be 1.8 or 3.3, got {env!r}") from None if la_voltage is not None: self.set_la_voltage(la_voltage) + self._auto_pin_directions() except BaseException: self.close() raise @@ -529,6 +526,7 @@ def wiring(self) -> Wiring: @wiring.setter def wiring(self, value: Union[Wiring, Mapping[str, Any], str, "os.PathLike[str]"]) -> None: self._wiring = Wiring.coerce(value) + self._auto_pin_directions() def _resolve_wiring(self) -> Wiring: env = os.environ.get(WIRING_ENV, "").strip() @@ -637,7 +635,10 @@ def save_wiring(self, wiring: Union[Wiring, Mapping[str, Any], str, None] = None "for a LAN pod keep the profile in a file and pass wiring=") api = self._require_server_api() api.put_wiring(api.resolve_device_id(self._device_name), profile.to_dict()) + changed = profile != self._wiring self._wiring = profile.with_changes(source="server") + if changed: + self._auto_pin_directions() return self._wiring def signal(self, name: str) -> GpioPin: @@ -2637,36 +2638,63 @@ def la_dir(self, la: Union[Pin, int, str, Sequence[Union[Pin, int, str]], None] def pin_directions(self, wiring: Optional[Wiring] = None) -> Dict[int, str]: """The DUT direction of every pin the wiring profile names, as :meth:`la_dir` takes it: - ``{5: "dut_output", 4: "dut_input", 1: "bidir", ...}``. - - Roles: ``uart_rx`` and ``spi_miso`` are DUT outputs, ``uart_tx``, ``spi_sclk``, - ``spi_mosi`` and ``spi_cs`` DUT inputs, the I2C pins ``bidir``. Signals: ``input`` (the - pod watches it) is a DUT output, ``output`` a DUT input, ``open_drain`` and ``bidir`` are - ``bidir``. SWD pins are left out.""" + ``{5: "dut_output", 4: "dut_input", 1: "bidir", ...}`` (:meth:`Wiring.pin_dut_dirs + `). + + Each role and signal has a default, and the profile's ``dut_dirs`` overrides it. Roles: + ``uart_rx`` and ``spi_miso`` are DUT outputs; ``uart_tx``, ``spi_sclk``, ``spi_mosi``, + ``spi_cs`` and ``swd_swclk`` DUT inputs; the I2C pins and ``swd_swdio`` ``bidir``. + Signals: ``input`` (the pod watches it) is a DUT output, ``output`` a DUT input, + ``open_drain`` and ``bidir`` are ``bidir``.""" w = wiring if wiring is not None else self.wiring - out: Dict[int, str] = {} - for key, d in _ROLE_DIRS.items(): - la = getattr(w, key, None) - if la: - out[int(la)] = d - for sig in w.signals: - out[int(sig.la)] = _SIGNAL_DIRS.get(sig.direction, "unknown") - return out + return w.pin_dut_dirs() def apply_pin_directions(self, wiring: Optional[Wiring] = None) -> List[LaPinState]: """Send :meth:`pin_directions` of the wiring profile to the pod (:meth:`la_dir`), one - command per direction, and return the pins' new state. The pod forgets them on a reboot; - call this after connecting or changing the wiring. Raises + command per direction, and mark the pins the profile doesn't name ``unknown``. Returns the + pins' new state. + + The SDK does this on its own when it connects to a pod with firmware 3.10.4+ and when + :attr:`wiring` changes (:attr:`pin_directions_status` says how it went); the pod forgets + the directions on a reboot. Raises :class:`~embeddedci.benchpod.errors.DirectionConflictError` when a pin the profile calls a - DUT output is driven right now.""" + DUT output is driven right now; the other directions are sent first.""" + dirs = self.pin_directions(wiring) by_dir: Dict[str, List[int]] = {} - for la, d in sorted(self.pin_directions(wiring).items()): - by_dir.setdefault(d, []).append(la) + for la in range(1, 15): + by_dir.setdefault(dirs.get(la, "unknown"), []).append(la) out: List[LaPinState] = [] - for d, las in by_dir.items(): - out += self.la_dir(las, d) + # "unknown" first, so a pin moving to another role never refuses on its old mark; + # dut_output last, so a driven pin refusing it leaves every other direction in place. + for d in ("unknown", "dut_input", "bidir", "dut_output"): + if by_dir.get(d): + out += self.la_dir(by_dir[d], d) return out + def _auto_pin_directions(self) -> None: + """Send the wiring's pin directions on connect and on a new profile, when the firmware + has ``la_dir``. Never raises: the outcome goes to :attr:`pin_directions_status`.""" + if getattr(self, "_transport", None) is None: + return + try: + if type(self._transport).__name__ == "SerialTransport": + self.pin_directions_status = "not sent: the USB console has no la_dir" + return + # The cached capabilities when there are some; else one status read that is not cached + # (the first real use of capabilities reads and caches them as before). + caps = self._caps if self._caps is not None else Capabilities.from_status( + _dict(self.status())) + if not caps.la_dir: + fw = caps.firmware_version or "unknown" + self.pin_directions_status = (f"not sent: this pod's firmware ({fw}) has no " + "la_dir (3.10.4+)") + return + self.apply_pin_directions() + self.pin_directions_status = "sent" + except Exception as exc: # a convenience on connect: report it, never fail the caller + self.pin_directions_status = f"not sent: {exc}" + _log.warning("pin directions from the wiring were not sent to the pod: %s", exc) + def gpio(self, la: Union[Pin, int, str], mode: GpioMode = "output", *, level: Optional[int] = None) -> GpioPin: """Use an LA channel as GPIO and return its :class:`~embeddedci.benchpod.gpio.GpioPin`. diff --git a/packages/embeddedci/src/embeddedci/benchpod/wiring.py b/packages/embeddedci/src/embeddedci/benchpod/wiring.py index 863262c..633a6c5 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/wiring.py +++ b/packages/embeddedci/src/embeddedci/benchpod/wiring.py @@ -30,6 +30,19 @@ "spi_sclk", "spi_mosi", "spi_miso", "spi_cs") #: Valid :attr:`Signal.direction` values. SIGNAL_DIRECTIONS: Tuple[str, ...] = ("input", "output", "open_drain", "bidir") +#: Valid :attr:`Wiring.dut_dirs` values: which way the DUT's signal goes. +DUT_DIRS: Tuple[str, ...] = ("dut_input", "dut_output", "bidir") +#: The default DUT direction of each role. ``uart_rx`` is the pod's receive line, so the DUT drives +#: it (a DUT output); ``uart_tx`` is the pod's transmit line (a DUT input). +ROLE_DUT_DIRS: Mapping[str, str] = { + "uart_rx": "dut_output", "uart_tx": "dut_input", "i2c_sda": "bidir", "i2c_scl": "bidir", + "swd_swclk": "dut_input", "swd_swdio": "bidir", + "spi_sclk": "dut_input", "spi_mosi": "dut_input", "spi_miso": "dut_output", "spi_cs": "dut_input", +} +#: The default DUT direction of a signal from its :attr:`Signal.direction` (seen from the pod): a +#: signal the pod watches (``input``) is a DUT output, one it drives (``output``) a DUT input. +SIGNAL_DUT_DIRS: Mapping[str, str] = {"input": "dut_output", "output": "dut_input", + "open_drain": "bidir", "bidir": "bidir"} _NAME = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,31}$") _SIGNAL_KEYS = ("name", "la", "direction", "active_low", "description") @@ -102,9 +115,14 @@ class Wiring: #: Where this profile came from: ``"defaults"``, ``"file"``, ``"server"`` or ``"dict"``. source: str = field(default="defaults", compare=False) version: int = WIRING_VERSION + #: The DUT direction of a role or signal where its default is wrong: role key or signal name + #: -> ``dut_input``, ``dut_output`` or ``bidir`` (see :meth:`pin_dut_dirs`). + dut_dirs: Mapping[str, str] = field(default_factory=dict) def __post_init__(self) -> None: object.__setattr__(self, "signals", tuple(self.signals)) + if isinstance(self.dut_dirs, Mapping): + object.__setattr__(self, "dut_dirs", dict(self.dut_dirs)) object.__setattr__(self, "extra", dict(self.extra)) errors = self._errors() if errors: @@ -125,7 +143,7 @@ def from_dict(cls, data: Mapping[str, Any], *, source: str = "dict", profile written by a newer server).""" if not isinstance(data, Mapping): raise ValueError("invalid wiring: a wiring profile must be a JSON object") - scalar = {f.name for f in fields(cls)} - {"signals", "extra", "source"} + scalar = {f.name for f in fields(cls)} - {"signals", "extra", "source", "dut_dirs"} kwargs: Dict[str, Any] = {} extra: Dict[str, Any] = {} unknown: List[str] = [] @@ -134,6 +152,8 @@ def from_dict(cls, data: Mapping[str, Any], *, source: str = "dict", kwargs[key] = value elif key == "signals": kwargs["signals"] = tuple(_parse_signals(value, unknown)) + elif key == "dut_dirs": + kwargs["dut_dirs"] = value elif key.startswith(_PASSTHROUGH_PREFIXES) or not strict: extra[key] = value else: @@ -184,10 +204,11 @@ def to_dict(self) -> Dict[str, Any]: """The profile's JSON object (every key, defaults included).""" out: Dict[str, Any] = {"version": self.version} for f in fields(self): - if f.name in ("signals", "extra", "source", "version"): + if f.name in ("signals", "extra", "source", "version", "dut_dirs"): continue out[f.name] = getattr(self, f.name) out["signals"] = [s.to_dict() for s in self.signals] + out["dut_dirs"] = dict(self.dut_dirs) out.update(self.extra) return out @@ -217,6 +238,24 @@ def pins(self) -> Dict[int, str]: """LA channel → the role or signal on it (unwired channels are absent).""" return {la: name for name, la in self.assignments()} + def dut_dir(self, name: str) -> str: + """The DUT direction of a role (``"uart_rx"``) or signal: its :attr:`dut_dirs` entry, else + the default (:data:`ROLE_DUT_DIRS`, or :data:`SIGNAL_DUT_DIRS` from the signal's + direction).""" + key = str(name).strip() + for k, d in self.dut_dirs.items(): + if k.lower() == key.lower(): + return d + if key.lower() in ROLE_KEYS: + return ROLE_DUT_DIRS[key.lower()] + return SIGNAL_DUT_DIRS.get(self.signal(key).direction, "dut_output") + + def pin_dut_dirs(self) -> Dict[int, str]: + """LA channel -> the DUT direction of what is wired to it (:meth:`dut_dir`), for every + wired role and signal: ``{5: "dut_output", 4: "dut_input", 1: "bidir", ...}``. This is what + the pod gets with ``la_dir``; unwired channels are absent.""" + return {la: self.dut_dir(name) for name, la in self.assignments()} + def la(self, name: str) -> int: """The LA channel of a role (``"uart_rx"``) or a signal (``"TRIGGER"``, case-insensitive).""" key = str(name).strip() @@ -251,13 +290,16 @@ def warnings(self) -> List[str]: return out def describe(self) -> str: - """A readable pin table: channel, what is wired to it, and its bias resistor.""" + """A readable pin table: channel, what is wired to it, its bias resistor and its DUT + direction.""" pins = self.pins() + dirs = self.pin_dut_dirs() lines = [f"LA I/O {self.la_voltage:g} V, target power eFuse {self.efuse}"] for la in range(1, 15): pull = PULL_OHMS.get(la) bias = (f"{pull} pull-{'down' if la in PULLDOWN_CHANNELS else 'up'}" if pull else "no pull") - lines.append(f"LA{la:<3} {pins.get(la, '-'):<12} {bias}") + line = f"LA{la:<3} {pins.get(la, '-'):<12} {bias:<16} {dirs.get(la, '')}" + lines.append(line.rstrip()) return "\n".join(lines + [f"warning: {w}" for w in self.warnings()]) # -- validation ------------------------------------------------------------- @@ -308,6 +350,16 @@ def _errors(self) -> List[str]: errs.append(f"signals[{i}].active_low must be true or false") if not isinstance(s.description, str) or len(s.description) > 120: errs.append(f"signals[{i}].description must be a string of at most 120 characters") + if not isinstance(self.dut_dirs, Mapping): + errs.append("dut_dirs must be an object of role or signal name -> direction") + else: + names = {s.name.lower() for s in self.signals if isinstance(s, Signal) + and isinstance(s.name, str)} + for k, d in self.dut_dirs.items(): + if not isinstance(k, str) or (k.lower() not in ROLE_KEYS and k.lower() not in names): + errs.append(f"dut_dirs.{k}: {k!r} is not a wiring role or a signal name") + elif d not in DUT_DIRS: + errs.append(f"dut_dirs.{k} must be one of {', '.join(DUT_DIRS)}, got {d!r}") users: Dict[int, str] = {} for name, la in self.assignments(): if not 1 <= la <= 14: diff --git a/packages/embeddedci/tests/api_surface.json b/packages/embeddedci/tests/api_surface.json index baaf8a9..2a2b0a2 100644 --- a/packages/embeddedci/tests/api_surface.json +++ b/packages/embeddedci/tests/api_surface.json @@ -346,6 +346,7 @@ "counts_to_volts": "(self, count: 'float') -> 'float'", "from_parameters": "(cls, params: 'Mapping[str, Any]') -> \"'Capabilities'\"", "from_status": "(cls, status: 'Mapping[str, Any]') -> \"'Capabilities'\"", + "la_dir": "property", "merge": "(self, other: \"'Capabilities'\") -> \"'Capabilities'\"" }, "Capture": { @@ -1175,17 +1176,20 @@ "signals", "extra", "source", - "version" + "version", + "dut_dirs" ], "assignments": "(self) -> 'Iterator[Tuple[str, int]]'", "coerce": "(cls, value: \"Union['Wiring', Mapping[str, Any], str, Path]\") -> \"'Wiring'\"", "defaults": "(cls) -> \"'Wiring'\"", "describe": "(self) -> 'str'", + "dut_dir": "(self, name: 'str') -> 'str'", "from_dict": "(cls, data: 'Mapping[str, Any]', *, source: 'str' = 'dict', strict: 'bool' = True) -> \"'Wiring'\"", "i2c_address": "property", "la": "(self, name: 'str') -> 'int'", "la_voltage": "property", "load": "(cls, path: 'Union[str, Path]') -> \"'Wiring'\"", + "pin_dut_dirs": "(self) -> 'Dict[int, str]'", "pins": "(self) -> 'Dict[int, str]'", "signal": "(self, name: 'str') -> 'Signal'", "to_dict": "(self) -> 'Dict[str, Any]'", diff --git a/packages/embeddedci/tests/test_i2c_master.py b/packages/embeddedci/tests/test_i2c_master.py index 7c89712..1251899 100644 --- a/packages/embeddedci/tests/test_i2c_master.py +++ b/packages/embeddedci/tests/test_i2c_master.py @@ -294,7 +294,8 @@ def test_la_pins_dir_and_warn(): bp, fake = pod() bp.la_dir(4, "dut_input") pins = {p.la: p for p in bp.la_pins()} - assert pins[4].dir == "dut_input" and pins[1].dir is None + # LA1 is i2c_sda in the wiring: bidir since connect. LA3 is not wired: unknown. + assert pins[4].dir == "dut_input" and pins[1].dir == "bidir" and pins[3].dir is None assert pins[3].warn == 3 and len(pins[3].warnings) == 2 and pins[4].warnings == [] @@ -307,10 +308,13 @@ def test_pin_directions_from_wiring(): assert dirs == {5: "dut_output", 4: "dut_input", 1: "bidir", 2: "bidir", 7: "dut_input", 8: "dut_input", 9: "dut_output", 10: "dut_input", 11: "dut_output", 12: "dut_input", 13: "bidir"} + fake.commands.clear() bp.apply_pin_directions(w) sent = [c for c in fake.commands if c["cmd"] == "la_dir"] + # Unwired pins first (unknown), DUT outputs last. + assert [c["dir"] for c in sent] == ["unknown", "dut_input", "bidir", "dut_output"] assert {c["dir"]: c["la"] for c in sent} == {"bidir": [1, 2, 13], "dut_input": [4, 7, 8, 10, 12], - "dut_output": [5, 9, 11]} + "dut_output": [5, 9, 11], "unknown": [3, 6, 14]} assert fake.dirs[11] == "dut_output" diff --git a/packages/embeddedci/tests/test_pin_directions.py b/packages/embeddedci/tests/test_pin_directions.py new file mode 100644 index 0000000..f9cf25f --- /dev/null +++ b/packages/embeddedci/tests/test_pin_directions.py @@ -0,0 +1,113 @@ +"""Pin directions by default: the wiring profile's DUT direction per role and signal (dut_dirs), +sent to the pod with la_dir on connect and on a new profile (firmware 3.10.4+).""" + +from __future__ import annotations + +import pytest + +from embeddedci.benchpod import BenchPod, DirectionConflictError, Wiring +from embeddedci.benchpod.wiring import Signal + +from test_i2c_master import BusPod + + +class OldPod(BusPod): + def __init__(self, version: str) -> None: + super().__init__() + self.version = version + + def status(self): + st = super().status() + st["version"] = self.version + return st + + +def _sent(fake): + return [c for c in fake.commands if c["cmd"] == "la_dir"] + + +# -- the profile ---------------------------------------------------------------------------------- + +def test_defaults_cover_every_wired_role(): + assert Wiring().pin_dut_dirs() == {5: "dut_output", 4: "dut_input", 1: "bidir", 2: "bidir", + 11: "dut_input", 12: "bidir"} + + +def test_dut_dirs_override_roles_and_signals_case_insensitively(): + w = Wiring(signals=(Signal("READY", 9), Signal("TRIG", 10, "output")), + dut_dirs={"uart_rx": "bidir", "ready": "dut_input"}) + assert w.dut_dir("uart_rx") == "bidir" and w.dut_dir("READY") == "dut_input" + assert w.dut_dir("TRIG") == "dut_input" and w.dut_dir("uart_tx") == "dut_input" + assert w.pin_dut_dirs()[9] == "dut_input" and w.pin_dut_dirs()[5] == "bidir" + + +@pytest.mark.parametrize("dirs, msg", [ + ({"uart_rx": "output"}, "dut_dirs.uart_rx must be one of dut_input, dut_output, bidir"), + ({"uart_rx": "unknown"}, "dut_dirs.uart_rx must be one of"), + ({"NOPE": "bidir"}, "'NOPE' is not a wiring role or a signal name"), +]) +def test_dut_dirs_are_validated(dirs, msg): + with pytest.raises(ValueError, match=msg): + Wiring(dut_dirs=dirs) + + +def test_dut_dirs_round_trip_and_old_profiles_stay_valid(): + w = Wiring.from_dict({"signals": [{"name": "READY", "la": 9}], "dut_dirs": {"READY": "bidir"}}) + assert w.to_dict()["dut_dirs"] == {"READY": "bidir"} + assert Wiring.from_dict(w.to_dict()) == w + assert Wiring.from_dict({"uart_rx": 5}).dut_dirs == {} + assert Wiring().to_dict()["dut_dirs"] == {} + + +def test_describe_shows_the_direction(): + text = Wiring().describe() + assert "uart_rx" in text and "dut_output" in text.splitlines()[5] + + +# -- pushed to the pod ------------------------------------------------------------------------------ + +def test_connect_sends_the_wiring_directions(): + fake = BusPod() + bp = BenchPod(transport=fake, wiring=Wiring(signals=(Signal("READY", 9),))) + assert bp.pin_directions_status == "sent" + assert _sent(fake) == [ + {"cmd": "la_dir", "la": [3, 6, 7, 8, 10, 13, 14], "dir": "unknown"}, + {"cmd": "la_dir", "la": [4, 11], "dir": "dut_input"}, + {"cmd": "la_dir", "la": [1, 2, 12], "dir": "bidir"}, + {"cmd": "la_dir", "la": [5, 9], "dir": "dut_output"}, + ] + # The protection is on: a push-pull GPIO on the DUT's TX line is refused. + with pytest.raises(DirectionConflictError): + bp.gpio(5, "output") + with pytest.raises(DirectionConflictError): + bp.signal("READY").configure("output") + + +@pytest.mark.parametrize("version", ["3.10.3", "3.9.6", ""]) +def test_older_firmware_is_skipped_quietly(version): + fake = OldPod(version) + bp = BenchPod(transport=fake, wiring=Wiring()) + assert _sent(fake) == [] + assert bp.pin_directions_status.startswith("not sent: this pod's firmware") + assert "la_dir (3.10.4+)" in bp.pin_directions_status + + +def test_a_new_profile_is_sent_and_moves_marks(): + fake = BusPod() + bp = BenchPod(transport=fake, wiring=Wiring()) + fake.commands.clear() + bp.wiring = Wiring(uart_rx=9, uart_tx=10) + assert bp.pin_directions_status == "sent" + assert fake.dirs[5] == "unknown" and fake.dirs[9] == "dut_output" and fake.dirs[10] == "dut_input" + + +def test_a_driven_pin_does_not_fail_the_connect_or_the_assignment(): + fake = BusPod() + bp = BenchPod(transport=fake, wiring=Wiring(uart_rx=None)) + bp.gpio(9, "output") # LA9 unwired: a push-pull GPIO is fine + bp.wiring = Wiring(uart_rx=9) # now the profile says LA9 is the DUT's TX + assert "direction conflict: LA9" in bp.pin_directions_status + # The rest still went out, DUT outputs last. + assert fake.dirs[4] == "dut_input" and fake.dirs[1] == "bidir" + with pytest.raises(DirectionConflictError): + bp.apply_pin_directions() diff --git a/packages/embeddedci/tests/test_sensors.py b/packages/embeddedci/tests/test_sensors.py index 6f58598..7f780ff 100644 --- a/packages/embeddedci/tests/test_sensors.py +++ b/packages/embeddedci/tests/test_sensors.py @@ -228,7 +228,8 @@ def test_old_firmware_gets_a_clear_error_before_anything_is_sent(): assert str(info.value) == ("this pod's firmware is too old for sensor emulation; " "update the firmware") assert info.value.feature == "sensors" - assert fake.commands == [] + # Only the pin directions the SDK sends from the wiring on connect. + assert [c for c in fake.commands if c["cmd"] != "la_dir"] == [] def test_start_defines_in_chunks_then_starts_with_wired_pins():