From 67cfa0a514c7de4605ed7360e15a81aa781e510e Mon Sep 17 00:00:00 2001
From: Christian Kolset <christian.kolset@gmail.com>
Date: Fri, 24 Apr 2026 11:51:26 -0600
Subject: Added difference serial protocols

---
 api_layers/protocols/base_protocol.py | 120 ++++++++++++++++++++++++++++++++++
 1 file changed, 120 insertions(+)
 create mode 100644 api_layers/protocols/base_protocol.py

(limited to 'api_layers/protocols/base_protocol.py')

diff --git a/api_layers/protocols/base_protocol.py b/api_layers/protocols/base_protocol.py
new file mode 100644
index 0000000..096151d
--- /dev/null
+++ b/api_layers/protocols/base_protocol.py
@@ -0,0 +1,120 @@
+"""
+api_layers/protocols/base_protocol.py
+
+Abstract base for serial instrument protocol layers.
+
+Background thread calls _poll() at poll_interval and stores results in
+self._cache — same threading model as ArduinoLayer.
+Simulation uses _simulate(t) instead.
+"""
+
+import threading
+import time
+from abc import ABC, abstractmethod
+from typing import Dict, List, Optional, Tuple
+
+
+class BaseProtocol(ABC):
+
+    def __init__(
+        self,
+        port:          str,
+        baud:          int,
+        poll_interval: float = 0.1,
+        simulate:      bool  = True,
+    ):
+        self.port          = port
+        self.baud          = baud
+        self.poll_interval = poll_interval
+        self.simulate      = simulate
+
+        self._cache:  Dict[str, float]          = {}
+        self._lock    = threading.Lock()
+        self._running = False
+        self._thread: Optional[threading.Thread] = None
+        self._ser     = None
+
+    # ── Public API ────────────────────────────────────────────────────────
+
+    def connect(self) -> bool:
+        if self.simulate:
+            self._running = True
+            self._thread  = threading.Thread(target=self._sim_loop, daemon=True)
+            self._thread.start()
+            return True
+        try:
+            import serial
+            self._ser = serial.Serial(
+                self.port, self.baud, timeout=1.0,
+                **self._serial_kwargs(),
+            )
+            time.sleep(0.05)
+            self._running = True
+            self._thread  = threading.Thread(target=self._poll_loop, daemon=True)
+            self._thread.start()
+            return True
+        except Exception as e:
+            print(f"[{type(self).__name__}] connect failed on {self.port}: {e}")
+            return False
+
+    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
+
+    def read(self) -> Dict[str, float]:
+        with self._lock:
+            return dict(self._cache)
+
+    @classmethod
+    def list_ports(cls) -> List[Tuple[str, str]]:
+        try:
+            import serial.tools.list_ports
+            return [(p.device, p.description) for p in serial.tools.list_ports.comports()]
+        except ImportError:
+            return []
+
+    # ── Internal ──────────────────────────────────────────────────────────
+
+    def _serial_kwargs(self) -> dict:
+        return {}
+
+    def _poll_loop(self) -> None:
+        while self._running:
+            try:
+                result = self._poll()
+                if result:
+                    with self._lock:
+                        self._cache.update(result)
+            except Exception as e:
+                print(f"[{type(self).__name__}] poll error: {e}")
+            time.sleep(self.poll_interval)
+
+    def _sim_loop(self) -> None:
+        t0 = time.time()
+        while self._running:
+            t = time.time() - t0
+            with self._lock:
+                self._cache = self._simulate(t)
+            time.sleep(self.poll_interval)
+
+    # ── Abstract ──────────────────────────────────────────────────────────
+
+    @abstractmethod
+    def _poll(self) -> Dict[str, float]:
+        """Read instrument, return {channel_id: value}."""
+
+    @abstractmethod
+    def _simulate(self, t: float) -> Dict[str, float]:
+        """Return simulated values at elapsed time t (seconds)."""
+
+    @abstractmethod
+    def write(self, channel_id: str, value) -> bool:
+        """Send command to instrument."""
-- 
cgit v1.2.3