NumpyInMemoryPublisher

class NumpyInMemoryPublisher(

maxlen: int,

value_dtype: Any = np.float32,

)

Bases: ChannelBufferPublisher

String-valued channels are stored in a plain deque; a float32 ring cannot hold them.

Attributes

channel_names	Return an insertion-ordered snapshot of all channel names currently in the buffer.
size_bytes	Return the current memory in bytes.

Methods

close	Close the publisher and clear all buffers with thread synchronization.
get	Return the trailing length samples for channel_name.
publish	Publish measurement data to the buffers with thread synchronization.

property size_bytes: int

Return the current memory in bytes.

property channel_names: tuple[str, ...]

Return an insertion-ordered snapshot of all channel names currently in the buffer.

Names are complete after one full daemon iteration: {name}.loop_time publishes every iteration, so call get_channel(“loop_time”, wait_for_new_samples=True) then read channel_names to ensure the complete set has arrived.

close() → None

Close the publisher and clear all buffers with thread synchronization.

get(

channel_name: str,

length: int = 1,

wait_for_new_samples: bool = False,

timeout: float = 10.0,

) → Measurement

Return the trailing length samples for channel_name.

With wait_for_new_samples=True blocks for length new samples; otherwise waits for the buffer to already hold length.

Raises:

• ChannelNotFoundError – Channel did not appear within timeout.

• ChannelValueTimeoutError – Values did not arrive within timeout.

publish(

data: Measurement | Command,

**kwargs,

)

Publish measurement data to the buffers with thread synchronization.