Andrey/templates
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Объедини Modbus в один тип с функциями и количеством 1–4

Browse Source
This commit is contained in:
Andrey
2026-10-23 16:15:11 +03:00
parent f17c124f0e
commit 41f5b879fc
14 changed files with 247 additions and 354 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/protocan/protocan.py
+50 -55
View File
@@ -47,9 +47,7 @@ class MsgType(IntEnum):
ANALOG = 0b0010
GENERAL_ADDRESS_SPACE = 0b0011
GAS_SLOT = 0b0100 # Тип 0x4 однозначно выбирает слотный GAS с маской вместо прежнего Modbus Coil.
MODBUS_DISCRETE = 0b0101
MODBUS_HOLDING = 0b0110
MODBUS_INPUT = 0b0111
MODBUS = 0b0101 # Единый тип 0x5 выбирает банк функцией 1…4 в Body; номера 0x6 и 0x7 освобождены.
ERROR = 0b1000
BOOT_CONTROL = 0b1001
BOOT_DATA_A = 0b1010
@@ -60,6 +58,13 @@ class MsgType(IntEnum):
PULSE = 0b1111
class ModbusFunction(IntEnum): # Функция единого типа Modbus выбирает один из четырёх банков.
COIL = 1 # Дискретные выходы соответствуют функции 1.
DISCRETE = 2 # Дискретные входы соответствуют функции 2.
HOLDING = 3 # Изменяемые 16-битные регистры соответствуют функции 3.
INPUT = 4 # Входные 16-битные регистры соответствуют функции 4.
class SettingsResult(IntEnum):
OK = 0x00
INVALID_DLC = 0x01
@@ -116,9 +121,7 @@ MSGTYPE_RU = {
MsgType.ANALOG: "Analog",
MsgType.GENERAL_ADDRESS_SPACE: "General Address Space",
MsgType.GAS_SLOT: "GAS Slot", # Отдельное имя показывает, что адрес в Body относится к слоту, а не к Modbus-регистру.
MsgType.MODBUS_DISCRETE: "Modbus Discrete",
MsgType.MODBUS_HOLDING: "Modbus Holding",
MsgType.MODBUS_INPUT: "Modbus Input",
MsgType.MODBUS: "Modbus", # Один тип охватывает Coil, Discrete, Holding и Input через поле функции.
MsgType.ERROR: "Error",
MsgType.BOOT_CONTROL: "Boot Control",
MsgType.BOOT_DATA_A: "Boot Data A",
@@ -348,9 +351,8 @@ def split_analog(body: int):
return body & 0xFFF, (body >> 12) & 0xF
def split_modbus(body: int):
"""MODBUS: RegCount[3:0], StrAdr[15:4]."""
return (body >> 4) & 0xFFF, body & 0xF
def split_modbus(body: int): # Возвращает адрес, количество и функцию в тех же единицах, что принимает упаковщик.
return (body >> 4) & 0xFFF, (body & 3) + 1, ((body >> 2) & 3) + 1 # Адрес занимает 12 бит, два остальных поля кодируются значением минус один.
def split_error(body: int):
@@ -370,8 +372,10 @@ def merge_analog(an_type: int, sensor_id: int) -> int:
return ((an_type & 0xF) << 12) | (sensor_id & 0xFFF)
def merge_modbus(str_adr: int, reg_count: int) -> int:
return ((str_adr & 0xFFF) << 4) | (reg_count & 0xF)
def merge_modbus(str_adr: int, reg_count: int, function: int) -> int: # Функция 1…4 явно задаёт банк единого сообщения Modbus.
if not 0 <= str_adr <= 0xFFF or not 1 <= reg_count <= 4 or not 1 <= function <= 4 or str_adr + reg_count > 0x1000: # Проверяем окно целиком, чтобы не усекать адреса и значения двухбитных полей.
raise ValueError("Modbus: адрес 0…4095, количество и функция 1…4; окно не должно выходить за 4095") # Ошибочный ввод не превращается в допустимый кадр другого банка.
return (str_adr << 4) | ((function - 1) << 2) | (reg_count - 1) # Функция занимает биты 3:2, количество — 1:0; оба wire-кода лежат в диапазоне 0…3.
def merge_error(info: int, code: int) -> int:
@@ -460,14 +464,16 @@ def decode(raw_id: int, data: bytes, ide: bool = True, rtr: bool = False) -> Dec
res.fields.append(("Слот 0x%04X, Word%d" % (pid.body, index), "0x%04X (%d)" % (values[index], values[index]))) # Полная координата сохраняет адрес слота и выбранную позицию.
res.summary = "GAS SLOT slot=0x%04X%s" % (pid.body, " mask=0x%04X" % values[0] if values else " (запрос)") # CAN-журнал явно показывает маску частичного изменения.
return res # Не передаём четыре слова старому обработчику адреса первого регистра.
if pid.msg_type == MsgType.MODBUS and len(source_data) > 8: # Проверяем реальную длину до усечения отображаемого classic CAN payload.
res.summary = "MODBUS: неверная длина кадра" # Нельзя выдавать девять и более октетов за допустимые восемь.
res.warnings.append("Modbus: classic CAN допускает не более 8 байт") # Сохраняем причину отказа для журнала и анализатора захвата.
return res # Не публикуем регистры из частично обрезанного кадра.
handler = {
MsgType.BROADCAST: _decode_broadcast,
MsgType.DISCRETE: _decode_discrete,
MsgType.ANALOG: _decode_analog,
MsgType.GENERAL_ADDRESS_SPACE: _decode_gas,
MsgType.MODBUS_DISCRETE: _decode_modbus_bit,
MsgType.MODBUS_HOLDING: _decode_modbus_reg,
MsgType.MODBUS_INPUT: _decode_modbus_reg,
MsgType.MODBUS: _decode_modbus, # Банк выбирается функцией из Body внутри общего обработчика.
MsgType.ERROR: _decode_error,
MsgType.BOOT_CONTROL: _decode_boot,
MsgType.BOOT_DATA_A: _decode_boot,
@@ -663,47 +669,36 @@ def _decode_gas(res: Decoded) -> None:
res.summary = "GAS: запрос по адресу 0x%04X" % start
def _decode_modbus_bit(res: Decoded) -> None:
str_adr, reg_count = split_modbus(res.id.body)
kind = "DISCRETE" # После назначения 0x4 слотному GAS битовым Modbus-сообщением остаётся только тип 0x5.
res.fields.append((" Modbus.StrAdr", "0x%03X (%d)" % (str_adr, str_adr)))
res.fields.append((" Modbus.RegCount", "%d" % reg_count))
expected = reg_count % 8 + 1
if res.dlc and res.dlc != expected:
res.warnings.append(
"Для %s ожидается DLC = RegCount %% 8 + 1 = %d, получено %d"
% (kind, expected, res.dlc))
if len(res.data) >= 2:
val = res.data[0] | (res.data[1] << 8)
bits = ["%d" % ((val >> i) & 1) for i in range(16)]
res.registers = [(str_adr, val)]
res.fields.append((" Значение", "0x%04X" % val))
res.fields.append((" Биты 0..15", " ".join(bits)))
res.summary = "MODBUS %s @0x%03X x%d = 0x%04X" % (kind, str_adr, reg_count, val)
else:
res.summary = "MODBUS %s @0x%03X x%d (запрос)" % (kind, str_adr, reg_count)
def _decode_modbus_reg(res: Decoded) -> None:
str_adr, reg_count = split_modbus(res.id.body)
kind = "HOLDING" if res.id.msg_type == MsgType.MODBUS_HOLDING else "INPUT"
res.fields.append((" Modbus.StrAdr", "0x%03X (%d)" % (str_adr, str_adr)))
res.fields.append((" Modbus.RegCount", "%d" % reg_count))
if res.dlc and res.dlc != reg_count * 2:
res.warnings.append(
"Для %s ожидается DLC = 2 * RegCount = %d, получено %d"
% (kind, reg_count * 2, res.dlc))
if reg_count > 4:
res.warnings.append("В одном кадре не более 4 регистров, в ID указано %d" % reg_count)
vals = regs_le(res.data)
res.registers = [(str_adr + i, v) for i, v in enumerate(vals)]
for adr, v in res.registers:
res.fields.append((" Reg 0x%04X" % adr, "0x%04X (%d)" % (v, v)))
if vals:
res.summary = "MODBUS %s @0x%03X: %s" % (
kind, str_adr, " ".join("%04X" % v for v in vals))
else:
res.summary = "MODBUS %s @0x%03X x%d (запрос)" % (kind, str_adr, reg_count)
def _decode_modbus(res: Decoded) -> None: # Единый тип 0x5 разбирается по номеру функции, а не по прежним трём MsgType.
if not res.ide or res.rtr: # Modbus поверх ProtoCAN принимает только расширенные data frames.
return # Общий декодер уже добавил предупреждение о неверном формате; данные такого кадра не публикуются.
str_adr, reg_count, function = split_modbus(res.id.body) # Извлекаем физические значения функции и количества 1…4.
kind = ("COIL", "DISCRETE", "HOLDING", "INPUT")[function - 1] # Порядок банков соответствует функциям Modbus 1, 2, 3 и 4.
res.fields.extend([(" Modbus.Function", "%d (%s)" % (function, kind)), (" Modbus.StrAdr", "0x%03X (%d)" % (str_adr, str_adr)), (" Modbus.RegCount", str(reg_count))]) # Журнал показывает функцию отдельно от единого типа сообщения.
res.summary = "MODBUS FC%d %s @0x%03X x%d" % (function, kind, str_adr, reg_count) # Общая подпись сохраняется также для ошибочного payload.
if str_adr + reg_count > 0x1000: # Проверяем конец окна относительно 12-битного адресного пространства.
res.warnings.append("Modbus: окно выходит за адрес 0xFFF") # Не отображаем выход за границу как корректные регистры.
return # Неверное окно не интерпретируется дальше.
expected = 1 if function <= 2 else reg_count * 2 # До четырёх битов помещаются в октет; каждому 16-битному регистру нужны два октета.
if not res.dlc: # Пустой payload обозначает запрос чтения от ПМ.
if res.id.route == Route.FROM_DEVICE: # Ответ устройства обязан содержать запрошенные значения.
res.warnings.append("Modbus: пустой ответ устройства") # Пустой ответ не маскируется под запрос.
res.summary += " (запрос)" # Запрос отображается без вымышленных нулевых значений.
return # Чтение не содержит регистров для декодирования.
if res.dlc != expected: # Длина должна точно соответствовать функции и количеству в Body.
res.warnings.append("Modbus FC%d: ожидается DLC=%d, получено %d" % (function, expected, res.dlc)) # Сообщаем ожидаемую длину без частичного разбора.
return # Некорректный payload не выдаётся за действительные данные.
if function <= 2: # Coil и Discrete передают значения младшими битами одного октета.
if res.data[0] >> reg_count: # Биты вне запрошенного количества должны оставаться нулевыми.
res.warnings.append("Modbus: установлены биты вне запрошенного количества") # Выявляем неверную битовую маску.
return # Не публикуем неверные значения.
values = [(res.data[0] >> index) & 1 for index in range(reg_count)] # Каждый следующий адрес соответствует следующему младшему биту.
else: # Holding и Input сохраняют порядок little-endian внутри каждого регистра.
values = regs_le(res.data) # Общий помощник извлекает проверенное число 16-битных значений.
res.registers = [(str_adr + index, value) for index, value in enumerate(values)] # Возвращаем отдельную координату для каждого бита или регистра.
for address, value in res.registers: # Детальная панель показывает все значения выбранного банка.
res.fields.append((" Reg 0x%04X" % address, "0x%04X (%d)" % (value, value))) # Одновременно доступны HEX и десятичное представление.
res.summary += ": " + " ".join("%04X" % value for value in values) # Краткая строка журнала включает полученные значения.
def _decode_error(res: Decoded) -> None: