diff options
Diffstat (limited to 'api_layers/nidaqmx_layer.py')
| -rw-r--r-- | api_layers/nidaqmx_layer.py | 181 |
1 files changed, 181 insertions, 0 deletions
diff --git a/api_layers/nidaqmx_layer.py b/api_layers/nidaqmx_layer.py new file mode 100644 index 0000000..ba40efb --- /dev/null +++ b/api_layers/nidaqmx_layer.py @@ -0,0 +1,181 @@ +""" +api_layers/nidaqmx_layer.py + +NI-DAQmx API abstraction layer. + + • When nidaqmx package + NI runtime are present → uses real hardware + • Otherwise → falls back to simulation + +Swap this layer by changing the `backend` parameter on AnalogInputDevice +or by subclassing NidaqmxLayer and overriding _hw_read(). + +Usage example: + from api_layers.nidaqmx_layer import NidaqmxLayer + layer = NidaqmxLayer(device_name="Dev1", channels=["ai0","ai1"], simulate=False) + layer.start() + values = layer.read() # {"ai0": 1.23, "ai1": -0.45} + layer.stop() +""" + +import math +import random +import time +from typing import Dict, List, Optional + +# ── Try to import real nidaqmx ────────────────────────────────────────────── +try: + import nidaqmx # type: ignore + from nidaqmx.constants import TerminalConfiguration # type: ignore + _NI_AVAILABLE = True +except ImportError: + _NI_AVAILABLE = False + + +class NidaqmxLayer: + """ + Thin wrapper around nidaqmx.Task for analog input. + + Parameters + ---------- + device_name : str + NI device identifier, e.g. "Dev1" + channels : list of str + Physical channel names relative to device, e.g. ["ai0", "ai1", "ai2"] + sample_rate : float + Samples per second (hardware mode only; ignored in sim) + min_val / max_val : float + Expected voltage range for hardware task configuration + simulate : bool + Force simulation even if nidaqmx is available + """ + + def __init__( + self, + device_name: str = "Dev1", + channels: List[str] = None, + sample_rate: float = 1000.0, + min_val: float = -10.0, + max_val: float = 10.0, + simulate: bool = True, + ): + self.device_name = device_name + self.channels = channels or ["ai0", "ai1", "ai2", "ai3"] + self.sample_rate = sample_rate + self.min_val = min_val + self.max_val = max_val + self.simulate = simulate or not _NI_AVAILABLE + + self._task = None + self._started = False + self._t0 = 0.0 + + # Sim waveform params per channel + self._sim_params = [ + { + "freq": 0.3 + i * 0.17, + "amp": (max_val - min_val) * 0.4, + "offset": (max_val + min_val) / 2, + "noise": 0.02, + "phase": i * 0.8, + } + for i in range(len(self.channels)) + ] + + # ── Lifecycle ─────────────────────────────────────────────────────── + + def start(self) -> bool: + """Configure and start acquisition. Returns True on success.""" + self._t0 = time.time() + if self.simulate: + self._started = True + return True + try: + self._task = nidaqmx.Task() + for ch in self.channels: + physical = f"{self.device_name}/{ch}" + self._task.ai_channels.add_ai_voltage_chan( + physical, + min_val=self.min_val, + max_val=self.max_val, + terminal_config=TerminalConfiguration.RSE, + ) + self._task.timing.cfg_samp_clk_timing( + rate=self.sample_rate, + sample_mode=nidaqmx.constants.AcquisitionType.CONTINUOUS, + samps_per_chan=int(self.sample_rate), + ) + self._task.start() + self._started = True + return True + except Exception as e: + print(f"[NidaqmxLayer] start() failed: {e}") + self._started = False + return False + + def stop(self) -> None: + self._started = False + if self._task is not None: + try: + self._task.stop() + self._task.close() + except Exception: + pass + self._task = None + + # ── Read ──────────────────────────────────────────────────────────── + + def read(self) -> Dict[str, float]: + """Return latest sample per channel as {channel_name: voltage}.""" + if not self._started: + return {} + if self.simulate: + return self._sim_read() + return self._hw_read() + + def _hw_read(self) -> Dict[str, float]: + """Read one sample per channel from hardware.""" + try: + samples = self._task.read(number_of_samples_per_channel=1) + # nidaqmx returns list-of-lists when multiple channels + if len(self.channels) == 1: + samples = [samples] + return {ch: float(samples[i][0]) for i, ch in enumerate(self.channels)} + except Exception as e: + print(f"[NidaqmxLayer] read() failed: {e}") + return {} + + def _sim_read(self) -> Dict[str, float]: + t = time.time() - self._t0 + result = {} + for i, ch in enumerate(self.channels): + p = self._sim_params[i] + val = p["amp"] * math.sin(2 * math.pi * p["freq"] * t + p["phase"]) + p["offset"] + val += random.gauss(0, p["noise"] * p["amp"]) + val = max(self.min_val, min(self.max_val, val)) + result[ch] = round(val, 5) + return result + + # ── Introspection ─────────────────────────────────────────────────── + + @staticmethod + def list_devices() -> List[str]: + """Return list of detected NI device names, or [] if unavailable.""" + if not _NI_AVAILABLE: + return [] + try: + system = nidaqmx.system.System.local() + return [d.name for d in system.devices] + except Exception: + return [] + + @property + def is_simulated(self) -> bool: + return self.simulate + + @property + def ni_available(self) -> bool: + return _NI_AVAILABLE + + def __repr__(self): + mode = "SIM" if self.simulate else "HW" + return f"<NidaqmxLayer {self.device_name} ch={self.channels} [{mode}]>" |