130 lines
5.2 KiB
Python
130 lines
5.2 KiB
Python
# Python-привязка исторического CAN-протокола Balsam к общему C99-ядру. Модуль представляет
|
|
# результат в объектах Python; двоичный контракт регистров остаётся в balsam_can.
|
|
|
|
"""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)
|