path: root/api_layers/arduino_layer.py

"""
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);
  }
}
*/
"""