`msight_core.data`

Data containers for various sensor data types.

:class:~msight_core.data.Data — Base class for all data containers. Handles registration, serialization to/from dict/msgpack/JSON, and per-field codecs.
:class:~msight_core.data.SensorData — Extends :class:~msight_core.data.Data with common sensor fields such as sensor_name, capture_timestamp,, creation_timestamp, frame_id, and device_name. Use this as the base for any sensor-originated payloads.
:class:~msight_core.data.SensorDataSequence — Sequence wrapper for grouping ordered sensor data frames.
:class:~msight_core.data.ImageData — Image container.
:class:~msight_core.data.DetectionResultsData — Container for detection outputs that associates a detection result object with its source image frame id and optional raw sensor payload.
:class:~msight_core.data.RoadUserListData — Stores a list of road-user points (e.g. pedestrian/vehicle positions). Includes a field codec for serialization of the list.
:class:~msight_core.data.BytesData — Generic opaque binary payload. Useful when you need to store or transmit arbitrary bytes without interpreting their contents.
:class:~msight_core.data.VideoData — Video container.
:class:~msight_core.data.PointCloudData — Point-cloud data container for lidar/depth sensors.

Examples

Import a specific class::

from msight_core.data import ImageData

Import multiple classes for type annotations::

from msight_core.data import Data, SensorData, BytesData

`Data` `dataclass`

Base class for all serializable MSight messages.

Responsibilities

Provide generic to_dict / from_dict implementations that add a data_type tag, handle nested Data instances, and support per-field codecs via __field_codecs__.
Provide convenience serialize / deserialize (MessagePack) and to_json / from_json helpers.
Maintain an in-memory registry mapping data_type -> class so deserialization uses fast dictionary lookup instead of dynamic imports.

`__init_subclass__(**kwargs)`

Set up subclass metadata and merge field codecs.

This hook runs when a subclass is created and performs two tasks: 1. Compute and cache a __data_type__ tag for the subclass and register it in Data.__registry__ for fast lookup during deserialization. 2. Merge __field_codecs__ from base classes into the subclass so codecs defined on parents won't be overridden in default behavior.

`to_dict()`

Convert this dataclass instance into a serializable dict.

The resulting dict contains a data_type key that identifies the concrete subclass. Fields marked with a FieldCodec will be transformed by their encoder prior to recursive encoding.

Returns:

Type	Description
`Dict[str, Any]`	A dictionary suitable for JSON/MessagePack encoding.

Raises:

Type	Description
`TypeError`	If the instance is not a dataclass.

`from_dict(payload)` `classmethod`

Construct a Data instance (or subclass) from a dict payload.

If called on the base Data class it will dispatch to the concrete subclass specified by the payload's data_type tag using the internal registry. When invoked on a concrete subclass it will instantiate that class, apply any per-field decode codecs, and recursively decode nested structures.

Parameters:

Name	Type	Description	Default
`payload`	`Dict[str, Any]`	A dictionary produced by `to_dict` or an equivalent source.	required

Returns:

Type	Description
`T`	An instance of `cls` (or the dispatched subclass for calls on `Data`).

Raises:

Type	Description
`TypeError`	If `payload` is not a dict.
`ValueError`	If `data_type` key is missing in the payload dict or unknown when dispatching from the base `Data` class.

`serialize()`

Serialize this object to MessagePack bytes.

Internally calls :meth:to_dict and packs the result using msgpack.packb with use_bin_type=True.

Returns:

Type	Description
`bytes`	Bytes containing the MessagePack-encoded representation.

`deserialize(data)` `classmethod`

Deserialize MessagePack bytes into a Data instance.

Parameters:

Name	Type	Description	Default
`data`	`bytes`	MessagePack-encoded bytes representing a serialized `Data` object.	required

Returns:

Type	Description
`T`	A concrete `Data` subclass instance reconstructed from the payload.

`to_json()`

Convert this object to a JSON string.

Uses to_dict and a custom BytesEncoder to ensure byte sequences are JSON serializable.

Returns:

Type	Description
`str`	A JSON string.

`from_json(data)` `classmethod`

Create a Data instance from a JSON string.

Parameters:

Name	Type	Description	Default
`data`	`str`	JSON string produced by :meth:`to_json` or an equivalent source.	required

Returns:

Type	Description
`T`	A concrete `Data` instance.

`SensorData` `dataclass`

Bases: Data

Base class for sensor data messages.

Shared fields across all sensor payloads.

Attributes:

Name	Type	Description
`sensor_name`	`Optional[str]`	Identifier for the sensor that produced the measurement (e.g. camera or lidar name).
`capture_timestamp`	`float`	POSIX timestamp (seconds, float) when the sensor actually captured the measurement. Defaults to the current time via `time.time()`.
`creation_timestamp`	`float`	POSIX timestamp (seconds, float) when the sensor data object was created. Defaults to the current time via `time.time()` and normally should not be set manually.
`frame_id`	`int`	Monotonically-increasing numeric id generated by :func:`generate_frame_id` (Snowflake-like). Useful to correlate frames across messages and processes. Defaults to a new id at creation time.
`device_name`	`Optional[str]`	Logical name of the device producing the data. Read from the `MSIGHT_EDGE_DEVICE_NAME` environment variable at module import time; Normally should not be set manually.

`time` `property`

ISO 8601 formatted capture timestamp.

`capture_time` `property`

ISO 8601 formatted capture timestamp. The same as time.

`creation_time` `property`

ISO 8601 formatted creation timestamp.

`SensorDataSequence` `dataclass`

Bases: SensorData

A sequence container for sensor messages.

The obj_list field may contain heterogeneous Data subclasses, allowing sequences of mixed message types.

`ImageData` `dataclass`

Bases: SensorData

Container for image sensor payloads.

The dataclass stores an image payload together with timestamps and other metadata inherited from :class:SensorData.

Attributes:

Name	Type	Description
`image`	`Optional[bytes]`	Encoded image bytes (e.g. JPEG) or raw bytes depending on `is_encoded`. Hidden from `repr` to avoid large binary dumps in logs.
`is_encoded`	`bool`	Whether `image` contains an encoded image format (True) or raw numpy bytes (False). When encoded, `to_array` will decode using OpenCV; when not encoded the consumer must know the original dtype/shape, which is provided in `size`.
`detection_results`	`Optional[DetectionResultsData]`	Optional detection output associated with this image/frame.
`size`	`Optional[tuple[int, int, int]]`	Shape of the original numpy array (height, width, channels). Required when `is_encoded` is False to reconstruct the array from raw bytes.

`decoded_image` `property`

Get the decoded image as a numpy array.

Returns:

Type	Description
`ndarray`	np.ndarray: The decoded image.

Raises:

Type	Description
`ValueError`	If the image cannot be decoded from encoded bytes.

`from_ndarray(image, sensor_name, capture_timestamp=None, creation_timestamp=None, is_encoded=True, detection_results=None, jpeg_quality=50)` `classmethod`

Create an ImageData instance from a numpy array.

Parameters:

Name	Type	Description	Default
`image`	`ndarray`	The image data as a numpy array.	required
`sensor_name`	`str`	The name of the sensor.	required
`capture_timestamp`	`float`	The capture timestamp. Defaults to None.	`None`
`creation_timestamp`	`float`	The creation timestamp. Defaults to None.	`None`
`is_encoded`	`bool`	Flag indicating if the image is encoded. Defaults to True. When flag is True, the function will encode the image to JPEG format.	`True`
`detection_results`	`Optional[DetectionResultsData]`	Detection results data. Defaults to None.	`None`
`jpeg_quality`	`int`	JPEG quality for encoding. Defaults to 50.	`50`

Returns:

Name	Type	Description
`ImageData`	`ImageData`	An instance of ImageData containing the image and metadata.

`to_ndarray()`

Decode JPEG bytes back into a numpy array.

`from_json(data)` `classmethod`

Deserialize a JSON string into a concrete Data subclass.

`DetectionResultsData` `dataclass`

Bases: SensorData

Container for detection outputs linked to a sensor frame.

This dataclass carries the detection result produced for an image/frame along with optional references to the originating sensor payload.

Attributes:

Name	Type	Description
`image_frame_id`	`Optional[int]`	Identifier of the image/frame the detection was produced for.
`detection_result`	`Optional[DetectionResultBase]`	Structured detection result object coming from `msight_base`. The object may include bounding boxes, classes, scores and additional metadata.
`raw_sensor_data`	`Optional[SensorData]`	Optional pointer to the raw sensor message (e.g. compressed image) that produced this detection. When present this allows consumers to fetch the original payload alongside detections.

`RoadUserListData` `dataclass`

Bases: SensorData

Holds a list of road-user points with custom serialization.

The road_user_list field contains a list of RoadUserPoint objects from msight_base. A :class:FieldCodec is attached to perform element-wise conversion to/from dicts during serialization so the list can be embedded in standard JSON/MessagePack payloads.

Attributes:

Name	Type	Description
`road_user_list`	`list[RoadUserPoint]`	List of road user points.

Notes

The module-level road_user_list_codec implements the codec used for road_user_list and is registered on the class via __field_codecs__.

`BytesData` `dataclass`

Bases: SensorData

Container for binary sensor payloads.

This dataclass stores an opaque bytes payload together with the standard sensor metadata inherited from :class:SensorData.

Attributes:

Name	Type	Description
`data`	`bytes`	Binary payload (e.g. compressed image or raw packet).

`VideoData` `dataclass`

Bases: SensorData

Container for encoded video payloads and optional metadata.

Fields

video: Optional[bytes] Encoded video bytes (e.g., MP4, H.264).

Optional[DetectionResultsData]

Optional detection results associated with the video. Use this to attach aggregated or per-video detection outputs.

Optional[List[int]]

Ordered list of frame identifiers corresponding to frames included in the encoded video. Helpful for mapping video segments back to original frame-level metadata.

Optional[List[float]]

Ordered list of capture timestamps (in seconds since epoch) for each frame included in the encoded video. Corresponds to the capture_timestamp field from individual image frames.

Notes

This container does not implement encoding/decoding helpers itself; use external utilities (e.g., OpenCV or ffmpeg wrappers) to produce encoded bytes prior to constructing a VideoData instance.

`PointCloudData` `dataclass`

Bases: SensorData

Container for structured point-cloud arrays.

Fields

points: np.ndarray A NumPy structured array containing per-point fields (e.g. x, y, z, intensity). The array is large and therefore is excluded from the repr to avoid huge console prints.

Optional[tuple]

The shape of the array (kept in sync with points.shape).

Optional[np.dtype]

The NumPy dtype describing the structure of points. When serializing, the dtype is converted to a compact JSON header via :func:pack_dtype.

int

Zstandard compression level used when encoding the raw bytes. Higher numbers increase compression ratio at the cost of CPU.

Notes

The class registers two field codecs via __field_codecs__:

points uses :data:point_cloud_codec to compress/decompress the raw bytes.
dtype uses :data:dtype_codec to emit the structured-dtype header.

Use :meth:from_ndarray to create instances from an existing NumPy array and rely on automatic shape/dtype synchronization performed in __post_init__.

`from_ndarray(points, sensor_name, capture_timestamp=None, creation_timestamp=None, compress_level=3)` `classmethod`

Create a PointCloudData from a NumPy array.

Parameters:

Name	Type	Description	Default
`points`	`ndarray`	Structured NumPy array containing point-cloud data.	required
`sensor_name`	`str`	Name of the sensor producing the data.	required
`capture_timestamp`	`float`	Optional capture timestamp (defaults to now).	`None`
`creation_timestamp`	`float`	Optional creation timestamp (defaults to now).	`None`
`compress_level`	`int`	Zstandard compression level to use on encoding.	`3`

Returns:

Name	Type	Description
`PointCloudData`	`PointCloudData`	Initialized dataclass instance.

msight_core.data

msight_core.data

Examples

Data dataclass

__init_subclass__(**kwargs)

to_dict()

from_dict(payload) classmethod

serialize()

deserialize(data) classmethod

to_json()

from_json(data) classmethod

SensorData dataclass

time property

capture_time property

creation_time property

SensorDataSequence dataclass

ImageData dataclass

decoded_image property

from_ndarray(image, sensor_name, capture_timestamp=None, creation_timestamp=None, is_encoded=True, detection_results=None, jpeg_quality=50) classmethod

to_ndarray()

from_json(data) classmethod

DetectionResultsData dataclass

RoadUserListData dataclass

BytesData dataclass

VideoData dataclass

Fields

Notes

PointCloudData dataclass

Fields

Notes

from_ndarray(points, sensor_name, capture_timestamp=None, creation_timestamp=None, compress_level=3) classmethod

`msight_core.data`

`msight_core.data`

`Data` `dataclass`

`__init_subclass__(**kwargs)`

`to_dict()`

`from_dict(payload)` `classmethod`

`serialize()`

`deserialize(data)` `classmethod`

`to_json()`

`from_json(data)` `classmethod`

`SensorData` `dataclass`

`time` `property`

`capture_time` `property`

`creation_time` `property`

`SensorDataSequence` `dataclass`

`ImageData` `dataclass`

`decoded_image` `property`

`from_ndarray(image, sensor_name, capture_timestamp=None, creation_timestamp=None, is_encoded=True, detection_results=None, jpeg_quality=50)` `classmethod`

`to_ndarray()`

`from_json(data)` `classmethod`

`DetectionResultsData` `dataclass`

`RoadUserListData` `dataclass`

`BytesData` `dataclass`

`VideoData` `dataclass`

`PointCloudData` `dataclass`

`from_ndarray(points, sensor_name, capture_timestamp=None, creation_timestamp=None, compress_level=3)` `classmethod`