InstroDAQ

InstroDAQ is a hardware abstraction layer (HAL) that provides a unified interface for data acquisition devices from multiple vendors. The key benefit is vendor-independent code: swap the driver instance you hand to InstroDAQ and the same configuration/acquisition code works across different hardware.

Supported Vendors

National Instruments (NI DAQmx) - Using NI-DAQmx driver
LabJack T-Series - T4, T7, T8 models using LJM driver
Measurement Computing (MCC) - USB and Ethernet DAQ devices using the MCC Universal Library (mcculw)
Keysight 34980A - Multifunction switch/measure unit via SCPI/VISA

Key Concepts

Lifecycle Pattern

The typical InstroDAQ workflow follow this pattern:

InstroDAQ(name, driver, ...) - Instantiate the DAQ with a vendor driver
open() - Establish connection to the hardware
Configure - Set up , timing, and other settings
start() - Begin hardware-timed acquisition (if using hardware timing)
Acquire data - Read/fetch measurements
stop() - End acquisition (if using hardware timing)
close() - Disconnect from hardware

Channels

InstroDAQ supports both analog and digital I/O:

Analog Input/Output: Measurements and generation with configurable ranges
Digital Input/Output: Line-based or port-based digital I/O

Aliases

Map vendor-specific physical channel names (e.g., “AIN0”) to logical names (e.g., “temperature_sensor”) Channel aliases are used as channel names when data is published to Nominal Core or Connect.

Timing Modes

InstroDAQ supports two acquisition modes: Software-Timed (Manual Polling)

You call read_analog() when you want a sample
No need to call start() or stop()
Use case: Low-frequency monitoring, event-driven sampling

Hardware-Timed (Buffered Acquisition)

Configure sample rate with configure_ai_sample_rate()
Call start() to begin background acquisition
Data automatically published via the background daemon, OR manually fetch with read_analog()
Use case: Mid to high frequency continuous monitoring with deterministic sample timing.

Creating a InstroDAQ Instance

Construct a vendor driver and pass it to InstroDAQ:

from instro.daq.drivers.labjack import LabJackTSeriesDriver
from instro.daq import InstroDAQ

# Create a LabJack DAQ
daq = InstroDAQ(
    name="myDAQ",
    driver=LabJackTSeriesDriver(device_id="440020473"),  # Vendor-specific resource identifier
)

Parameters

name: A name for this DAQ instance. Used as a prefix for channel names when publishing to Nominal Core.
driver: A concrete DAQDriverBase subclass instance. Each driver owns its own transport (vendor SDK or SCPI/VISA) and accepts vendor-specific connection parameters at construction.
publishers: List of publishers to attach
dataset_rid: Shorthand to add a NominalCorePublisher

Vendor-Specific Driver Examples

from instro.daq import InstroDAQ

# National Instruments (device name from NI-MAX)
from instro.daq.drivers.ni import NIDAQDriver

daq = InstroDAQ(name="niDAQ", driver=NIDAQDriver(device_id="Dev1"))

# LabJack (serial number, device name, or IP address)
from instro.daq.drivers.labjack import LabJackTSeriesDriver

daq = InstroDAQ(name="ljDAQ", driver=LabJackTSeriesDriver(device_id="440020473"))

# Measurement Computing (MCC device unique ID, optionally suffixed with a board number)
from instro.daq.drivers.mcc import MCCDriver

daq = InstroDAQ(
    name="mccDAQ",
    driver=MCCDriver(device_id="344371:0"),  # "<unique_id>" or "<unique_id>:<board_number>"
)

# Keysight 34980A (SCPI/VISA device: driver owns its VisaDriver)
from instro.daq.drivers import Keysight34980A

daq = InstroDAQ(
    name="keysightDAQ",
    driver=Keysight34980A("TCPIP0::<IP_ADDRESS>::INSTR"),
)

Configuring Channels

Analog Input Channels

Configure analog input channels with measurement range and logical names:

from instro.daq.types import Direction

daq.configure_analog_channel(
    direction=Direction.INPUT,
    physical_channel="AIN0",  # Vendor-specific channel name
    alias="temperature_sensor",  # Your logical name (used for publishing)
    range_min=0.0,  # Minimum voltage (V)
    range_max=5.0   # Maximum voltage (V)
)

The physical_channel naming convention depends on your DAQ vendor:

NI DAQmx: fully qualified device/channel, for example “Dev1/ai0”, “Dev1/ai1”.
LabJack: “AIN0”, “AIN1”, etc.
MCC: Integer channel index as a string, for example "0", "1".
Keysight: Depends on module slot and channel configuration

Refer to your device’s documentation for channel naming.

Scalers

It’s common for the data being read by an analog input channel to need scaling to represent a real-world physical phenomenon. For example, a 0-5 volt sensor measuring pressure. InstroDAQ supports adding a Scaler object when you configure your analog channel. The Measurement published and returned by read_analog will contain these scaled values. Example of a 0-5V pressure sensor that measures 0-3000 psia.

from instro.daq.types import Direction
from instro.daq.scaling import LinearScaler

daq.configure_analog_channel(
    direction=Direction.INPUT,
    physical_channel="AIN0",  # Vendor-specific channel name
    alias="pressure_sensor",  # Your logical name (used for publishing)
    range_min=0,  # Minimum voltage (V)
    range_max=5,   # Maximum voltage (V)
    scaler = LinearScaler(gain = 600, offset = 0, units = "psia")
)

You can cascade multiple scalers together using the ScalerPipeline scaler. For example, a thermocouple that’s fed into an amplifier and then the DAQ will require two stages of scaling.

Scaling out the amplifier to get to the actual voltage seen across the thermocouple terminals.
Scaling the voltage seen across the thermocouple terminals to temperature.

from instro.daq.scaling import ScalerPipeline, ReverseLinearScaler
from instro.daq.scaling.thermocouple import ThermocoupleSensor

scaler_pipeline = ScalerPipeline(ReverseLinearScaler(gain=<amplifier gain>, offset=<voltage offset>, units="V"), ThermocoupleSensor(type = <TC_type>, cjc_temp = <cjc temp in Celsius>))

You can create your own scaler by subclassing Scaler and implementing scale and units methods.

from instro.daq.scaling import Scaler

class TheAnswer(Scaler):

   def scale(self, raw: float | int) -> float:
       return raw * 42

   @property
   def units(self) -> str:
       return "everything"

Analog Output Channels

Configure analog output channels:

from instro.daq.types import Direction

daq.configure_analog_channel(
        direction=Direction.OUTPUT, physical_channel="DAC0", alias=f"ao_0", range_min=0, range_max=5
    )

The physical_channel naming convention depends on your DAQ vendor:

NI DAQmx: fully qualified device/channel, for example “Dev1/ao0”, “Dev1/ao1”.
LabJack: “DAC0”, “DAC1”, etc.
MCC: Integer channel index as a string, for example "0", "1".
Keysight: Depends on module slot and channel configuration

Refer to your device’s documentation for channel naming.

Digital Channels

Before configuring a digital channel, you need to specify the following parameters to match your application and DAQ hardware: InstroDAQ exposes two methods: configure_digital_line and configure_digital_port.

direction: Use Direction.INPUT for digital input or Direction.OUTPUT for output.
physical_channel: The physical line or port on your DAQ device (e.g., "5101", "5101/2" for Keysight, or "FIRSTPORTA", "FIRSTPORTA/0" for MCC). This name is vendor-specific. Refer to your device documentation.
logic: Sets whether the channel treats a HIGH or LOW physical level as “True”. This is required for correct logic interpretation.
logic_level (optional): Specifies the voltage threshold (in volts) used to distinguish HIGH from LOW, if your device supports changing this.
alias (optional): A logical name for the channel, helpful for clarity and for use with publishers.
port_width (port only): Port width in bits (8/16/32/64), required by configure_digital_port.

For example, to configure a digital input line with a custom logic threshold:

from instro.daq.types import Direction, Logic

daq.configure_digital_line(
    direction=Direction.INPUT,
    physical_channel="5101/2",
    alias="limit_switch",
    logic=Logic.HIGH,
    logic_level=2.0    # Optional: 2V threshold for HIGH, if supported by hardware
)

If configuring an entire port:

from instro.daq.types import Direction, DigitalPortWidth, Logic

daq.configure_digital_port(
    direction=Direction.OUTPUT,
    physical_channel="5101",
    logic=Logic.HIGH,
    port_width=DigitalPortWidth.WIDTH_8  # E.g., for 8-bit port
)

Refer to your DAQ device’s documentation for available physical channel names and supported features.

Hardware-Timed Sample Rate

For continuous hardware-timed acquisition, configure the sample rate.

daq.configure_ai_sample_rate(
    sample_rate=1000.0,  # Hz
    samples_per_channel=500  # Optional: samples per read (defaults to sample_rate/10)
)

The samples_per_channel parameter determines how many samples, per channel, are returned on every call to read_analog().

The lower the samples_per_channel, the more responsive and lower latency your app will be, but may not be able to keep up with the sample rate.
The ratio of sample_rate to samples_per_channel determines how often data will be fetched from the DAQ buffer.
- Example, if sample_rate is 1000 and samples_per_channel is 500, you’ll see 500 sample batches of 1000Hz data twice a second, for every channel.
The default for samples_per_channel, if left unset, is dynamically set to enable fetching batches 10 times per second. This is a reasonable balance between reliably keeping up with the data stream and app responsiveness.

Hardware Timing ConstraintsDifferent DAQ devices have different timing capabilities:

Multiplexed DAQs (e.g., LabJack T4/T7, most NI DAQ devices, most MCC devices): Maximum per-channel rate decreases with more channels
Simultaneous DAQs (e.g., LabJack T8): All channels sampled simultaneously
Sample rate limits: Check your device specifications

Analog Input

Software-Timed Acquisition

For manual, on-demand sampling:

from instro.daq.types import Direction
import time

daq.open()

# Configure channels
daq.configure_analog_channel(
    direction=Direction.INPUT,
    physical_channel="AIN0",
    alias="sensor_1",
    range_min=0,
    range_max=5
)

# Read samples on demand
for i in range(10):
    measurement = daq.read_analog()
    print(f"Value: {measurement.latest}")
    time.sleep(1)  # Wait 1 second between reads

daq.close()

The read_analog() method returns a Measurement object (or list of Measurement objects for multiple channels) containing:

channel_data: Dictionary mapping channel aliases to lists of values
timestamps: List of timestamps (nanoseconds since epoch)
value: Property returning the first value (convenience for single-value reads)

Hardware-Timed Acquisition with Background Fetching

See Two ways to get data for more information regarding background fetching of measurements. For continuous high-speed acquisition.

import time
from instro.daq.drivers.labjack import LabJackTSeriesDriver
from instro.daq import InstroDAQ
from instro.daq.types import Direction
from instro.lib.publishers import NominalCorePublisher

DATASET_RID = "ri.catalog.main.dataset.<your-dataset-rid>"

# Create DAQ with publisher
daq = InstroDAQ(
    name="myDAQ",
    driver=LabJackTSeriesDriver(device_id="440020473"),
)
daq.add_publisher(NominalCorePublisher(dataset_rid=DATASET_RID))

daq.open()

# Configure analog input channels
daq.configure_analog_channel(
    direction=Direction.INPUT,
    physical_channel="AIN0",
    alias="sensor_1",
    range_min=0,
    range_max=5
)

# Configure hardware timing
daq.configure_ai_sample_rate(
    sample_rate=100,  # 100 Hz
)

# Start background acquisition
daq.start()

# Data is now flowing to publishers automatically
while True:
    try:
        sensor_1 = daq.get_channel(
            channel_name = "myDAQ.sensor_1",
            length = 10,
            wait_for_latest = True)  # This will block for the next 10 samples from the background daemon
        print("Stopping acquisition")
        break
    except KeyboardInterrupt:
        print("Exiting main loop")
        break

daq.stop()
daq.close()

In this mode:

start() begins hardware-timed acquisition in the background daemon
stop() ends the background acquisition

Important Note about PublishersData is published as a direct result of an instrument method being called.For example, when you call read_analog(), this not only returns the DAQ data but also causes all attached Publishers to publish the measurements automatically.Therefore the background daemon, which is calling instrument methods, is publishing the data!

Hardware-Timed Acquisition with Manual Fetching

For hardware-timed acquisition where you control when to fetch buffered data. Disable the Background Daemon and manually call read_analog().

# Disable automatic background publishing
daq.background_enable = False

daq.open()

# Configure channels and timing (same as above)
daq.configure_analog_channel(...)
daq.configure_ai_sample_rate(sample_rate=100, samples_per_channel=10)

# Start hardware acquisition (fills buffer)
daq.start()

# Manually fetch from buffer when ready
for i in range(10):
    measurement = daq.read_analog()
    print(f"Fetched {len(measurement.timestamps)} samples")
    time.sleep(0.1)

daq.stop()
daq.close()

Failing to fetch samples from the hardware buffer at a reasonable rate will cause the DAQ data buffer to fill up and data will be either lost or an exception will be raised.

When background_enable = False, calling read_analog() during hardware-timed acquisition fetches from the hardware buffer rather than triggering a new conversion.

Analog Output

Software-Timed Generation

For manual, on demand updates of set points:

from instro.daq.types import Direction

daq.open()

daq.configure_analog_channel(
    direction=Direction.OUTPUT,
    physical_channel=CHANNEL_0,
    alias=f"ao_0",
    range_min=0,
    range_max=5
)
daq.write_analog_value("ao_0", 2.2)

daq.close()

Digital I/O

Reading Digital Lines

from instro.daq.types import Direction, Logic

daq.configure_digital_line(
    direction=Direction.INPUT,
    physical_channel="DIO0",
    alias="limit_switch",
    logic=Logic.HIGH
)

measurement = daq.read_digital_line(channel="limit_switch")
print(f"Digital state: {measurement.latest}")

Software Timed Generation

For manual, on-demand setting of analog output channels:

daq.open()

daq.configure_analog_channel(
    direction=Direction.OUTPUT, physical_channel=CHANNEL_0, alias=f"ao_0", range_min=0, range_max=5
)
daq.configure_analog_channel(
    direction=Direction.OUTPUT, physical_channel=CHANNEL_1, alias=f"ao_1", range_min=0, range_max=5
)

daq.write_analog_value("ao_0", 2.2)
daq.write_analog_value("ao_1", 3.4)

daq.close()

Writing Digital Lines

from instro.daq.types import Direction, Logic

daq.configure_digital_line(
    direction=Direction.OUTPUT,
    physical_channel="DIO1",
    alias="enable_signal",
    logic=Logic.HIGH
)

# Write high (1)
daq.write_digital_line(channel="enable_signal", data=1)

# Write low (0)
daq.write_digital_line(channel="enable_signal", data=0)

Reading and Writing Digital Ports

For devices that expose digital I/O as parallel ports (multiple lines read or written together), use read_digital_port() and write_digital_port(). Configure the channel with a port_width that matches the hardware port, then read or write the full port as a single integer value.

from instro.daq.types import Direction, DigitalPortWidth, Logic

# Configure an 8-bit digital output port
daq.configure_digital_port(
    direction=Direction.OUTPUT,
    physical_channel="FIRSTPORTA",
    alias="relay_bank",
    logic=Logic.HIGH,
    port_width=DigitalPortWidth.WIDTH_8,
)

# Write all 8 bits at once (0x0F turns on lines 0-3, off 4-7)
daq.write_digital_port(channel="relay_bank", data=0x0F)

# Configure an 8-bit digital input port and read its value
daq.configure_digital_port(
    direction=Direction.INPUT,
    physical_channel="FIRSTPORTB",
    alias="status_bits",
    logic=Logic.HIGH,
    port_width=DigitalPortWidth.WIDTH_8,
)

measurement = daq.read_digital_port(channel="status_bits")
print(f"Port value: {int(measurement.latest)}")

Port-based I/O is implemented for MCC, NI DAQmx, and Keysight devices. On LabJack, write_digital_port() and read_digital_port() raise NotImplementedError; use write_digital_line() / read_digital_line() to address individual lines instead. Keysight groups a single port into one channel of at most 32 bits, so DigitalPortWidth.WIDTH_64 is rejected — configure a 64-bit span as two channels.

Complete Examples

Example 1: Multi-Channel Software-Timed Acquisition

import time
from instro.daq.drivers.labjack import LabJackTSeriesDriver
from instro.daq import InstroDAQ
from instro.daq.types import Direction
from instro.lib.publishers import NominalCorePublisher

DATASET_RID = "ri.catalog.main.dataset.<your-dataset-rid>"

daq = InstroDAQ(
    name="myDAQ",
    driver=LabJackTSeriesDriver(device_id="440020473"),
)
daq.add_publisher(NominalCorePublisher(dataset_rid=DATASET_RID))

daq.open()

try:
    # Configure multiple channels
    for i, channel in enumerate(["AIN0", "AIN1"]):
        daq.configure_analog_channel(
            direction=Direction.INPUT,
            physical_channel=channel,
            alias=f"ch_{i}",
            range_min=0,
            range_max=5
        )

    # Read 5 times with 1 second between reads
    for i in range(5):
        measurement = daq.read_analog()
        print(measurement)
        time.sleep(1)  # Software sleep until time for next sample

except Exception as e:
    print(f"Exception: {e}")
finally:
    daq.close()

Example 2: Hardware-Timed Continuous Acquisition

import time
from instro.daq.drivers.labjack import LabJackTSeriesDriver
from instro.daq import InstroDAQ
from instro.daq.types import Direction
from instro.lib.publishers import NominalCorePublisher

DATASET_RID = "ri.catalog.main.dataset.<your-dataset-rid>"

daq = InstroDAQ(
    name="myDAQ",
    driver=LabJackTSeriesDriver(device_id="440020473"),
    dataset_rid=DATASET_RID,
)

daq.open()

# Configure channel
daq.configure_analog_channel(
    direction=Direction.INPUT,
    physical_channel="AIN0",
    alias="ch_1",
    range_min=0,
    range_max=5
)

# Configure hardware timing for 100 Hz acquisition
daq.configure_ai_sample_rate(
    sample_rate=100,
    samples_per_channel=10
)

# Start background acquisition
daq.start()

# Let it run until interrupted
while True:
    try:
        time.sleep(1)
    except KeyboardInterrupt:
        print("Exiting main loop")
        break

daq.stop()
daq.close()

Published channels

Every measurement/command call produces a channel keyed under {name}.{descriptor}, where {name} is the constructor argument and {descriptor} is the row below. For analog, digital, and relay operations, {alias} is the alias you assigned with configure_*_channel(alias=...) (or the physical channel name if no alias was provided). If you depended on the pre-v1.0 form for digital channels (bare {alias} with no {name}. prefix and no .cmd suffix), pass legacy_naming=True.

Method	Descriptor	Type
`read_analog()` (per channel)	`{alias}`	telemetry
`write_analog_value(channel)`	`{alias}.cmd`	command
`read_digital_line(channel)` / `read_digital_port(channel)`	`{alias}`	telemetry
`write_digital_line(channel, data)` / `write_digital_port(channel, data)`	`{alias}.cmd`	command
`close_relay(channel)`	`{alias}.cmd` (value `"CLOSED"`)	command
`open_relay(channel)`	`{alias}.cmd` (value `"OPEN"`)	command
`get_points_in_buffer()`	`buffer`	telemetry

The base Instrument background daemon additionally publishes per-iteration diagnostic telemetry on loop_time and daemon_work_time. Those descriptors are shared across all instruments and are not affected by legacy_naming.

Method Reference

Method	Purpose
`InstroDAQ(name, driver, publishers=None, legacy_naming=False, **kwargs)`	Construct an InstroDAQ instance around a vendor driver
`open()`	Establish connection to the DAQ device
`close()`	Disconnect from device and close publishers
`configure_analog_channel(direction, physical_channel, alias, range_min, range_max)`	Configure an analog I/O channel
`configure_digital_line(direction, physical_channel, logic, logic_level=None, alias=None)`	Configure a single digital I/O line
`configure_digital_port(direction, physical_channel, logic, port_width, logic_level=None, alias=None)`	Configure a digital I/O port (multi-line)
`configure_ai_sample_rate(sample_rate, samples_per_channel=None)`	Set hardware timing for analog input
`start()`	Begin hardware-timed acquisition (background daemon)
`stop()`	End hardware-timed acquisition
`read_analog()`	Read analog input (SW-timed) or fetch from buffer (HW-timed)
`read_digital_line(channel)`	Read digital input line
`write_digital_line(channel, data)`	Write to digital output line
`read_digital_port(channel)`	Read all lines on a digital input port as a single integer
`write_digital_port(channel, data)`	Write all lines on a digital output port as a single integer
`add_publisher(publisher)`	Attach a publisher for data routing
`get_points_in_buffer()`	Get number of samples in hardware buffer

Vendor Independence

The power of InstroDAQ is that the same code works across different vendors. To switch vendors, just swap the driver instance:

from instro.daq.drivers.labjack import LabJackTSeriesDriver
from instro.daq.drivers.ni import NIDAQDriver
from instro.daq import InstroDAQ

# LabJack version
daq = InstroDAQ(name="myDAQ", driver=LabJackTSeriesDriver(device_id="440020473"))

# Switch to NI DAQmx: same configuration code works!
daq = InstroDAQ(name="myDAQ", driver=NIDAQDriver(device_id="Dev1"))

The rest of your code (channel configuration, data acquisition, etc.) remains identical.

Driver Development

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 will manage the background daemon for the hardware-timed acquisition.
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_ai_channel(channel): Configure an analog input channel on the hardware; record on self._ai_channels[channel.alias].
configure_ao_channel(channel): Configure an analog output channel; record on self._ao_channels[channel.alias]. Default raises NotImplementedError for drivers that don’t support AO.
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.

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_channel`	`read_analog`, `fetch_analog`, `start`, …
`_ao_channels`	`ao_channels`	`dict[str, AnalogChannel]`	`configure_ao_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_channel(self, channel: AnalogChannel) -> 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 configure_ao_channel 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 call configure_ai_sample_rate() 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

​InstroDAQ

​Supported Vendors

​Key Concepts

​Lifecycle Pattern

​Channels

​Aliases

​Timing Modes

​Creating a InstroDAQ Instance

​Parameters

​Vendor-Specific Driver Examples

​Configuring Channels

​Analog Input Channels

​Scalers

​Analog Output Channels

​Digital Channels

​Hardware-Timed Sample Rate

​Analog Input

​Software-Timed Acquisition

​Hardware-Timed Acquisition with Background Fetching

​Hardware-Timed Acquisition with Manual Fetching

​Analog Output

​Software-Timed Generation

​Digital I/O

​Reading Digital Lines

​Software Timed Generation

​Writing Digital Lines

​Reading and Writing Digital Ports

​Complete Examples

​Example 1: Multi-Channel Software-Timed Acquisition

​Example 2: Hardware-Timed Continuous Acquisition

​Published channels

​Method Reference

​Vendor Independence

​Driver Development

​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