Core API Reference

This section documents the core components of MsgCenterPy.

MessageInstance

class msgcenterpy.core.message_instance.MessageInstance(inner_data: T, message_type: MessageType, metadata: FormatMetadata | None = None)

Bases: TypeInfoProvider, ABC, Generic[T]

统一消息实例基类

property fields: FieldAccessor

__getitem__(field_name: str) → Any

支持通过下标访问字段

__setitem__(field_name: str, value: Any) → None

支持通过下标设置字段

__contains__(field_name: str) → bool

支持in操作符检查字段是否存在

__init__(inner_data: T, message_type: MessageType, metadata: FormatMetadata | None = None)

to(target_type: MessageType, **kwargs: Any) → MessageInstance[Any]

直接转换到目标类型

abstract classmethod import_from_envelope(data: MessageEnvelope, **kwargs: Any) → MessageInstance[Any]

从统一信封字典创建实例（仅接受 data 一个参数）。

abstract export_to_envelope(**kwargs: Any) → MessageEnvelope

导出为字典格式

abstract get_python_dict() → Dict[str, Any]

获取当前所有的字段名和对应的python可读值

abstract set_python_dict(value: Dict[str, Any], **kwargs: Any) → bool

设置所有字段的值

get_json_schema() → Dict[str, Any]

生成当前消息实例的JSON Schema，委托给FieldAccessor递归处理

to_ros2(type_hint: str | Type[Any], **kwargs: Any) → ROS2MessageInstance

转换到ROS2实例。传入必备的类型提示，

to_json_schema(**kwargs: Any) → JSONSchemaMessageInstance

转换到JSON Schema实例

MessageType

class msgcenterpy.core.types.MessageType(value)

Bases: Enum

Supported message types

ROS2 = 'ros2'

PYDANTIC = 'pydantic'

DATACLASS = 'dataclass'

JSON = 'json'

JSON_SCHEMA = 'json_schema'

DICT = 'dict'

YAML = 'yaml'

TypeInfo

class msgcenterpy.core.type_info.TypeInfo(field_name: str, field_path: str, standard_type: ~msgcenterpy.core.type_converter.StandardType, python_type: ~typing.Type, original_type: ~typing.Any, _outdated: bool = False, current_value: ~typing.Any | None = None, default_value: ~typing.Any | None = None, constraints: ~typing.List[~msgcenterpy.core.type_info.TypeConstraint] = <factory>, is_array: bool = False, array_size: int | None = None, _element_type_info: ~msgcenterpy.core.type_info.TypeInfo | None = None, is_object: bool = False, object_fields: ~typing.Dict[str, ~msgcenterpy.core.type_info.TypeInfo] = <factory>, metadata: ~typing.Dict[str, ~typing.Any] = <factory>)

Bases: object

Detailed type information including standard type, Python type and constraints

field_name: str

field_path: str

standard_type: StandardType

python_type: Type

original_type: Any

property outdated: bool

current_value: Any = None

default_value: Any = None

constraints: List[TypeConstraint]

is_array: bool = False

array_size: int | None = None

property element_type_info: TypeInfo | None

is_object: bool = False

object_fields: Dict[str, TypeInfo]

metadata: Dict[str, Any]

property python_value_from_standard_type: Any

add_constraint(constraint_type: ConstraintType, value: Any, description: str | None = None) → None

Add constraint

get_constraint(constraint_type: ConstraintType) → TypeConstraint | None

Get constraint of specified type

has_constraint(constraint_type: ConstraintType) → bool

Check if constraint of specified type exists

get_constraint_value(constraint_type: ConstraintType) → Any

Get value of specified constraint

validate_value(value: Any) → bool

Validate value according to constraints

to_json_schema_property(include_constraints: bool = True) → Dict[str, Any]

Convert to JSON Schema property definition

convert_value(value: Any, target_standard_type: StandardType | None = None) → Any

Convert value to current type or specified target type

get_value_info() → Dict[str, Any]

Get detailed information about current value

clone() → TypeInfo

Create deep copy of TypeInfo

__init__(field_name: str, field_path: str, standard_type: ~msgcenterpy.core.type_converter.StandardType, python_type: ~typing.Type, original_type: ~typing.Any, _outdated: bool = False, current_value: ~typing.Any | None = None, default_value: ~typing.Any | None = None, constraints: ~typing.List[~msgcenterpy.core.type_info.TypeConstraint] = <factory>, is_array: bool = False, array_size: int | None = None, _element_type_info: ~msgcenterpy.core.type_info.TypeInfo | None = None, is_object: bool = False, object_fields: ~typing.Dict[str, ~msgcenterpy.core.type_info.TypeInfo] = <factory>, metadata: ~typing.Dict[str, ~typing.Any] = <factory>) → None

FieldAccessor

class msgcenterpy.core.field_accessor.FieldAccessor(data: Any, type_info_provider: TypeInfoProvider, parent: FieldAccessor | None, field_name: str)

Bases: object

字段访问器，支持类型转换和约束验证的统一字段访问接口 只需要getitem和setitem，外部必须通过字典的方式来赋值

property parent_msg_center: FieldAccessor | None

property full_path_from_root: str

property root_accessor_msg_center: FieldAccessor

获取根访问器

property value: Any

property type_info: TypeInfo | None

__init__(data: Any, type_info_provider: TypeInfoProvider, parent: FieldAccessor | None, field_name: str)

初始化字段访问器

Parameters:

• data – 要访问的数据对象

• type_info_provider – 类型信息提供者

• parent – 父字段访问器，用于嵌套访问

• field_name – 当前访问器对应的字段名（用于构建路径）

get_sub_type_info(field_name: str) → TypeInfo | None

获取字段的类型信息，通过获取字段的accessor

__getitem__(field_name: str) → FieldAccessor

获取字段访问器，支持嵌套访问

__setitem__(field_name: str, value: Any) → None

设置字段值，支持类型转换和验证

__getattr__(field_name: str) → FieldAccessor | Any

支持通过属性访问字段，用于嵌套访问如 accessor.pose.position.x

__setattr__(field_name: str, value: Any) → None

支持通过属性设置字段值，用于嵌套赋值如 accessor.pose.position.x = 1.0

clear_cache(field_name: str | None = None) → None

失效字段相关的缓存

clear_type_info() → None

清空所有缓存

get_nested_field_accessor(path: str, separator: str = '.') → FieldAccessor

set_nested_value(path: str, value: Any, separator: str = '.') → None

get_json_schema() → Dict[str, Any]

原有的递归生成 JSON Schema 逻辑

update_from_dict(source_data: Dict[str, Any]) → None

递归更新嵌套字典

Parameters:

source_data – 源数据字典

MessageCenter

class msgcenterpy.core.message_center.MessageCenter

Bases: object

Message Center singleton class that manages all message types and instances

classmethod get_instance() → MessageCenter

Get MessageCenter singleton instance

__init__() → None

Private constructor, use get_instance() to get singleton

get_instance_class(message_type: MessageType) → Type[MessageInstance]

Get instance class for the specified message type

convert(source: MessageInstance, target_type: MessageType, override_properties: Dict[str, Any], **kwargs: Any) → MessageInstance

Convert message types

TypeConverter

class msgcenterpy.core.type_converter.TypeConverter

Bases: object

类型转换器，负责不同数据源类型之间的转换和标准化

PYTHON_TO_STANDARD = {<class 'NoneType'>: StandardType.NULL, <class 'bool'>: StandardType.BOOLEAN, <class 'bytearray'>: StandardType.BYTES, <class 'bytes'>: StandardType.BYTES, <class 'datetime.datetime'>: StandardType.DATETIME, <class 'decimal.Decimal'>: StandardType.DECIMAL, <class 'dict'>: StandardType.OBJECT, <class 'float'>: StandardType.FLOAT, <class 'int'>: StandardType.INTEGER, <class 'list'>: StandardType.ARRAY, <class 'str'>: StandardType.STRING, <class 'tuple'>: StandardType.ARRAY}

STANDARD_TO_PYTHON = {StandardType.ANY: <class 'object'>, StandardType.ARRAY: <class 'list'>, StandardType.BOOL: <class 'bool'>, StandardType.BOOLEAN: <class 'bool'>, StandardType.BOUNDED_ARRAY: <class 'list'>, StandardType.BOUNDED_SEQUENCE: <class 'list'>, StandardType.BYTE: <class 'int'>, StandardType.BYTES: <class 'bytes'>, StandardType.CHAR: <class 'str'>, StandardType.DATETIME: <class 'datetime.datetime'>, StandardType.DECIMAL: <class 'decimal.Decimal'>, StandardType.DOUBLE: <class 'float'>, StandardType.DURATION: <class 'float'>, StandardType.FLOAT: <class 'float'>, StandardType.FLOAT32: <class 'float'>, StandardType.FLOAT64: <class 'float'>, StandardType.INT16: <class 'int'>, StandardType.INT32: <class 'int'>, StandardType.INT64: <class 'int'>, StandardType.INT8: <class 'int'>, StandardType.INTEGER: <class 'int'>, StandardType.NULL: <class 'NoneType'>, StandardType.OBJECT: <class 'dict'>, StandardType.OCTET: <class 'int'>, StandardType.SEQUENCE: <class 'list'>, StandardType.STRING: <class 'str'>, StandardType.TIME: <class 'datetime.datetime'>, StandardType.UINT16: <class 'int'>, StandardType.UINT32: <class 'int'>, StandardType.UINT64: <class 'int'>, StandardType.UINT8: <class 'int'>, StandardType.UNBOUNDED_ARRAY: <class 'list'>, StandardType.UNBOUNDED_SEQUENCE: <class 'list'>, StandardType.UNKNOWN: <class 'object'>, StandardType.WCHAR: <class 'str'>, StandardType.WSTRING: <class 'str'>}

ROS2_TO_STANDARD = {'bool': StandardType.BOOL, 'boolean': StandardType.BOOLEAN, 'byte': StandardType.BYTE, 'char': StandardType.CHAR, 'double': StandardType.DOUBLE, 'duration': StandardType.DURATION, 'float': StandardType.FLOAT32, 'float32': StandardType.FLOAT32, 'float64': StandardType.FLOAT64, 'generic_bool': StandardType.BOOLEAN, 'generic_float': StandardType.FLOAT, 'generic_int': StandardType.INTEGER, 'generic_string': StandardType.STRING, 'int16': StandardType.INT16, 'int32': StandardType.INT32, 'int64': StandardType.INT64, 'int8': StandardType.INT8, 'long': StandardType.INT64, 'long double': StandardType.DOUBLE, 'long long': StandardType.INT64, 'octet': StandardType.OCTET, 'short': StandardType.INT16, 'string': StandardType.STRING, 'time': StandardType.TIME, 'uint16': StandardType.UINT16, 'uint32': StandardType.UINT32, 'uint64': StandardType.UINT64, 'uint8': StandardType.UINT8, 'unsigned long': StandardType.UINT64, 'unsigned long long': StandardType.UINT64, 'unsigned short': StandardType.UINT16, 'wchar': StandardType.WCHAR, 'wstring': StandardType.WSTRING}

JSON_SCHEMA_TO_STANDARD = {'array': StandardType.ARRAY, 'boolean': StandardType.BOOLEAN, 'integer': StandardType.INTEGER, 'null': StandardType.NULL, 'number': StandardType.FLOAT, 'object': StandardType.OBJECT, 'string': StandardType.STRING}

STANDARD_TO_JSON_SCHEMA = {StandardType.ANY: 'string', StandardType.ARRAY: 'array', StandardType.BOOL: 'boolean', StandardType.BOOLEAN: 'boolean', StandardType.BOUNDED_ARRAY: 'array', StandardType.BOUNDED_SEQUENCE: 'array', StandardType.BYTE: 'integer', StandardType.BYTES: 'string', StandardType.CHAR: 'string', StandardType.DATETIME: 'string', StandardType.DECIMAL: 'number', StandardType.DOUBLE: 'number', StandardType.DURATION: 'number', StandardType.FLOAT: 'number', StandardType.FLOAT32: 'number', StandardType.FLOAT64: 'number', StandardType.INT16: 'integer', StandardType.INT32: 'integer', StandardType.INT64: 'integer', StandardType.INT8: 'integer', StandardType.INTEGER: 'integer', StandardType.NULL: 'null', StandardType.OBJECT: 'object', StandardType.OCTET: 'integer', StandardType.SEQUENCE: 'array', StandardType.STRING: 'string', StandardType.TIME: 'string', StandardType.UINT16: 'integer', StandardType.UINT32: 'integer', StandardType.UINT64: 'integer', StandardType.UINT8: 'integer', StandardType.UNBOUNDED_ARRAY: 'array', StandardType.UNBOUNDED_SEQUENCE: 'array', StandardType.UNKNOWN: 'string', StandardType.WCHAR: 'string', StandardType.WSTRING: 'string'}

ARRAY_TYPECODE_TO_STANDARD = {'B': StandardType.UINT8, 'H': StandardType.UINT16, 'I': StandardType.UINT32, 'L': StandardType.UINT64, 'b': StandardType.INT8, 'd': StandardType.FLOAT64, 'f': StandardType.FLOAT32, 'h': StandardType.INT16, 'i': StandardType.INT32, 'l': StandardType.INT64}

ARRAY_TYPECODE_TO_PYTHON = {'B': <class 'int'>, 'H': <class 'int'>, 'I': <class 'int'>, 'L': <class 'int'>, 'b': <class 'int'>, 'd': <class 'float'>, 'f': <class 'float'>, 'h': <class 'int'>, 'i': <class 'int'>, 'l': <class 'int'>}

Python Type

classmethod python_type_to_standard(python_type: Type) → StandardType

将Python类型转换为标准类型

classmethod standard_to_python_type(standard_type: StandardType) → Type

将标准类型转换为Python类型

classmethod ros2_type_str_to_standard(ros2_type_str: str) → StandardType

将ROS2类型字符串转换为标准类型

classmethod rosidl_definition_to_standard(definition_type: Any) → StandardType

classmethod array_typecode_to_standard(typecode: str) → StandardType

将array.array的typecode转换为标准类型

classmethod json_schema_type_to_standard(json_type: str) → StandardType

将JSON Schema类型转换为标准类型

classmethod standard_type_to_json_schema_type(standard_type: StandardType) → str

将StandardType转换为JSON Schema类型字符串

classmethod convert_to_python_value_with_standard_type(value: Any, target_standard_type: StandardType) → Any

将值转换为指定的标准类型对应的Python值

Envelopes

class msgcenterpy.core.envelope.Properties

Bases: TypedDict

ros_msg_cls_path: str

ros_msg_cls_namespace: str

json_schema: Dict[str, Any]

class msgcenterpy.core.envelope.FormatMetadata

Bases: TypedDict

Additional metadata for source format, optional.

Examples: field statistics, original type descriptions, field type mappings, etc.

current_format: str

source_cls_name: str

source_cls_module: str

properties: Properties

class msgcenterpy.core.envelope.MessageEnvelope

Bases: TypedDict

Unified message envelope format.

• version: Protocol version

• format: Source format (MessageType.value)

• type_info: Type information (applicable for ROS2, Pydantic, etc.)

• content: Normalized message content (dictionary)

• metadata: Additional metadata

version: str

format: str

content: Dict[str, Any]

metadata: FormatMetadata

msgcenterpy.core.envelope.create_envelope(*, format_name: str, content: Dict[str, Any], metadata: FormatMetadata) → MessageEnvelope