Added difference serial protocols

1 files changed, 95 insertions, 0 deletions

diff --git a/api_layers/protocols/mark10.py b/api_layers/protocols/mark10.py
new file mode 100644
index 0000000..1b5bf1c
--- /dev/null
+++ b/api_layers/protocols/mark10.py
@@ -0,0 +1,95 @@
+"""
+api_layers/protocols/mark10.py
+
+Mark-10 ASCII protocol layer (Series 5 primary, compatible with Series 4/3).
+
+Commands (sent with CR terminator):
+  ?   — request current reading
+  Z   — zero the gauge
+  U   — cycle units (lb → kgF → N → ozF → …)
+
+Response format: "+0.1234 kgF" (sign, value, space, unit suffix)
+
+Channels exposed:
+  force      — current force reading (in instrument's selected unit)
+  unit_code  — numeric index into UNITS list (lb=0, kgF=1, N=2, ozF=3)
+"""
+
+import math
+import re
+from typing import Dict, Optional
+
+from api_layers.protocols.base_protocol import BaseProtocol
+
+
+UNITS = ["lb", "kgF", "N", "ozF"]
+
+_RESP_RE = re.compile(r"([+-]?\d+\.?\d*(?:[eE][+-]?\d+)?)\s*([a-zA-Z]*)")
+
+
+class Mark10Layer(BaseProtocol):
+
+    def __init__(
+        self,
+        port:          str,
+        baud:          int   = 115200,
+        poll_interval: float = 0.05,
+        simulate:      bool  = True,
+    ):
+        super().__init__(port, baud, poll_interval, simulate)
+        self._unit = "N"
+
+    # ── Protocol ──────────────────────────────────────────────────────────
+
+    def _poll(self) -> Dict[str, float]:
+        try:
+            self._ser.write(b"?\r")
+            resp = self._ser.readline().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 {}
+        val  = float(m.group(1))
+        unit = m.group(2).strip() or self._unit
+        if unit in UNITS:
+            self._unit = unit
+        return {
+            "force":     val,
+            "unit_code": float(UNITS.index(self._unit) if self._unit in UNITS else 2),
+        }
+
+    def _simulate(self, t: float) -> Dict[str, float]:
+        cycle = t % 30.0
+        ramp  = cycle * 6.67 if cycle < 15.0 else (30.0 - cycle) * 6.67
+        return {
+            "force":     ramp + math.sin(t * 10.0) * 0.3,
+            "unit_code": float(UNITS.index("N")),
+        }
+
+    def write(self, channel_id: str, value) -> bool:
+        cmd_map = {
+            "zero":         b"Z\r",
+            "cycle_units":  b"U\r",
+        }
+        cmd = cmd_map.get(channel_id)
+        if cmd is None or not self._ser:
+            return False
+        try:
+            self._ser.write(cmd)
+            return True
+        except Exception:
+            return False
+
+    # ── Convenience ───────────────────────────────────────────────────────
+
+    def zero(self) -> None:
+        if self._ser:
+            self._ser.write(b"Z\r")
+
+    def cycle_units(self) -> None:
+        if self._ser:
+            self._ser.write(b"U\r")