Enhance protocol layers: implement I/O locking for thread safety in serial communication

5 files changed, 35 insertions, 21 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 67be0d5..6bc85f0 100644
--- a/api_layers/protocols/cml.py
+++ b/api_layers/protocols/cml.py
@@ -73,9 +73,10 @@ class CMLLayer(BaseProtocol):
             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()
+        with self._io_lock:
+            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]:
@@ -127,7 +128,10 @@ class CMLLayer(BaseProtocol):
                 frame = f"{cmd}{data}\r".encode()
             else:
                 frame = f"#{motor.address}{cmd}{data}\r".encode()
-            self._ser.write(frame)
+            with self._io_lock:
+                self._ser.write(frame)
+                self._ser.flush()
+                self._ser.readline()  # discard echo-back so it doesn't desync the next poll read
             return True
         except Exception:
             return False
diff --git a/api_layers/protocols/mark10.py b/api_layers/protocols/mark10.py
index 4a97892..b7dacb4 100644
--- a/api_layers/protocols/mark10.py
+++ b/api_layers/protocols/mark10.py
@@ -43,8 +43,9 @@ class Mark10Layer(BaseProtocol):
 
     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.write(b"?\r")
+                resp = self._ser.readline().decode(errors="replace").strip()
             return self._parse(resp)
         except Exception:
             return {}
@@ -83,7 +84,8 @@ class Mark10Layer(BaseProtocol):
         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
@@ -92,8 +94,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 68d644e..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,9 +143,10 @@ 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:
diff --git a/api_layers/protocols/scpi.py b/api_layers/protocols/scpi.py
index 526ed21..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
@@ -77,7 +78,8 @@ class SCPILayer(BaseProtocol):
             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
@@ -87,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})"