From e9c5ec90b89e45284b7de3372514c936463959a7 Mon Sep 17 00:00:00 2001 From: Andrey Date: Tue, 1 Sep 2026 12:00:53 +0300 Subject: [PATCH] refactor(python): centralize ProtoCAN decoder --- python/protocan/protocan.py | 239 ++++++++++++++++++++++++++++++------ 1 file changed, 204 insertions(+), 35 deletions(-) diff --git a/python/protocan/protocan.py b/python/protocan/protocan.py index 540c44e..94a7264 100644 --- a/python/protocan/protocan.py +++ b/python/protocan/protocan.py @@ -5,7 +5,7 @@ битовые поля укладываются от младших бит к старшим. Модуль не импортирует Qt: разбор переносим и проверяется host-тестами. -Источник контракта — CAN_to_RS485/docs/PROTOCOL.md, §1. +Источник контракта — ``templates/c/set-protocol``. """ from __future__ import annotations @@ -14,6 +14,15 @@ from dataclasses import dataclass, field from enum import IntEnum from typing import List, Optional +try: + from .native import get_native_core + _SHARED_PROTOCOL = get_native_core() +except (ImportError, OSError, RuntimeError): + _SHARED_PROTOCOL = None + + +EXTENDED_ID_MASK = 0x1FFFFFFF + # -------------------------------------------------------------------------- # Перечисления протокола @@ -44,9 +53,21 @@ class MsgType(IntEnum): BOOT_DATA_B = 0b1011 BOOT_STATUS = 0b1100 BOOT_DISCOVERY = 0b1101 + SETTINGS = 0b1110 PULSE = 0b1111 +class SettingsResult(IntEnum): + OK = 0x00 + INVALID_DLC = 0x01 + INVALID_ROM_CRC = 0x02 + NOT_FOUND = 0x03 + ALREADY_ASSIGNED = 0x04 + EEPROM_ERROR = 0x05 + INVALID_LOCATION = 0x06 + BUSY = 0x07 + + class BroadcastType(IntEnum): STATUS = 0 ONOFF = 1 @@ -100,9 +121,32 @@ MSGTYPE_RU = { MsgType.BOOT_DATA_B: "Boot Data B", MsgType.BOOT_STATUS: "Boot Status", MsgType.BOOT_DISCOVERY: "Boot Discovery", + MsgType.SETTINGS: "Settings", MsgType.PULSE: "Pulse", } +#: Имена адресов из листа «Интерфейс протокола» исходного XLSX. +#: Для остальных Device Type таблица задаёт только диапазон Device 0x0..0xF, +#: но не присваивает приборам имён. +DEVICE_TYPE_NAMES = { + 0x0: "Верхний уровень", +} + +DEVICE_ADDRESS_NAMES = { + (0x7, 0xD): "configurator", +} + +SETTINGS_RESULT_RU = { + SettingsResult.OK: "успех", + SettingsResult.INVALID_DLC: "неверный DLC", + SettingsResult.INVALID_ROM_CRC: "неверный family code или CRC ROM", + SettingsResult.NOT_FOUND: "ROM не найден на 1-Wire", + SettingsResult.ALREADY_ASSIGNED: "ROM уже назначен другой локации", + SettingsResult.EEPROM_ERROR: "ошибка EEPROM", + SettingsResult.INVALID_LOCATION: "неверная локация", + SettingsResult.BUSY: "устройство занято", +} + BROADCAST_RU = { BroadcastType.STATUS: "запрос статуса", BroadcastType.ONOFF: "вкл/выкл пульса", @@ -145,44 +189,113 @@ def _name(enum_cls, value, table): return "%s (%s)" % (item.name, table.get(item, "")) +def device_name(device_type: int, device: int) -> str: + """Имя пары Device Type/Device из исходной таблицы протокола.""" + return DEVICE_ADDRESS_NAMES.get( + (device_type, device), DEVICE_TYPE_NAMES.get(device_type, "")) + + # -------------------------------------------------------------------------- # Идентификатор # -------------------------------------------------------------------------- @dataclass class ProtoCanId: - """Разобранный 29-битный идентификатор.""" + """Разобранный 29-битный идентификатор. + + Имена полей повторяют исходную таблицу ``Протокол CAN и ОАП.xlsx``: + ``Priority | ПМ | Device Type | Device | Msg Type | Body``. Свойства + ``route``, ``device_id`` и ``msg_body`` оставлены как совместимые + псевдонимы для старого кода. + """ raw: int - msg_body: int + body: int msg_type: int - device_id: int + device: int device_type: int - route: int + pm: int priority: int @staticmethod def parse(raw: int) -> "ProtoCanId": - raw &= 0x1FFFFFFF + raw &= EXTENDED_ID_MASK + if _SHARED_PROTOCOL is not None: + priority, pm, device_type, device, msg_type, body = ( + _SHARED_PROTOCOL.id_unpack(raw)) + return ProtoCanId(raw, body, msg_type, device, device_type, pm, + priority) return ProtoCanId( raw=raw, - msg_body=raw & 0xFFFF, + body=raw & 0xFFFF, msg_type=(raw >> 16) & 0xF, - device_id=(raw >> 20) & 0xF, + device=(raw >> 20) & 0xF, device_type=(raw >> 24) & 0x7, - route=(raw >> 27) & 0x1, + pm=(raw >> 27) & 0x1, priority=(raw >> 28) & 0x1, ) @staticmethod - def build(priority: int, route: int, device_type: int, device_id: int, - msg_type: int, msg_body: int) -> int: + def build(priority: int, pm: Optional[int] = None, + device_type: Optional[int] = None, device: Optional[int] = None, + msg_type: Optional[int] = None, body: Optional[int] = None, + **legacy: int) -> int: + """Собирает ID из полей протокола. + + Старые именованные аргументы ``route``, ``device_id`` и + ``msg_body`` принимаются для обратной совместимости. + """ + aliases = ( + ("route", "pm", pm), + ("device_id", "device", device), + ("msg_body", "body", body), + ) + resolved = [] + for old_name, new_name, value in aliases: + old_value = legacy.pop(old_name, None) + if value is not None and old_value is not None: + raise TypeError("одновременно заданы %s и %s" % + (new_name, old_name)) + resolved.append(value if value is not None else old_value) + if legacy: + raise TypeError("неизвестное поле: %s" % next(iter(legacy))) + pm, device, body = resolved + required = { + "pm": pm, + "device_type": device_type, + "device": device, + "msg_type": msg_type, + "body": body, + } + missing = [name for name, value in required.items() if value is None] + if missing: + raise TypeError("не заданы поля: %s" % ", ".join(missing)) + assert pm is not None and device_type is not None and device is not None + assert msg_type is not None and body is not None + if _SHARED_PROTOCOL is not None: + return _SHARED_PROTOCOL.id_pack( + priority, pm, device_type, device, msg_type, body) return (((priority & 0x1) << 28) - | ((route & 0x1) << 27) + | ((pm & 0x1) << 27) | ((device_type & 0x7) << 24) - | ((device_id & 0xF) << 20) + | ((device & 0xF) << 20) | ((msg_type & 0xF) << 16) - | (msg_body & 0xFFFF)) + | (body & 0xFFFF)) + + @property + def route(self) -> int: + """Совместимый псевдоним поля ``ПМ``.""" + return self.pm + + @property + def device_id(self) -> int: + """Совместимый псевдоним поля ``Device``.""" + return self.device + + @property + def msg_body(self) -> int: + """Совместимый псевдоним поля ``Body``.""" + return self.body @property def msg_type_name(self) -> str: @@ -191,6 +304,11 @@ class ProtoCanId: except ValueError: return "резерв 0x%X" % self.msg_type + @property + def device_name(self) -> str: + """Табличное имя адреса Device Type/Device, если оно задано.""" + return device_name(self.device_type, self.device) + # -------------------------------------------------------------------------- # Раскладки MsgBody @@ -284,14 +402,17 @@ def decode(raw_id: int, data: bytes, ide: bool = True, rtr: bool = False) -> Dec data = bytes(data[:8]) res = Decoded(id=pid, dlc=len(data), data=data, ide=ide, rtr=rtr) - res.fields.append(("Priority", "%d — %s" % ( + res.fields.append(("Priority [28]", "%d — %s" % ( pid.priority, PRIORITY_RU.get(Priority(pid.priority), "")))) - res.fields.append(("Route", "%d — %s" % ( - pid.route, ROUTE_RU.get(Route(pid.route), "")))) - res.fields.append(("DeviceType", "%d (0b{:03b})".format(pid.device_type) % pid.device_type)) - res.fields.append(("DeviceID", "%d (0b{:04b})".format(pid.device_id) % pid.device_id)) - res.fields.append(("MsgType", "0x%X — %s" % (pid.msg_type, pid.msg_type_name))) - res.fields.append(("MsgBody", "0x%04X" % pid.msg_body)) + res.fields.append(("ПМ [27]", "%d — %s" % ( + pid.pm, ROUTE_RU.get(Route(pid.pm), "")))) + res.fields.append(("Device Type [26:24]", "%d (0b{:03b})".format( + pid.device_type) % pid.device_type)) + res.fields.append(("Device [23:20]", "%d (0b{:04b})".format( + pid.device) % pid.device)) + res.fields.append(("Msg Type [19:16]", "0x%X — %s" % ( + pid.msg_type, pid.msg_type_name))) + res.fields.append(("Body [15:0]", "0x%04X" % pid.body)) if not ide: res.warnings.append("Стандартный ID: ProtoCAN использует расширенный (29 бит)") @@ -313,6 +434,7 @@ def decode(raw_id: int, data: bytes, ide: bool = True, rtr: bool = False) -> Dec MsgType.BOOT_DATA_B: _decode_boot, MsgType.BOOT_STATUS: _decode_boot, MsgType.BOOT_DISCOVERY: _decode_boot, + MsgType.SETTINGS: _decode_settings, MsgType.PULSE: _decode_pulse, } try: @@ -328,10 +450,11 @@ def decode(raw_id: int, data: bytes, ide: bool = True, rtr: bool = False) -> Dec def _decode_boot(res: Decoded) -> None: + """Расшифровывает служебные кадры ProtoCAN Boot Protocol 1.0.""" kind = MsgType(res.id.msg_type) if kind == MsgType.BOOT_CONTROL: - command = res.id.msg_body & 0xFF - session = res.id.msg_body >> 8 + command = res.id.body & 0xFF + session = res.id.body >> 8 names = {1: "IDENTIFY", 2: "ENTER_BOOT", 3: "BEGIN_IMAGE", 4: "BEGIN_COMPAT", 5: "ERASE", 6: "VERIFY", 7: "COMMIT", 8: "CONFIRM", 9: "REBOOT", 10: "ABORT", 11: "QUERY_PROGRESS"} @@ -339,10 +462,13 @@ def _decode_boot(res: Decoded) -> None: names.get(command, "0x%02X" % command), session) elif kind in (MsgType.BOOT_DATA_A, MsgType.BOOT_DATA_B): res.summary = "BOOT DATA slot %s, block=%d" % ( - "A" if kind == MsgType.BOOT_DATA_A else "B", res.id.msg_body) + "A" if kind == MsgType.BOOT_DATA_A else "B", res.id.body) elif kind == MsgType.BOOT_STATUS: - res.summary = "BOOT STATUS command=0x%02X, session=%d" % ( - res.id.msg_body & 0xFF, res.id.msg_body >> 8) + command = res.id.body & 0xFF + session = res.id.body >> 8 + status = res.data[0] if res.data else None + res.summary = "BOOT STATUS command=0x%02X, session=%d%s" % ( + command, session, " status=0x%02X" % status if status is not None else "") else: res.summary = "BOOT DISCOVERY" if res.data[:4] == b"PONG" and len(res.data) == 8: @@ -350,7 +476,7 @@ def _decode_boot(res: Decoded) -> None: def _decode_broadcast(res: Decoded) -> None: - body, btype = split_broadcast(res.id.msg_body) + body, btype = split_broadcast(res.id.body) res.fields.append((" Broadcast.Type", "0x%03X — %s" % ( btype, _name(BroadcastType, btype, BROADCAST_RU)))) res.fields.append((" Broadcast.Body", "0x%X (%d)" % (body, body))) @@ -428,7 +554,7 @@ def _check_datetime(res: Decoded, h, m, s, yy, mo, dd, wd) -> None: def _decode_discrete(res: Decoded) -> None: - body, dtype = split_discrete(res.id.msg_body) + body, dtype = split_discrete(res.id.body) res.fields.append((" Discrete.Type", "0x%X — %s" % ( dtype, _name(DiscreteType, dtype, DISCRETE_RU)))) res.fields.append((" Discrete.Body", "0x%03X (%d)" % (body, body))) @@ -445,7 +571,7 @@ def _decode_discrete(res: Decoded) -> None: def _decode_analog(res: Decoded) -> None: - sensor_id, atype = split_analog(res.id.msg_body) + sensor_id, atype = split_analog(res.id.body) res.fields.append((" Analog.Type", "0x%X — %s" % ( atype, _name(AnalogType, atype, ANALOG_RU)))) res.fields.append((" Analog.SensorID", "%d (0x%03X)" % (sensor_id, sensor_id))) @@ -469,7 +595,7 @@ def _decode_analog(res: Decoded) -> None: def _decode_gas(res: Decoded) -> None: - start = res.id.msg_body + start = res.id.body res.fields.append((" Адрес первого регистра", "0x%04X (%d)" % (start, start))) if res.dlc % 2: res.warnings.append("DLC = %d нечётный: регистры GAS передаются парами байт" % res.dlc) @@ -492,7 +618,7 @@ def _decode_gas(res: Decoded) -> None: def _decode_modbus_bit(res: Decoded) -> None: - str_adr, reg_count = split_modbus(res.id.msg_body) + str_adr, reg_count = split_modbus(res.id.body) kind = "COIL" if res.id.msg_type == MsgType.MODBUS_COIL else "DISCRETE" res.fields.append((" Modbus.StrAdr", "0x%03X (%d)" % (str_adr, str_adr))) res.fields.append((" Modbus.RegCount", "%d" % reg_count)) @@ -513,7 +639,7 @@ def _decode_modbus_bit(res: Decoded) -> None: def _decode_modbus_reg(res: Decoded) -> None: - str_adr, reg_count = split_modbus(res.id.msg_body) + 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)) @@ -535,7 +661,7 @@ def _decode_modbus_reg(res: Decoded) -> None: def _decode_error(res: Decoded) -> None: - code, info = split_error(res.id.msg_body) + code, info = split_error(res.id.body) res.fields.append((" Error.Code", "0x%02X (%d)" % (code, code))) res.fields.append((" Error.Info", "0x%02X (%d)" % (info, info))) res.summary = "ERROR code=0x%02X info=0x%02X" % (code, info) @@ -545,6 +671,49 @@ def _decode_error(res: Decoded) -> None: res.warnings.append("ERROR передаётся с DLC = 0, получено %d" % res.dlc) +def _decode_settings(res: Decoded) -> None: + """SETTINGS: Body[15:8] = сборка Z, Body[7:0] = позиция Y.""" + assembly = (res.id.body >> 8) & 0xFF + position = res.id.body & 0xFF + res.fields += [ + (" SETTINGS.Assembly Z", "%d (0x%02X)" % (assembly, assembly)), + (" SETTINGS.Position Y", "%d (0x%02X)" % (position, position)), + ] + + if res.id.pm == Route.FROM_PM: + if res.dlc == 0: + operation = "GET" + elif res.dlc == 8 and not any(res.data): + operation = "CLEAR" + res.fields.append((" ROM", "00 00 00 00 00 00 00 00")) + elif res.dlc == 8: + operation = "WRITE/REPLACE" + res.fields.append((" ROM", " ".join("%02X" % b for b in res.data))) + else: + operation = "INVALID" + res.warnings.append("Запрос SETTINGS должен иметь DLC = 0 или 8") + else: + if res.dlc == 8: + operation = "RESPONSE" + res.fields.append((" ROM", " ".join("%02X" % b for b in res.data))) + elif res.dlc == 1: + operation = "ERROR" + code = res.data[0] + try: + result = SettingsResult(code) + result_name = "%s — %s" % (result.name, SETTINGS_RESULT_RU[result]) + except ValueError: + result_name = "неизвестный код" + res.fields.append((" SETTINGS.Result", "0x%02X — %s" % + (code, result_name))) + else: + operation = "INVALID" + res.warnings.append("Ответ SETTINGS должен иметь DLC = 1 или 8") + + res.summary = "SETTINGS %s: сборка %d, позиция %d" % ( + operation, assembly, position) + + def _decode_pulse(res: Decoded) -> None: if res.dlc >= 1: res.summary = "PULSE, счётчик %d" % res.data[0] @@ -552,5 +721,5 @@ def _decode_pulse(res: Decoded) -> None: else: res.summary = "PULSE (без счётчика)" res.warnings.append("PULSE передаётся с DLC = 1") - if res.id.msg_body: - res.warnings.append("MsgBody у PULSE должен быть 0, получено 0x%04X" % res.id.msg_body) + if res.id.body: + res.warnings.append("Body у PULSE должен быть 0, получено 0x%04X" % res.id.body)