135 lines
5.6 KiB
Python
135 lines
5.6 KiB
Python
"""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
|