"""Balsam 167 legacy CAN register decoder backed by the shared C99 core.""" from __future__ import annotations from dataclasses import dataclass from .native import NativeProtocolUnavailable, get_native_protocol BASE_ID = 0x00BA0000 DATA_OFFSET = 0x10 NODE_COUNT = 13 PULT_COMMAND_ID = 0x00745009 PULT_DATA_ID = 0x00745019 @dataclass(frozen=True) class BalsamRegister: address: int value: int name: str @property def signed_value(self) -> int: return self.value if self.value < 0x8000 else self.value - 0x10000 @dataclass(frozen=True) class BalsamFrame: can_id: int device: int device_name: str from_device: bool start_address: int present_mask: int registers: tuple[BalsamRegister, ...] @property def route(self) -> str: return ("%s → ПМ" % self.device_name if self.from_device else "ПМ → %s" % self.device_name) def command_description(self, value: int) -> str: """Command register 127; named UMP bits from BALZAM_167 automats.h.""" if value == 0: return "нет активных команд" names = ({0x0004: "сохранение уставок", 0x0010: "установка нулевого уровня"} if 1 <= self.device <= 7 else {}) if self.device in (5, 6): names.update({0x0040: "останов заряда", 0x0080: "пуск заряда", 0x0100: "инициализация ЦАП", 0x8000: "сброс УМП"}) return ", ".join(names.get(1 << bit, "бит %d" % bit) for bit in range(16) if value & (1 << bit)) @property def summary(self) -> str: direction = "данные" if self.from_device else "команда" values = ", ".join( "%s=0x%04X (%d)" % (item.name or "R%04X" % item.address, item.value, item.signed_value) + (" — " + self.command_description(item.value) if not self.from_device and item.address == 0x7F else "") for item in self.registers ) or "нет отмеченных регистров" return "BALZAM · %s · %s · %s" % (self.route, direction, values) def is_balsam_id(can_id: int) -> bool: if can_id & 0x1FFFFFFF in (PULT_COMMAND_ID, PULT_DATA_ID): return True relative = (can_id & 0x1FFFFFFF) - BASE_ID return (0 <= relative < NODE_COUNT or DATA_OFFSET <= relative < DATA_OFFSET + NODE_COUNT) def decode(can_id: int, data: bytes, native=None) -> BalsamFrame | None: """Decode wire order: BE value2, value3, mask/address, value1.""" if native is None: try: native = get_native_protocol() except NativeProtocolUnavailable: return _decode_fallback(can_id, data) status, decoded = native.balsam_decode(can_id, bytes(data)) if status == 0: return None if status == -2: raise ValueError("BALZAM CAN frame must contain exactly 8 data bytes") if status != 1 or decoded is None: raise ValueError("invalid BALZAM CAN frame") device, direction, mask, start, values = decoded registers = tuple( BalsamRegister(start + index, values[index], native.balsam_register_name(device, start + index)) for index in range(3) if mask & (4 >> index) ) return BalsamFrame(can_id & 0x1FFFFFFF, device, native.balsam_device_name(device), direction == 1, start, mask, registers) def _decode_fallback(can_id: int, data: bytes) -> BalsamFrame | None: if not is_balsam_id(can_id): return None if len(data) != 8: raise ValueError("BALZAM CAN frame must contain exactly 8 data bytes") relative = (can_id & 0x1FFFFFFF) - BASE_ID device = (relative & 0x0F) + 1 from_device = relative >= DATA_OFFSET if can_id & 0x1FFFFFFF in (PULT_COMMAND_ID, PULT_DATA_ID): device = 10 from_device = (can_id & 0x1FFFFFFF) == PULT_DATA_ID header = int.from_bytes(data[4:6], "big") values = tuple(int.from_bytes(data[offset:offset + 2], "big") for offset in (6, 0, 2)) names = { 1: "Трансформатор 1", 2: "Трансформатор 2", 3: "Силовой блок 1", 4: "Силовой блок 2", 5: "УМП 1", 6: "УМП 2", 7: "Двигатель", 8: "ВЭП", 9: "Задатчик", 10: "Пульт (МПУ)", 13: "Терминал", } start, mask = header & 0x1FFF, (header >> 13) & 7 registers = tuple(BalsamRegister(start + i, values[i], "Команды" if start + i == 0x7F else "") for i in range(3) if mask & (4 >> i)) return BalsamFrame(can_id & 0x1FFFFFFF, device, names.get(device, "Узел %d" % device), from_device, start, mask, registers)