runtimepy.net.arbiter.struct package

Module contents

A module implementing runtime struct interfaces.

class runtimepy.net.arbiter.struct.TimestampedStruct(name: str, config: dict[str, str | int | float | bool | None | dict[str, str | int | float | bool | None] | list[str | int | float | bool | None]], markdown: str = None)

Bases: RuntimeStruct

A bast struct with a timestamp field.

init_env() → None

Initialize this sample environment.

log_level_channel: bool = False

poll() → None

Update this instance’s timestamp.

process_datagram(data: bytes) → Iterator[int]

Process an array message.

sequence: Uint16Primitive

time_keeper() → int

Get a timestamp value using a default method.

timestamp: Uint64Primitive

update_single(data: bytes) → int

Update this struct instance and return the nanosecond timestamp.

class runtimepy.net.arbiter.struct.UdpStructTransceiver(transport: DatagramTransport, protocol: UdpQueueProtocol, **kwargs)

Bases: UdpConnection, Generic[T]

A connection that can send and receive arrays of structs.

assign_app_rx(pattern: str, app: AppInfo) → T | None

Attempt to assign a receive struct to this instance.

assign_app_tx(pattern: str, app: AppInfo) → T | None

Attempt to assign a transmit struct to this instance.

assign_rx(instance: T) → None

Assign a receive struct.

assign_tx(instance: T) → None

Assign a struct to this connection.

capture(sample: bool = True, flush: bool = False) → None

Sample this struct and possibly send telemetry.

framer_tx: SerializableFramer

handle_update(timestamp_ns: int, instance: T, addr: tuple[str, int]) → None

Handle individual struct updates.

async process_datagram(data: bytes, addr: tuple[str, int]) → bool

Process an array of struct instances.

struct_kind: type[T]

struct_rx: T | None = None

struct_tx: T | None = None