158 lines
6.9 KiB
Python
158 lines
6.9 KiB
Python
# Обмен с историческим терминалом TMS. Формат обычных ответов и телеметрии зависит от
|
|
# используемого адаптера; состояние parser сохраняется между фрагментами, а профиль
|
|
# контроллера выбирается вызывающим приложением.
|
|
|
|
"""Thin host binding for legacy TMS2812 requests and ordinary responses.
|
|
|
|
The wire implementation is c/set-protocol/src/tms2812.c, also used by JNI.
|
|
This is not the SETProtocol v2 wire format.
|
|
"""
|
|
import ctypes as ct
|
|
|
|
from .native import get_native_protocol, NativeProtocolUnavailable
|
|
from .tms2812 import crc16_modbus
|
|
|
|
PEEK = 56
|
|
POKE = 57
|
|
PING_ADDRESS = 0x100
|
|
|
|
|
|
def _api(name, arguments, result=ct.c_size_t):
|
|
try:
|
|
function = getattr(get_native_protocol().lib, "pcan_abi_tms2812_" + name)
|
|
except AttributeError as error:
|
|
raise NativeProtocolUnavailable("Пересоберите templates: отсутствует TMS2812 terminal API") from error
|
|
function.argtypes = arguments
|
|
function.restype = result
|
|
return function
|
|
|
|
|
|
def _range(value, maximum, name):
|
|
if not isinstance(value, int) or not 0 <= value <= maximum:
|
|
raise ValueError(f"{name} вне диапазона 0…{maximum}")
|
|
|
|
|
|
def build_fixed_request(controller, command, payload):
|
|
_range(controller, 255, "Адрес контроллера")
|
|
_range(command, 255, "Команда")
|
|
payload = bytes(payload)
|
|
output = ct.create_string_buffer(len(payload) + 4)
|
|
function = _api("build_fixed", [ct.c_uint8, ct.c_uint8, ct.c_void_p, ct.c_size_t, ct.c_void_p, ct.c_size_t])
|
|
size = function(controller, command, payload, len(payload), output, len(output))
|
|
if not size:
|
|
raise ValueError("Не удалось сформировать команду TMS2812")
|
|
return output.raw[:size]
|
|
|
|
|
|
def _memory(controller, command, address, value):
|
|
_range(controller, 255, "Адрес контроллера")
|
|
_range(address, 0xFFFFFFFF, "Адрес слова")
|
|
_range(value, 0xFFFF, "Значение")
|
|
output = ct.create_string_buffer(12)
|
|
function = _api("build_memory", [ct.c_uint8, ct.c_uint8, ct.c_uint32, ct.c_uint16, ct.c_void_p, ct.c_size_t])
|
|
size = function(controller, command, address, value, output, len(output))
|
|
if not size:
|
|
raise ValueError("Не удалось сформировать команду памяти TMS2812")
|
|
return output.raw[:size]
|
|
|
|
|
|
def request(controller, address, value=None):
|
|
return _memory(controller, PEEK if value is None else POKE, address, 0 if value is None else value)
|
|
|
|
|
|
def ping(controller):
|
|
return _memory(controller, 0, 0, 0)
|
|
|
|
|
|
def build_test_request(address, duration, pause, digits):
|
|
_range(address, 255, "Адрес контроллера")
|
|
_range(duration, 65535, "Длительность")
|
|
_range(pause, 65535, "Пауза")
|
|
digits = bytes(digits)
|
|
if len(digits) != 12:
|
|
raise ValueError("CMD_TEST_ALL: требуется 12 байт управления")
|
|
output = ct.create_string_buffer(20)
|
|
function = _api("build_test", [ct.c_uint8, ct.c_uint16, ct.c_uint16, ct.c_void_p, ct.c_size_t, ct.c_void_p, ct.c_size_t])
|
|
size = function(address, duration, pause, digits, len(digits), output, len(output))
|
|
if not size:
|
|
raise ValueError("Не удалось сформировать CMD_TEST_ALL")
|
|
return output.raw[:size]
|
|
|
|
|
|
def validate_response(raw, controller, command, full_size, allow_short_tail=False):
|
|
_range(controller, 255, "Адрес контроллера")
|
|
_range(command, 255, "Команда")
|
|
if full_size < 6:
|
|
raise ValueError("Неверная длина ответа TMS2812")
|
|
function = _api("validate_response", [ct.c_void_p, ct.c_size_t, ct.c_uint8, ct.c_uint8, ct.c_size_t, ct.c_int], ct.c_int)
|
|
status = function(bytes(raw), len(raw), controller, command, full_size, bool(allow_short_tail))
|
|
errors = {-1: "Неверный формат ответа TMS2812", -3: "Неверная длина ответа TMS2812",
|
|
-4: "CRC ответа TMS2812 не совпадает", -5: "Неверный адрес или команда ответа TMS2812"}
|
|
if status:
|
|
raise ValueError(errors.get(status, f"Ошибка TMS2812: {status}"))
|
|
|
|
|
|
def decode_reply(raw, controller, command):
|
|
if command not in (PEEK, POKE):
|
|
raise ValueError("Ожидается CMD_PEEK или CMD_POKE")
|
|
validate_response(raw, controller, command, 10 if command == PEEK else 6, True)
|
|
return int.from_bytes(raw[2:4], "little") if command == PEEK else None
|
|
|
|
|
|
def memory_response(raw, controller, command):
|
|
"""Find a PEEK/POKE response in a UART buffer, as Android feedLegacy does.
|
|
|
|
Returns None until data plus CRC are present. A short valid result still
|
|
needs the transport's settle timer before the next request. Ignore noise,
|
|
an optional preliminary ACK, and bytes after the complete response.
|
|
"""
|
|
_range(controller, 255, "Адрес контроллера")
|
|
if command not in (PEEK, POKE):
|
|
raise ValueError("Ожидается CMD_PEEK или CMD_POKE")
|
|
header = bytes((controller, command))
|
|
start = bytes(raw).find(header)
|
|
if start < 0:
|
|
return None
|
|
data = bytes(raw[start:])
|
|
size = 10 if command == PEEK else 6
|
|
if size > 6:
|
|
for ack_size in (6, 4):
|
|
if data[ack_size:ack_size + 2] != header:
|
|
continue
|
|
try:
|
|
validate_response(data[:ack_size], controller, command, 6, True)
|
|
except ValueError:
|
|
continue
|
|
data = data[ack_size:]
|
|
break
|
|
if len(data) >= size:
|
|
packet = data[:size]
|
|
validate_response(packet, controller, command, size)
|
|
return packet
|
|
if len(data) >= size - 2:
|
|
try:
|
|
validate_response(data, controller, command, size, True)
|
|
except ValueError:
|
|
return None
|
|
return data
|
|
return None
|
|
|
|
|
|
def bus_steps(write, tetrad, register, data=0):
|
|
"""Two bus register operations planned by the shared C core."""
|
|
_range(tetrad, 14, "Адрес шины")
|
|
_range(register, 15, "Регистр шины")
|
|
_range(data, 0xFFFF, "Данные шины")
|
|
function = _api("bus_step", [ct.c_int, ct.c_uint, ct.c_uint, ct.c_uint,
|
|
ct.c_uint16, ct.POINTER(ct.c_uint32), ct.POINTER(ct.c_uint16),
|
|
ct.POINTER(ct.c_uint8)], ct.c_int)
|
|
result = []
|
|
for step in range(2):
|
|
address, value, command = ct.c_uint32(), ct.c_uint16(), ct.c_uint8()
|
|
status = function(bool(write), step, tetrad, register, data,
|
|
ct.byref(address), ct.byref(value), ct.byref(command))
|
|
if status:
|
|
raise ValueError(f"Ошибка планирования шины: {status}")
|
|
result.append((address.value, None if command.value == PEEK else value.value))
|
|
return result
|