1 files changed, 73 insertions, 0 deletions
diff --git a/api_layers/protocols/dscusb.py b/api_layers/protocols/dscusb.py
new file mode 100644
index 0000000..5601338
--- /dev/null
+++ b/api_layers/protocols/dscusb.py
@@ -0,0 +1,73 @@
+"""
+api_layers/protocols/dscusb.py
+
+DSCUSB displacement transducer (extensometer) ASCII protocol layer.
+
+Command (sent with CR terminator):
+  !001:SYS?   — request current reading ("001" = device address; hardcoded
+                for now — fine for one unit per COM port, which is the only
+                configuration seen so far)
+
+Response: a plain ASCII number, no unit suffix — read up to CR with
+Serial.read_until(b"\\r") rather than readline(), since pyserial's
+readline() looks for LF by default and this device only terminates with
+CR. Using readline() here is the failure mode that was actually observed:
+port opens and shows CONNECTED fine, but zero lines are ever received,
+because readline() never sees the LF it's waiting for. Same reasoning as
+Mark-10 (api_layers/protocols/mark10.py) — also CR-only.
+
+Reports the raw device reading, unit unconverted (no unit is reported by
+the instrument itself). If you need e.g. inches→mm, apply a Scale/Offset
+Channel operation downstream (Configuration > Channels) rather than
+hardcoding a conversion here — keeps this layer honest about what the
+device actually said. See docs/developer-new-protocols.md.
+
+Channels exposed:
+  extension — current raw reading
+"""
+
+import math
+import re
+from typing import Dict
+
+from api_layers.protocols.base_protocol import BaseProtocol
+
+_QUERY    = b"!001:SYS?\r"
+_RESP_RE  = re.compile(r"[+-]?\d+\.?\d*(?:[eE][+-]?\d+)?")
+
+
+class DSCUSBLayer(BaseProtocol):
+
+    def __init__(
+        self,
+        port:          str,
+        baud:          int   = 115200,
+        poll_interval: float = 0.05,
+        simulate:      bool  = True,
+    ):
+        super().__init__(port, baud, poll_interval, simulate)
+
+    # ── Protocol ──────────────────────────────────────────────────────────
+
+    def _poll(self) -> Dict[str, float]:
+        try:
+            with self._io_lock:
+                self._ser.reset_input_buffer()
+                self._ser.write(_QUERY)
+                self._ser.flush()
+                resp = self._ser.read_until(b"\r").decode(errors="replace").strip()
+            return self._parse(resp)
+        except Exception:
+            return {}
+
+    def _parse(self, resp: str) -> Dict[str, float]:
+        m = _RESP_RE.search(resp)
+        if not m:
+            return {}
+        return {"extension": float(m.group())}
+
+    def _simulate(self, t: float) -> Dict[str, float]:
+        return {"extension": math.sin(t * 0.4) * 2.0 + 2.0}
+
+    def write(self, channel_id: str, value) -> bool:
+        return False   # no write-capable channels on this device