From 67cfa0a514c7de4605ed7360e15a81aa781e510e Mon Sep 17 00:00:00 2001
From: Christian Kolset <christian.kolset@gmail.com>
Date: Fri, 24 Apr 2026 11:51:26 -0600
Subject: Added difference serial protocols

---
 api_layers/protocols/modbus_rtu.py | 158 +++++++++++++++++++++++++++++++++++++
 1 file changed, 158 insertions(+)
 create mode 100644 api_layers/protocols/modbus_rtu.py

(limited to 'api_layers/protocols/modbus_rtu.py')

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
-- 
cgit v1.2.3