Instrument

Base instrument interface and background daemon support.

logger module-attribute

logger = getLogger(__name__)

Instrument

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

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

Parameters:

• name

  (str) –

  Channel-name prefix for published data.

• publishers

  (list[Publisher] | None, default: None ) –

  Publishers that receive emitted Measurement/Command data.

• background_config

  (BackgroundDaemonConfig | None, default: None ) –

  Background-daemon settings; default if omitted.

• legacy_naming

  (bool, default: False ) –

  When True, publish channels under pre-v1.0 names (e.g. main.ch1_v instead of main.ch1.voltage for PSU). Categories with no v1.0 rename (DMM, Modbus) ignore the flag. Scheduled for removal in v2.0.

• **kwargs

  –

  Default tags applied to every emitted Measurement/Command. Pass dataset_rid="<rid>" to auto-create a NominalCorePublisher (uses the on-disk 'default' Nominal credential).

name instance-attribute

name = name

legacy_naming instance-attribute

legacy_naming = legacy_naming

publishers instance-attribute

publishers = publishers or []

default_tags instance-attribute

default_tags: dict[str, str] = {}

background_interval property writable

background_interval

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

add_publisher

add_publisher(publisher: Publisher)

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

publish

publish(data: Measurement | Command, **kwargs)

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

_package_command

_package_command(
    channel: str, data: float | bool | str, timestamp: int, **kwargs
) -> Command

Build a single-channel 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

_package_measurement(
    channel: str, data: float | bool, timestamp: int, **kwargs
) -> Measurement

Build a single-channel 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

open() -> None

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

close

close() -> None

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.

add_background_daemon_function

add_background_daemon_function(method: Callable, *args, **kwargs)

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

start

start()

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

stop

stop()

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

get_channel

get_channel(
    channel_name: str,
    length: int = 1,
    wait_for_new_samples: bool = False,
    timeout: float = 10.0,
) -> Measurement

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

  (str) –

  Name of the channel to retrieve.

• length

  (int, default: 1 ) –

  Number of trailing samples to return.

• wait_for_new_samples

  (bool, default: False ) –

  Block until at least length new values arrive.

• timeout

  (float, default: 10.0 ) –

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

Raises:

• RuntimeError –

  No background buffer; start() was not called.

• ChannelNotFoundError –

  channel had no values and no data appeared before timeout. wait_for_new_samples=True and channel did not appear within timeout.

• ChannelValueTimeoutError –

  wait_for_new_samples=True and values did not arrive within timeout.

get_single_channel_value

get_single_channel_value(channel_name: str) -> float | None

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

  (str) –

  Name of the channel to retrieve.

Raises:

• RuntimeError –

  No background buffer; start() was not called.

define_background_daemon

define_background_daemon(method: Callable, *args, **kwargs)

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

publish_command

publish_command(func: Callable) -> Callable

Decorator that publishes the Command returned by an instrument method.

Channel naming is the call site's responsibility; the wrong return type raises rather than silently publishing as the wrong kind.

publish_measurement

publish_measurement(func: Callable) -> Callable

Decorator that publishes the Measurement (or list) returned by an instrument method.

Lists are published item by item. None passes through unpublished (e.g. a measurement that could not be obtained). Channel naming is the call site's responsibility.