> ## Documentation Index
> Fetch the complete documentation index at: https://instro.nominal.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Custom Instruments

> Add a driver to an existing instrument type, or build a whole new instrument type by subclassing `Instrument`

<Tip>
  **Looking to add a new vendor or model to a supported instrument type?**

  Jump to that category's driver-development section below: [Oscilloscope (Scope)](#oscilloscope-scope), [Power Supply (PSU)](#power-supply-psu), [Digital Multimeter (DMM)](#digital-multimeter-dmm), [Data Acquisition (DAQ)](#data-acquisition-daq), [Arbitrary Waveform Generator (AWG)](#arbitrary-waveform-generator-awg), [Electronic Load (ELoad)](#electronic-load-eload), [Flow Controller](#flow-controller), or [I2C](/library/protocols/i2c/overview#driver-development).

  If your instrument doesn't fit any supported type at all, see [Generic Custom Instrumentation](#generic-custom-instrumentation) below instead — that's for building an entirely new instrument type using `Instrument`.
</Tip>

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

1. **Expose a protocol-native constructor**: accept a `visa_resource` string or `VisaConfig`.
2. **Own transport setup**: create and store the transport internally. Do not require callers to pass a `VisaDriver`.
3. **Own lifecycle**: implement `open()` and `close()`.
4. **Map commands**: translate each abstract method into vendor commands.
5. **Parse responses**: convert responses to the expected types (`float`, enums, `WaveformData`).
6. **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:

```python theme={null}
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](/library/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](/library/transports/visa) for the full transport reference.

### Implementation Example: Keysight 1200 X-Series Driver

```python theme={null}
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:

```python theme={null}
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:

1. **Expose a protocol-native constructor**: accept inputs like `visa_resource`, `host`, `port`, `unit_id`, `interface`, or `node_id`, depending on the instrument.
2. **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.
3. **Own lifecycle**: implement `open()` and `close()` by opening and closing the underlying transport.
4. **Map commands**: translate each abstract method into vendor-specific commands.
5. **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:

```python theme={null}
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](/library/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](/library/transports/visa) 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:

```python theme={null}
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}")
```

<Tip>
  **Vendor SCPI Variations**

  Different 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.
</Tip>

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

```python theme={null}
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:

```python theme={null}
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:

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

### DMMDriverBase Interface

All DMM drivers subclass `DMMDriverBase` and implement these abstract methods:

```python theme={null}
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](/library/transports/visa) 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:

```python theme={null}
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...
```

<Tip>
  **Vendor SCPI Variations**

  Different 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.
</Tip>

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

```python theme={null}
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](#daqdriverbase-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:

1. **Hardware connection lifecycle**: Implement `open()` and `close()` for establishing and terminating hardware connections.
2. **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](#driver-owned-state) below.
3. **Channel configuration**: Translate generic channel configurations into vendor-specific setup.
4. **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.
5. **Data format conversion**: Convert vendor-specific data formats to `Measurement` objects.
6. **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](#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](/daq#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

```python theme={null}
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

```python theme={null}
# 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:

```python theme={null}
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:

1. **Expose a protocol-native constructor**: accept inputs like `visa_resource`, `host`, `port`, depending on the instrument.
2. **Own transport setup**: create and store the transport internally. Do not require users to pass a `VisaDriver` or other transport object.
3. **Own lifecycle**: implement `open()` and `close()` by opening and closing the underlying transport.
4. **Map commands**: translate each abstract method into vendor-specific commands.
5. **Parse responses**: convert instrument responses to the expected Python types (`Waveform`, `float`, `bool`, `ModulationType`, etc.).
6. **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`:

```python theme={null}
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](/library/transports/visa) 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:

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
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:

1. **Expose a protocol-native constructor**: accept inputs like `visa_resource`, `host`, `port`, `unit_id`, `interface`, or `node_id`, depending on the instrument.
2. **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.
3. **Own lifecycle**: implement `open()` and `close()` by opening and closing the underlying transport.
4. **Map commands**: translate each abstract method into vendor-specific commands.
5. **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:

```python theme={null}
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.

`VisaDriver` owns the resource lock. Concurrent `write` / `query` calls against the same driver are serialized automatically.

See the [VisaDriver guide](/library/transports/visa) 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:

```python theme={null}
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:

```python theme={null}
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}")
```

<Tip>
  **Vendor SCPI Variations**

  Different electronic load vendors use different SCPI command sets. Always consult your electronic load's programming manual for the correct SCPI syntax.
</Tip>

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

```python theme={null}
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`:

```python theme={null}
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:

1. **Expose a protocol-native constructor**: accept inputs like `visa_resource`, `host`, `port`, or `device_id`, depending on the instrument.
2. **Own transport setup**: create and store the transport internally. Do not require users to pass a `VisaDriver` or other transport object.
3. **Own lifecycle**: implement `open()` and `close()` by opening and closing the underlying transport.
4. **Map commands**: translate each abstract method into vendor-specific commands.
5. **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:

```python theme={null}
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](/library/transports/visa) 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.

```python theme={null}
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:]),
        )
```

<Tip>
  **Alicat ASCII protocol, not SCPI**

  Alicat 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](https://documents.alicat.com/manuals/Gas_Flow_Controller_Manual.pdf) for the full command reference.
</Tip>

### Using a Custom Driver

Construct `InstroFlowController` with your own driver instance:

```python theme={null}
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](#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`](https://github.com/nominal-io/instro/blob/main/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.

```python theme={null}
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`.

<Note>
  `_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](https://nominal-io.github.io/instro/reference/instrument/).
</Note>

```python theme={null}
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.

```python theme={null}
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

```python theme={null}
# `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:

```python theme={null}
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()`:

```python theme={null}
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](/library/library#backwards-compatible-channel-naming-legacy_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.