108 lines
4.1 KiB
Python
108 lines
4.1 KiB
Python
"""Thin Python port of the shared C99 PM35/TMS320F28335 protocol."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from .native import get_native_protocol
|
||
|
||
|
||
REGISTER_COUNT = 128
|
||
DEFAULT_CONTROLLER = 16
|
||
DEFAULT_BAUD_RATE = 115_200
|
||
|
||
|
||
def _core():
|
||
return get_native_protocol()
|
||
|
||
|
||
def crc16_modbus(data: bytes, crc: int = 0xFFFF) -> int:
|
||
if crc != 0xFFFF:
|
||
raise ValueError("Произвольное начальное значение CRC не входит в протокол ПМ35")
|
||
return _core().periph28335_crc16(bytes(data))
|
||
|
||
|
||
def with_crc(payload: bytes) -> bytes:
|
||
return _core().periph28335_append_crc(bytes(payload))
|
||
|
||
|
||
def build_read_registers(controller: int, start: int, count: int) -> bytes:
|
||
_range("Адрес контроллера", controller, 0xFF)
|
||
_range("Начальный регистр", start, 0xFFFF)
|
||
if not 1 <= count <= REGISTER_COUNT or start + count > REGISTER_COUNT:
|
||
raise ValueError("Диапазон регистров должен находиться в 0..127")
|
||
return _core().periph28335_build_read(controller, start, count)
|
||
|
||
|
||
def build_write_register(controller: int, address: int, value: int) -> bytes:
|
||
_range("Адрес контроллера", controller, 0xFF)
|
||
_range("Адрес регистра", address, REGISTER_COUNT - 1)
|
||
_range("Значение", value, 0xFFFF)
|
||
return _core().periph28335_build_write(controller, address, value)
|
||
|
||
|
||
def build_command(controller: int, command_index: int) -> bytes:
|
||
_range("Адрес контроллера", controller, 0xFF)
|
||
if not 0 <= command_index <= 16:
|
||
raise ValueError("Номер команды должен быть в диапазоне 0..16")
|
||
return _core().periph28335_build_command(controller, command_index)
|
||
|
||
|
||
def expected_read_response_size(count: int) -> int:
|
||
if not 1 <= count <= REGISTER_COUNT:
|
||
raise ValueError("Число регистров должно быть в диапазоне 1..128")
|
||
size = _core().periph28335_expected_read_size(count)
|
||
if size == 0:
|
||
raise ValueError("SETProtocol отклонил размер ответа ПМ35")
|
||
return size
|
||
|
||
|
||
def decode_read_response(data: bytes, count: int,
|
||
controller: int | None = None) -> tuple[int, ...]:
|
||
expected = expected_read_response_size(count)
|
||
if len(data) != expected:
|
||
raise ValueError(f"Ожидалось {expected} байт, получено {len(data)}")
|
||
expected_controller = data[0] if controller is None and data else controller
|
||
if expected_controller is None:
|
||
raise ValueError("Пустой ответ ПМ35")
|
||
status, values = _core().periph28335_decode_read(
|
||
bytes(data), expected_controller, count)
|
||
errors = {
|
||
-1: "Неверные аргументы ответа",
|
||
-2: "Диапазон регистров вне 0..127",
|
||
-3: "Неверная длина данных ответа",
|
||
-4: "Ошибка CRC ответа",
|
||
-5: "Неверный заголовок ответа",
|
||
-6: "Недостаточный буфер ответа",
|
||
}
|
||
if status != 0:
|
||
raise ValueError(errors.get(status, f"Ошибка ответа ПМ35: {status}"))
|
||
return values
|
||
|
||
|
||
def validate_write_response(data: bytes, request: bytes) -> bool:
|
||
return _core().periph28335_validate_write(bytes(data), bytes(request)) == 0
|
||
|
||
|
||
def bits_lsb_first(value: int) -> tuple[bool, ...]:
|
||
_range("Значение", value, 0xFFFF)
|
||
return tuple(bool(value & (1 << bit)) for bit in range(16))
|
||
|
||
|
||
def word_from_bits(bits) -> int:
|
||
values = tuple(bool(item) for item in bits)
|
||
if len(values) != 16:
|
||
raise ValueError("Должно быть ровно 16 бит")
|
||
return sum(1 << bit for bit, checked in enumerate(values) if checked)
|
||
|
||
|
||
def signed_word(value: int) -> int:
|
||
_range("Значение", value, 0xFFFF)
|
||
return value if value < 0x8000 else value - 0x10000
|
||
|
||
|
||
def _range(name: str, value: int, maximum: int) -> None:
|
||
if not 0 <= value <= maximum:
|
||
raise ValueError(f"{name} вне диапазона 0..{maximum}")
|
||
|
||
|
||
PROJECT_COMMANDS = _core().periph28335_catalog()
|