runtimepy.codec.protocol package#

Submodules#

runtimepy.codec.protocol.base module#

A module implementing an interface to build communication protocols.

class runtimepy.codec.protocol.base.FieldSpec(name: str, kind: str, enum: str | int | None = None, array_length: int | None = None)[source]#

Bases: NamedTuple

Information specifying a protocol field.

array_length: int | None#

Alias for field number 3

asdict() dict[str, str | int | float | bool | None | dict[str, str | int | float | bool | None] | list[str | int | float | bool | None]][source]#

Obtain a dictionary representing this instance.

enum: str | int | None#

Alias for field number 2

kind: str#

Alias for field number 1

name: str#

Alias for field number 0

class runtimepy.codec.protocol.base.ProtocolBase(enum_registry: EnumRegistry, names: NameRegistry = None, fields: BitFieldsManager = None, build: list[tuple[int, str] | FieldSpec | tuple[str, int]] = None, identifier: int = 1, byte_order: ByteOrder | str | int = ByteOrder.NETWORK, serializables: dict[str, list[Serializable]] = None, alias: str = None)[source]#

Bases: PrimitiveArray

A class for defining runtime communication protocols.

add_bit_fields(name: str, kind: type[Int8Primitive | Int16Primitive | Int32Primitive | Int64Primitive | Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive | HalfPrimitive | FloatPrimitive | DoublePrimitive | BooleanPrimitive] | str | Int8Primitive | Int16Primitive | Int32Primitive | Int64Primitive | Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive | HalfPrimitive | FloatPrimitive | DoublePrimitive | BooleanPrimitive = 'uint8') Iterator[BitFields][source]#

Add a bit-fields primitive to the protocol.

add_field(name: str, kind: type[Int8Primitive | Int16Primitive | Int32Primitive | Int64Primitive | Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive | HalfPrimitive | FloatPrimitive | DoublePrimitive | BooleanPrimitive] | str | Int8Primitive | Int16Primitive | Int32Primitive | Int64Primitive | Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive | HalfPrimitive | FloatPrimitive | DoublePrimitive | BooleanPrimitive | None = None, enum: str | int = None, serializable: Serializable = None, array_length: int = None, track: bool = True) None[source]#

Add a new field to the protocol.

add_serializable(name: str, serializable: Serializable, array_length: int = None) None[source]#

Add a serializable instance.

register_name(name: str) int[source]#

Register the field name.

set(name: str, val: int | float | bool | str, index: int = 0) None[source]#

Set a value of a field belonging to the protocol.

trace_size(logger: Logger | LoggerAdapter[Any]) None[source]#

Log a size trace.

value(name: str, resolve_enum: bool = True, index: int = 0) int | float | bool | str[source]#

Get the value of a field belonging to the protocol.

runtimepy.codec.protocol.json module#

A protocol extension for importing and exporting JSON.

class runtimepy.codec.protocol.json.JsonProtocol(enum_registry: EnumRegistry, names: NameRegistry = None, fields: BitFieldsManager = None, build: list[tuple[int, str] | FieldSpec | tuple[str, int]] = None, identifier: int = 1, byte_order: ByteOrder | str | int = ByteOrder.NETWORK, serializables: dict[str, list[Serializable]] = None, alias: str = None)[source]#

Bases: ProtocolBase

A class for defining runtime communication protocols.

export_json(resolve_enum: bool = True) dict[str, dict[str, str | int | float | bool | None | dict[str, str | int | float | bool | None] | list[str | int | float | bool | None]]][source]#

Export this protocol’s data to JSON.

classmethod import_json(data: dict[str, dict[str, str | int | float | bool | None | dict[str, str | int | float | bool | None] | list[str | int | float | bool | None]]], serializables: dict[str, list[Serializable]] = None) T[source]#

Create a bit-fields manager from JSON data.

meta_bytes(resolve_enum: bool = True) bytes[source]#

Get protocol metadata as a bytes object.

meta_str(resolve_enum: bool = True) str[source]#

Get protocol metadata as a string..

write_meta(stream: BinaryIO, resolve_enum: bool = True) int[source]#

Write protocol metadata to a stream.

Module contents#

A module implementing an interface to build communication protocols.

class runtimepy.codec.protocol.Protocol(enum_registry: EnumRegistry, names: NameRegistry = None, fields: BitFieldsManager = None, build: list[tuple[int, str] | FieldSpec | tuple[str, int]] = None, identifier: int = 1, byte_order: ByteOrder | str | int = ByteOrder.NETWORK, serializables: dict[str, list[Serializable]] = None, alias: str = None)[source]#

Bases: JsonProtocol

A class for defining runtime communication protocols.

class runtimepy.codec.protocol.ProtocolFactory[source]#

Bases: ABC

A class implementing an interface for creating runtime instances of a prototypes that are unique to the implementing class.

abstractmethod classmethod initialize(protocol: Protocol) None[source]#

Initialize this protocol.

initialized = False#
classmethod instance() Protocol[source]#

Create an instance of this factory’s protocol.

protocol: Protocol = <runtimepy.codec.protocol.Protocol object>#
classmethod singleton() Protocol[source]#

Get a shared single instance of a protocol for this class.