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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
"""
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
if self.simulate:
return True
try:
int_val = round((float(value) - ch.offset) / ch.scale)
return self._write_register(ch.register, int_val)
except Exception:
return False
|
