> ## 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.
# Data Acquisition (DAQ)
> Using InstroDAQ for vendor-independent data acquisition
# 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:
1. **`InstroDAQ(name, driver, ...)`** - Instantiate the DAQ with a vendor driver
2. **`open()`** - Establish connection to the hardware
3. **Configure** - Set up channels, timing, and other settings
4. **`start()`** - Begin hardware-timed acquisition (if using hardware timing)
5. **Acquire data** - Read/fetch measurements
6. **`stop()`** - End acquisition (if using hardware timing)
7. **`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 an InstroDAQ Instance
Construct a vendor driver and pass it to `InstroDAQ`:
```python theme={null}
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
```python theme={null}
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"), # "" or ":"
)
# Keysight 34980A (SCPI/VISA device: driver owns its VisaDriver)
from instro.daq.drivers import Keysight34980A
daq = InstroDAQ(
name="keysightDAQ",
driver=Keysight34980A("TCPIP0::::INSTR"),
)
```
## Configuring Channels
### Analog Input Channels
Configure analog input channels with measurement range and logical names:
```python theme={null}
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.
```python theme={null}
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.
1. Scaling out the amplifier to get to the actual voltage seen across the thermocouple terminals.
2. Scaling the voltage seen across the thermocouple terminals to temperature.
```python theme={null}
from instro.daq.scaling import ScalerPipeline, ReverseLinearScaler
from instro.daq.scaling.thermocouple import ThermocoupleSensor
scaler_pipeline = ScalerPipeline(ReverseLinearScaler(gain=, offset=, units="V"), ThermocoupleSensor(type = , cjc_temp = ))
```
You can create your own scaler by subclassing `Scaler` and implementing `scale` and `units` methods.
```python theme={null}
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:
```python theme={null}
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:
```python theme={null}
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:
```python theme={null}
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.
```python theme={null}
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 Constraints**
Different 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](/instrumentation/examples/daq/daq_multi_rate_ni).
## Analog Input
### Software-Timed Acquisition
For manual, on-demand sampling:
```python theme={null}
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)
### Hardware-Timed Acquisition with Background Fetching
See [Two ways to get data](/instrumentation/overview#two-ways-to-get-data) for more information regarding background fetching of measurements.
For continuous high-speed acquisition.
```python theme={null}
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."
# 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_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 Publishers**
Data 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()`.
```python theme={null}
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) 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:
```python theme={null}
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
```python theme={null}
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:
```python theme={null}
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
```python theme={null}
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.
```python theme={null}
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.
## Complete Examples
### Example 1: Multi-Channel Software-Timed Acquisition
```python theme={null}
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."
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
```python theme={null}
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."
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`](/instrumentation/library#backwards-compatible-channel-naming-legacy_naming).
| 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:
```Python theme={null}
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](#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` will manage the background daemon for the hardware-timed acquisition.
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_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
```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_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._[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
```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 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