DequeInMemoryPublisher#

class DequeInMemoryPublisher(maxlen: int)[source]#

Bases: ChannelBufferPublisher

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.