path: root/api_layers/arduino_layer.py

"""
api_layers/arduino_layer.py

Arduino serial API layer.

Protocol:  Arduino sends newline-terminated lines, e.g.:
    "A0:1.23,A1:4.56\n"     key:value pairs  (default)
    "1.23,4.56\n"            plain CSV
    {"A0":1.23,"A1":4.56}    JSON

connect() is non-blocking — serial open is done, then a background
thread reads continuously.  The 2-second Arduino-reset wait is done
inside the thread so the UI never freezes.

Swap protocol by subclassing and overriding _parse_line().
"""

import math
import random
import threading
import time
from typing import Dict, List, Optional, Tuple


try:
    import serial                      # type: ignore
    import serial.tools.list_ports    # type: ignore
    _SERIAL_AVAILABLE = True
except ImportError:
    _SERIAL_AVAILABLE = False


class ArduinoLayer:

    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      = 1.0,
        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   # stored exactly as given — no override

        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
        self._last_error: str = ""          # surfaced to UI for diagnosis

        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:
        """
        Open the connection.
        Simulation: starts waveform thread immediately → returns True.
        Hardware:   opens serial port synchronously (fast), then starts
                    read thread which handles the Arduino reset wait.
        Returns True on success, False on failure.
        Check self.last_error for the reason on failure.
        """
        self._t0 = time.time()
        self._last_error = ""

        if self.simulate:
            self._running = True
            self._thread  = threading.Thread(
                target=self._sim_loop, daemon=True, name="ArduinoSim"
            )
            self._thread.start()
            return True

        if not _SERIAL_AVAILABLE:
            self._last_error = "pyserial not installed — run: pip install pyserial"
            print(f"[ArduinoLayer] {self._last_error}")
            return False

        try:
            self._ser = serial.Serial()
            self._ser.port     = self.port
            self._ser.baudrate = self.baud
            self._ser.timeout  = self.timeout
            self._ser.open()                      # raises SerialException on failure
        except Exception as e:
            self._last_error = str(e)
            print(f"[ArduinoLayer] connect() failed on {self.port}: {e}")
            self._ser = None
            return False

        self._running = True
        self._thread  = threading.Thread(
            target=self._read_loop, daemon=True, name=f"Arduino-{self.port}"
        )
        self._thread.start()
        return True

    def disconnect(self) -> None:
        self._running = False
        if self._thread:
            self._thread.join(timeout=2.0)
            self._thread = None
        if self._ser:
            try:
                self._ser.close()
            except Exception:
                pass
            self._ser = None
        with self._lock:
            self._cache.clear()

    @property
    def is_connected(self) -> bool:
        if self.simulate:
            return self._running
        return self._ser is not None and self._ser.is_open

    @property
    def last_error(self) -> str:
        return self._last_error

    # ── Read / Write ──────────────────────────────────────────────────────

    def read(self) -> Dict[str, float]:
        with self._lock:
            return dict(self._cache)

    def write(self, pin: str, value: int) -> bool:
        if self.simulate:
            return True
        if self._ser and self._ser.is_open:
            try:
                self._ser.write(f"W:{pin}:{int(bool(value))}\n".encode())
                return True
            except Exception as e:
                print(f"[ArduinoLayer] write() failed: {e}")
        return False

    # ── Background threads ────────────────────────────────────────────────

    def _read_loop(self):
        """
        Hardware read thread.
        Waits 2 s for Arduino reset, then reads lines continuously.
        All errors are caught so the thread never crashes silently.
        """
        # Wait for Arduino to reset after serial open
        deadline = time.time() + 2.5
        while time.time() < deadline and self._running:
            time.sleep(0.05)

        if not self._running:
            return

        # Flush any garbage from reset
        try:
            self._ser.reset_input_buffer()
        except Exception:
            pass

        consecutive_errors = 0
        while self._running:
            try:
                if not self._ser or not self._ser.is_open:
                    break
                raw = self._ser.readline()
                if not raw:
                    continue
                line = raw.decode("utf-8", errors="replace").strip()
                if not line:
                    continue
                parsed = self._parse_line(line)
                if parsed:
                    with self._lock:
                        self._cache.update(parsed)
                    consecutive_errors = 0
            except Exception as e:
                consecutive_errors += 1
                if consecutive_errors <= 3:
                    print(f"[ArduinoLayer] read error: {e}")
                if consecutive_errors > 20:
                    print(f"[ArduinoLayer] too many errors, stopping read loop")
                    break
                time.sleep(0.1)

    def _sim_loop(self):
        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"])
                update[pin] = round(max(0.0, min(5.0, val)), 4)
            with self._lock:
                self._cache.update(update)
            time.sleep(0.05)

    # ── Protocol ─────────────────────────────────────────────────────────

    def _parse_line(self, line: str) -> Dict[str, float]:
        """
        Parse common Arduino output formats:
          "A0:1.23,A1:4.56"    → key:value pairs
          "1.23,4.56"          → positional CSV mapped to analog_pins
          {"A0":1.23}          → JSON
        """
        line = line.strip()
        result: Dict[str, float] = {}

        # JSON
        if line.startswith("{"):
            try:
                import json
                d = json.loads(line)
                return {k: float(v) for k, v in d.items()}
            except Exception:
                return {}

        # Key:value CSV
        for token in line.split(","):
            token = token.strip()
            if not token:
                continue
            if ":" in token:
                parts = token.split(":", 1)
                try:
                    result[parts[0].strip()] = float(parts[1].strip())
                except (ValueError, IndexError):
                    pass
            else:
                idx = len(result)
                if idx < len(self.analog_pins):
                    try:
                        result[self.analog_pins[idx]] = float(token)
                    except ValueError:
                        pass
        return result

    # ── Utilities ─────────────────────────────────────────────────────────

    @staticmethod
    def list_ports() -> List[Tuple[str, str]]:
        """
        Return list of (device, description) for all detected serial ports.
        Returns [] if pyserial is not installed.
        """
        if not _SERIAL_AVAILABLE:
            return []
        try:
            return [
                (p.device, p.description or "")
                for p in serial.tools.list_ports.comports()
            ]
        except Exception as e:
            print(f"[ArduinoLayer] list_ports() error: {e}")
            return []

    @staticmethod
    def is_pyserial_available() -> bool:
        return _SERIAL_AVAILABLE

    def __repr__(self):
        mode = "SIM" if self.simulate else f"HW:{self.port}@{self.baud}"
        return f"<ArduinoLayer {mode} pins={self.analog_pins}>"