# ChannelBufferPublisher

## class ChannelBufferPublisher(maxlen: int)[source]

Bases: [`ABC`](https://docs.python.org/3/library/abc.html#abc.ABC "(in Python v3.15)")

Base class for channel buffer publishers that store measurements in memory.

Attributes

| [`channel_names`](https://dev.nominal.io/instro/ref/library/instro.lib.publishers.channel_buffer.ChannelBufferPublisher.md#instro.lib.publishers.channel_buffer.ChannelBufferPublisher.channel_names "instro.lib.publishers.channel_buffer.ChannelBufferPublisher.channel_names") | Return an insertion-ordered snapshot of all channel names currently in the buffer. |
| --- | --- |
| [`size_bytes`](https://dev.nominal.io/instro/ref/library/instro.lib.publishers.channel_buffer.ChannelBufferPublisher.md#instro.lib.publishers.channel_buffer.ChannelBufferPublisher.size_bytes "instro.lib.publishers.channel_buffer.ChannelBufferPublisher.size_bytes") | Return the current memory in bytes. |

Methods

| [`close`](https://dev.nominal.io/instro/ref/library/instro.lib.publishers.channel_buffer.ChannelBufferPublisher.md#instro.lib.publishers.channel_buffer.ChannelBufferPublisher.close "instro.lib.publishers.channel_buffer.ChannelBufferPublisher.close") | Close the publisher and clear all buffers with thread synchronization. |
| --- | --- |
| [`get`](https://dev.nominal.io/instro/ref/library/instro.lib.publishers.channel_buffer.ChannelBufferPublisher.md#instro.lib.publishers.channel_buffer.ChannelBufferPublisher.get "instro.lib.publishers.channel_buffer.ChannelBufferPublisher.get") | Return the trailing `length` samples for `channel_name`. |
| [`publish`](https://dev.nominal.io/instro/ref/library/instro.lib.publishers.channel_buffer.ChannelBufferPublisher.md#instro.lib.publishers.channel_buffer.ChannelBufferPublisher.publish "instro.lib.publishers.channel_buffer.ChannelBufferPublisher.publish") | Publish measurement data to the buffers with thread synchronization. |

### publish( data: Measurement | Command, **kwargs, )[source]

Publish measurement data to the buffers with thread synchronization.

### get( channel_name: str, length: int = 1, wait_for_new_samples: bool = False, timeout: float = 10.0, ) → Measurement[source]

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**](https://dev.nominal.io/instro/ref/library/instro.lib.publishers.channel_buffer.ChannelNotFoundError.md#instro.lib.publishers.channel_buffer.ChannelNotFoundError "instro.lib.publishers.channel_buffer.ChannelNotFoundError") – Channel did not appear within `timeout`.
    - [**ChannelValueTimeoutError**](https://dev.nominal.io/instro/ref/library/instro.lib.publishers.channel_buffer.ChannelValueTimeoutError.md#instro.lib.publishers.channel_buffer.ChannelValueTimeoutError "instro.lib.publishers.channel_buffer.ChannelValueTimeoutError") – Values did not arrive within `timeout`.

### close() → None[source]

Close the publisher and clear all buffers with thread synchronization.

### 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.

### abstract property size_bytes: int

Return the current memory in bytes.