Исправить порядок слов BALZAM CAN и добавить поддержку пульта МПУ
This commit is contained in:
@@ -10,6 +10,8 @@ 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)
|
||||
@@ -33,25 +35,46 @@ class BalsamFrame:
|
||||
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.device_name, direction, values)
|
||||
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 one abstract Balsam frame: BE mask/address plus three BE words."""
|
||||
"""Decode wire order: BE value2, value3, mask/address, value1."""
|
||||
if native is None:
|
||||
try:
|
||||
native = get_native_protocol()
|
||||
@@ -82,17 +105,22 @@ def _decode_fallback(can_id: int, data: bytes) -> BalsamFrame | None:
|
||||
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")
|
||||
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 (2, 4, 6))
|
||||
for offset in (6, 0, 2))
|
||||
names = {
|
||||
1: "Трансформатор 1", 2: "Трансформатор 2",
|
||||
3: "Силовой блок 1", 4: "Силовой блок 2", 5: "УМП 1", 6: "УМП 2",
|
||||
7: "Двигатель", 8: "ВЭП", 9: "Задатчик", 13: "Терминал",
|
||||
7: "Двигатель", 8: "ВЭП", 9: "Задатчик", 10: "Пульт (МПУ)", 13: "Терминал",
|
||||
}
|
||||
start, mask = header & 0x1FFF, (header >> 13) & 7
|
||||
registers = tuple(BalsamRegister(start + i, values[i], "")
|
||||
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),
|
||||
relative >= DATA_OFFSET, start, mask, registers)
|
||||
from_device, start, mask, registers)
|
||||
|
||||
@@ -31,12 +31,15 @@ def decode(raw_id: int, data: bytes, *, ide: bool = True,
|
||||
)
|
||||
result = pc.Decoded(pid, len(data), bytes(data), ide=ide, rtr=rtr,
|
||||
protocol="BALZAM 167", device_label=frame.device_name,
|
||||
message_label="BALZAM регистры")
|
||||
message_label=("BALZAM данные" if frame.from_device
|
||||
else "BALZAM команды"))
|
||||
result.summary = frame.summary
|
||||
result.fields = [
|
||||
("Протокол", "BALZAM 167 legacy eCAN"),
|
||||
("Узел", "%d — %s" % (frame.device, frame.device_name)),
|
||||
("Направление", "от устройства" if frame.from_device else "к устройству"),
|
||||
("Направление", frame.route),
|
||||
("Отправитель", frame.device_name if frame.from_device else "ПМ"),
|
||||
("Получатель", "ПМ" if frame.from_device else frame.device_name),
|
||||
("Начальный адрес", "0x%04X" % frame.start_address),
|
||||
("Маска трёх слов", "0b%s" % format(frame.present_mask, "03b")),
|
||||
]
|
||||
@@ -46,6 +49,8 @@ def decode(raw_id: int, data: bytes, *, ide: bool = True,
|
||||
"0x%04X %s" % (item.address, item.name or "регистр"),
|
||||
"0x%04X (%d)" % (item.value, item.signed_value),
|
||||
))
|
||||
if not frame.from_device and item.address == 0x7F:
|
||||
result.fields.append(("Команды ПМ", frame.command_description(item.value)))
|
||||
if frame.present_mask == 0:
|
||||
result.warnings.append("В адресном слове не отмечен ни один регистр")
|
||||
return result
|
||||
|
||||
Reference in New Issue
Block a user