summaryrefslogtreecommitdiff
path: root/api_layers/protocols
diff options
context:
space:
mode:
Diffstat (limited to 'api_layers/protocols')
-rw-r--r--api_layers/protocols/base_protocol.py1
-rw-r--r--api_layers/protocols/cml.py120
-rw-r--r--api_layers/protocols/mark10.py32
-rw-r--r--api_layers/protocols/modbus_rtu.py18
-rw-r--r--api_layers/protocols/scpi.py15
5 files changed, 126 insertions, 60 deletions
diff --git a/api_layers/protocols/base_protocol.py b/api_layers/protocols/base_protocol.py
index 096151d..bb45d6b 100644
--- a/api_layers/protocols/base_protocol.py
+++ b/api_layers/protocols/base_protocol.py
@@ -30,6 +30,7 @@ class BaseProtocol(ABC):
self._cache: Dict[str, float] = {}
self._lock = threading.Lock()
+ self._io_lock = threading.Lock() # guards self._ser — shared by poll thread and write() callers
self._running = False
self._thread: Optional[threading.Thread] = None
self._ser = None
diff --git a/api_layers/protocols/cml.py b/api_layers/protocols/cml.py
index dbee2fa..113afe7 100644
--- a/api_layers/protocols/cml.py
+++ b/api_layers/protocols/cml.py
@@ -1,36 +1,38 @@
"""
api_layers/protocols/cml.py
-CoolMuscle Language (CML) protocol layer.
+CoolMuscle CM1-C ASCII (CML) protocol layer.
-For CoolMuscle CM-series servo motors over RS-232 (single axis)
-or RS-485 (multi-drop, up to 31 axes).
+Wire format — see docs/CM1-C_ASCII_Command_Cheatsheet.md:
+ <CMD>.<motor_id>[=<value>]\\r
+ e.g. "S0.1=20\\r" (set speed register), "?97.1\\r" (query speed),
+ "^.1\\r" (execute), "(.1\\r" (enable), ").1\\r" (disable), "].1\\r" (stop)
-Frame format
-─────────────
- RS-232 (address=0):
- Command: <CMD>[<data>]\\r
- Response: <CMD>[<data>]\\r (echo-back)
+Motor ID is always appended explicitly as ".<id>" on every command — there is
+no address-less / prefix-free mode.
- RS-485 (address 1-31):
- Command: #<addr><CMD>[<data>]\\r
- Response: *<addr><CMD>[<data>]\\r
+Terminator is CR only (default K70 setting). Replies are read up to that CR
+with Serial.read_until(b"\\r") rather than readline(), since pyserial's
+readline() looks for LF by default and would otherwise block for the full
+serial timeout on every single query.
Read commands (query current state):
- TP — Tell absolute position (encoder counts, signed)
- TV — Tell velocity (counts/second, signed)
- TC — Tell current (% rated × 10, unsigned)
- TS — Tell status word (hex flags)
+ TP — current position (?96, pulses, signed)
+ TV — current speed (?97, unit set by K37, signed)
+ TC — averaged current (?98, % rated, unsigned)
+ TS — motor status (?99, bit field — see cheatsheet)
Write commands (control):
- ME — Motor Enable
- MD — Motor Disable
- MA<value> — Move Absolute (encoder counts)
- MR<value> — Move Relative (encoder counts)
- VS<value> — Velocity Setpoint
+ ME — Motor Enable ("(")
+ MD — Motor Disable (")")
+ ST — Immediate stop ("]")
+ VS<value> — Velocity setpoint: sets a large P0 target (direction via sign
+ of S0) then executes with "^" — continuous rotation until
+ stopped (S0=0 or ST/MD).
+ MA<value> — Move absolute: sets P0 to <value> then executes with "^".
Channel IDs follow pattern: <motor_id>_<CMD>
- e.g. "M1_TP", "M1_TV", "M2_TC"
+ e.g. "M1_TP", "M1_VS", "M2_TC"
"""
import math
@@ -41,15 +43,19 @@ from typing import Dict, List, Optional
from api_layers.protocols.base_protocol import BaseProtocol
-_READ_CMDS = ["TP", "TV", "TC", "TS"]
-_RESP_RE = re.compile(r"[*@]?\d*([A-Z]{2})([\s\S]*)")
-_NUM_RE = re.compile(r"[+-]?\d+")
+_QUERY_CMDS = {"TP": "?96", "TV": "?97", "TC": "?98", "TS": "?99"}
+_NUM_RE = re.compile(r"[+-]?\d+\.?\d*")
+
+# Position target for velocity-mode continuous rotation — direction comes
+# from the sign of S0, not this value, so it just needs to be far enough
+# away that the move never completes on its own.
+_CONTINUOUS_POSITION = 1_000_000_000
@dataclass
class CMLMotor:
motor_id: str
- address: int = 1 # 0 = RS-232 (no address prefix)
+ address: int = 1 # CM1-C motor ID, always sent as ".<address>"
read_cmds: List[str] = field(default_factory=lambda: ["TP", "TV", "TC"])
@@ -66,16 +72,27 @@ class CMLLayer(BaseProtocol):
super().__init__(port, baud, poll_interval, simulate)
self.motors = motors or [CMLMotor("M1", address=1)]
- # ── Protocol ──────────────────────────────────────────────────────────
+ # ── Low-level I/O ─────────────────────────────────────────────────────
- def _send_query(self, motor: CMLMotor, cmd: str) -> Optional[float]:
- if motor.address == 0:
- frame = f"{cmd}\r".encode()
- else:
- frame = f"#{motor.address}{cmd}\r".encode()
+ def _send(self, address: int, cmd: str, value: Optional[int] = None) -> None:
+ suffix = f"={value}" if value is not None else ""
+ frame = f"{cmd}.{address}{suffix}\r".encode()
+ self._ser.reset_input_buffer()
self._ser.write(frame)
self._ser.flush()
- resp = self._ser.readline().decode(errors="replace").strip()
+
+ def _read_reply(self) -> str:
+ return self._ser.read_until(b"\r").decode(errors="replace").strip()
+
+ # ── Protocol ──────────────────────────────────────────────────────────
+
+ def _send_query(self, motor: CMLMotor, cmd: str) -> Optional[float]:
+ query = _QUERY_CMDS.get(cmd)
+ if query is None:
+ return None
+ with self._io_lock:
+ self._send(motor.address, query)
+ resp = self._read_reply()
return _parse_cml_response(resp)
def _poll(self) -> Dict[str, float]:
@@ -110,8 +127,7 @@ class CMLLayer(BaseProtocol):
return result
def write(self, channel_id: str, value) -> bool:
- # channel_id: "<motor_id>_<CMD>[<data>]"
- # e.g. "M1_ME", "M1_MA" (value carries position)
+ # channel_id: "<motor_id>_<CMD>" e.g. "M1_ME", "M1_VS" (value carries speed)
parts = channel_id.split("_", 1)
if len(parts) != 2:
return False
@@ -119,21 +135,41 @@ class CMLLayer(BaseProtocol):
motor = next((m for m in self.motors if m.motor_id == motor_id), None)
if motor is None:
return False
+ if self.simulate:
+ return True
try:
- data = "" if cmd in ("ME", "MD") else str(int(value))
- if motor.address == 0:
- frame = f"{cmd}{data}\r".encode()
- else:
- frame = f"#{motor.address}{cmd}{data}\r".encode()
- self._ser.write(frame)
+ with self._io_lock:
+ if cmd == "ME":
+ self._send(motor.address, "(")
+ self._read_reply()
+ elif cmd == "MD":
+ self._send(motor.address, ")")
+ self._read_reply()
+ elif cmd == "ST":
+ self._send(motor.address, "]")
+ self._read_reply()
+ elif cmd == "VS":
+ self._send(motor.address, "P0", _CONTINUOUS_POSITION)
+ self._read_reply()
+ self._send(motor.address, "S0", int(value))
+ self._read_reply()
+ self._send(motor.address, "^")
+ self._read_reply()
+ elif cmd == "MA":
+ self._send(motor.address, "P0", int(value))
+ self._read_reply()
+ self._send(motor.address, "^")
+ self._read_reply()
+ else:
+ return False
return True
except Exception:
return False
def _parse_cml_response(resp: str) -> Optional[float]:
- """Extract numeric value from CML response like '*1TP+001234'."""
- m = _NUM_RE.search(resp[3:] if resp and resp[0] in "*@#" else resp)
+ """Extract the first numeric value from a CM1-C reply."""
+ m = _NUM_RE.search(resp)
if m:
return float(m.group())
return None
diff --git a/api_layers/protocols/mark10.py b/api_layers/protocols/mark10.py
index 1b5bf1c..48ed927 100644
--- a/api_layers/protocols/mark10.py
+++ b/api_layers/protocols/mark10.py
@@ -10,6 +10,11 @@ Commands (sent with CR terminator):
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)
@@ -39,12 +44,20 @@ class Mark10Layer(BaseProtocol):
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:
- self._ser.write(b"?\r")
- resp = self._ser.readline().decode(errors="replace").strip()
+ 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 {}
@@ -76,10 +89,15 @@ class Mark10Layer(BaseProtocol):
"cycle_units": b"U\r",
}
cmd = cmd_map.get(channel_id)
- if cmd is None or not self._ser:
+ if cmd is None:
+ return False
+ if self.simulate:
+ return True
+ if not self._ser:
return False
try:
- self._ser.write(cmd)
+ with self._io_lock:
+ self._ser.write(cmd)
return True
except Exception:
return False
@@ -88,8 +106,10 @@ class Mark10Layer(BaseProtocol):
def zero(self) -> None:
if self._ser:
- self._ser.write(b"Z\r")
+ with self._io_lock:
+ self._ser.write(b"Z\r")
def cycle_units(self) -> None:
if self._ser:
- self._ser.write(b"U\r")
+ with self._io_lock:
+ self._ser.write(b"U\r")
diff --git a/api_layers/protocols/modbus_rtu.py b/api_layers/protocols/modbus_rtu.py
index 9d371b1..33f6877 100644
--- a/api_layers/protocols/modbus_rtu.py
+++ b/api_layers/protocols/modbus_rtu.py
@@ -91,10 +91,11 @@ class ModbusRTULayer(BaseProtocol):
def _read_registers(self, fc: int, start: int, count: int) -> List[int]:
req = _frame(struct.pack(">BBHH", self.slave_addr, fc, start, count))
- self._ser.write(req)
- time.sleep(0.005)
- n_bytes = 5 + 2 * count
- resp = self._ser.read(n_bytes)
+ with self._io_lock:
+ self._ser.write(req)
+ time.sleep(0.005)
+ n_bytes = 5 + 2 * count
+ resp = self._ser.read(n_bytes)
if len(resp) < n_bytes:
raise IOError(f"Short response {len(resp)}/{n_bytes} bytes")
crc_recv = struct.unpack("<H", resp[-2:])[0]
@@ -142,15 +143,18 @@ class ModbusRTULayer(BaseProtocol):
def _write_register(self, register: int, value: int) -> bool:
req = _frame(struct.pack(">BBHH", self.slave_addr, 0x06, register, value & 0xFFFF))
- self._ser.write(req)
- time.sleep(0.005)
- resp = self._ser.read(8)
+ with self._io_lock:
+ self._ser.write(req)
+ time.sleep(0.005)
+ resp = self._ser.read(8)
return len(resp) == 8
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:
return False
+ if self.simulate:
+ return True
try:
int_val = round((float(value) - ch.offset) / ch.scale)
return self._write_register(ch.register, int_val)
diff --git a/api_layers/protocols/scpi.py b/api_layers/protocols/scpi.py
index 730f8e0..fca5ba4 100644
--- a/api_layers/protocols/scpi.py
+++ b/api_layers/protocols/scpi.py
@@ -54,8 +54,9 @@ class SCPILayer(BaseProtocol):
if not ch.query.strip():
continue
try:
- self._ser.write(f"{ch.query.strip()}\n".encode())
- resp = self._ser.readline().decode(errors="replace").strip()
+ with self._io_lock:
+ 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
@@ -73,9 +74,12 @@ class SCPILayer(BaseProtocol):
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
+ if self.simulate:
+ return True
try:
cmd = ch.write_cmd.format(value=value)
- self._ser.write(f"{cmd}\n".encode())
+ with self._io_lock:
+ self._ser.write(f"{cmd}\n".encode())
return True
except Exception:
return False
@@ -85,8 +89,9 @@ class SCPILayer(BaseProtocol):
if not self._ser:
return "(not connected)"
try:
- self._ser.write(b"*IDN?\n")
- return self._ser.readline().decode(errors="replace").strip()
+ with self._io_lock:
+ self._ser.write(b"*IDN?\n")
+ return self._ser.readline().decode(errors="replace").strip()
except Exception as e:
return f"(error: {e})"
generated by cgit v1.2.3 (git 2.46.0) at 2026-09-13 17:25:17 +0000