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

refactor(python): centralize ProtoCAN decoder

Browse Source
This commit is contained in:
Andrey
2026-09-01 12:00:53 +03:00
parent 00a0f944c4
commit e9c5ec90b8
1 changed files with 204 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

239
python/protocan/protocan.py
View File

@@ -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)