`msight_core.data.base`

`FieldCodec`

Wrapper for per-field encode/decode functions.

This helper class provides an encoder/decoder pair that can be attached to a dataclass field via the __field_codecs__ mapping on a Data subclass, to customize serialization and deserialization of specific fields.

Attributes:

Name	Type	Description
`encode`		Callable that converts a Python value into a JSON/msgpack-safe representation. If `context` is True the function will receive the parent object as a second argument.
`decode`		Callable that converts an encoded value back into a Python object. If `context` is True the function will receive the raw payload (or parent) as a second argument.
`context`		Whether encoder/decoder expect (value, context) instead of single-argument call signature.

`init(encode, decode, context=False)`

Create a FieldCodec.

Parameters:

Name	Type	Description	Default
`encode`	`Callable[[Any], Any]`	Function(value) or Function(value, context) that returns a serializable representation of `value`.	required
`decode`	`Callable[[Any], Any]`	Function(encoded) or Function(encoded, context) that returns the original Python value.	required
`context`	`bool`	If True, pass contextual information (parent object or payload) as a second argument to encode/decode.	`False`

`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.

`get_class_path(cls)`

Return the fully-qualified class path used as a type tag.

The returned string is used as the data_type tag embedded in serialized payloads. It is intentionally the module + qualname so the tag remains stable even if the class is moved between modules during refactors.

Parameters:

Name	Type	Description	Default
`cls`	`type`	Class object to create a tag for.	required

Returns:

Type	Description
`str`	A string of the form `{module}.{qualname}`.

`generate_frame_id()`

Generate a unique frame identifier.

Uses a Snowflake-like generator to produce monotonically-increasing 64-bit IDs suitable for frame identification across processes. The worker id is derived from the current process id to reduce collision probability when multiple processes are running on the same host.

Returns:

Type	Description
`int`	An integer representing a unique frame id.

msight_core.data.base