InstroDMM

class InstroDMM(

name: str | None = None,

driver: DMMDriverBase | None = None,

publishers: list[Publisher] | None = None,

config: DMMConfig | dict | Path | str | None = None,

autostart: bool = False,

**kwargs,

)

Bases: Instrument

Digital multimeter instrument. Call set_measurement_function then read, or a single read_<function>.

Initialize an InstroDMM.

Provide either config or driver, not both.

Parameters:

• name

  Channel-name prefix for published data. Falls back to config.device.name when config is given.

• driver

  Concrete DMM driver; owns its own transport:

  dmm = InstroDMM(
      "main",
      driver=Agilent34401A("ASRL3::INSTR"),
  )
  

• publishers

  Publishers that receive emitted Measurement/Command data. Combined with any publishers declared in config.

• config

  A DMMConfig, a dict, or a path to a JSON config file. Its measurement block is applied through the public setters on open().

• autostart

  When True, open the connection and start background polling. Requires config with a measurement block, since polling cannot start without a measurement function.

• **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).

Attributes

background_interval	Seconds between background daemon iterations (0 = no wait).
channel_names	Return an insertion-ordered snapshot of all published channel names.

Methods

add_background_daemon_function	Append method to the daemon's call list.
add_publisher	Register a publisher to receive this instrument's Measurement/Command data.
close	Close the underlying driver and stop the daemon.
define_background_daemon	Replace all daemon functions with a single method (called with the given args).
get_channel	Return the most recent length samples for channel_name from the in-memory buffer.
get_single_channel_value	Return the most recent sample for channel_name from the in-memory buffer.
open	Open the underlying driver and apply any configured measurement state.
publish	Fan data out to every configured publisher; kwargs pass through.
read	Trigger a measurement under the configured function and return it.
read_ac_current	Read AC current (amperes), selecting the function first if needed.
read_ac_voltage	Read AC voltage (volts), selecting the function first if needed.
read_dc_current	Read DC current (amperes), selecting the function first if needed.
read_dc_voltage	Read DC voltage (volts), selecting the function first if needed.
read_four_wire_resistance	Read 4-wire resistance (ohms), selecting the function first if needed.
read_resistance	Read 2-wire resistance (ohms), selecting the function first if needed.
set_aperture_nplc	Set integration time as a number of power-line cycles.
set_aperture_seconds	Set integration time (aperture) in seconds.
set_digits	Set resolution in digits.
set_measurement_function	Configure the DMM for function (DC voltage, resistance, etc.).
set_range	Set manual range; None selects auto-range.
start	Start the background daemon thread.
stop	Signal the background daemon to stop and join it.

property background_interval

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

start() → None

Start the background daemon thread.

Raises:

ValueError – set_measurement_function has not been called.

open() → None

Open the underlying driver and apply any configured measurement state.

close() → None

Close the underlying driver and stop the daemon.

set_measurement_function(

function: MeasurementFunction,

**kwargs,

) → Command

Configure the DMM for function (DC voltage, resistance, etc.).

set_digits(n: int, **kwargs) → Command

Set resolution in digits. Requires set_measurement_function() to have been called.

set_aperture_seconds(

seconds: float,

**kwargs,

) → Command

Set integration time (aperture) in seconds. Requires set_measurement_function() first.

set_aperture_nplc(

nplc: float,

**kwargs,

) → Command

Set integration time as a number of power-line cycles.

Higher NPLC integrates over more AC noise (better DC/DCI/resistance accuracy) at the cost of measurement rate. NPLC=100 is the typical highest-accuracy setting. Requires set_measurement_function first.

set_range(

value: float | None,

**kwargs,

) → Command

Set manual range; None selects auto-range. Requires set_measurement_function first.

Publishes range_mode.cmd ("AUTO"/"MANUAL") on every call and range.cmd (float) only when a manual range is supplied — splitting them keeps each channel single-typed, which the Nominal streaming backend requires.

read(**kwargs) → Measurement

Trigger a measurement under the configured function and return it. Requires set_measurement_function first.

read_dc_voltage(**kwargs) → Measurement

Read DC voltage (volts), selecting the function first if needed.

read_ac_voltage(**kwargs) → Measurement

Read AC voltage (volts), selecting the function first if needed.

read_dc_current(**kwargs) → Measurement

Read DC current (amperes), selecting the function first if needed.

read_ac_current(**kwargs) → Measurement

Read AC current (amperes), selecting the function first if needed.

read_resistance(**kwargs) → Measurement

Read 2-wire resistance (ohms), selecting the function first if needed.

read_four_wire_resistance(

**kwargs,

) → Measurement

Read 4-wire resistance (ohms), selecting the function first if needed.

add_background_daemon_function(

method: Callable,

*args,

**kwargs,

)

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

add_publisher(

publisher: Publisher,

)

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

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 – No background buffer; start() was not called.

define_background_daemon(

method: Callable,

*args,

**kwargs,

)

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

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

  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 – 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(

channel_name: str,

) → float | str | 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

Name of the channel to retrieve.

Raises:

RuntimeError – No background buffer; start() was not called.

publish(

data: Measurement | Command,

**kwargs,

)

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

stop()

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