1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
"""
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)
Replies are read up to CR with Serial.read_until(b"\\r") rather than
readline(), since pyserial's readline() looks for LF by default and the
gauge only terminates with CR — readline() would otherwise block for the
full serial timeout on every poll.
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"
@property
def current_unit(self) -> str:
"""Last unit suffix seen in a gauge reply (e.g. "N", "kgF")."""
return self._unit
# ── Protocol ──────────────────────────────────────────────────────────
def _poll(self) -> Dict[str, float]:
try:
with self._io_lock:
self._ser.reset_input_buffer()
self._ser.write(b"?\r")
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 {}
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:
return False
if self.simulate:
return True
if not self._ser:
return False
try:
with self._io_lock:
self._ser.write(cmd)
return True
except Exception:
return False
# ── Convenience ───────────────────────────────────────────────────────
def zero(self) -> None:
if self._ser:
with self._io_lock:
self._ser.write(b"Z\r")
def cycle_units(self) -> None:
if self._ser:
with self._io_lock:
self._ser.write(b"U\r")
|
