""" 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""