stream¶

This module is a data acquisition module that captures video streams from Miniscopes based on the Miniscope-SAMD-Framework firmware. The firmware repository will be published in future updates but is currently under development and private.

Command¶

After installation and customizing device configurations and mio.devices.stream.config.StreamDevConfig if necessary, run the command described in CLI Usage.

One example of this command is the following:

$ mio stream capture -c .path/to/device/config.yml -o output_filename.avi -m

A window displaying the image transferred from the Miniscope and a graph plotting metadata (-m option) should pop up. Additionally, the indexes of captured frames and their statuses will be logged in the terminal. The MIO_STREAM_HEADER_PLOT_KEY defines plotted header fields (see .env.sample).

Prerequisites¶

Data capture hardware: Opal Kelly XEM7310-A75 FPGA board (connected via USB)
Supported Operating Systems: MacOS or Ubuntu PC (To do: specify version)
Imaging hardware: Miniscope based on the Miniscope-SAMD-Framework firmware. Hardware modules for feeding in data into the data capture hardware are also needed but these will be specified in future updates.

Device configuration¶

A YAML file is used to configure Stream DAQ based on the device configuration. The device configuration needs to match the imaging and data capture hardware for proper operation. This file is used to set up hardware, define data formats, and set data preambles. The contents of this YAML file will be parsed into a mio.devices.stream.config.StreamDevConfig, which then configures the Stream DAQ.

FPGA (Opal Kelly) configuration¶

The bitstream field in the device configuration yaml file specifies the image that will be uploaded to the opal kelly board. This file needs to be placed in mio.devices.

Bitstream file nomenclature¶

Name format of the bitstream files and directory: [DEVICE_DIR]/USBInterface-[CLOCK_FREQUENCY]-[INPUT_PIN]_[IO_VOLTAGE]-[ENCODING_POLARITY]

DEVICE_DIR: FPGA module name. The current package supports XEM7310-A75 (Opal kelly).
CLOCK_FREQUENCY: Manchester encoding clock frequency. For the current FPGA (XEM7310-A75), this frequency can be configured to 100 / i MHz where i is an integer.
INPUT_PIN: Signal input pin. J2_2 means signal goes into second pin of J2 pin headers in the hardware module.
IO_VOLTAGE: I/O voltage of the FPGA INPUT_PIN. The current package supports 3.3V input.
ENCODING_POLARITY: Manchester encoding convention, which corresponds to bit polarity. The current package supports IEEE convention Manchester encoding.

Example device configuration¶

Below is an example configuration YAML file. More examples can be found in mio.data.config.

# capture device. "OK" (Opal Kelly) or "UART"
device: "OK"

# bitstream file to upload to Opal Kelly board
bitstream: "XEM7310-A75/USBInterface-8_33mhz-J2_2-3v3-IEEE.bit"

# COM port and baud rate is only required for UART mode
port: null
baudrate: null

# Preamble for each data buffer.
preamble: 0x12345678

# Image format. StreamDevice will calculate buffer size, etc. based on these parameters
frame_width: 200
frame_height: 200
pix_depth: 8

# Buffer data format. These have to match the firmware value
header_len: 384 # 12 * 32 (in bits)
buffer_block_length: 10
block_size: 512
num_buffers: 32
dummy_words: 10

# Flags to flip bit/byte order when recovering headers and data. See model document for details.
reverse_header_bits: True
reverse_header_bytes: True
reverse_payload_bits: True
reverse_payload_bytes: True

adc_scale:
  ref_voltage: 1.1
  bitdepth: 8
  battery_div_factor: 5
  vin_div_factor: 11.3

runtime:
  serial_buffer_queue_size: 10
  frame_buffer_queue_size: 5
  image_buffer_queue_size: 5
  csv:
    buffer: 100
  plot:
    keys:
      - timestamp
      - buffer_count
      - frame_buffer_count
      - battery_voltage
      - input_voltage
    update_ms: 1000
    history: 500

Base classes used by streaming miniscopes, like the miniscope zero and MSUS

class mio.devices.stream.StreamBufferHeader(*, linked_list: int, frame_num: int, buffer_count: int, frame_buffer_count: int, write_buffer_count: int, dropped_buffer_count: int, timestamp: int, write_timestamp: int, pixel_count: int, battery_voltage_raw: int, input_voltage_raw: int, runtime_metadata: RuntimeMetadata = <factory>)¶

Refinements of BufferHeader for StreamDevice

POSITIONS: ClassVar[dict[str, int]] = {'battery_voltage_raw': 9, 'buffer_count': 2, 'dropped_buffer_count': 5, 'frame_buffer_count': 3, 'frame_num': 1, 'input_voltage_raw': 10, 'linked_list': 0, 'pixel_count': 7, 'timestamp': 6, 'write_buffer_count': 4, 'write_timestamp': 8}¶

property adc_scaling: ADCScaling | None¶: ADCScaling applied to voltage readings

property battery_voltage: float¶: Scaled battery voltage in Volts.

battery_voltage_raw: int¶

classmethod csv_header_cols() → list[str]¶

Return the standardized column names for CSV output.

This ensures consistent column ordering across all StreamBufferHeader instances when writing to CSV files.

Parameters:: header_format – The StreamBufferHeaderFormat instance to get column ordering from
Returns:: Column names in the order they should appear in CSV output
Return type:: list[str]

classmethod from_sequence(vals: Sequence, construct: bool = False, runtime_metadata: RuntimeMetadata = None) → Self¶

Instantiate a stream buffer header from linearized values (eg. in an ndarray or list), an associated format that tells us what index the model values are found in that data, and runtime metadata container.

Parameters:

vals (list, numpy.ndarray) – Indexable values to cast to the header model
construct (bool) – If True , use model_construct() to create the model instance (ie. without validation, but faster). Default: False
runtime_metadata (RuntimeMetadata, optional) – Runtime metadata to attach to the header.

Returns:

StreamBufferHeader

property input_voltage: float¶: Scaled input voltage in Volts.

input_voltage_raw: int¶

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_dump_all(warning: bool = False) → dict¶

Return a dictionary of the model values, including runtime metadata if available.

Returns:: Dictionary of model values
Return type:: dict

model_post_init(context: Any, /) → None¶

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

self – The BaseModel instance.
context – The context.

pixel_count: int¶

runtime_metadata: RuntimeMetadata¶

class mio.devices.stream.StreamDevConfig(*, id: Annotated[str, _PydanticGeneralMetadata(pattern='[\\w\\-\\/#]+')], mio_model: Annotated[str, AfterValidator(func=_is_identifier)] = None, mio_version: str = '0.0.1.dev476+g19de29a', device: Literal['OK', 'UART'], bitstream: Path | None = None, port: str | None = None, baudrate: int | None = None, frame_width: int, frame_height: int, fs: int = 20, preamble: bytes, header_len: int, pix_depth: int = 8, buffer_block_length: int, block_size: int, num_buffers: int, reverse_header_bits: bool = False, reverse_header_bytes: bool = False, reverse_payload_bits: bool = False, reverse_payload_bytes: bool = False, dummy_words: int = 0, adc_scale: ADCScaling | None = ADCScaling(ref_voltage=1.1, bitdepth=8, battery_div_factor=5.0, vin_div_factor=11.3), runtime: StreamDevRuntime = StreamDevRuntime(serial_buffer_queue_size=10, frame_buffer_queue_size=5, image_buffer_queue_size=5, queue_put_timeout=5, plot=StreamPlotterConfig(keys=['timestamp', 'buffer_count', 'frame_buffer_count'], update_ms=1000, history=500), csvwriter=CSVWriterConfig(buffer=100), ntp_server=None, ntp_max_offset_seconds=0.01, ber_test_n_buffers=32767))¶

Format model used to parse DAQ configuration yaml file (examples are in ./config) The model attributes are key-value pairs needed for reconstructing frames from data streams.

Parameters:

device (str) – Interface hardware used for receiving data. Current options are “OK” (Opal Kelly XEM 7310) and “UART” (generic UART-USB converters). Only “OK” is supported at the moment.
bitstream (str, optional) – Required when device is “OK”. The configuration bitstream file to upload to the Opal Kelly board. This uploads a Manchester decoder HDL and different bitstream files are required to configure different data rates and bit polarity. This is a binary file synthesized using Vivado, and details for generating this file will be provided in later updates.
port (str, optional) – Required when device is “UART”. COM port connected to the UART-USB converter.
baudrate (Optional[int]) – Required when device is “UART”. Baudrate of the connection to the UART-USB converter.
frame_width (int) – Frame width of transferred image. This is used to reconstruct image.
frame_height (int) – Frame height of transferred image. This is used to reconstruct image.
fs (int) – Framerate of acquired stream
preamble (str) – 32-bit preamble used to locate the start of each buffer. The header and image data follows this preamble. This is used as a hex but imported as a string because yaml doesn’t support hex format.
header_len (int, optional) – Length of header in bits. (For 32-bit words, 32 * number of words) This is useful when not all the variable/words in the header are defined in MetadataHeaderFormat. The user is responsible to ensure that header_len is larger than the largest bit position defined in MetadataHeaderFormat otherwise unexpected behavior might occur.
pix_depth (int, optional) – Bit-depth of each pixel, by default 8.
buffer_block_length (int) – Defines the data buffer structure. This value needs to match the Miniscope firmware. Number of blocks per each data buffer. This is required to calculate the number of pixels contained in one data buffer.
block_size (int) – Defines the data buffer structure. This value needs to match the Miniscope firmware. Number of 32-bit words per data block. This is required to calculate the number of pixels contained in one data buffer.
num_buffers (int) – Defines the data buffer structure. This value needs to match the Miniscope firmware. This is the number of buffers that the source microcontroller cycles around. This isn’t strictly required for data reconstruction but useful for debugging.
reverse_header_bits (bool, optional) – If True, reverse the bits within each byte of the header. Default is False.
reverse_header_bytes (bool, optional) – If True, reverse the byte order within each 32-bit word of the header. This is used for handling endianness in systems where the byte order needs to be swapped. Default is False.
reverse_payload_bits (bool, optional) – If True, reverse the bits within each byte of the payload. Default is False.
reverse_payload_bytes (bool, optional) – If True, reverse the byte order within each 32-bit word of the payload. This is used for handling endianness in systems where the byte order needs to be swapped. Default is False.
dummy_words (int, optional) – Number of 32-bit dummy words in the header. This is used to stabilize clock recovery in FPGA Manchester decoder. This value does not have a meaning for image recovery.
..todo:: – Move port (for USART) to a user config area. This should make this pure device config.

adc_scale: ADCScaling | None¶

baudrate: int | None¶

bitstream: Path | None¶

block_size: int¶

buffer_block_length: int¶

property buffer_npix: list[int]¶

List of pixel counts per buffer for a complete frame.

A frame is split across multiple buffers. This returns a list where each element is the number of pixels in that buffer. The last buffer may have fewer pixels (the remainder).

device: Literal['OK', 'UART']¶

dummy_words: int¶

classmethod ensure_exists(value: Path | None) → Path | None¶: If a bitstream file has been provided, ensure it exists

frame_height: int¶

frame_width: int¶

fs: int¶

header_len: int¶

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context: Any, /) → None¶

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

self – The BaseModel instance.
context – The context.

num_buffers: int¶

pix_depth: int¶

port: str | None¶

preamble: bytes¶

classmethod preamble_to_bytes(value: str | bytes | int) → bytes¶

Cast preamble to bytes.

Parameters:: value (str, bytes, int) – Recast from str (in yaml like preamble: "0x12345" ) or int (in yaml like preamble: 0x12345
Returns:: bytes

property px_per_buffer: int¶: Number of pixels per buffer

classmethod resolve_relative(value: Path) → Path¶: If we are given a relative path to a bitstream, resolve it relative to the device path

reverse_header_bits: bool¶

reverse_header_bytes: bool¶

reverse_payload_bits: bool¶

reverse_payload_bytes: bool¶

runtime: StreamDevRuntime¶

class mio.devices.stream.StreamDevRuntime(*, serial_buffer_queue_size: int = 10, frame_buffer_queue_size: int = 5, image_buffer_queue_size: int = 5, queue_put_timeout: int = 5, plot: StreamPlotterConfig | None = StreamPlotterConfig(keys=['timestamp', 'buffer_count', 'frame_buffer_count'], update_ms=1000, history=500), csvwriter: CSVWriterConfig | None = CSVWriterConfig(buffer=100), ntp_server: str | None = None, ntp_max_offset_seconds: float = 0.01, ber_test_n_buffers: int = 32767)¶

Runtime configuration for StreamDevice

Included within StreamDevConfig to separate config that is not unique to the device, but how that device is controlled at runtime.

ber_test_n_buffers: int¶

csvwriter: CSVWriterConfig | None¶

frame_buffer_queue_size: int¶

image_buffer_queue_size: int¶

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

ntp_max_offset_seconds: float¶

ntp_server: str | None¶

plot: StreamPlotterConfig | None¶

queue_put_timeout: int¶

serial_buffer_queue_size: int¶

class mio.devices.stream.StreamDevice(device_config: StreamDevConfig | Path | PathLike[str] | Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[_PydanticGeneralMetadata(pattern='[\\w\\-\\/#]+')])])¶

A combined class for configuring and reading frames from a UART and FPGA source. Supported devices and required inputs are described in StreamDevConfig model documentation. This function’s entry point is the main function, which should be used from the stream_image_capture command installed with the package. Example configuration yaml files are stored in /mio/config/.

Examples

$ mio stream capture -c path/to/config.yml -o output_filename.avi Connected to XEM7310-A75 Succesfully uploaded /mio/mio/interfaces/selected_bitfile.bit FrontPanel is supported

Todo

Make it fast and understandable.

alive_processes() → list[Process]¶

Return a list of alive processes.

Returns:: List of alive processes.
Return type:: List[multiprocessing.Process]

property buffer_npix: list[int]¶: List of pixels per buffer for a frame

Entry point to start frame capture.

Parameters:

source (Literal[uart, fpga]) – Device source.
read_length (Optional[int], optional) – Passed to _fpga_recv() when source == “fpga”, by default None.
video (Path, optional) – If present, a path to an output video file
video_kwargs (dict, optional) – kwargs passed to init_video()
metadata (Path, optional) – Save metadata information during capture.
binary (Path, optional) – Save raw binary directly from okDev to file, if present. Note that binary is captured in append mode, rather than rewriting an existing file.
show_video (bool, optional) – If True, display the video in real-time.
show_metadata (bool, optional) – If True, show metadata information during capture.
mode (Literal["image", "ber"], optional) – Capture mode. "image" (default) reconstructs frames; "ber" runs a PRBS bit-error-rate test on the incoming data stream.
ber_output (Path, optional) – When mode == "ber", JSON file to write the BER summary to.

Raises:

ValueError – If source is not in (“uart”, “fpga”).

property nbuffer_per_fm: int¶: Number of buffers per frame, computed from buffer_npix

prbs15_ber(serial_buffer_queue: Queue, n_buffers: int = 100) → dict[str, float | int]¶

Measure bit-error-rate (BER) on the communication link using PRBS-15 (pseudo-random binary sequence; standard pattern for link tests).

Unlike typical continuous-stream BER tests, this preserves the buffer framing of normal image capture and substitutes PRBS-15 for the pixel payload, so the same data path that delivers images is what’s being measured.

Each buffer’s payload is seeded with the device’s buffer_count (mod 2^15, zero remapped to 1). The host regenerates the matching sequence and XORs it against the first pixel_count bytes of payload; trailing bytes (dummies, merged-buffer tails) are excluded so they don’t inflate the count. Errors and bits accumulate across up to n_buffers buffers.

Returns:: Run summary: buffers received, bits and errors compared, cumulative ber, buffer_count range, and per-window snapshots.
Return type:: dict

class mio.devices.stream.StreamPlotterConfig(*, keys: list[str], update_ms: int = 1000, history: int = 500)¶

Configuration for mio.plots.headers.StreamPlotter

history: int¶

keys: list[str]¶

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

update_ms: int¶

mio.devices.stream.iter_buffers(source: Iterator[bytes], preamble: Bits, pre_first: bool = True, capture_binary: Path | None = None) → Generator[bytes, None, None]¶

Given some iterator that yields bytes (like a camera device), yield buffers from that iterator as bytes objects split by the preamble delimiter.

Parameters:

source (Iterator[bytes]) – The iterator that yields bytes
preamble (Bits) – The delimiter bit series to split buffers by
pre_first (bool | None) – Whether preamble/header is returned at the beginning of each buffer, by default True.
capture_binary (Path | None) – save binary directly from the okDev to the supplied path, if present.