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 follows 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 acquisition (if using hardware timing or the software-timed daemon)
Acquire data - Read/fetch measurements
stop() - End acquisition (if started)
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 three acquisition modes: Software-Timed (Manual Polling)

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

Software-Timed (Background Daemon)

Configure the polling rate with configure_sw_sample_rate()
Call start() to launch the background daemon, which reads every configured analog and digital input channel once per period and publishes the results
Use case: Low-frequency continuous monitoring without a hardware sample clock

Hardware-Timed (Buffered Acquisition)

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

Hardware and software timing are mutually exclusive on one instance: configuring the second mode raises TimingConfigException. To run both against one device, build a separate InstroDAQ per mode with non-overlapping channels: see the NI hardware- and software-timed example.

Creating an 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.

Typed Channel Configuration

Alongside the generic configure_analog_channel / configure_digital_line, InstroDAQ exposes measurement-specific configuration methods. Each builds a typed channel (AnalogVoltageChannel, AnalogCurrentChannel, AnalogThermocoupleChannel, or a DigitalLineChannel) and delegates to a dedicated driver method:

Method	Channel type	Driver method
`configure_voltage_input(physical_channel, *, alias, range_min, range_max, scaler, terminal_config)`	`AnalogVoltageChannel`	`configure_ai_voltage_channel`
`configure_voltage_output(physical_channel, *, alias, range_min, range_max, scaler)`	`AnalogVoltageChannel`	`configure_ao_voltage_channel`
`configure_current_input(physical_channel, *, alias, range_min, range_max, scaler)`	`AnalogCurrentChannel`	`configure_ai_current_channel`
`configure_current_output(physical_channel, *, alias, range_min, range_max, scaler)`	`AnalogCurrentChannel`	`configure_ao_current_channel`
`configure_thermocouple_input(physical_channel, tc_type, *, unit, alias, range_min, range_max, scaler, cjc_source, cjc_temp, cjc_channel, tc_input_scaler)`	`AnalogThermocoupleChannel`	`configure_ai_thermocouple_channel`
`configure_digital_input(physical_channel, *, logic, logic_level, alias)`	`DigitalLineChannel`	`configure_di_line_channel`
`configure_digital_output(physical_channel, *, logic, logic_level, alias)`	`DigitalLineChannel`	`configure_do_line_channel`

Driver support. NI-DAQmx implements every typed method above. LabJack T-Series implements the typed voltage, thermocouple, and digital methods; current input/output is not supported by LabJack hardware, so the driver points to the measuring current app note.

LabJack thermocouple channels

The LabJack driver registers a thermocouple’s AIN# in the scan list as raw volts and converts each sample to temperature on read with LJM’s TCVoltsToTemp and the devices internal CJC. More info about using thermocouples with a LabJack device can be found here.

from instro.daq.scaling.thermocouple import TC_TYPE, TC_UNIT

daq.configure_thermocouple_input("AIN0", TC_TYPE.K, alias="tc0", unit=TC_UNIT.CELSIUS)

Cold-junction compensation depends on cjc_source:

INTERNAL (default): the driver reads the device’s cold-junction sensor alongside the thermocouple. The T8 streams the per-channel TEMPERATURE# sensor, the T7 streams the internal sensor’s raw volts (AIN14), and the T4 snapshots TEMPERATURE_DEVICE_K before each software-timed read (the TEMPERATURE_DEVICE_K register cannot be streamed).
CONSTANT: pass the reference junction temperature via cjc_temp, expressed in the channel’s unit.
CHANNEL: not yet supported by this driver.

The T4 snapshots CJC once when the stream starts, so its thermocouple readings can drift over a long hardware-timed acquisition.

To back external signal conditioning out of the measured volts before conversion, pass a tc_input_scaler when configuring the channel. The T4’s 12-bit ADC cannot resolve a bare thermocouple, so LabJack recommends using an LJTick-InAmp: the T4 defaults tc_input_scaler to the InAmp’s x51 gain and 1.25 V offset jumpers.

from instro.daq.scaling.scaling import ReverseLinearScaler

daq.configure_thermocouple_input(
    "AIN4",
    TC_TYPE.K,
    unit=TC_UNIT.CELSIUS,
    tc_input_scaler=ReverseLinearScaler(gain=51, offset=1.25, units="V"),
)

unit is required and is a TC_UNIT member. It governs cjc_temp, range_min/range_max, and the returned readings. A channel scaler applies to the converted temperatures.

daq.configure_thermocouple_input(
    "AIN0", TC_TYPE.K, alias="tc0", unit=TC_UNIT.FAHRENHEIT, cjc_source="CONSTANT", cjc_temp=77.0
)

Hardware-Timed Sample Rate

For continuous hardware-timed acquisition, configure the sample rate with configure_hw_sample_rate(). It applies to every configured input channel.

daq.configure_hw_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

A single InstroDAQ instance carries one analog input sample rate. To run multiple NI DAQmx tasks at different hardware sample rates, create one InstroDAQ instance per task: see the NI multi-rate acquisition example.

Software-Timed Sample Rate

For continuous acquisition without a hardware sample clock, configure a software-timed polling rate instead. The background daemon then paces read_analog() / read_digital() calls at 1 / sample_rate.

daq.configure_sw_sample_rate(
    sample_rate=10.0,  # Hz, paces the background daemon loop
)

The device never starts a buffered acquisition in this mode: each daemon iteration triggers an immediate conversion, timestamped when the read returns. The configured rate is a ceiling, not a guarantee. If a DAQ read takes longer than the period, the daemon runs as fast as the reads allow.

Analog Input

Software-Timed Acquisition (Manual Polling)

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)
values: Property returning all values as a list (convenience for single-channel reads)
latest: Property returning the most recent value (convenience for single-channel reads)

Software-Timed Acquisition with the Background Daemon

For continuous software-timed acquisition, configure a polling rate and start() the background daemon. The daemon calls read_analog() once per period and publishes the results.

from instro.daq.types import Direction

daq.open()

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

# Pace the background daemon at 10 Hz (no hardware clock is involved)
daq.configure_sw_sample_rate(sample_rate=10)

daq.start()

# Pull the latest sample from the channel buffer whenever you need it
measurement = daq.get_channel("sensor_1", wait_for_new_samples=True)
print(f"Value: {measurement.latest}")

daq.stop()
daq.close()

While the daemon is running, read() and read_batch() serve analog and digital channels from the daemon’s channel buffer, so the same read code works whether or not the daemon is running. read_analog() and read_digital() raise instead: the daemon owns the hardware reads. Call get_channel() directly for more control over buffered reads (sample count, blocking for new samples, timeouts). See the software-timed example.

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_hw_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_new_samples = 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, start without the background daemon and manually call read_analog().

daq.open()

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

# Start hardware acquisition (fills buffer) without spinning the background daemon
daq.start(background=False)

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

With start(background=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. On NI DAQmx, port_width is checked against the port’s physical line count and a mismatch raises ValueError — pass the DigitalPortWidth that matches the hardware port.

Unified Read and Write

Reading by Alias

read() accepts a single alias and returns that channel’s Measurement. read_batch() accepts a list of aliases, or None for every configured input channel, and returns a dict keyed by alias, with each value the channel’s Measurement. Analog aliases are served from one batched analog read; digital aliases are read per line or port. While the background daemon is running, both analog and digital aliases are served from the daemon’s channel buffer (latest sample) instead of triggering a hardware read; call get_channel() directly when you need more control (sample count, blocking for new samples, timeouts).

daq.configure_voltage_input(physical_channel="ai0", alias="v0")
daq.configure_digital_input(physical_channel="DIO0", alias="limit_switch")

# A single channel
measurement = daq.read("v0")
print(measurement.latest)

# Several channels of mixed type
result = daq.read_batch(["v0", "limit_switch"])
print(result["v0"].latest, result["limit_switch"].latest)

# Every configured input channel
result = daq.read_batch()

Writing by Alias

write() writes a single value to a single alias and returns the resulting Command. write_batch() writes values[i] to channels[i] and returns a list of Command objects. Both route each alias to analog or digital output by channel type. By default, the first write that fails at the device logs a warning and raises RuntimeError naming the failed channel; later channels in the batch are not written. Pass continue_on_failed_write=True to log the failure and continue with the remaining channels instead; the returned list then contains only the successful commands. Every write emits a per-channel log line: <alias> -> succeeded at debug level, or <alias> -> failed: <reason> at warning level.

daq.configure_voltage_output(physical_channel="ao0", alias="ao0")
daq.configure_digital_output(physical_channel="DIO1", alias="enable")

# A single channel
daq.write("ao0", 2.2)

# Several channels, paired by index
daq.write_batch(["ao0", "enable"], [2.2, 1])

Values may be float, int, or bool, and every channel and value is validated before anything is written: each alias must be configured as an output channel (unknown aliases raise KeyError), aliases may appear only once per batch, analog outputs require a finite number within the channel’s configured range_min/range_max, digital lines require 0 or 1 (in any of the three forms), and digital ports require an integer. An invalid value, a duplicate alias, or a channels/values length mismatch raises ValueError with no channels written. Digital values are coerced to int. Relays are not routed through write(); use open_relay() and close_relay().

Batched Digital Input

read_digital() mirrors read_analog(): it reads every configured digital input channel in one driver call, dispatching a software-timed sample or a hardware-timed buffer fetch based on the configured timing mode. Lines and ports are both covered — a line channel carries 0/1, a port channel carries its whole N-bit integer:

# One driver round-trip for every configured digital input
measurement = daq.read_digital()
print(measurement.channel_data)  # {"myDAQ.limit_switch": [1.0], "myDAQ.status_bits": [15.0]}

read_digital() returns one Measurement per timebase cluster. There is no batched digital write; use write_digital_line() and write_digital_port().

No packaged driver implements the read_digital / fetch_digital driver methods yet, and read_batch() still reads digital aliases per line or port.

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_hw_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
`read(channel)` / `read_batch(channels)` (per channel)	`{alias}`	telemetry
`write(channel, value)` / `write_batch(channels, values)` (per channel)	`{alias}.cmd`	command
`read_digital()` (all DI channels in one Measurement)	`{alias}`	telemetry
`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_voltage_input(physical_channel, ...)`	Configure a typed analog voltage input channel
`configure_voltage_output(physical_channel, ...)`	Configure a typed analog voltage output channel
`configure_current_input(physical_channel, ...)`	Configure a typed analog current input channel
`configure_current_output(physical_channel, ...)`	Configure a typed analog current output channel
`configure_thermocouple_input(physical_channel, tc_type, ...)`	Configure a thermocouple input channel
`configure_digital_input(physical_channel, ...)`	Configure a typed digital input line
`configure_digital_output(physical_channel, ...)`	Configure a typed digital output line
`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_hw_sample_rate(sample_rate, samples_per_channel=None)`	Set hardware timing
`configure_sw_sample_rate(sample_rate)`	Set the software-timed polling rate
`start()`	Begin acquisition (device task, background daemon, or both)
`stop()`	End acquisition and stop the background daemon
`read_analog()`	Read analog input (SW-timed) or fetch from buffer (HW-timed)
`read(channel)`	Read one input channel by alias; returns `Measurement`
`read_batch(channels=None)`	Read input channel(s) by alias; returns `{alias: Measurement}`
`write(channel, value)`	Write value to one output channel by alias; returns `Command`
`write_batch(channels, values, continue_on_failed_write=False)`	Write values to output channels by alias, paired by index; returns `list[Command]`. By default raises on the first failed write; `continue_on_failed_write=True` logs, skips, and continues
`read_digital()`	Read every digital input channel, line and port (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 manages the background daemon for both timing modes; software-timed pacing never reaches the driver.
Data format conversion: Convert vendor-specific data formats to Measurement objects.
Constraint validation: Handle vendor-specific hardware constraints and raise clear exceptions when the vendor library cannot.

DAQDriverBase Interface

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

Connection Management

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

Channel Configuration

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

configure_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.

The analog methods below back InstroDAQ’s typed channel configuration. Of the packaged drivers, only NI-DAQmx and LabJack T-Series implement them.

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.

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() / fetch_digital(): Sample every configured DI channel in one call, lines and ports alike
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 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

​InstroDAQ

​Supported Vendors

​Key Concepts

​Lifecycle Pattern

​Channels

​Aliases

​Timing Modes

​Creating an InstroDAQ Instance

​Parameters

​Vendor-Specific Driver Examples

​Configuring Channels

​Analog Input Channels

​Scalers

​Analog Output Channels

​Digital Channels

​Typed Channel Configuration

​LabJack thermocouple channels

​Hardware-Timed Sample Rate

​Software-Timed Sample Rate

​Analog Input

​Software-Timed Acquisition (Manual Polling)

​Software-Timed Acquisition with the Background Daemon

​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

​Unified Read and Write

​Reading by Alias

​Writing by Alias

​Batched Digital Input

​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