runtimepy.primitives.serializable package

runtimepy.primitives.serializable package#

Submodules#

runtimepy.primitives.serializable.base module#

A module defining a base interface fore serializable objects.

class runtimepy.primitives.serializable.base.Serializable(byte_order: ByteOrder = ByteOrder.NETWORK, chain: T = None)[source]#

Bases: ABC

An interface for serializable objects.

add_to_end(chain: T, array_length: int = None) → list[T][source]#

Add a new serializable to the end of this chain.

assign(chain: T) → None[source]#

Assign a next serializable.

chain_bytes() → bytes[source]#

Get the fully encoded chain.

copy() → T[source]#

Create a copy of a serializable instance.

copy_without_chain() → T[source]#

A method for copying instances without chain references.

property end: Serializable#

Get the end of this chain.

from_stream(stream: BinaryIO, timestamp_ns: int = None) → int[source]#

Update this serializable from a stream.

length() → int[source]#

Get the full length of this chain.

length_trace(alias: str = None) → str[source]#

Get a length-tracing string for this instance.

resolve_alias(alias: str = None) → str[source]#

Resolve a possible alias string.

size: int#
to_stream(stream: BinaryIO) → int[source]#

Write this serializable to a stream.

abstractmethod update(data: bytes, timestamp_ns: int = None) → int[source]#

Update this serializable from a bytes instance.

update_chain(data: bytes, timestamp_ns: int = None) → int[source]#

Update this serializable from a bytes instance.

update_str(data: str, timestamp_ns: int = None) → int[source]#

Update this serializable from string data.

update_with(other: T, timestamp_ns: int = None) → int[source]#

Update this instance from another of the same type.

runtimepy.primitives.serializable.fixed module#

A module implementing a fixed-size bytes serializable.

class runtimepy.primitives.serializable.fixed.FixedChunk(data: bytes, chain: Serializable = None)[source]#

Bases: Serializable

A simple fixed-size serializable chunk.

update(data: bytes, timestamp_ns: int = None) → int[source]#

Update this serializable from a bytes instance.

runtimepy.primitives.serializable.framer module#

A module implementing a message framing interface for serializables.

class runtimepy.primitives.serializable.framer.SerializableFramer(instance: Serializable, mtu: int)[source]#

Bases: object

A class implementing a serializable message framer.

capture(sample: bool = True, flush: bool = False) → bytes | None[source]#

Optionally sample this struct and attempt to resolve a full or flushed frame.

elements: int#
raw: bytes#
reset() → None[source]#

Reset this framer’s state

set_mtu(mtu: int, logger: Logger | LoggerAdapter[Any] = None, protocol_overhead: int = 0) → int[source]#

Set a new maximum transmission unit for this framer.

runtimepy.primitives.serializable.prefixed module#

A module implementing a variable-size bytes serializable, using an integer primitive prefix to determine the size of the chunk portion.

class runtimepy.primitives.serializable.prefixed.PrefixedChunk(prefix: Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive, byte_order: ByteOrder = ByteOrder.NETWORK, chain: Serializable = None)[source]#

Bases: Serializable

A simple integer-prefixed chunk serializable.

classmethod create(prefix: type[Int8Primitive | Int16Primitive | Int32Primitive | Int64Primitive | Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive | HalfPrimitive | FloatPrimitive | DoublePrimitive | BooleanPrimitive] | str = 'uint16', chain: Serializable = None) → T[source]#

Create a prefixed chunk.

update(data: bytes, timestamp_ns: int = None) → int[source]#

Update this serializable from a bytes instance.

Module contents#

A module defining an interface for serializable objects.

class runtimepy.primitives.serializable.FixedChunk(data: bytes, chain: Serializable = None)[source]#

Bases: Serializable

A simple fixed-size serializable chunk.

update(data: bytes, timestamp_ns: int = None) → int[source]#

Update this serializable from a bytes instance.

class runtimepy.primitives.serializable.PrefixedChunk(prefix: Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive, byte_order: ByteOrder = ByteOrder.NETWORK, chain: Serializable = None)[source]#

Bases: Serializable

A simple integer-prefixed chunk serializable.

classmethod create(prefix: type[Int8Primitive | Int16Primitive | Int32Primitive | Int64Primitive | Uint8Primitive | Uint16Primitive | Uint32Primitive | Uint64Primitive | HalfPrimitive | FloatPrimitive | DoublePrimitive | BooleanPrimitive] | str = 'uint16', chain: Serializable = None) → T[source]#

Create a prefixed chunk.

update(data: bytes, timestamp_ns: int = None) → int[source]#

Update this serializable from a bytes instance.

class runtimepy.primitives.serializable.Serializable(byte_order: ByteOrder = ByteOrder.NETWORK, chain: T = None)[source]#

Bases: ABC

An interface for serializable objects.

add_to_end(chain: T, array_length: int = None) → list[T][source]#

Add a new serializable to the end of this chain.

assign(chain: T) → None[source]#

Assign a next serializable.

chain_bytes() → bytes[source]#

Get the fully encoded chain.

copy() → T[source]#

Create a copy of a serializable instance.

copy_without_chain() → T[source]#

A method for copying instances without chain references.

property end: Serializable#

Get the end of this chain.

from_stream(stream: BinaryIO, timestamp_ns: int = None) → int[source]#

Update this serializable from a stream.

length() → int[source]#

Get the full length of this chain.

length_trace(alias: str = None) → str[source]#

Get a length-tracing string for this instance.

resolve_alias(alias: str = None) → str[source]#

Resolve a possible alias string.

size: int#
to_stream(stream: BinaryIO) → int[source]#

Write this serializable to a stream.

abstractmethod update(data: bytes, timestamp_ns: int = None) → int[source]#

Update this serializable from a bytes instance.

update_chain(data: bytes, timestamp_ns: int = None) → int[source]#

Update this serializable from a bytes instance.

update_str(data: str, timestamp_ns: int = None) → int[source]#

Update this serializable from string data.

update_with(other: T, timestamp_ns: int = None) → int[source]#

Update this instance from another of the same type.

Contents