summaryrefslogtreecommitdiff
path: root/api_layers/arduino_layer.py
diff options
context:
space:
mode:
Diffstat (limited to 'api_layers/arduino_layer.py')
-rw-r--r--api_layers/arduino_layer.py295
1 files changed, 295 insertions, 0 deletions
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);
+ }
+}
+*/
+"""
generated by cgit v1.2.3 (git 2.46.0) at 2026-09-13 20:32:34 +0000