diff options
Diffstat (limited to 'api_layers/protocols')
18 files changed, 615 insertions, 0 deletions
diff --git a/api_layers/protocols/__init__.py b/api_layers/protocols/__init__.py new file mode 100644 index 0000000..6fda325 --- /dev/null +++ b/api_layers/protocols/__init__.py @@ -0,0 +1 @@ +"""api_layers/protocols — serial instrument protocol layers.""" diff --git a/api_layers/protocols/__pycache__/__init__.cpython-312.pyc b/api_layers/protocols/__pycache__/__init__.cpython-312.pyc Binary files differnew file mode 100644 index 0000000..e3d3dea --- /dev/null +++ b/api_layers/protocols/__pycache__/__init__.cpython-312.pyc diff --git a/api_layers/protocols/__pycache__/__init__.cpython-314.pyc b/api_layers/protocols/__pycache__/__init__.cpython-314.pyc Binary files differnew file mode 100644 index 0000000..c12aff7 --- /dev/null +++ b/api_layers/protocols/__pycache__/__init__.cpython-314.pyc diff --git a/api_layers/protocols/__pycache__/base_protocol.cpython-312.pyc b/api_layers/protocols/__pycache__/base_protocol.cpython-312.pyc Binary files differnew file mode 100644 index 0000000..faba1c8 --- /dev/null +++ b/api_layers/protocols/__pycache__/base_protocol.cpython-312.pyc diff --git a/api_layers/protocols/__pycache__/base_protocol.cpython-314.pyc b/api_layers/protocols/__pycache__/base_protocol.cpython-314.pyc Binary files differnew file mode 100644 index 0000000..7f87d61 --- /dev/null +++ b/api_layers/protocols/__pycache__/base_protocol.cpython-314.pyc diff --git a/api_layers/protocols/__pycache__/cml.cpython-312.pyc b/api_layers/protocols/__pycache__/cml.cpython-312.pyc Binary files differnew file mode 100644 index 0000000..fa580d6 --- /dev/null +++ b/api_layers/protocols/__pycache__/cml.cpython-312.pyc diff --git a/api_layers/protocols/__pycache__/cml.cpython-314.pyc b/api_layers/protocols/__pycache__/cml.cpython-314.pyc Binary files differnew file mode 100644 index 0000000..84c94a2 --- /dev/null +++ b/api_layers/protocols/__pycache__/cml.cpython-314.pyc diff --git a/api_layers/protocols/__pycache__/mark10.cpython-312.pyc b/api_layers/protocols/__pycache__/mark10.cpython-312.pyc Binary files differnew file mode 100644 index 0000000..f31dcb9 --- /dev/null +++ b/api_layers/protocols/__pycache__/mark10.cpython-312.pyc diff --git a/api_layers/protocols/__pycache__/mark10.cpython-314.pyc b/api_layers/protocols/__pycache__/mark10.cpython-314.pyc Binary files differnew file mode 100644 index 0000000..c2ce6dc --- /dev/null +++ b/api_layers/protocols/__pycache__/mark10.cpython-314.pyc diff --git a/api_layers/protocols/__pycache__/modbus_rtu.cpython-312.pyc b/api_layers/protocols/__pycache__/modbus_rtu.cpython-312.pyc Binary files differnew file mode 100644 index 0000000..1947863 --- /dev/null +++ b/api_layers/protocols/__pycache__/modbus_rtu.cpython-312.pyc diff --git a/api_layers/protocols/__pycache__/modbus_rtu.cpython-314.pyc b/api_layers/protocols/__pycache__/modbus_rtu.cpython-314.pyc Binary files differnew file mode 100644 index 0000000..dd104cb --- /dev/null +++ b/api_layers/protocols/__pycache__/modbus_rtu.cpython-314.pyc diff --git a/api_layers/protocols/__pycache__/scpi.cpython-312.pyc b/api_layers/protocols/__pycache__/scpi.cpython-312.pyc Binary files differnew file mode 100644 index 0000000..e2a2e83 --- /dev/null +++ b/api_layers/protocols/__pycache__/scpi.cpython-312.pyc diff --git a/api_layers/protocols/__pycache__/scpi.cpython-314.pyc b/api_layers/protocols/__pycache__/scpi.cpython-314.pyc Binary files differnew file mode 100644 index 0000000..971d9ed --- /dev/null +++ b/api_layers/protocols/__pycache__/scpi.cpython-314.pyc diff --git a/api_layers/protocols/base_protocol.py b/api_layers/protocols/base_protocol.py new file mode 100644 index 0000000..096151d --- /dev/null +++ b/api_layers/protocols/base_protocol.py @@ -0,0 +1,120 @@ +""" +api_layers/protocols/base_protocol.py + +Abstract base for serial instrument protocol layers. + +Background thread calls _poll() at poll_interval and stores results in +self._cache — same threading model as ArduinoLayer. +Simulation uses _simulate(t) instead. +""" + +import threading +import time +from abc import ABC, abstractmethod +from typing import Dict, List, Optional, Tuple + + +class BaseProtocol(ABC): + + def __init__( + self, + port: str, + baud: int, + poll_interval: float = 0.1, + simulate: bool = True, + ): + self.port = port + self.baud = baud + self.poll_interval = poll_interval + self.simulate = simulate + + self._cache: Dict[str, float] = {} + self._lock = threading.Lock() + self._running = False + self._thread: Optional[threading.Thread] = None + self._ser = None + + # ── Public API ──────────────────────────────────────────────────────── + + def connect(self) -> bool: + if self.simulate: + self._running = True + self._thread = threading.Thread(target=self._sim_loop, daemon=True) + self._thread.start() + return True + try: + import serial + self._ser = serial.Serial( + self.port, self.baud, timeout=1.0, + **self._serial_kwargs(), + ) + time.sleep(0.05) + self._running = True + self._thread = threading.Thread(target=self._poll_loop, daemon=True) + self._thread.start() + return True + except Exception as e: + print(f"[{type(self).__name__}] connect failed on {self.port}: {e}") + return False + + def disconnect(self) -> None: + self._running = False + if self._thread: + self._thread.join(timeout=2.0) + self._thread = None + if self._ser: + try: + self._ser.close() + except Exception: + pass + self._ser = None + + def read(self) -> Dict[str, float]: + with self._lock: + return dict(self._cache) + + @classmethod + def list_ports(cls) -> List[Tuple[str, str]]: + try: + import serial.tools.list_ports + return [(p.device, p.description) for p in serial.tools.list_ports.comports()] + except ImportError: + return [] + + # ── Internal ────────────────────────────────────────────────────────── + + def _serial_kwargs(self) -> dict: + return {} + + def _poll_loop(self) -> None: + while self._running: + try: + result = self._poll() + if result: + with self._lock: + self._cache.update(result) + except Exception as e: + print(f"[{type(self).__name__}] poll error: {e}") + time.sleep(self.poll_interval) + + def _sim_loop(self) -> None: + t0 = time.time() + while self._running: + t = time.time() - t0 + with self._lock: + self._cache = self._simulate(t) + time.sleep(self.poll_interval) + + # ── Abstract ────────────────────────────────────────────────────────── + + @abstractmethod + def _poll(self) -> Dict[str, float]: + """Read instrument, return {channel_id: value}.""" + + @abstractmethod + def _simulate(self, t: float) -> Dict[str, float]: + """Return simulated values at elapsed time t (seconds).""" + + @abstractmethod + def write(self, channel_id: str, value) -> bool: + """Send command to instrument.""" diff --git a/api_layers/protocols/cml.py b/api_layers/protocols/cml.py new file mode 100644 index 0000000..dbee2fa --- /dev/null +++ b/api_layers/protocols/cml.py @@ -0,0 +1,139 @@ +""" +api_layers/protocols/cml.py + +CoolMuscle Language (CML) protocol layer. + +For CoolMuscle CM-series servo motors over RS-232 (single axis) +or RS-485 (multi-drop, up to 31 axes). + +Frame format +───────────── + RS-232 (address=0): + Command: <CMD>[<data>]\\r + Response: <CMD>[<data>]\\r (echo-back) + + RS-485 (address 1-31): + Command: #<addr><CMD>[<data>]\\r + Response: *<addr><CMD>[<data>]\\r + +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) + +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 + +Channel IDs follow pattern: <motor_id>_<CMD> + e.g. "M1_TP", "M1_TV", "M2_TC" +""" + +import math +import re +from dataclasses import dataclass, field +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+") + + +@dataclass +class CMLMotor: + motor_id: str + address: int = 1 # 0 = RS-232 (no address prefix) + read_cmds: List[str] = field(default_factory=lambda: ["TP", "TV", "TC"]) + + +class CMLLayer(BaseProtocol): + + def __init__( + self, + port: str, + baud: int = 38400, + motors: List[CMLMotor] = None, + poll_interval: float = 0.1, + simulate: bool = True, + ): + super().__init__(port, baud, poll_interval, simulate) + self.motors = motors or [CMLMotor("M1", address=1)] + + # ── Protocol ────────────────────────────────────────────────────────── + + 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() + self._ser.write(frame) + self._ser.flush() + resp = self._ser.readline().decode(errors="replace").strip() + return _parse_cml_response(resp) + + def _poll(self) -> Dict[str, float]: + result: Dict[str, float] = {} + for motor in self.motors: + for cmd in motor.read_cmds: + try: + val = self._send_query(motor, cmd) + if val is not None: + result[f"{motor.motor_id}_{cmd}"] = val + except Exception: + pass + return result + + def _simulate(self, t: float) -> Dict[str, float]: + result: Dict[str, float] = {} + for i, motor in enumerate(self.motors): + phase = i * 1.0 + for cmd in motor.read_cmds: + key = f"{motor.motor_id}_{cmd}" + if cmd == "TP": + val = math.sin(t * 0.2 + phase) * 10000.0 + elif cmd == "TV": + val = math.cos(t * 0.2 + phase) * 2000.0 + elif cmd == "TC": + val = abs(math.sin(t * 0.4 + phase)) * 500.0 + elif cmd == "TS": + val = 0.0 + else: + val = math.sin(t + i) * 100.0 + result[key] = val + 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) + parts = channel_id.split("_", 1) + if len(parts) != 2: + return False + motor_id, cmd = parts + motor = next((m for m in self.motors if m.motor_id == motor_id), None) + if motor is None: + return False + 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) + 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) + if m: + return float(m.group()) + return None 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") diff --git a/api_layers/protocols/modbus_rtu.py b/api_layers/protocols/modbus_rtu.py new file mode 100644 index 0000000..9d371b1 --- /dev/null +++ b/api_layers/protocols/modbus_rtu.py @@ -0,0 +1,158 @@ +""" +api_layers/protocols/modbus_rtu.py + +Modbus RTU over RS-232 / RS-485. Pure Python — no external library. + +Supported function codes: + FC03 Read Holding Registers + FC04 Read Input Registers + FC06 Write Single Register + +Register data types: + uint16 — unsigned 16-bit (1 register) + int16 — signed 16-bit (1 register) + float32 — IEEE-754 float (2 registers, big-endian) + int32 — signed 32-bit (2 registers, big-endian) +""" + +import math +import struct +import time +from dataclasses import dataclass +from typing import Dict, List, Optional + +from api_layers.protocols.base_protocol import BaseProtocol + + +# ── CRC-16 (Modbus polynomial 0xA001) ──────────────────────────────────────── + +def _crc16(data: bytes) -> int: + crc = 0xFFFF + for byte in data: + crc ^= byte + for _ in range(8): + crc = (crc >> 1) ^ 0xA001 if crc & 1 else crc >> 1 + return crc + + +def _frame(payload: bytes) -> bytes: + return payload + struct.pack("<H", _crc16(payload)) + + +# ── Channel definition ──────────────────────────────────────────────────────── + +@dataclass +class ModbusChannel: + channel_id: str + name: str + register: int # 0-based register address + function_code: int = 0x03 # FC03=holding, FC04=input + data_type: str = "uint16" # uint16 | int16 | float32 | int32 + scale: float = 1.0 + offset: float = 0.0 + unit: str = "" + + +# ── Protocol layer ──────────────────────────────────────────────────────────── + +class ModbusRTULayer(BaseProtocol): + + def __init__( + self, + port: str, + baud: int = 9600, + slave_addr: int = 1, + channels: List[ModbusChannel] = None, + parity: str = "N", # N / E / O + stopbits: int = 1, + poll_interval: float = 0.1, + simulate: bool = True, + ): + super().__init__(port, baud, poll_interval, simulate) + self.slave_addr = slave_addr + self.channels = channels or [] + self._parity = parity + self._stopbits = stopbits + + def _serial_kwargs(self) -> dict: + import serial + parity_map = { + "N": serial.PARITY_NONE, + "E": serial.PARITY_EVEN, + "O": serial.PARITY_ODD, + } + return { + "parity": parity_map.get(self._parity, serial.PARITY_NONE), + "stopbits": self._stopbits, + "bytesize": 8, + } + + # ── Read ────────────────────────────────────────────────────────────── + + 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) + if len(resp) < n_bytes: + raise IOError(f"Short response {len(resp)}/{n_bytes} bytes") + crc_recv = struct.unpack("<H", resp[-2:])[0] + if crc_recv != _crc16(resp[:-2]): + raise ValueError("CRC mismatch") + n_data = resp[2] + return list(struct.unpack(f">{n_data // 2}H", resp[3:3 + n_data])) + + def _decode(self, ch: ModbusChannel, regs: List[int]) -> float: + dt = ch.data_type + if dt == "uint16": + raw = regs[0] + elif dt == "int16": + raw = regs[0] if regs[0] < 0x8000 else regs[0] - 0x10000 + elif dt in ("float32", "int32"): + combined = (regs[0] << 16) | regs[1] + if dt == "float32": + raw = struct.unpack(">f", struct.pack(">I", combined))[0] + else: + raw = combined if combined < 0x80000000 else combined - 0x100000000 + else: + raw = regs[0] + return float(raw) * ch.scale + ch.offset + + def _poll(self) -> Dict[str, float]: + result: Dict[str, float] = {} + for ch in self.channels: + count = 2 if ch.data_type in ("float32", "int32") else 1 + try: + regs = self._read_registers(ch.function_code, ch.register, count) + result[ch.channel_id] = self._decode(ch, regs) + except Exception: + pass + return result + + def _simulate(self, t: float) -> Dict[str, float]: + return { + ch.channel_id: ( + (math.sin(t * (0.5 + i * 0.3)) * 100 + 100) * ch.scale + ch.offset + ) + for i, ch in enumerate(self.channels) + } + + # ── Write ───────────────────────────────────────────────────────────── + + 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) + 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 + try: + int_val = round((float(value) - ch.offset) / ch.scale) + return self._write_register(ch.register, int_val) + except Exception: + return False 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 |