Metadata-Version: 2.4
Name: nidaqwrapper
Version: 0.1.0
Summary: Unified NI-DAQmx Python Wrapper
Author: Tibor Barsi
License-Expression: MIT
Project-URL: Homepage, https://github.com/tibor-barsi/nidaqwrapper
Project-URL: Repository, https://github.com/tibor-barsi/nidaqwrapper
Keywords: nidaqmx,daq,data-acquisition,national-instruments,measurement
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: Topic :: Scientific/Engineering
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Operating System :: OS Independent
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: numpy>=1.20
Requires-Dist: nidaqmx>=0.8.0
Requires-Dist: pyTrigger>=0.3.0
Requires-Dist: tomli>=1.0; python_version < "3.11"
Provides-Extra: examples
Requires-Dist: jupyter; extra == "examples"
Requires-Dist: matplotlib; extra == "examples"
Dynamic: license-file

# nidaqwrapper

Unified NI-DAQmx Python Wrapper.

A Python package that provides a clean, high-level interface to NI-DAQmx hardware. It consolidates analog input, analog output, and digital I/O into a single package with two layers: task classes for channel configuration and orchestrator classes for acquisition lifecycle management.

The architecture uses direct delegation -- `nidaqmx.Task` is the single source of truth. No intermediate state is maintained; every channel addition and timing configuration delegates immediately to the underlying driver.

## Installation

```bash
pip install nidaqwrapper
```

**Requires NI-DAQmx drivers installed on the system.** The `nidaqmx` Python package communicates with the NI-DAQmx C driver, which must be installed separately from [ni.com](https://www.ni.com/en/support/downloads/drivers/download.ni-daq-mx.html).

For development:

```bash
git clone https://github.com/tibor-barsi/nidaqwrapper.git
cd nidaqwrapper
pip install -e ".[dev]"
```

## Quick Start

Accelerometer acquisition with software triggering:

```python
from nidaqwrapper import AITask, DAQHandler

# Define an analog input task with two accelerometer channels
task = AITask('vibration_test', sample_rate=25600)
task.add_channel('accel_x', device_ind=0, channel_ind=0,
                 sensitivity=100, sensitivity_units='mV/g', units='g')
task.add_channel('accel_y', device_ind=0, channel_ind=1,
                 sensitivity=100, sensitivity_units='mV/g', units='g')

# Use DAQHandler for triggered acquisition
wrapper = DAQHandler()
wrapper.configure(task_in=task)
wrapper.connect()
wrapper.set_trigger(n_samples=25600, trigger_channel=0,
                    trigger_level=0.5, trigger_type='up', presamples=2560)
data = wrapper.acquire()  # shape: (25600, 2)
wrapper.disconnect()
```

## Features

- **Analog input** (`AITask`) -- voltage, accelerometer (IEPE), force (IEPE), and custom linear scales
- **Analog output** (`AOTask`) -- voltage generation with continuous buffer regeneration
- **Digital I/O** (`DITask` / `DOTask`) -- on-demand single-sample and clocked continuous modes
- **Single-task handler** (`DAQHandler`) -- configure, connect, acquire/generate, disconnect lifecycle with software triggering via pyTrigger
- **Multi-task synchronization** (`MultiHandler`) -- hardware-triggered finite acquisition and validated multi-task pipelines
- **TOML configuration** -- `save_config()` / `from_config()` for portable, human-readable task definitions with device aliases
- **Device discovery** -- `list_devices()`, `list_tasks()`, `get_connected_devices()` for hardware enumeration
- **Raw task injection** -- `from_task()` on all task classes wraps pre-configured `nidaqmx.Task` objects
- **Context manager support** -- automatic resource cleanup with `with` statements
- **Thread safety** -- `DAQHandler` and `MultiHandler` use per-instance `RLock` for concurrent access

## Usage Examples

### Load an NI MAX task by name

```python
from nidaqwrapper import DAQHandler

wrapper = DAQHandler()
wrapper.configure(task_in='MyInputTask', task_out='MyOutputTask')
wrapper.connect()
data = wrapper.acquire()
wrapper.disconnect()
```

### Digital output

```python
from nidaqwrapper import DOTask

with DOTask('relay_control') as do:
    do.add_channel('relays', lines='Dev1/port0/line0:3')
    do.start()
    do.write([True, False, True, False])
```

### Digital input

```python
from nidaqwrapper import DITask

with DITask('switches') as di:
    di.add_channel('sw', lines='Dev1/port0/line0:3')
    di.start()
    state = di.read()  # array of bool values, one per line
```

### Analog output (continuous waveform)

```python
import numpy as np
from nidaqwrapper import AOTask

task = AOTask('sig_gen', sample_rate=10000)
task.add_channel('ao_0', device_ind=0, channel_ind=0)
task.start()

t = np.linspace(0, 1, 10000)
signal = np.sin(2 * np.pi * 10 * t)  # 10 Hz sine
task.generate(signal)
# ...
task.clear_task()
```

### TOML configuration (portable across machines)

```python
from nidaqwrapper import AITask

# Save task configuration
task = AITask('vibration', sample_rate=25600)
task.add_channel('ch0', device_ind=0, channel_ind=0,
                 sensitivity=100, sensitivity_units='mV/g', units='g')
task.save_config('vibration.toml')
task.clear_task()

# Recreate the same task on another machine
task = AITask.from_config('vibration.toml')
task.start()
```

The generated TOML file uses device aliases, so only the `[devices]` section needs editing when moving between machines:

```toml
[task]
name = "vibration"
sample_rate = 25600
type = "input"

[devices]
dev0 = "cDAQ1Mod1"  # NI 9234

[[channels]]
name = "ch0"
device = "dev0"
channel = 0
sensitivity = 100
sensitivity_units = "mV/g"
units = "g"
```

### Multi-task synchronized acquisition

```python
from nidaqwrapper import MultiHandler

adv = MultiHandler()
adv.configure(input_tasks=[task1, task2])
adv.connect()
adv.set_trigger(n_samples=25600, trigger_channel=0, trigger_level=0.5)
data = adv.acquire()
adv.disconnect()
```

### Context manager (automatic cleanup)

```python
from nidaqwrapper import DAQHandler

with DAQHandler(task_in='MyTask') as wrapper:
    wrapper.connect()
    wrapper.set_trigger(n_samples=1000, trigger_channel=0, trigger_level=0.1)
    data = wrapper.acquire()
# Resources automatically cleaned up
```

### Wrap a raw nidaqmx.Task

```python
import nidaqmx
from nidaqwrapper import AITask

raw_task = nidaqmx.Task('external')
raw_task.ai_channels.add_ai_voltage_chan('Dev1/ai0')
raw_task.timing.cfg_samp_clk_timing(rate=25600)

wrapped = AITask.from_task(raw_task)
# Use wrapped task with DAQHandler or read directly
data = wrapped.acquire()  # shape: (n_channels, n_samples)
raw_task.close()  # Caller retains ownership
```

## API Reference

### Task Classes

| Class | Module | Purpose |
|-------|--------|---------|
| `AITask` | `ai_task` | Analog input -- channels, timing, acquisition |
| `AOTask` | `ao_task` | Analog output -- channels, timing, generation |
| `DITask` | `digital` | Digital input -- on-demand and clocked reads |
| `DOTask` | `digital` | Digital output -- on-demand and clocked writes |

### Orchestrators

| Class | Module | Purpose |
|-------|--------|---------|
| `DAQHandler` | `handler` | Single-task handler with software triggering, auto-reconnection |
| `MultiHandler` | `multi_handler` | Multi-task orchestrator with hardware trigger validation |

### Utility Functions

| Function | Purpose |
|----------|---------|
| `list_devices()` | List connected NI-DAQmx devices with product types |
| `list_tasks()` | List tasks saved in NI MAX |
| `get_connected_devices()` | Get set of connected device name strings |
| `get_task_by_name(name)` | Load a pre-configured task from NI MAX |
| `UNITS` | Dict mapping unit strings (`'g'`, `'mV/g'`, `'V'`, etc.) to nidaqmx constants |

## Data Format

The public API uses `(n_samples, n_channels)` for all multi-channel data. Internal transposition to nidaqmx's `(n_channels, n_samples)` layout is handled automatically.

- `DAQHandler.acquire()` returns `(n_samples, n_channels)` or a dict
- `DAQHandler.read_all_available()` returns `(n_samples, n_channels)`
- `DAQHandler.read()` returns `(n_channels,)` -- single sample
- `AITask.acquire()` returns `(n_channels, n_samples)` -- internal format
- `AOTask.generate(signal)` accepts `(n_samples, n_channels)` or `(n_samples,)`

## Requirements

- Python >= 3.9
- NI-DAQmx drivers (system-level installation)
- numpy >= 1.20
- nidaqmx >= 0.8.0
- pyTrigger >= 0.3.0
- tomli >= 1.0 (Python < 3.11 only; Python 3.11+ uses built-in `tomllib`)

## Testing

nidaqwrapper uses a three-tier test strategy:

| Tier | Command | Requirements |
|------|---------|-------------|
| Mocked | `uv run pytest` | None (default) |
| Simulated | `uv run pytest -m simulated -v` | NI-DAQmx driver + simulated device |
| Hardware | `uv run pytest -m hardware -v` | Physical NI hardware |

The mocked tier (630 tests) runs by default and requires no NI-DAQmx driver. The simulated tier uses the real driver with simulated devices to catch API contract violations. The hardware tier validates real-world timing and physical signals.

See [TESTING.md](TESTING.md) for detailed setup instructions, troubleshooting, and how to configure simulated devices.

## License

MIT License -- Copyright (c) 2026 Tibor Barsi and contributors
