"""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 @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 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) for item in self.registers ) or "нет отмеченных регистров" return "BALZAM · %s · %s · %s" % (self.device_name, direction, values) def is_balsam_id(can_id: int) -> bool: 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 one abstract Balsam frame: BE mask/address plus three BE words.""" 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 header = int.from_bytes(data[:2], "big") values = tuple(int.from_bytes(data[offset:offset + 2], "big") for offset in (2, 4, 6)) names = { 1: "Трансформатор 1", 2: "Трансформатор 2", 3: "Силовой блок 1", 4: "Силовой блок 2", 5: "УМП 1", 6: "УМП 2", 7: "Двигатель", 8: "ВЭП", 9: "Задатчик", 13: "Терминал", } start, mask = header & 0x1FFF, (header >> 13) & 7 registers = tuple(BalsamRegister(start + i, values[i], "") for i in range(3) if mask & (4 >> i)) return BalsamFrame(can_id & 0x1FFFFFFF, device, names.get(device, "Узел %d" % device), relative >= DATA_OFFSET, start, mask, registers)