# Instrument

## class Instrument( name: str, publishers: list[Publisher] | None = None, background_config: BackgroundDaemonConfig | None = None, legacy_naming: bool = False, **kwargs, )[source]

Bases: [`object`](https://docs.python.org/3/builtins/functions.html#object "(in Python v3.15)")

Base class for Nominal instrument interfaces. Owns publishing and the background daemon.

Initialize an Instrument.

Parameters:
:   - **name**

      Channel-name prefix for published data.
    - **publishers**

      Publishers that receive emitted Measurement/Command data.
    - **background_config**

      Background-daemon settings; default if omitted.
    - **legacy_naming**

      When True, publish channels under their pre-v1.0 names
      (e.g. `main.ch1_v` instead of `main.ch1.voltage` for PSU) for
      dashboards or recorded datasets keyed on those names. The setting is
      all-or-nothing per instrument. PSU, ELoad, I2C, InstroScope, and DAQ
      digital channels honor the flag; the categories with no v1.0 rename
      (DMM, Modbus, DAQ analog and relay channels) ignore it. Scheduled for
      removal in v2.0.
    - ****kwargs**

      Default tags applied to every emitted
      [`Measurement`](https://dev.nominal.io/instro/ref/library/instro.lib.types.Measurement.md#instro.lib.types.Measurement "instro.lib.types.Measurement") or
      [`Command`](https://dev.nominal.io/instro/ref/library/instro.lib.types.Command.md#instro.lib.types.Command "instro.lib.types.Command").
      Pass `dataset_rid="<rid>"` to auto-create a NominalCorePublisher
      (uses the on-disk ‘default’ Nominal credential).

Attributes

| [`background_interval`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument.background_interval "instro.lib.instrument.Instrument.background_interval") | Seconds between background daemon iterations (0 = no wait). |
| --- | --- |
| [`channel_names`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument.channel_names "instro.lib.instrument.Instrument.channel_names") | Return an insertion-ordered snapshot of all published channel names. |

Methods

| [`add_background_daemon_function`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument.add_background_daemon_function "instro.lib.instrument.Instrument.add_background_daemon_function") | Append `method` to the daemon's call list. |
| --- | --- |
| [`add_publisher`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument.add_publisher "instro.lib.instrument.Instrument.add_publisher") | Register a publisher to receive this instrument's Measurement/Command data. |
| [`close`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument.close "instro.lib.instrument.Instrument.close") | Stop the background daemon and close every publisher. |
| [`define_background_daemon`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument.define_background_daemon "instro.lib.instrument.Instrument.define_background_daemon") | Replace all daemon functions with a single `method` (called with the given args). |
| [`get_channel`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument.get_channel "instro.lib.instrument.Instrument.get_channel") | Return the most recent `length` samples for `channel_name` from the in-memory buffer. |
| [`get_single_channel_value`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument.get_single_channel_value "instro.lib.instrument.Instrument.get_single_channel_value") | Return the most recent sample for `channel_name` from the in-memory buffer. |
| [`open`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument.open "instro.lib.instrument.Instrument.open") | No-op base hook. |
| [`publish`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument.publish "instro.lib.instrument.Instrument.publish") | Fan `data` out to every configured publisher; `kwargs` pass through. |
| [`start`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument.start "instro.lib.instrument.Instrument.start") | Start the background daemon thread. |
| [`stop`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument.stop "instro.lib.instrument.Instrument.stop") | Signal the background daemon to stop and join it. |
| [`_package_command`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument._package_command "instro.lib.instrument.Instrument._package_command") | Build a single-channel [`Command`](https://dev.nominal.io/instro/ref/library/instro.lib.types.Command.md#instro.lib.types.Command "instro.lib.instrument.Command") namespaced under this instrument. |
| [`_package_measurement`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument._package_measurement "instro.lib.instrument.Instrument._package_measurement") | Build a single-channel [`Measurement`](https://dev.nominal.io/instro/ref/library/instro.lib.types.Measurement.md#instro.lib.types.Measurement "instro.lib.instrument.Measurement") namespaced under this instrument. |

### add_publisher( publisher: Publisher, )[source]

Register a publisher to receive this instrument’s Measurement/Command data.

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

Fan `data` out to every configured publisher; `kwargs` pass through.

### _package_command( channel: str, data: float | bool | str, timestamp: int, **kwargs, ) → Command[source]

Build a single-channel [`Command`](https://dev.nominal.io/instro/ref/library/instro.lib.types.Command.md#instro.lib.types.Command "instro.lib.instrument.Command") namespaced under this instrument.

The published channel key is `{self.name}.{channel}`. The caller writes the
full descriptor including any `.cmd` suffix, so the literal published name
appears at the call site (e.g. `f"ch{channel}.voltage.cmd"`).

### _package_measurement( channel: str, data: float | bool | str, timestamp: int, **kwargs, ) → Measurement[source]

Build a single-channel [`Measurement`](https://dev.nominal.io/instro/ref/library/instro.lib.types.Measurement.md#instro.lib.types.Measurement "instro.lib.instrument.Measurement") namespaced under this instrument.

The published channel key is `{self.name}.{channel}`. The caller writes the
full descriptor, so the literal published name appears at the call site.

### open() → None[source]

No-op base hook. Category subclasses override to open their transport.

### close() → None[source]

Stop the background daemon and close every publisher.

Category subclasses override to also close their transport, calling
`super().close()` to keep daemon and publisher cleanup running.

### property background_interval

Seconds between background daemon iterations (0 = no wait).

### add_background_daemon_function( method: Callable, *args, **kwargs, )[source]

Append `method` to the daemon’s call list. Use `define_background_daemon` to replace instead.

### start()[source]

Start the background daemon thread. No-op if already running.

### stop()[source]

Signal the background daemon to stop and join it. No-op if not running.

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

Return the most recent `length` samples for `channel_name` from the in-memory buffer.

If the channel does not exist yet, or no sample is available, the code will always block until `timeout` expires.

Parameters:
:   - **channel_name**

      Name of the channel to retrieve.
    - **length**

      Number of trailing samples to return.
    - **wait_for_new_samples**

      Block until at least `length` new values arrive.
    - **timeout**

      Seconds to wait when insufficient data exists or `wait_for_new_samples=True`.

Raises:
:   - [**RuntimeError**](https://docs.python.org/3/builtins/exceptions.html#RuntimeError "(in Python v3.15)") – No background buffer; `start()` was not called.
    - [**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 had no values and no data appeared before `timeout`.
      `wait_for_new_samples=True` and 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") – `wait_for_new_samples=True` and values did not arrive within `timeout`.

### get_single_channel_value( channel_name: str, ) → float | str | None[source]

Return the most recent sample for `channel_name` from the in-memory buffer.

This will not wait, if data is not available then `None` is returned.

Parameters:
:   **channel_name**

    Name of the channel to retrieve.

Raises:
:   [**RuntimeError**](https://docs.python.org/3/builtins/exceptions.html#RuntimeError "(in Python v3.15)") – No background buffer; `start()` was not called.

### property channel_names: tuple[str, ...]

Return an insertion-ordered snapshot of all published channel names.

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.

Raises:
:   [**RuntimeError**](https://docs.python.org/3/builtins/exceptions.html#RuntimeError "(in Python v3.15)") – No background buffer; `start()` was not called.

### define_background_daemon( method: Callable, *args, **kwargs, )[source]

Replace all daemon functions with a single `method` (called with the given args).