summaryrefslogtreecommitdiff
path: root/api_layers
diff options
context:
space:
mode:
Diffstat (limited to 'api_layers')
-rw-r--r--api_layers/__init__.py5
-rw-r--r--api_layers/__pycache__/__init__.cpython-314.pycbin0 -> 324 bytes
-rw-r--r--api_layers/__pycache__/arduino_layer.cpython-314.pycbin0 -> 15070 bytes
-rw-r--r--api_layers/__pycache__/nidaqmx_layer.cpython-314.pycbin0 -> 10251 bytes
-rw-r--r--api_layers/arduino_layer.py295
-rw-r--r--api_layers/nidaqmx_layer.py181
6 files changed, 481 insertions, 0 deletions
diff --git a/api_layers/__init__.py b/api_layers/__init__.py
new file mode 100644
index 0000000..242d6a9
--- /dev/null
+++ b/api_layers/__init__.py
@@ -0,0 +1,5 @@
+# api_layers/__init__.py
+from api_layers.nidaqmx_layer import NidaqmxLayer
+from api_layers.arduino_layer import ArduinoLayer
+
+__all__ = ["NidaqmxLayer", "ArduinoLayer"]
diff --git a/api_layers/__pycache__/__init__.cpython-314.pyc b/api_layers/__pycache__/__init__.cpython-314.pyc
new file mode 100644
index 0000000..88ad51b
--- /dev/null
+++ b/api_layers/__pycache__/__init__.cpython-314.pyc
Binary files differ
diff --git a/api_layers/__pycache__/arduino_layer.cpython-314.pyc b/api_layers/__pycache__/arduino_layer.cpython-314.pyc
new file mode 100644
index 0000000..186b224
--- /dev/null
+++ b/api_layers/__pycache__/arduino_layer.cpython-314.pyc
Binary files differ
diff --git a/api_layers/__pycache__/nidaqmx_layer.cpython-314.pyc b/api_layers/__pycache__/nidaqmx_layer.cpython-314.pyc
new file mode 100644
index 0000000..92f0ff3
--- /dev/null
+++ b/api_layers/__pycache__/nidaqmx_layer.cpython-314.pyc
Binary files differ
diff --git a/api_layers/arduino_layer.py b/api_layers/arduino_layer.py
new file mode 100644
index 0000000..dea2783
--- /dev/null
+++ b/api_layers/arduino_layer.py
@@ -0,0 +1,295 @@
+"""
+api_layers/arduino_layer.py
+
+Arduino serial API layer.
+
+Protocol (default): Arduino sends newline-terminated CSV strings:
+ "A0:1.23,A1:4.56,A2:0.12\n" (analog)
+ "D2:1,D3:0,D4:1\n" (digital)
+
+The Arduino firmware sketch is provided at the bottom of this file
+as a multi-line string for reference / deployment.
+
+Swap for a different protocol by subclassing ArduinoLayer and
+overriding `_parse_line()` and `_build_write_cmd()`.
+
+Usage:
+ from api_layers.arduino_layer import ArduinoLayer
+ layer = ArduinoLayer(port="COM3", baud=115200, simulate=False)
+ if layer.connect():
+ data = layer.read() # {"A0": 3.14, "A1": 1.07, ...}
+ layer.write("D13", 1)
+ layer.disconnect()
+"""
+
+import math
+import random
+import re
+import threading
+import time
+from typing import Dict, List, Optional, Tuple
+
+
+# ── Try pyserial ─────────────────────────────────────────────────────────────
+try:
+ import serial # type: ignore
+ import serial.tools.list_ports # type: ignore
+ _SERIAL_AVAILABLE = True
+except ImportError:
+ _SERIAL_AVAILABLE = False
+
+
+class ArduinoLayer:
+ """
+ Serial communication layer for Arduino-based DAQ nodes.
+
+ Supports:
+ - Auto-detect available serial ports
+ - Configurable baud rate / timeout
+ - Background read thread with latest-value cache
+ - Digital output writes
+ - Full simulation mode (no hardware required)
+ """
+
+ DEFAULT_ANALOG_PINS = ["A0", "A1", "A2", "A3", "A4", "A5"]
+ DEFAULT_DIGITAL_PINS = ["D2", "D3", "D4", "D5", "D6", "D7"]
+
+ def __init__(
+ self,
+ port: str = "COM3",
+ baud: int = 115200,
+ timeout: float = 0.5,
+ analog_pins: List[str] = None,
+ digital_pins: List[str] = None,
+ simulate: bool = True,
+ ):
+ self.port = port
+ self.baud = baud
+ self.timeout = timeout
+ self.analog_pins = analog_pins or self.DEFAULT_ANALOG_PINS
+ self.digital_pins = digital_pins or []
+ self.simulate = simulate or not _SERIAL_AVAILABLE
+
+ self._ser: Optional[object] = None
+ self._cache: Dict[str, float] = {}
+ self._lock = threading.Lock()
+ self._running = False
+ self._thread: Optional[threading.Thread] = None
+ self._t0 = 0.0
+
+ # Sim waveform params
+ self._sim_params = {
+ pin: {
+ "freq": 0.1 + i * 0.13,
+ "amp": 2.5,
+ "offset": 2.5,
+ "noise": 0.01,
+ "phase": i * 1.1,
+ }
+ for i, pin in enumerate(self.analog_pins)
+ }
+
+ # ── Lifecycle ────────────────────────────────────────────────────────
+
+ def connect(self) -> bool:
+ self._t0 = time.time()
+ if self.simulate:
+ self._running = True
+ self._thread = threading.Thread(target=self._sim_loop, daemon=True)
+ self._thread.start()
+ return True
+ if not _SERIAL_AVAILABLE:
+ print("[ArduinoLayer] pyserial not installed.")
+ return False
+ try:
+ self._ser = serial.Serial(
+ port=self.port, baudrate=self.baud, timeout=self.timeout
+ )
+ time.sleep(2.0) # Allow Arduino reset
+ self._ser.reset_input_buffer()
+ self._running = True
+ self._thread = threading.Thread(target=self._read_loop, daemon=True)
+ self._thread.start()
+ return True
+ except Exception as e:
+ print(f"[ArduinoLayer] connect() failed: {e}")
+ return False
+
+ def disconnect(self) -> None:
+ self._running = False
+ if self._thread:
+ self._thread.join(timeout=2.0)
+ if self._ser:
+ try:
+ self._ser.close()
+ except Exception:
+ pass
+ self._ser = None
+
+ # ── Read / Write ────────────────────────────────────────────────────
+
+ def read(self) -> Dict[str, float]:
+ """Return cached latest values for all pins."""
+ with self._lock:
+ return dict(self._cache)
+
+ def write(self, pin: str, value: int) -> bool:
+ """
+ Send digital write command to Arduino.
+ Format sent: "W:D13:1\n"
+ """
+ if self.simulate:
+ return True
+ if self._ser and self._ser.is_open:
+ try:
+ cmd = f"W:{pin}:{int(bool(value))}\n"
+ self._ser.write(cmd.encode())
+ return True
+ except Exception as e:
+ print(f"[ArduinoLayer] write() failed: {e}")
+ return False
+
+ # ── Background threads ───────────────────────────────────────────────
+
+ def _read_loop(self):
+ """Background thread: reads lines from serial port."""
+ while self._running and self._ser and self._ser.is_open:
+ try:
+ line = self._ser.readline().decode("utf-8", errors="replace").strip()
+ if line:
+ parsed = self._parse_line(line)
+ with self._lock:
+ self._cache.update(parsed)
+ except Exception:
+ time.sleep(0.05)
+
+ def _sim_loop(self):
+ """Background thread: generates simulated waveforms."""
+ while self._running:
+ t = time.time() - self._t0
+ update = {}
+ for pin, p in self._sim_params.items():
+ val = p["amp"] * math.sin(2 * math.pi * p["freq"] * t + p["phase"])
+ val += p["offset"]
+ val += random.gauss(0, p["noise"] * p["amp"])
+ # Clamp to 0-5V (Arduino ADC range)
+ update[pin] = round(max(0.0, min(5.0, val)), 4)
+ with self._lock:
+ self._cache.update(update)
+ time.sleep(0.05)
+
+ # ── Protocol helpers ─────────────────────────────────────────────────
+
+ def _parse_line(self, line: str) -> Dict[str, float]:
+ """
+ Parse "A0:1.23,A1:4.56,D2:1" → {"A0": 1.23, "A1": 4.56, "D2": 1.0}
+ Also handles plain CSV "1.23,4.56,0.12" mapped to analog_pins in order.
+ """
+ result: Dict[str, float] = {}
+ # Key:value pairs
+ for token in line.split(","):
+ token = token.strip()
+ if ":" in token:
+ parts = token.split(":", 1)
+ try:
+ result[parts[0].strip()] = float(parts[1].strip())
+ except ValueError:
+ pass
+ else:
+ # plain CSV fallback
+ try:
+ idx = len(result)
+ if idx < len(self.analog_pins):
+ result[self.analog_pins[idx]] = float(token)
+ except ValueError:
+ pass
+ return result
+
+ def _build_write_cmd(self, pin: str, value: int) -> str:
+ return f"W:{pin}:{int(bool(value))}\n"
+
+ # ── Utilities ────────────────────────────────────────────────────────
+
+ @staticmethod
+ def list_ports() -> List[str]:
+ """Return available serial port names."""
+ if not _SERIAL_AVAILABLE:
+ return []
+ return [p.device for p in serial.tools.list_ports.comports()]
+
+ @property
+ def is_simulated(self) -> bool:
+ return self.simulate
+
+ def __repr__(self):
+ mode = "SIM" if self.simulate else f"HW:{self.port}@{self.baud}"
+ return f"<ArduinoLayer {mode} pins={self.analog_pins}>"
+
+
+# ════════════════════════════════════════════════════════════════════════════
+# Arduino Firmware Reference Sketch
+# ════════════════════════════════════════════════════════════════════════════
+ARDUINO_SKETCH = """
+/*
+ * LabDAQ Arduino Firmware
+ * Upload this to your Arduino to communicate with the Python DAQ system.
+ *
+ * Protocol:
+ * SEND (Arduino → PC): "A0:3.14,A1:2.71,A2:1.41,D2:1,D3:0\\n"
+ * RECV (PC → Arduino): "W:D13:1\\n" to set digital outputs
+ *
+ * Analog values are converted from 10-bit ADC (0-1023) to 0.0-5.0 V.
+ */
+
+const int ANALOG_PINS[] = {A0, A1, A2, A3, A4, A5};
+const int DIGITAL_IN[] = {2, 3, 4};
+const int DIGITAL_OUT[] = {5, 6, 7, 13};
+const int N_ANALOG = 6;
+const int N_DIG_IN = 3;
+const int N_DIG_OUT = 4;
+const int SEND_INTERVAL = 50; // ms between transmissions
+
+unsigned long lastSend = 0;
+
+void setup() {
+ Serial.begin(115200);
+ for (int i = 0; i < N_DIG_IN; i++) pinMode(DIGITAL_IN[i], INPUT_PULLUP);
+ for (int i = 0; i < N_DIG_OUT; i++) pinMode(DIGITAL_OUT[i], OUTPUT);
+}
+
+void loop() {
+ // ── Handle incoming commands ────────────────────────────
+ if (Serial.available()) {
+ String cmd = Serial.readStringUntil('\\n');
+ cmd.trim();
+ if (cmd.startsWith("W:")) {
+ // W:D13:1 → set pin 13 HIGH
+ int colon1 = cmd.indexOf(':', 2);
+ int colon2 = cmd.indexOf(':', colon1 + 1);
+ if (colon1 > 0 && colon2 > 0) {
+ String pinStr = cmd.substring(colon1 + 1, colon2);
+ int val = cmd.substring(colon2 + 1).toInt();
+ int pin = pinStr.substring(1).toInt(); // strip 'D'
+ digitalWrite(pin, val ? HIGH : LOW);
+ }
+ }
+ }
+
+ // ── Transmit data ────────────────────────────────────────
+ unsigned long now = millis();
+ if (now - lastSend >= SEND_INTERVAL) {
+ lastSend = now;
+ String out = "";
+ for (int i = 0; i < N_ANALOG; i++) {
+ float v = analogRead(ANALOG_PINS[i]) * (5.0 / 1023.0);
+ out += "A" + String(i) + ":" + String(v, 3);
+ if (i < N_ANALOG - 1) out += ",";
+ }
+ for (int i = 0; i < N_DIG_IN; i++) {
+ out += ",D" + String(DIGITAL_IN[i]) + ":" + String(!digitalRead(DIGITAL_IN[i]));
+ }
+ Serial.println(out);
+ }
+}
+*/
+"""
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}]>"
generated by cgit v1.2.3 (git 2.46.0) at 2026-09-13 19:24:04 +0000