Looking to add a new vendor or model to a supported instrument type?Jump to that category’s driver-development section below: Oscilloscope (Scope), Power Supply (PSU), Digital Multimeter (DMM), Data Acquisition (DAQ), Arbitrary Waveform Generator (AWG), Electronic Load (ELoad), Flow Controller, or I2C.If your instrument doesn’t fit any supported type at all, see Generic Custom Instrumentation below instead — that’s for building an entirely new instrument type using Instrument.

Driver Development

Oscilloscope (Scope)

Overview

This section is for developers implementing InstroScope support for oscilloscopes that aren’t supported out of the box. A scope driver subclasses ScopeDriverBase and owns its transport. Its responsibilities:

Expose a protocol-native constructor: accept a visa_resource string or VisaConfig.
Own transport setup: create and store the transport internally. Do not require callers to pass a VisaDriver.
Own lifecycle: implement open() and close().
Map commands: translate each abstract method into vendor commands.
Parse responses: convert responses to the expected types (float, enums, WaveformData).
Drain the error queue: implement check_errors(). InstroScope calls it between setup commands and blocking queries, so a pending syntax error raises instead of hanging a data query.

Channels are 1-indexed analog input numbers throughout.

ScopeDriverBase Interface

All scope drivers subclass ScopeDriverBase and implement these abstract methods:

def open(self) -> None: ...
def close(self) -> None: ...
def check_errors(self) -> None: ...

# Channel vertical settings
def set_vertical_scale(self, volts_per_div: float, channel: int) -> None: ...
def get_vertical_scale(self, channel: int) -> float: ...
def set_vertical_offset(self, offset: float, channel: int) -> None: ...
def get_vertical_offset(self, channel: int) -> float: ...
def set_coupling(self, coupling: Coupling, channel: int) -> None: ...
def get_coupling(self, channel: int) -> Coupling: ...
def set_probe_attenuation(self, factor: float, channel: int) -> None: ...
def get_probe_attenuation(self, channel: int) -> float: ...

# Horizontal (timebase) and sample rate
def set_horizontal_scale(self, seconds_per_div: float) -> None: ...
def get_horizontal_scale(self) -> float: ...
def get_sample_rate(self) -> float: ...

# Acquisition
def set_acquisition_mode(self, mode: AcquisitionMode) -> None: ...
def get_acquisition_mode(self) -> AcquisitionMode: ...
def set_average_count(self, count: int) -> None: ...
def get_average_count(self) -> int: ...
def run(self) -> None: ...
def stop(self) -> None: ...
def single(self) -> None: ...
def digitize(self, timeout: float) -> None: ...
def get_acquisition_state(self) -> AcquisitionState: ...

# Waveform data and measurements
def fetch_waveform(self, channel: int) -> WaveformData: ...
def measure(self, measurement_type: ScopeMeasurementType, channel: int) -> float: ...

# Trigger
def set_trigger_source(self, channel: int) -> None: ...
def set_trigger_type(self, trigger_type: TriggerType) -> None: ...
def set_trigger_level(self, level: float) -> None: ...
def set_trigger_slope(self, slope: TriggerSlope) -> None: ...
def set_trigger_mode(self, mode: TriggerMode) -> None: ...
def force_trigger(self) -> None: ...
def get_trigger_status(self) -> TriggerStatus: ...

# File operations
def save_screenshot(self, filepath: str, to_instrument: bool = False) -> bytes: ...
def save_settings(self, name: str, to_instrument: bool = False) -> bytes: ...
def load_settings(self, name: str, from_instrument: bool = False) -> None: ...

setup_measurement(measurement_type, channel) is an optional override (default no-op). Implement it for instruments that compute measurements during acquisition (e.g. Tektronix), where the measurement slot must exist at trigger time. Raise NotImplementedError for AcquisitionMode values the scope does not support (e.g. the Keysight 1200X has no ENVELOPE mode). See Exceptions for unsupported-feature handling. For VISA-backed drivers, create a VisaDriver internally and use self._visa.write(command) and self._visa.query(command) for all I/O. VisaDriver owns the resource lock and serializes concurrent calls. See the VisaDriver guide for the full transport reference.

Implementation Example: Keysight 1200 X-Series Driver

from __future__ import annotations

import math

from instro.lib.transports.visa import VisaConfig, VisaDriver
from instro.scope import ScopeDriverBase

# IEEE-488.2 sentinel returned when a measurement has no valid result.
_VENDOR_INVALID_MEASUREMENT = 9.91e37


class Keysight1200X(ScopeDriverBase):
    """SCPI driver for Keysight InfiniiVision 1200 X-Series oscilloscopes."""

    def __init__(self, visa_resource: str | VisaConfig) -> None:
        self._visa = VisaDriver(visa_resource)
        self._trigger_source: int | None = None

    def open(self) -> None:
        self._visa.open()

    def close(self) -> None:
        self._visa.close()

    def check_errors(self) -> None:
        while True:
            resp = self._visa.query(":SYSTem:ERRor?")
            parts = resp.split(",", 1)
            code = int(parts[0])
            if code == 0:
                return
            msg = parts[1].strip().strip('"') if len(parts) > 1 else "Unknown error"
            raise RuntimeError(f"Keysight SCPI error {code}: {msg}")

    def set_vertical_scale(self, volts_per_div: float, channel: int) -> None:
        self._visa.write(f":CHANnel{channel}:SCALe {volts_per_div}")

    def get_vertical_scale(self, channel: int) -> float:
        return float(self._visa.query(f":CHANnel{channel}:SCALe?"))

    # ...remaining channel, timebase, acquisition, trigger, and file methods...

Power Supply (PSU)

This section is for developers implementing InstroPSU support for power supplies that aren’t supported out of the box.

Overview

Driver developers subclass PSUDriverBase and own whatever transport their instrument needs. The caller chooses a concrete driver, and that concrete driver exposes connection parameters that make sense for its protocol:

psu = InstroPSU(
    name="labPSU",
    driver=MyVendorPSU(host="10.0.0.42", unit_id=1),
    num_channels=1,
)

The driver is responsible for translating InstroPSU’s vendor-independent API (set_voltage, get_voltage, output_enable, …) into vendor-specific commands.

Driver Responsibilities

A PSU driver must:

Expose a protocol-native constructor: accept inputs like visa_resource, host, port, unit_id, interface, or node_id, depending on the instrument.
Own transport setup: create and store the transport internally. Do not require users to pass a VisaDriver, socket client, Modbus client, or other transport object.
Own lifecycle: implement open() and close() by opening and closing the underlying transport.
Map commands: translate each abstract method into vendor-specific commands.
Parse responses: convert instrument responses to the expected Python types (float, bool, etc.).

PSUDriverBase Interface

All PSU drivers subclass PSUDriverBase and implement these abstract methods:

def open(self) -> None:
    """Open the driver's underlying transport and perform any vendor handshake."""

def close(self) -> None:
    """Close the driver's underlying transport."""

def set_voltage(self, voltage: float, channel: int) -> None:
    """Set the output voltage (volts) on `channel`."""

def get_voltage(self, channel: int) -> float:
    """Query and return the measured output voltage in volts. Output may vary from voltage setpoint outside of constant voltage mode."""

def set_current_limit(self, current_limit: float, channel: int) -> None:
    """Set the current limit (amperes) on `channel`."""

def get_current(self, channel: int) -> float:
    """Query and return the measured output current in amperes. Output may vary from current-limit setpoint outside of constant current mode."""

def output_enable(self, enable: bool, channel: int) -> None:
    """Enable or disable the output on `channel`."""

def get_output_status(self, channel: int) -> bool:
    """Query and return the output enable status on `channel`."""

def get_voltage_setpoint(self, channel: int) -> float:
    """Query and return the configured voltage setpoint in volts (not the measured output). Output may vary from actual measured voltage outside of constant voltage mode."""

def get_current_setpoint(self, channel: int) -> float:
    """Query and return the configured current-limit setpoint in amperes (not the measured output). Output may vary from actual measured current outside of constant current mode."""

def get_operating_mode(self, channel: int) -> OperatingMode:
    """Query whether `channel` is regulating in constant voltage, constant current, or off."""

def set_overvoltage_protection_level(self, voltage: float, channel: int) -> None:
    """Set the overvoltage protection threshold (volts) on `channel`."""

def get_overvoltage_protection_level(self, channel: int) -> float:
    """Query and return the overvoltage protection threshold in volts."""

def set_overvoltage_protection_enabled(self, enabled: bool, channel: int) -> None:
    """Enable or disable overvoltage protection on `channel`."""

def get_overvoltage_protection_enabled(self, channel: int) -> bool:
    """Query and return whether overvoltage protection is enabled."""

def set_overvoltage_protection_delay(self, delay: float, channel: int) -> None:
    """Set the overvoltage protection trip delay (seconds) on `channel`."""

def get_overvoltage_protection_delay(self, channel: int) -> float:
    """Query and return the overvoltage protection trip delay in seconds."""

def set_overcurrent_protection_level(self, current: float, channel: int) -> None:
    """Set the overcurrent protection threshold (amperes) on `channel`."""

def get_overcurrent_protection_level(self, channel: int) -> float:
    """Query and return the overcurrent protection threshold in amperes."""

def set_overcurrent_protection_enabled(self, enabled: bool, channel: int) -> None:
    """Enable or disable overcurrent protection on `channel`."""

def get_overcurrent_protection_enabled(self, channel: int) -> bool:
    """Query and return whether overcurrent protection is enabled."""

def set_remote_sense_enabled(self, enabled: bool, channel: int) -> None:
    """Enable or disable remote sense on `channel`."""

def get_remote_sense_enabled(self, channel: int) -> bool:
    """Query and return the remote sense enable status on `channel`."""

Required output-control and measurement methods on PSUDriverBase are declared with @abc.abstractmethod. Optional protection and sense methods default to NotImplementedError; override the ones the driver implements. See Exceptions for unsupported-feature handling.

Talking to the Instrument

Concrete drivers should hide transport details behind private attributes. For VISA-backed drivers, create a VisaDriver internally and use it for all I/O:

self._visa.write(command): Send a SCPI command (no response expected).
self._visa.query(command): Send a SCPI query and receive the response string.

VisaDriver owns the resource lock. Concurrent write / query calls against the same driver are serialized automatically. Use with self._visa.lock(): when a write and its error check need to execute atomically. See the VisaDriver guide for the full transport reference, covering configuration, terminators, timeouts, serial settings, and the raw-byte I/O path. For non-VISA instruments, follow the same shape with the protocol client your driver needs. The important part is that the public constructor describes the instrument connection, while the transport object remains an implementation detail.

Implementation Example: B&K Precision Single-Channel Driver

Here’s a representative driver implementation for the B&K Precision 9115-series single-channel supplies:

from instro.psu import PSUDriverBase
from instro.lib.transports import VisaConfig, VisaDriver


class BK9115(PSUDriverBase):
    """SCPI mapping for the B&K Precision 9115-series single-channel power supplies."""

    def __init__(self, visa_resource: str | VisaConfig) -> None:
        self._visa = VisaDriver(visa_resource)

    def open(self) -> None:
        self._visa.open()

    def close(self) -> None:
        self._visa.close()

    def set_voltage(self, voltage: float, channel: int) -> None:
        self._write_checked(f"VOLT {voltage:.3f}")

    def get_voltage(self, channel: int) -> float:
        return self._query_checked_float("MEAS:VOLT?")

    def set_current_limit(self, current_limit: float, channel: int) -> None:
        self._write_checked(f"CURR {current_limit:.3f}")

    def get_current(self, channel: int) -> float:
        return self._query_checked_float("MEAS:CURR?")

    def output_enable(self, enable: bool, channel: int) -> None:
        self._write_checked("OUTP:STAT ON" if enable else "OUTP:STAT OFF")

    def get_output_status(self, channel: int) -> bool:
        with self._visa.lock():
            resp = self._visa.query("OUTP:STAT?")
            self._check_errors()
        return resp == "1"

    def _write_checked(self, command: str) -> None:
        with self._visa.lock():
            self._visa.write(command)
            self._check_errors()

    def _query_checked_float(self, command: str) -> float:
        with self._visa.lock():
            value = self._visa.query(command)
            self._check_errors()
            return float(value)

    def _check_errors(self) -> None:
        err = self._visa.query("SYST:ERR?")
        if not err.startswith("0"):
            raise RuntimeError(f"BK PSU reported error: {err}")

Vendor SCPI VariationsDifferent PSU vendors use different SCPI command sets:

Siglent: CH<n>:VOLT, MEAS:VOLT? CH<n>, SYST:ERR? (positive-zero prefix)
Keysight E36100: VOLT, MEAS:VOLT?, SYST:ERR? (single channel)
Rigol: :SOUR<n>:VOLT, :MEAS:VOLT? CH<n>, :SYST:ERR?
TDK Lambda Genesys / white-label Agilent/Keysight N5700: VOLT, MEAS:VOLT?, SYSTEM:ERROR? (single channel)

Always consult your PSU’s programming manual for the correct SCPI syntax.

Using a Custom Driver

For drivers that aren’t shipped in the library, construct InstroPSU with your own driver instance. The driver should accept connection settings directly and create its transport internally:

from instro.psu import InstroPSU
from instro.psu import PSUDriverBase
from instro.lib.transports import VisaDriver


class MyCustomPSUDriver(PSUDriverBase):
    """Custom driver for my lab's proprietary PSU."""

    def __init__(self, visa_resource: str) -> None:
        self._visa = VisaDriver(visa_resource)

    def open(self) -> None:
        self._visa.open()

    def close(self) -> None:
        self._visa.close()

    def set_voltage(self, voltage: float, channel: int) -> None:
        self._visa.write(f"VOLTAGE {voltage}")

    # ... implement other required methods ...

    def get_output_status(self, channel: int) -> bool:
        return self._visa.query("OUTP?") == "ON"


psu = InstroPSU(
    name="labPSU",
    driver=MyCustomPSUDriver(visa_resource="<VISA_ADDRESS>"),
    num_channels=1,
)

psu.open()
psu.set_voltage(12.0, channel=1)
psu.close()

Summary

Driver development requires careful mapping of vendor-specific behavior to the unified InstroPSU interface. Focus on:

Subclassing PSUDriverBase
Designing a constructor around natural connection parameters for the instrument
Hiding transport construction inside the driver
Implementing all abstract methods on PSUDriverBase
Using the correct vendor protocol or command syntax

Digital Multimeter (DMM)

This section is for developers implementing InstroDMM support for DMMs that aren’t supported out of the box.

Overview

Driver developers subclass DMMDriverBase and own whatever transport their instrument needs. The caller chooses a concrete driver, and that concrete driver exposes connection parameters that make sense for its protocol:

dmm = InstroDMM(
    name="labDMM",
    driver=MyVendorDMM(host="10.0.0.42"),
)

The driver is responsible for translating InstroDMM’s vendor-independent API (set_measurement_function, measure_dc_voltage, …) into vendor-specific commands.

Driver Responsibilities

A DMM driver must:

Expose a protocol-native constructor: accept inputs like visa_resource, host, port, depending on the instrument.
Own transport setup: create and store the transport internally. Do not require users to pass a VisaDriver, socket client, or other transport object.
Own lifecycle: implement open() and close() by opening and closing the underlying transport.
Map commands: translate each abstract method into vendor-specific commands.
Parse responses: convert instrument responses to the expected Python types (float).

DMMDriverBase Interface

All DMM drivers subclass DMMDriverBase and implement these abstract methods:

def open(self) -> None:
    """Open the driver's underlying transport and perform any vendor handshake."""

def close(self) -> None:
    """Close the driver's underlying transport."""

def set_measurement_function(self, function: MeasurementFunction) -> None:
    """Configure the active measurement function."""

def measure_dc_voltage(self) -> float: ...
def measure_ac_voltage(self) -> float: ...
def measure_dc_current(self) -> float: ...
def measure_ac_current(self) -> float: ...
def measure_resistance(self) -> float: ...

Optional overrides (raise NotImplementedError if not supported):

set_digits(n): Set resolution in digits.
set_aperture_seconds(seconds): Set integration time in seconds. (Not implemented by the bundled drivers.)
Per-function range setters: set_dc_voltage_range, set_ac_voltage_range, set_dc_current_range, set_ac_current_range, set_two_wire_resistance_range, set_four_wire_resistance_range. Each takes value: float | None (None = auto). InstroDMM dispatches to the right one based on the active measurement function, so the driver never needs to know which function is active.
Per-function NPLC setters: set_dc_voltage_nplc, set_ac_voltage_nplc, set_dc_current_nplc, set_ac_current_nplc, set_two_wire_resistance_nplc, set_four_wire_resistance_nplc. Same dispatch shape as the range setters.
measure_four_wire_resistance(): Measure 4-wire resistance.

The per-function range/NPLC split keeps InstroDMM’s _measurement_config.function as the single source of truth. The driver never receives or tracks the active function, so the wrong-function-passed-by-mistake class of bug doesn’t exist.

Talking to the Instrument

Concrete drivers should hide transport details behind private attributes. For VISA-backed drivers, create a VisaDriver internally and use it for all I/O:

self._visa.write(command): Send a SCPI command (no response expected).
self._visa.query(command): Send a SCPI query and receive the response string.

VisaDriver owns the resource lock. Concurrent write / query calls against the same driver are serialized automatically. See the VisaDriver guide for the full transport reference, covering configuration, terminators, timeouts, serial settings, and the raw-byte I/O path. For non-VISA instruments, follow the same shape with the protocol client your driver needs. The important part is that the public constructor describes the instrument connection, while the transport object remains an implementation detail.

Implementation Example: Agilent 34401A Driver

Here is the abridged shape of the Agilent 34401A driver:

from instro.dmm import DMMDriverBase
from instro.dmm.types import MeasurementFunction
from instro.lib.transports import VisaConfig, VisaDriver


class Agilent34401A(DMMDriverBase):
    """SCPI mapping for the Agilent/HP/Keysight 34401A DMM."""

    def __init__(self, visa_resource: str | VisaConfig) -> None:
        self._visa = VisaDriver(visa_resource)
        self._range: float | None = None
        self._resolution: float | None = None

    def open(self) -> None:
        self._visa.open()
        self._visa.write("*CLS")
        self._visa.write("SYST:REM")

    def close(self) -> None:
        self._visa.close()

    def set_measurement_function(self, function: MeasurementFunction) -> None:
        # Dispatch to the vendor-specific measurement; the 34401A configures via MEAS commands.
        ...

    def measure_dc_voltage(self) -> float:
        return float(self._visa.query("MEAS:VOLT:DC?").strip())

    # The 34401A uses one shared range cache regardless of function, so every
    # per-function range setter delegates to the same private slot.
    def _store_range(self, value: float | None) -> None:
        self._range = value

    set_dc_voltage_range = _store_range
    set_ac_voltage_range = _store_range
    set_dc_current_range = _store_range
    set_ac_current_range = _store_range
    set_two_wire_resistance_range = _store_range
    set_four_wire_resistance_range = _store_range

    # ...other measure_* methods, set_digits, _check_errors...

Vendor SCPI VariationsDifferent DMM vendors use different SCPI command sets:

Agilent 34401A: MEAS:VOLT:DC?, MEAS:CURR:DC?
Keithley 2400: :SENS:FUNC 'VOLT', :READ?

Always consult your DMM’s programming manual for the correct SCPI syntax.

Using a Custom Driver

For drivers that aren’t shipped in the library, construct InstroDMM with your own driver instance. The driver should accept connection settings directly and create its transport internally:

from instro.dmm import MeasurementFunction, InstroDMM
from instro.dmm import DMMDriverBase
from instro.lib.transports import VisaDriver


class MyCustomDMMDriver(DMMDriverBase):
    """Custom driver for my lab's proprietary DMM."""

    def __init__(self, visa_resource: str) -> None:
        self._visa = VisaDriver(visa_resource)

    def open(self) -> None:
        self._visa.open()

    def close(self) -> None:
        self._visa.close()

    def set_measurement_function(self, function: MeasurementFunction) -> None:
        self._visa.write(f"FUNC {function.value}")

    def measure_dc_voltage(self) -> float:
        return float(self._visa.query("MEAS:VOLT?"))

    # ... implement other required methods ...

    def _check_errors(self) -> None:
        err = self._visa.query("ERR?")
        if err != "OK":
            raise RuntimeError(f"DMM error: {err}")


dmm = InstroDMM(
    name="labDMM",
    driver=MyCustomDMMDriver(visa_resource="<VISA_ADDRESS>"),
)

dmm.open()
dmm.set_measurement_function(MeasurementFunction.DC_VOLTAGE)
measurement = dmm.read()
dmm.close()

Summary

Driver development requires careful mapping of vendor-specific behavior to the unified InstroDMM interface. Focus on:

Subclassing DMMDriverBase
Designing a constructor around natural connection parameters for the instrument
Hiding transport construction inside the driver
Implementing all abstract methods on DMMDriverBase (and optional overrides where supported)
Using the correct vendor protocol or command syntax

Data Acquisition (DAQ)

This section is for developers implementing InstroDAQ support for DAQ devices / vendors that are not supported out of the box.

Overview

Driver developers implement the DAQDriverBase abstract interface to add support for new DAQ vendors. The driver is responsible for translating InstroDAQ’s vendor-independent API calls into vendor-specific hardware operations, ensuring users get consistent behavior regardless of the underlying hardware (within its capabilities).

Driver Responsibilities

A DAQ driver shall:

Hardware connection lifecycle: Implement open() and close() for establishing and terminating hardware connections.
Channel and state tracking: The driver is the single source of truth for what’s configured. DAQDriverBase.__init__ initializes the private dicts (_ai_channels, _ao_channels, _di_channels, _do_channels, _relay_channels) and the private timing-config slots (_ai_hw_timing_config, plus AO/DI/DO slots); concrete drivers call super().__init__(), then populate those privates inside their configure_* methods and read from them in read_analog, fetch_analog, start, write_analog_value, etc. InstroDAQ exposes the same state to end users via read-only @property accessors that hand back frozen snapshots (daq.ai_channels, daq.ai_hw_timing_config, …) — no duplication. See Driver-owned state below.
Channel configuration: Translate generic channel configurations into vendor-specific setup.
Timing abstraction: Implement both software-timed reads and hardware-timed buffered acquisition. InstroDAQ manages the background daemon for both timing modes; software-timed pacing never reaches the driver.
Data format conversion: Convert vendor-specific data formats to Measurement objects.
Constraint validation: Handle vendor-specific hardware constraints and raise clear exceptions when the vendor library cannot.

DAQDriverBase Interface

All drivers must implement the DAQDriverBase abstract base class. Key methods:

Connection Management

open(): Establish connection to the hardware device
close(): Disconnect from the hardware device

Channel Configuration

Each of the methods below must (a) program the device and (b) record the resulting channel on the driver’s own private dict (self._ai_channels, self._di_channels, etc.) so later calls (read_analog, start, etc.) can find it. See Driver-owned state for the full rationale.

configure_di_line_channel(physical_channel, logic, ...): Parse, program, and register a digital input line. Record on self._di_channels[channel.alias].
configure_do_line_channel(physical_channel, logic, ...): Parse, program, and register a digital output line. Record on self._do_channels[channel.alias].
configure_di_port_channel(physical_channel, logic, port_width, ...): Parse, program, and register a digital input port. Record on self._di_channels[channel.alias]. Default raises NotImplementedError for line-only devices.
configure_do_port_channel(physical_channel, logic, port_width, ...): Parse, program, and register a digital output port. Record on self._do_channels[channel.alias]. Default raises NotImplementedError for line-only devices.

The analog methods below back InstroDAQ’s typed channel configuration; the support matrix there lists per-driver support. configure_ai_voltage_channel is required; the others default to NotImplementedError for hardware that lacks the capability.

configure_ai_voltage_channel(channel) / configure_ao_voltage_channel(channel): Configure a typed voltage input/output channel from an AnalogVoltageChannel; record on self._ai_channels / self._ao_channels.
configure_ai_current_channel(channel) / configure_ao_current_channel(channel): Configure a typed current input/output channel from an AnalogCurrentChannel; record on self._ai_channels / self._ao_channels.
configure_ai_thermocouple_channel(channel): Configure a thermocouple input channel from an AnalogThermocoupleChannel (thermocouple type plus cold-junction config); record on self._ai_channels.
configure_ai_channel(channel) / configure_ao_channel(channel): Deprecated. Implement configure_ai_voltage_channel / configure_ao_voltage_channel instead. Both default to NotImplementedError, so a new driver leaves them alone.

Timing Configuration

configure_ai_hw_timing(hw_timing_config): Program hardware-timed sampling on the device, then record on self._ai_hw_timing_config so start/fetch_analog can read it back.

Acquisition Control

start(): Begin hardware-timed data acquisition
- Start hardware sampling and buffer filling
stop(): End hardware-timed acquisition
- Stop sampling and flush buffers

Data Operations

read_analog() -> Any: Perform software-timed analog read
- Trigger immediate conversion and return data
fetch_analog() -> Any: Fetch samples from hardware-timed acquisition buffer
- Used during hardware-timed acquisition to retrieve buffered samples; reads timing from self.ai_hw_timing_config
read_digital_line(channel) -> int: Read a digital input line
write_digital_line(channel, data): Write to a digital output line
read_digital_port(channel) -> int: Read a digital input port
write_digital_port(channel, data): Write to a digital output port
- Port read/write are required of every driver; LabJack T-Series raises NotImplementedError

Data Conversion

_read_to_measurements(response, channel_list, daq_name, default_tags, **kwargs) -> list[Measurement]: Convert vendor data format to InstroDAQ Measurement objects
- Maps vendor data to channel names
- Creates Measurement objects for publishing

Properties

points_in_buffer (int): Number of samples currently in the hardware buffer

Implementation Considerations

Resource Handling

Each concrete driver owns its own transport and accepts the connection parameters it needs at construction:

Vendor SDK drivers (NI-DAQmx, LabJack LJM, MCC): Accept a device_id string (device name, serial number, or IP address) and use the vendor library directly.
SCPI/VISA drivers (e.g., Keysight 34980A): Accept a VISA resource string or a VisaConfig and compose a VisaDriver internally for all SCPI command/query operations.

Driver-owned state

The driver is the single source of truth for every channel and every timing config that has been configured. It holds that state in private dicts/slots that only the configure_* path mutates. InstroDAQ does not hold its own copies — daq.ai_channels, daq.ai_hw_timing_config, and friends are read-only @property accessors that delegate straight to the driver’s read-only accessors, which hand back frozen snapshots captured at call time. There is no back-channel into InstroDAQ, and the driver does not import InstroDAQ. To keep every driver consistent, DAQDriverBase.__init__ initializes the private dicts and slots; concrete drivers call super().__init__() once and then populate them inside their configure_* methods.

State on every driver

DAQDriverBase.__init__ initializes the private storage below — populate it inside the matching configure_* method, and read from it (via the private attribute) anywhere the driver needs to know what’s configured. Each private dict has a matching read-only @property (ai_channels, ao_channels, …) that returns a frozen snapshot for external consumers; drivers use the private form internally.

Private attribute	Read-only accessor	Type	Populated in	Read by (typical)
`_ai_channels`	`ai_channels`	`dict[str, AnalogChannel]`	`configure_ai_voltage_channel` / `configure_ai_current_channel` / `configure_ai_thermocouple_channel`	`read_analog`, `fetch_analog`, `start`, …
`_ao_channels`	`ao_channels`	`dict[str, AnalogChannel]`	`configure_ao_voltage_channel` / `configure_ao_current_channel`	`write_analog_value`
`_di_channels`	`di_channels`	`dict[str, DigitalChannel]`	`configure_di_line_channel` / `configure_di_port_channel`	digital-input read paths
`_do_channels`	`do_channels`	`dict[str, DigitalChannel]`	`configure_do_line_channel` / `configure_do_port_channel`	digital-output write paths
`_relay_channels`	`relay_channels`	`dict[str, RelayChannel]`	`define_relay_channel` (base default already records; overrides must too)	`open_relay` / `close_relay`
`_ai_hw_timing_config`	`ai_hw_timing_config`	`HWTimingConfig \| None`	`configure_ai_hw_timing`	`start`, `fetch_analog`
`_ao_hw_timing_config` / `_di_hw_timing_config` / `_do_hw_timing_config`	`ao/di/do_hw_timing_config`	`HWTimingConfig \| None`	(forward-compat; unused today)	—
`points_in_buffer`	`points_in_buffer`	`int`	`fetch_analog` (per-call)	`InstroDAQ.get_points_in_buffer`

Add vendor-specific state alongside these as needed (NI keeps an nidaqmx.Task per ChannelType; MCC caches a ULRange per channel; LabJack holds its LJM handle — all live on the driver instance, declared in the driver’s own __init__ after super().__init__()).

Pattern

class MyDriver(DAQDriverBase):
    def __init__(self, resource: str) -> None:
        super().__init__()  # initializes _ai_channels, _ao_channels, _di_channels,
                            # _do_channels, _relay_channels, _*_hw_timing_config slots,
                            # and points_in_buffer.
        self._transport = SomeTransport(resource)  # vendor-specific state

    def configure_ai_voltage_channel(self, channel: AnalogVoltageChannel) -> None:
        ...  # program the device
        self._ai_channels[channel.alias] = channel

    def configure_ai_hw_timing(self, hw_timing_config: HWTimingConfig) -> None:
        ...  # program the device
        self._ai_hw_timing_config = hw_timing_config

    def fetch_analog(self) -> Any:
        config = self._ai_hw_timing_config  # read driver-owned state
        for ch in self._ai_channels.values():
            ...

The same self._<dict>[channel.alias] = channel pattern applies to the other typed analog methods and the four digital configure methods (configure_di_line_channel, configure_do_line_channel, configure_di_port_channel, configure_do_port_channel). DAQDriverBase.define_relay_channel’s default builds a RelayChannel and records it on self._relay_channels already — only override if your hardware needs different parsing, and ensure your override records too.

How InstroDAQ exposes this

# In DAQDriverBase — read-only accessors over the private storage:
@property
def ai_channels(self) -> Mapping[str, AnalogChannel]:
    """Frozen snapshot of configured AI channels, keyed by alias."""
    return MappingProxyType(dict(self._ai_channels))

# In InstroDAQ — delegate to the driver's read-only accessors:
@property
def ai_channels(self) -> Mapping[str, AnalogChannel]:
    return self._driver.ai_channels

@property
def ai_hw_timing_config(self) -> HWTimingConfig | None:
    return self._driver.ai_hw_timing_config
# ... and so on for ao_channels, di_channels, do_channels, relay_channels,
# ao_hw_timing_config, di_hw_timing_config, do_hw_timing_config.

@property
def channels(self) -> tuple[DAQChannel, ...]:
    """Frozen snapshot of every configured AI/AO/DI/DO channel (excludes relays)."""
    return self._driver.channels

End users read daq.ai_channels as a mapping, but it is a frozen snapshot: the returned MappingProxyType rejects writes, the channels inside are frozen dataclasses, and the snapshot does not change when later configure_* calls run — read the property again to see new state. The only sanctioned way to change configuration is the configure_* / define_* path, which programs the device and then records the channel. Reaching the driver via daq.driver exposes the same read-only accessors, not the private dicts.

Example: Vendor differences in timing

Different vendors configure timing differently, but InstroDAQ users must always configure a timing mode before start().

NI-DAQmx: Sample rate is configured on the device prior to starting an acquisition. The driver implements this in configure_ai_hw_timing() by calling DAQmx’s task.timing.cfg_samp_clk_timing() and then stores the config on self._ai_hw_timing_config.
LabJack LJM: Sample rate is configured when ljm.eStreamStart() is called. The driver records the HWTimingConfig in configure_ai_hw_timing() and reads it back from self._ai_hw_timing_config inside start().

Driver’s Job: Implement these differences so the user interface is consistent at the InstroDAQ level.

Summary

Driver development requires careful abstraction of vendor-specific behaviors to provide the unified InstroDAQ interface. Focus on:

Implementing all DAQDriverBase abstract methods
Abstracting timing configuration differences between vendors
Validating hardware constraints and providing clear error messages
Converting vendor data formats to Measurement objects correctly
Managing state in a thread-safe manner for background acquisition

Arbitrary Waveform Generator (AWG)

This section is for developers implementing InstroAWG support for waveform generators that aren’t supported out of the box.

Overview

Driver developers subclass AWGDriverBase and own whatever transport their instrument needs. The caller chooses a concrete driver, and that concrete driver exposes connection parameters that make sense for its protocol:

awg = InstroAWG(
    name="labAWG",
    driver=MyVendorAWG(host="10.0.0.42"),
    num_channels=1,
)

The driver is responsible for translating InstroAWG’s vendor-independent API (set_waveform, set_amplitude, output_enable, …) into vendor-specific commands.

Driver Responsibilities

An AWG driver must:

Expose a protocol-native constructor: accept inputs like visa_resource, host, port, depending on the instrument.
Own transport setup: create and store the transport internally. Do not require users to pass a VisaDriver or other transport object.
Own lifecycle: implement open() and close() by opening and closing the underlying transport.
Map commands: translate each abstract method into vendor-specific commands.
Parse responses: convert instrument responses to the expected Python types (Waveform, float, bool, ModulationType, etc.).
Validate hardware constraints: if the instrument only supports a subset of waveform shapes, modulation types, or carrier/modulator combinations, raise ValueError for unsupported combinations rather than silently misprogramming the instrument.

AWGDriverBase Interface

All AWG drivers subclass AWGDriverBase. The required methods are declared @abc.abstractmethod:

def open(self) -> None:
    """Open the underlying transport."""

def close(self) -> None:
    """Close the underlying transport."""

def set_waveform(self, channel: int, waveform: Waveform) -> None:
    """Program channel with the waveform definition; raise ValueError if the definition is unsupported."""

def get_waveform(self, channel: int) -> Waveform:
    """Get the current waveform on channel; drivers may return the last-programmed definition if not readable."""

def set_amplitude(self, channel: int, amplitude: float, unit: AmplitudeMeasurementUnit) -> None:
    """Set the output amplitude on channel."""

def get_amplitude(self, channel: int) -> tuple[float, AmplitudeMeasurementUnit]:
    """Get the current output amplitude and voltage unit on channel."""

def set_offset(self, channel: int, offset: float) -> None:
    """Set the DC offset (volts) on channel."""

def get_offset(self, channel: int) -> float:
    """Get the DC offset (volts) on channel."""

def output_enable(self, channel: int, enable: bool) -> None:
    """Enable or disable the output on channel."""

def get_output_state(self, channel: int) -> bool:
    """Return True if the output on channel is enabled."""

Optional overrides (raise NotImplementedError if not supported):

set_output_load(channel, load) / get_output_load(channel): Set or read the output load impedance; None means high-Z.
align_phase(): Sync the phase of all channels.
set_modulation(channel, mod_type, shape, magnitude): Configure a channel’s modulation. Call modulation_enable() to activate it.
modulation_enable(channel, enable): Enable or disable modulation on channel.
get_modulation_type(channel): Read back the active modulation type from the instrument.
get_modulation_state(channel): Read back whether modulation is enabled from the instrument.
set_burst(channel, burst_type) / get_burst_type(channel): Configure or read back a channel’s burst type (NCYCLE/GATED/INFINITE). Call burst_enable() to activate it.
burst_enable(channel, enable) / get_burst_state(channel): Enable/disable burst mode, or read back whether it’s enabled.
set_burst_trigger(channel, source) / get_burst_trigger(channel): Set or read back the burst trigger source (INTERNAL/EXTERNAL/MANUAL).
fire_burst_trigger(channel): Fire a burst trigger on channel immediately. Requires the trigger source already set to MANUAL; call set_burst_trigger() first otherwise.
set_burst_delay(channel, delay_s) / get_burst_delay(channel): Set or read back the burst trigger delay in seconds.
set_burst_gate_polarity(channel, gate_polarity) / get_burst_gate_polarity(channel): Set or read back the gate polarity (NORM/INV) for GATED bursts.
set_burst_ncycles(channel, n_cycles) / get_burst_ncycles(channel): Set or read back the number of cycles per trigger for NCYCLE bursts.
set_burst_period(channel, period) / get_burst_period(channel): Set or read back the internal burst period in seconds.
set_sweep(channel, sweep_type) / get_sweep_type(channel): Configure or read back a channel’s sweep type (LINEAR/LOG/STEP).
sweep_enable(channel, enable) / get_sweep_state(channel): Enable or disable sweep mode, or read back whether it’s enabled.
set_sweep_trigger(channel, source) / get_sweep_trigger(channel): Set or read back the sweep trigger source (INTERNAL/EXTERNAL/MANUAL).
set_sweep_start_freq(channel, frequency_hz) / get_sweep_start_freq(channel): Set or read the sweep start frequency.
set_sweep_end_freq(channel, frequency_hz) / get_sweep_end_freq(channel): Set or read the sweep end frequency.
set_sweep_time(channel, sweep_time) / get_sweep_time(channel): Set or read the sweep time.
set_sweep_start_hold_time(channel, hold_time) / get_sweep_start_hold_time(channel): Set or read the sweep start hold time.
set_sweep_stop_hold_time(channel, hold_time) / get_sweep_stop_hold_time(channel): Set or read the sweep stop hold time.
set_sweep_return_time(channel, return_time) / get_sweep_return_time(channel): Set or read the sweep return time.
fire_sweep_trigger(channel): Fire a sweep trigger on channel immediately. Requires the trigger source already set to MANUAL; call set_sweep_trigger() first otherwise.

Modulation

set_modulation() only configures modulation on a specific channel and does not enable modulation. Only exception is the Keysight 33521B driver: set_modulation will temporarily disable modulation, if previously enabled, to change modulation type. Enable state is restored once set_modulation finishes successfully. In error cases, set_modulation could leave modulation disabled. To enable or disable modulation, use modulation_enable(). Only one modulation type is enabled at a time. get_modulation_type returns the most previous modulation type set by the user.

Talking to the Instrument

Concrete drivers should hide transport details behind private attributes. For VISA-backed drivers, create a VisaDriver internally and use it for all I/O:

self._visa.write(command): Send a SCPI command (no response expected).
self._visa.query(command): Send a SCPI query and receive the response string.

VisaDriver owns the resource lock. Concurrent write / query calls against the same driver are serialized automatically. Use with self._visa.lock(): when a sequence of writes and their error check need to execute atomically. See the VisaDriver guide for the full transport reference, covering configuration, terminators, timeouts, serial settings, and the raw-byte I/O path. For non-VISA instruments, follow the same shape with the protocol client your driver needs. The important part is that the public constructor describes the instrument connection, while the transport object remains an implementation detail.

Implementation Example: RigolDG1022Z Driver

Here’s an abridged shape of the Rigol DG1022Z driver:

from instro.awg.awg import AWGDriverBase
from instro.awg.types import AmplitudeMeasurementUnit, Sine, Waveform
from instro.lib.transports import VisaConfig, VisaDriver


class RigolDG1022Z(AWGDriverBase):
    """SCPI driver for the Rigol DG1022Z two-channel arbitrary waveform generator."""

    def __init__(self, visa_resource: str | VisaConfig) -> None:
        self._visa = VisaDriver(visa_resource)

    def open(self) -> None:
        self._visa.open()

    def close(self) -> None:
        self._visa.close()

    def set_waveform(self, channel: int, waveform: Waveform) -> None:
        with self._visa.lock():
            if isinstance(waveform, Sine):
                self._visa.write(f":SOUR{channel}:FUNC SIN")
                self._visa.write(f":SOUR{channel}:FREQ {waveform.frequency_hz}")
                self._visa.write(f":SOUR{channel}:PHAS {waveform.phase_deg % 360.0}")
            # ... other Waveform shapes ...
            else:
                raise ValueError(f"unsupported waveform definition {type(waveform).__name__}")
            self._check_errors()  # one check per call, not one per write — see note above

    def set_amplitude(self, channel: int, amplitude: float, unit: AmplitudeMeasurementUnit) -> None:
        if unit is AmplitudeMeasurementUnit.VP:
            raise ValueError("the DG1022Z has no VP amplitude unit; convert to VPP, VRMS, or DBM")
        with self._visa.lock():
            self._visa.write(f":SOUR{channel}:VOLT:UNIT {unit.value}")
            self._visa.write(f":SOUR{channel}:VOLT {amplitude}")
            self._check_errors()

    # ... other required and optional methods ...

    def _check_errors(self) -> None:
        err = self._visa.query(":SYST:ERR?")
        code = err.strip().split(",", 1)[0].lstrip("+")
        if code != "0":
            raise RuntimeError(f"Rigol DG1022Z reported error: {err.strip()}")

Using a Custom Driver

For drivers that aren’t shipped in the library, construct InstroAWG with your own driver instance. The driver should accept connection settings directly and create its transport internally:

from instro.awg import InstroAWG
from instro.awg.awg import AWGDriverBase
from instro.awg.types import AmplitudeMeasurementUnit, Waveform, Sine
from instro.lib.transports import VisaDriver


class MyCustomAWGDriver(AWGDriverBase):
    """Custom driver for my lab's proprietary AWG."""

    def __init__(self, visa_resource: str) -> None:
        self._visa = VisaDriver(visa_resource)

    def open(self) -> None:
        self._visa.open()

    def close(self) -> None:
        self._visa.close()

    def set_waveform(self, channel: int, waveform: Waveform) -> None:
        if isinstance(waveform, Sine):
            self._visa.write(f"CH{channel}:WAVE SIN,{waveform.frequency_hz}")
            self._check_errors()
        else:
            raise ValueError(f"unsupported waveform definition {type(waveform).__name__}")

    # ... implement other required methods ...

    def get_output_state(self, channel: int) -> bool:
        return self._visa.query(f"CH{channel}:OUTP?") == "ON"

    def _check_errors(self) -> None:
        err = self._visa.query("ERR?")
        if err != "OK":
            raise RuntimeError(f"AWG error: {err}")


awg = InstroAWG(
    name="labAWG",
    driver=MyCustomAWGDriver(visa_resource="<VISA_ADDRESS>"),
    num_channels=1,
)

awg.open()
awg.set_waveform(1, Sine(frequency_hz=1000.0))
awg.close()

Summary

Driver development requires careful mapping of vendor-specific behavior to the unified InstroAWG interface. Focus on:

Subclassing AWGDriverBase
Designing a constructor around natural connection parameters for the instrument
Hiding transport construction inside the driver
Implementing all abstract methods on AWGDriverBase (and optional overrides where supported)
Using the correct vendor protocol or command syntax
Converting instrument responses to the expected Python types
Validating carrier/modulator/waveform compatibility your instrument actually supports, raising ValueError rather than misprogramming the instrument
Querying and reporting errors from the instrument’s error queue where one exists
Testing with actual hardware to ensure commands work as expected

Electronic Load (ELoad)

This section is for developers implementing InstroELoad support for electronic loads that aren’t supported out of the box.

Overview

Driver developers subclass ELoadDriverBase and own whatever transport their instrument needs. The caller chooses a concrete driver, and that concrete driver exposes connection parameters that make sense for its protocol:

eload = InstroELoad(
    name="labLoad",
    driver=MyVendorLoad(host="10.0.0.42", unit_id=1),
)

The driver is responsible for translating InstroELoad’s vendor-independent API (set_mode, set_level, get_voltage, …) into vendor-specific commands.

Driver Responsibilities

An electronic load driver must:

Expose a protocol-native constructor: accept inputs like visa_resource, host, port, unit_id, interface, or node_id, depending on the instrument.
Own transport setup: create and store the transport internally. Do not require users to pass a VisaDriver, socket client, Modbus client, or other transport object.
Own lifecycle: implement open() and close() by opening and closing the underlying transport.
Map commands: translate each abstract method into vendor-specific commands.
Parse responses: convert instrument responses to the expected Python types (float, bool, etc.).

ELoadDriverBase Interface

All electronic load drivers subclass ELoadDriverBase and implement these abstract methods:

def open(self) -> None:
    """Open the driver's underlying transport and perform any vendor handshake."""

def close(self) -> None:
    """Close the driver's underlying transport."""

def set_mode(self, mode: LoadMode, channel: int) -> None:
    """Set the operating mode."""

def set_level(self, mode: LoadMode, value: float, channel: int, curr_limit: float | None) -> None:
    """Set the operating level.

    Args:
        mode: Current operating mode.
        value: Level value (units depend on mode: A, V, Ω, or W).
        channel: Channel number (default is 1).
        curr_limit: Optional current limit (used in CV mode for protection).
    """

def set_range(self, mode: LoadMode, value: float, channel: int) -> None:
    """Set the operating range."""

def set_slewrate(self, direction: SlewRateDirection, rate: float, channel: int) -> None:
    """Set the current slew rate."""

def output_enable(self, enable: bool, channel: int) -> None:
    """Enable or disable the load input."""

def short_output(self, enable: bool, channel: int) -> None:
    """Enable or disable the load short."""

def get_voltage(self, channel: int) -> float:
    """Query and return the measured input voltage in volts."""

def get_current(self, channel: int) -> float:
    """Query and return the measured input current in amperes."""

Error checking is not part of the base contract: if your vendor exposes an error queue, add a private _check_errors() helper and call it from your write/query paths (see the representative driver below).

Talking to the Instrument

Concrete drivers should hide transport details behind private attributes. For VISA-backed drivers, create a VisaDriver internally and use it for all I/O:

self._visa.write(command): Send a SCPI command (no response expected).
self._visa.query(command): Send a SCPI query and receive the response string.

class MyModbusELoad(ELoadDriverBase):
    def __init__(self, host: str, *, port: int = 502, unit_id: int = 1) -> None:
        self._host = host
        self._port = port
        self._unit_id = unit_id
        self._client = None

The important part is that the public constructor describes the instrument connection, while the transport object remains an implementation detail.

Implementation Example: B&K Precision Driver

Here’s the complete driver implementation for B&K Precision 85xx Series electronic loads:

from instro.eload import ELoadDriverBase
from instro.eload.types import LoadMode, SlewRateDirection
from instro.lib.transports import VisaConfig, VisaDriver


def loadmode_to_unit(mode: LoadMode) -> str:
    """Convert LoadMode enum to SCPI unit keyword."""
    return {
        LoadMode.CC: "CURR",
        LoadMode.CV: "VOLT",
        LoadMode.CP: "POW",
        LoadMode.CR: "RES",
    }[mode]


class BK85XXB(ELoadDriverBase):
    """SCPI mapping for B&K Precision 85xx Series electronic loads."""

    def __init__(self, visa_resource: str | VisaConfig) -> None:
        self._visa = VisaDriver(visa_resource)

    def open(self) -> None:
        """Open transport and put the instrument in remote mode."""
        self._visa.open()
        self._visa.write("SYST:REM")

    def close(self) -> None:
        self._visa.close()

    def set_mode(self, mode: LoadMode, channel: int) -> None:
        self._visa.write(f"FUNCtion {loadmode_to_unit(mode)}")

    def set_level(self, mode: LoadMode, value: float, channel: int, curr_limit: float | None) -> None:
        if mode is LoadMode.CV:
            # TODO: Implement CV→CC protection based on curr_limit
            pass
        self._visa.write(f"{loadmode_to_unit(mode)} {value}")

    def set_range(self, mode: LoadMode, value: float, channel: int) -> None:
        # The 85xx series only exposes :RANGe for CC and CV; CP/CR are auto-ranged.
        if mode not in (LoadMode.CC, LoadMode.CV):
            raise NotImplementedError(
                f"BK85XXB only exposes :RANGe for CC and CV; {mode.value} is auto-ranged from the level value"
            )
        self._visa.write(f"{loadmode_to_unit(mode)}:RANGe {value}")

    def set_slewrate(self, direction: SlewRateDirection, rate: float, channel: int) -> None:
        self._visa.write(f"CURRent:SLEW:{direction.value} {rate}")

    def output_enable(self, enable: bool, channel: int) -> None:
        self._visa.write(f"INPut {int(enable)}")

    def short_output(self, enable: bool, channel: int) -> None:
        self._visa.write(f"INPut:SHORt {int(enable)}")
        self.output_enable(enable, channel)

    def get_current(self, channel: int) -> float:
        return float(self._visa.query("MEASure:CURRent?"))

    def get_voltage(self, channel: int) -> float:
        return float(self._visa.query("MEASure:VOLTage?"))

    def _check_errors(self) -> None:
        err = self._visa.query("SYST:ERR?")
        if not err.startswith("0"):
            raise RuntimeError(f"BK85XXB reported error: {err}")

Vendor SCPI VariationsDifferent electronic load vendors use different SCPI command sets. Always consult your electronic load’s programming manual for the correct SCPI syntax.

Using a Custom Driver

For drivers that aren’t shipped in the library, construct InstroELoad with your own driver instance. The driver should accept connection settings directly and create its transport internally:

from instro.eload import LoadMode, InstroELoad
from instro.eload import ELoadDriverBase
from instro.lib.transports import VisaDriver


class MyCustomELoadDriver(ELoadDriverBase):
    """Custom driver for my lab's proprietary electronic load."""

    def __init__(self, visa_resource: str) -> None:
        self._visa = VisaDriver(visa_resource)

    def open(self) -> None:
        self._visa.open()

    def close(self) -> None:
        self._visa.close()

    def set_mode(self, mode: LoadMode, channel: int) -> None:
        self._visa.write(f"MODE {mode.value}")

    def set_level(self, mode: LoadMode, value: float, channel: int, curr_limit: float | None) -> None:
        self._visa.write(f"LEVEL {value}")

    # ... implement other required methods ...

    def _check_errors(self) -> None:
        err = self._visa.query("ERR?")
        if err != "OK":
            raise RuntimeError(f"Electronic load error: {err}")


eload = InstroELoad(
    name="labELoad",
    driver=MyCustomELoadDriver(visa_resource="<VISA_ADDRESS>"),
)

eload.open()
eload.set_mode(LoadMode.CC, channel=1)
eload.set_level(5.0, channel=1)
eload.close()

Summary

Driver development requires careful mapping of vendor-specific behavior to the unified InstroELoad interface. Focus on:

Subclassing ELoadDriverBase
Designing a constructor around natural connection parameters for the instrument
Hiding transport construction inside the driver
Implementing all abstract methods on ELoadDriverBase
Using the correct vendor protocol or command syntax
Converting instrument responses to the expected Python types

Flow Controller

This section is for developers implementing InstroFlowController support for flow controllers that aren’t shipped in the library.

Overview

Driver developers subclass FlowControllerDriverBase and own whatever transport their instrument needs. The caller chooses a concrete driver and passes it to InstroFlowController:

fc = InstroFlowController(
    name="fc",
    driver=MyVendorFlowController(visa_resource="ASRL7::INSTR"),
)

The driver translates InstroFlowController’s vendor-independent API (get_flow_data, set_setpoint, select_gas, tare_flow) into vendor-specific commands.

Driver Responsibilities

A flow controller driver must:

Expose a protocol-native constructor: accept inputs like visa_resource, host, port, or device_id, depending on the instrument.
Own transport setup: create and store the transport internally. Do not require users to pass a VisaDriver or other transport object.
Own lifecycle: implement open() and close() by opening and closing the underlying transport.
Map commands: translate each abstract method into vendor-specific commands.
Parse responses: convert instrument responses into FlowData or the expected return type.

FlowControllerDriverBase Interface

All flow controller drivers subclass FlowControllerDriverBase and implement these abstract methods:

def open(self) -> None:
    """Open the driver's underlying transport."""

def close(self) -> None:
    """Close the driver's underlying transport. Idempotent."""

def get_flow_data(self) -> FlowData:
    """Read a full measurement frame from the device."""

def set_setpoint(self, setpt: float) -> float:
    """Command a new flow setpoint in the device's configured engineering units."""

def select_gas(self, gas_name: str) -> str:
    """Select the active gas by name; driver resolves the device-internal number."""

def tare_flow(self) -> FlowData:
    """Zero the flow reading. Device must have zero flow when called."""

All six methods are @abc.abstractmethod. FlowData (from instro.flowcontroller.types) carries pressure, temperature, vol_flow, mass_flow, setpoint, gas, and status_flags.

Talking to the Instrument

Concrete drivers should hide transport details behind private attributes. For VISA-backed drivers (including RS-232/serial), create a VisaDriver internally and use it for all I/O:

self._visa.write(command): Send a command (no response expected).
self._visa.query(command): Send a command and receive the response string.
self._visa.read(): Read one line from the device buffer.

VisaDriver owns the resource lock. Use with self._visa.lock(): when a sequence of writes and reads must execute atomically. See the VisaDriver guide for the full transport reference, covering configuration, terminators, timeouts, and serial settings.

Implementation Example: Alicat MC-Series

AlicatMC is the reference driver. It communicates over RS-232 using Alicat’s proprietary ASCII polling protocol — not SCPI — so commands are terse single-character identifiers rather than IEEE 488.2-style mnemonics.

from instro.unstable.flowcontroller import FlowControllerDriverBase
from instro.unstable.flowcontroller.types import FlowData
from instro.lib.transports.visa import SerialConfig, TerminatorConfig, VisaConfig, VisaDriver


class AlicatMC(FlowControllerDriverBase):
    """Alicat MC-series mass-flow controller over RS-232 ASCII polling."""

    def __init__(self, visa_resource: str | VisaConfig, device_id: str = "A") -> None:
        self.unit_id = device_id
        if isinstance(visa_resource, str):
            visa_resource = VisaConfig(
                visa_resource=visa_resource,
                serial_config=SerialConfig(baud_rate=19200),
                terminator=TerminatorConfig(read="\r", write="\r"),
            )
        self._visa = VisaDriver(visa_resource)

    def open(self) -> None:
        self._visa.open()

    def close(self) -> None:
        self._visa.close()

    def get_flow_data(self) -> FlowData:
        # Poll: send the unit ID alone to request a measurement frame
        response = self._query_checked(self.unit_id)
        return self._parse_flowdata(response)

    def set_setpoint(self, setpt: float) -> float:
        response = self._query_checked(f"{self.unit_id}s{setpt}")
        return self._parse_flowdata(response).setpoint

    def tare_flow(self) -> FlowData:
        response = self._query_checked(f"{self.unit_id}v")
        return self._parse_flowdata(response)

    def select_gas(self, gas_name: str) -> str:
        # Resolves name → device gas number via list_gas_types(), then sends '{id}g{number}'
        ...

    def _query_checked(self, command: str) -> str:
        response = self._visa.query(command)
        if response == "?":
            raise RuntimeError(f"Device returned '?' for command {command!r}")
        return response

    def _parse_flowdata(self, response: str) -> FlowData:
        # Frame order: UnitID AbsPressure Temp VolumetricFlow MassFlow Setpoint Gas [status...]
        fields = response.split()
        return FlowData(
            pressure=float(fields[1]),
            temperature=float(fields[2]),
            vol_flow=float(fields[3]),
            mass_flow=float(fields[4]),
            setpoint=float(fields[5]),
            gas=fields[6],
            status_flags=set(fields[7:]),
        )

Alicat ASCII protocol, not SCPIAlicat MC-series devices use a proprietary ASCII polling protocol. Commands are single-character identifiers (e.g. {id} to poll, {id}s{setpt} to set a setpoint, {id}v to tare, {id}g{n} to select a gas) rather than SCPI mnemonics. An unrecognized command returns ?. Consult the Alicat Gas Flow Controller Manual for the full command reference.

Using a Custom Driver

Construct InstroFlowController with your own driver instance:

from instro.unstable.flowcontroller import InstroFlowController, FlowControllerDriverBase
from instro.unstable.flowcontroller.types import FlowData


class MyFlowControllerDriver(FlowControllerDriverBase):

    def __init__(self, visa_resource: str) -> None:
        # create transport internally
        ...

    def open(self) -> None: ...
    def close(self) -> None: ...
    def get_flow_data(self) -> FlowData: ...
    def set_setpoint(self, setpt: float) -> float: ...
    def select_gas(self, gas_name: str) -> str: ...
    def tare_flow(self) -> FlowData: ...


fc = InstroFlowController(
    name="fc",
    driver=MyFlowControllerDriver(visa_resource="<VISA_ADDRESS>"),
)

fc.open()
fc.set_setpoint(50.0)
fc.close()

Summary

Driver development requires careful mapping of vendor-specific behavior to the unified InstroFlowController interface. Focus on:

Subclassing FlowControllerDriverBase
Designing a constructor around natural connection parameters for the instrument
Hiding transport construction inside the driver
Implementing all six abstract methods on FlowControllerDriverBase
Returning correctly typed FlowData from measurement methods
Parsing and raising instrument errors appropriately

Generic Custom Instrumentation

instro provides a framework for developing instruments that do not ship with the library. This lets an engineer with domain knowledge of an instrument type develop new functionality within a paradigm that grants the features of other instruments within instro. This is different than custom driver development within an existing instrument type — see Driver Development above for that. For example, if there is no built-in RF Power Meter instrument type but you want the instro experience, creating a custom instrument type using Instrument is the path forward.

Walkthrough: building a `SimpleTempController` instrument

This section builds a custom instrument end to end. It’s a simulated temperature controller (like a benchtop PID controller or thermal chamber) whose reading lags the commanded setpoint as it warms up or cools down. Standard library only, no hardware, no extra dependencies. The full file lives at examples/custom/simple_temp_controller.py. The example covers everything a custom instrument needs:

a @publish_measurement method that emits data
a @publish_command method that changes instrument state and visibly affects the next several reads
a background daemon that polls the measurement on an interval

About the “device”

A real temperature controller has two pieces of state: the current temperature (what its sensor measures right now) and the setpoint (the temperature you’ve told it to hold). When you change the setpoint, the controller’s heater or cooler kicks on, and the current temperature lags behind, closing in on the new target over the next several reads rather than snapping to it. The simulation fakes that lag in one line. Each tick, the current temperature moves 20% of the way toward the setpoint, plus a small amount of noise. That’s enough for the set_target_temperature command to have a dramatic, observable effect. You’ll watch the temperature climb from room temperature toward 50°C, then toward 75°C, in real time.

1. Subclass `Instrument`

Every custom instrument is a subclass of Instrument. The base class owns the publisher list, the background-daemon thread, default tags, and the channel buffer. Your subclass gets all of them for free.

import random
import time

from instro.lib import Command, Instrument, Measurement
from instro.lib.instrument import publish_command, publish_measurement


class SimpleTempController(Instrument):
    def __init__(self, name: str, **kwargs):
        super().__init__(name, **kwargs)
        self._temperature_c = 20.0  # current temperature (starts at room temp)
        self._setpoint_c = 20.0     # commanded target
        self.background_interval = 0.5
        self.add_background_daemon_function(self.read_temperature)

    def open(self) -> None:
        # Establish your device connection here (open a socket, VISA session, etc.).
        super().open()

    def close(self) -> None:
        # `super().close()` stops the daemon and closes attached publishers.
        # Tear down your device connection after that.
        super().close()

The base constructor takes a name (used as the channel-name prefix), and any extra **kwargs become default tags. self.background_interval sets the daemon’s polling cadence in seconds. Two methods register what the daemon runs each tick: add_background_daemon_function(method, *args, **kwargs) appends one call to the daemon’s list (call it once per method you want polled), and define_background_daemon(method, *args, **kwargs) is the replace-all variant that clears the list and registers a single method. Override open() and close() to manage the connection to your device. The simulated controller has nothing to open, so both overrides just call super(). This is the seam where a real instrument would call socket.create_connection, open a VISA session, claim a USB interface, etc. The base close() already stops the background daemon and closes attached publishers, so user-side overrides only need to add their own teardown.

2. Publish a `Measurement` with `@publish_measurement`

Any method decorated with @publish_measurement must return a Measurement (or list[Measurement], or None). The decorator publishes the returned object to every attached publisher automatically. Your method just has to build it. The _package_measurement(channel, value, timestamp, **tags) helper on Instrument does the build for you: it prefixes the channel with {self.name}., passes a str value through untouched and coerces anything else to float, merges in self.default_tags, and returns a single-channel Measurement.

_package_measurement and _package_command are stable, supported extension points for Instrument subclass authors. The leading underscore marks them as protected (for subclasses, not external callers), which is standard Python. Both are documented on the Instrument reference page.

class SimpleTempController(Instrument):
    ...

    @publish_measurement
    def read_temperature(self, **kwargs) -> Measurement:
        # Each tick: move 20% of the way toward the setpoint, plus a little noise.
        drift = (self._setpoint_c - self._temperature_c) * 0.2
        self._temperature_c += drift + random.uniform(-0.1, 0.1)
        return self._package_measurement("temperature_c", self._temperature_c, time.time_ns(), **kwargs)

Every call to read_temperature() advances the simulated temperature one step and publishes the new value under {name}.temperature_c. With the constructor above, the background daemon calls this on its own every 500 ms. That’s where the visible “warming up” comes from. Any publisher you attach (e.g. NominalCorePublisher) receives every sample.

3. Publish a `Command` with `@publish_command`

Commands are for changing instrument state, like setpoints, mode switches, and configuration changes. A @publish_command-decorated method returns a Command and the decorator publishes it. Use the parallel helper _package_command(channel, value, timestamp, **tags) to build one. Pass the descriptor with a trailing .cmd so the published channel name has it.

class SimpleTempController(Instrument):
    ...

    @publish_command
    def set_target_temperature(self, value: float, **kwargs) -> Command:
        self._setpoint_c = value
        return self._package_command("setpoint_c.cmd", value, time.time_ns(), **kwargs)

controller.set_target_temperature(50.0) does two things. It changes the target the simulated controller will close in on, and it publishes a controller.setpoint_c.cmd record with the new value. The next several read_temperature() calls will show the temperature climbing. The command literally drives the measurement. If set_target_temperature raised before reaching _package_command (say, a bounds check that rejected value=999), the decorator wouldn’t fire. No record would land on your dashboard, and the previous setpoint would still be in effect. The decorator only publishes on a clean return.

4. Use the instrument

# `with` calls open() on entry and close() on exit: close() stops the daemon
# and closes publishers even if an exception escapes the block.
with SimpleTempController(name="controller") as controller:
    controller.start()  # start the background daemon

    for target in (25, 50, 75):
        controller.set_target_temperature(target)
        time.sleep(5)

For each target, the loop sends the command and lets the background daemon collect samples for five seconds (about ten ticks at the 500 ms cadence). The base Instrument keeps an in-memory channel buffer of every published sample, so controller.get_channel("controller.temperature_c").latest returns the most recent reading if you want to inspect it from your script:

latest = controller.get_channel("controller.temperature_c").latest
print(f"current={latest:.1f}")

To stream the same data to Nominal Core instead of (or in addition to) the in-memory buffer, attach a publisher before start():

from instro.lib.publishers import NominalCorePublisher

controller.add_publisher(NominalCorePublisher("<dataset_rid>"))

Both controller.temperature_c (telemetry) and controller.setpoint_c.cmd (command) now stream to Nominal automatically. No changes to the instrument class itself.

Patterns to copy

Channel naming. Writes get a trailing .cmd (e.g. "setpoint_c.cmd"); reads do not, whatever the value’s type (e.g. "temperature_c" for a numeric reading, "operating_mode" for a categorical one). The _package_* helpers prepend {self.name}. so multiple instances stay namespaced. See Backwards-compatible channel naming for the full convention.
One type per channel. A given channel’s value is always numeric or always a string across its lifetime, never both — publishing a bool/float to a channel that has ever carried a str (or vice versa) is a bug in the calling instrument, not something _package_measurement checks for you.
Pass .value, not the enum member, for a categorical read. @publish_measurement raises TypeError if a returned Measurement’s single-value channel holds anything other than int/float/str — so self._package_measurement(channel, some_enum_member.value, timestamp), not some_enum_member. This catches a forgotten .value at the call site.
Lifecycle. open() / close() bracket the device connection. start() / stop() bracket the background daemon. close() calls stop() internally and also tears down attached publishers, so a single try / finally around close() is enough to clean up everything. Instrument is also a context manager. with SimpleTempController(name="controller") as controller: calls open() on entry and close() on exit, including when an exception escapes the block, which is the pattern the example above uses. Override open() / close() and you get context-manager support for free, with no need to implement __enter__ / __exit__ yourself. For instruments that share a transport between threads (the daemon polling while user code sends commands), also serialize I/O behind a lock. instro’s built-in categories (PSU, ELoad, DMM, Modbus, …) all do this.
Decorator guarantees. @publish_command raises TypeError if the method returns anything other than a Command. @publish_measurement does the same for Measurement | list[Measurement] | None. Wrong return type fails loudly at the call site instead of silently producing the wrong kind of record.
Errors don’t get published. If your @publish_* method raises before returning, the decorator never fires: the instrument’s prior state is left untouched and no spurious record lands on your dashboard.
Coupled commands and measurements. set_target_temperature() mutates state that read_temperature() consumes. This is the most common shape for an instrument: a setpoint or mode-change command that drives the next round of measurements.

​Driver Development

​Oscilloscope (Scope)

​Overview

​ScopeDriverBase Interface

​Implementation Example: Keysight 1200 X-Series Driver

​Power Supply (PSU)

​Overview

​Driver Responsibilities

​PSUDriverBase Interface

​Talking to the Instrument

​Implementation Example: B&K Precision Single-Channel Driver

​Using a Custom Driver

​Summary

​Digital Multimeter (DMM)

​Overview

​Driver Responsibilities

​DMMDriverBase Interface

​Talking to the Instrument

​Implementation Example: Agilent 34401A Driver

​Using a Custom Driver

​Summary

​Data Acquisition (DAQ)

​Overview

​Driver Responsibilities

​DAQDriverBase Interface

​Connection Management

​Channel Configuration

​Timing Configuration

​Acquisition Control

​Data Operations

​Data Conversion

​Properties

​Implementation Considerations

​Resource Handling

​Driver-owned state

State on every driver

Pattern

How InstroDAQ exposes this

Example: Vendor differences in timing

​Summary

​Arbitrary Waveform Generator (AWG)

​Overview

​Driver Responsibilities

​AWGDriverBase Interface

​Modulation

​Talking to the Instrument

​Implementation Example: RigolDG1022Z Driver

​Using a Custom Driver

​Summary

​Electronic Load (ELoad)

​Overview

​Driver Responsibilities

​ELoadDriverBase Interface

​Talking to the Instrument

​Implementation Example: B&K Precision Driver

​Using a Custom Driver

​Summary

​Flow Controller

​Overview

​Driver Responsibilities

​FlowControllerDriverBase Interface

​Talking to the Instrument

​Implementation Example: Alicat MC-Series

​Using a Custom Driver

​Summary

​Generic Custom Instrumentation

​Walkthrough: building a SimpleTempController instrument

​About the “device”

​1. Subclass Instrument

​2. Publish a Measurement with @publish_measurement

​3. Publish a Command with @publish_command

​4. Use the instrument

​Patterns to copy

Driver Development

Oscilloscope (Scope)

Overview

ScopeDriverBase Interface

Implementation Example: Keysight 1200 X-Series Driver

Power Supply (PSU)

Overview

Driver Responsibilities

PSUDriverBase Interface

Talking to the Instrument

Implementation Example: B&K Precision Single-Channel Driver

Using a Custom Driver

Summary

Digital Multimeter (DMM)

Overview

Driver Responsibilities

DMMDriverBase Interface

Talking to the Instrument

Implementation Example: Agilent 34401A Driver

Using a Custom Driver

Summary

Data Acquisition (DAQ)

Overview

Driver Responsibilities

DAQDriverBase Interface

Connection Management

Channel Configuration

Timing Configuration

Acquisition Control

Data Operations

Data Conversion

Properties

Implementation Considerations

Resource Handling

Driver-owned state

Summary

Arbitrary Waveform Generator (AWG)

Overview

Driver Responsibilities

AWGDriverBase Interface

Modulation

Talking to the Instrument

Implementation Example: RigolDG1022Z Driver

Using a Custom Driver

Summary

Electronic Load (ELoad)

Overview

Driver Responsibilities

ELoadDriverBase Interface

Talking to the Instrument

Implementation Example: B&K Precision Driver

Using a Custom Driver

Summary

Flow Controller

Overview

Driver Responsibilities

FlowControllerDriverBase Interface

Talking to the Instrument

Implementation Example: Alicat MC-Series

Using a Custom Driver

Summary

Generic Custom Instrumentation

Walkthrough: building a `SimpleTempController` instrument

About the “device”

1. Subclass `Instrument`

2. Publish a `Measurement` with `@publish_measurement`

3. Publish a `Command` with `@publish_command`

4. Use the instrument

Patterns to copy