1 files changed, 102 insertions, 0 deletions
diff --git a/api_layers/protocols/scpi.py b/api_layers/protocols/scpi.py
new file mode 100644
index 0000000..730f8e0
--- /dev/null
+++ b/api_layers/protocols/scpi.py
@@ -0,0 +1,102 @@
+"""
+api_layers/protocols/scpi.py
+
+SCPI (Standard Commands for Programmable Instruments) protocol layer.
+
+Each channel maps to a query string.  The layer sends each query in
+sequence and parses the numeric response.
+
+Terminator: LF (\\n) by default; instruments also accept CR+LF.
+Response parsing strips unit suffixes — "+3.14159 V" → 3.14159.
+
+Example channel setup:
+  SCPIChannel("V1", "Voltage",  query="MEAS:VOLT?",         unit="V")
+  SCPIChannel("I1", "Current",  query="MEAS:CURR?",         unit="A")
+  SCPIChannel("T1", "Temp",     query="SENS:TEMP:DATA?",    unit="°C")
+"""
+
+import math
+import re
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional
+
+from api_layers.protocols.base_protocol import BaseProtocol
+
+
+@dataclass
+class SCPIChannel:
+    channel_id: str
+    name:       str
+    query:      str          # e.g. "MEAS:VOLT?" or "MEAS:VOLT? (@1)"
+    unit:       str  = ""
+    scale:      float = 1.0
+    write_cmd:  str  = ""    # e.g. "VOLT {value}" — set to send writes
+
+
+class SCPILayer(BaseProtocol):
+
+    def __init__(
+        self,
+        port:          str,
+        baud:          int              = 9600,
+        channels:      List[SCPIChannel] = None,
+        poll_interval: float            = 0.2,
+        simulate:      bool             = True,
+    ):
+        super().__init__(port, baud, poll_interval, simulate)
+        self.channels = channels or []
+
+    # ── Protocol ──────────────────────────────────────────────────────────
+
+    def _poll(self) -> Dict[str, float]:
+        result: Dict[str, float] = {}
+        for ch in self.channels:
+            if not ch.query.strip():
+                continue
+            try:
+                self._ser.write(f"{ch.query.strip()}\n".encode())
+                resp = self._ser.readline().decode(errors="replace").strip()
+                val  = _parse_numeric(resp)
+                if val is not None:
+                    result[ch.channel_id] = val * ch.scale
+            except Exception:
+                pass
+        return result
+
+    def _simulate(self, t: float) -> Dict[str, float]:
+        return {
+            ch.channel_id: math.sin(t * (0.3 + i * 0.2)) * 5.0 + i * 2.0
+            for i, ch in enumerate(self.channels)
+        }
+
+    def write(self, channel_id: str, value) -> bool:
+        ch = next((c for c in self.channels if c.channel_id == channel_id), None)
+        if ch is None or not ch.write_cmd:
+            return False
+        try:
+            cmd = ch.write_cmd.format(value=value)
+            self._ser.write(f"{cmd}\n".encode())
+            return True
+        except Exception:
+            return False
+
+    def query_idn(self) -> str:
+        """Send *IDN? and return instrument identification string."""
+        if not self._ser:
+            return "(not connected)"
+        try:
+            self._ser.write(b"*IDN?\n")
+            return self._ser.readline().decode(errors="replace").strip()
+        except Exception as e:
+            return f"(error: {e})"
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+_NUM_RE = re.compile(r"[+-]?\d+\.?\d*(?:[eE][+-]?\d+)?")
+
+
+def _parse_numeric(resp: str) -> Optional[float]:
+    """Extract first float from SCPI response, ignoring unit suffixes."""
+    m = _NUM_RE.search(resp)
+    return float(m.group()) if m else None