Microcontroller.py#

Serial interface for MCU-mediated FPGA measurement.

Provides the Microcontroller class that communicates with a microcontroller over a serial link to program an ICE40 FPGA and retrieve fitness-evaluation data (pulse counts or ADC waveforms).

Note

Serial protocol details were inferred from code and may need verification against MCU firmware.

class Hardware.Microcontroller.Microcontroller(fpga: str, logger: Logger, config: MicrocontrollerConfig)[source]#

Driver for a serial-connected MCU that programs and reads from an FPGA.

Opens a persistent serial connection on construction and provides async methods to request waveform or pulse-count measurements.

Concurrency note:#

The async methods in this class use blocking pyserial calls and do NOT yield to the event loop. As a result, asyncio.gather() over multiple Microcontroller instances currently runs them sequentially, not in parallel.

TODO: Migrate serial I/O to serial_asyncio (https://pypi.org/project/serial-asyncio/) so that serial reads genuinely suspend and allow other coroutines to run. This is the server-side component — when the architecture moves to a server-client model, this class (or its replacement) will run on the server and use serial_asyncio.

Per-Icestick exclusivity:#

Each Microcontroller instance manages exactly one Icestick via one serial port. iceprog holds exclusive USB access to the device during compile(). Concurrent calls to request_measurement() on the SAME Microcontroller instance will corrupt each other. The server must serialize calls per-instance (e.g. one asyncio.Semaphore(1) per Microcontroller, or a dedicated worker coroutine per device). Calls on DIFFERENT Microcontroller instances (different Icesticks) can run concurrently. See .claude/docs/hardware_concurrency.md.

__init__(fpga: str, logger: Logger, config: MicrocontrollerConfig)[source]#

Initialize the serial connection to the MCU.

Parameters:
  • fpga (str) – FPGA device identifier passed to icepack/iceprog.

  • logger (Logger) – Logger instance for event and warning messages.

  • config (MicrocontrollerConfig) – Serial port settings.

get_available_FPGAs() list[str][source]#

Return the list of FPGA device identifiers managed by this MCU.

async measure_pulses(samples: int) list[int][source]#

Collect multiple pulse-count samples from the FPGA.

Parameters:

samples (int) – Number of times to call measure_pulses_once.

Returns:

Concatenated pulse counts from all sample rounds.

Return type:

list[int]

async measure_pulses_once() list[int][source]#

Perform a single pulse-count read from the MCU.

Sends command '1' and polls the serial line for a numeric response, retrying up to 5 times on timeout. Returns -1 on timeout and -2 on parse failure.

Returns:

Parsed pulse count(s) from a single read cycle.

Return type:

list[int]

async measure_signal() list[int][source]#

Capture an ADC waveform from the FPGA via the MCU.

Sends command '2' to initiate ADC capture, then reads lines between START and FINISHED delimiters.

Returns:

Integer ADC samples (~10 us apart).

Return type:

list[int]

async request_measurement(measurement: Measurement) Measurement[source]#

Compile the circuit and perform the requested measurement.

Dispatches to measure_signal or measure_pulses based on measurement.data_request. The result (or error) is stored in measurement.result as a Success or Failure.

Parameters:

measurement (Measurement) – Measurement descriptor specifying circuit and data type.

Returns:

The same object with result populated.

Return type:

Measurement

class Hardware.Microcontroller.MicrocontrollerConfig(usb_path: str, serial_baud: int, read_timeout: float)[source]#

Serial connection parameters for the MCU.

usb_path#

Device path to the serial port (e.g. /dev/ttyUSB0).

Type:

str

serial_baud#

Baud rate for the serial connection.

Type:

int

read_timeout#

Timeout in seconds for serial read operations.

Type:

float