devices.base

Base device classes

class mio.devices.base.BufferHeader(*, 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)

Container for the data stream’s header.

When converting from a sequence (like a numpy array or list of bytes), the POSITIONS mapping is responsible for mapping sequential values to keys, as different devices have the fields in different orders.

Not all keys must be set in POSITIONS - extra keys may be provided e.g. at runtime while constructing the object by passing **kwargs to from_sequence()

POSITIONS: ClassVar[dict[str, int]] = {'buffer_count': 2, 'dropped_buffer_count': 5, 'frame_buffer_count': 3, 'frame_num': 1, 'linked_list': 0, 'timestamp': 6, 'write_buffer_count': 4, 'write_timestamp': 7}
buffer_count: 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]

dropped_buffer_count: int
frame_buffer_count: int
frame_num: int
classmethod from_sequence(vals: Sequence, construct: bool = False, **kwargs: dict[str, Any]) → Self

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

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

  • **kwargs – Additional kwargs that are not specified in POSITIONS . Values in kwargs overwrite those derived from vals .

Returns:

BufferHeader

linked_list: int
model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

timestamp: int
write_buffer_count: int
write_timestamp: int
class mio.devices.base.Device(config: DeviceConfig | Path | PathLike[str] | Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[_PydanticGeneralMetadata(pattern='[\\w\\-\\/#]+')])])

Abstract device parent class

config_cls

alias of DeviceConfig

device_name: ClassVar[str] = 'device'
header_cls

alias of BufferHeader