Исправлено api для пк на поддержку версии мк 1.4.0, внедрены pytest, .dll; обновлен main.py

parent fb56a0ac
"""
lta_cli_v4.py
Python API для CLI протокола LTA27 (v1.4.0)
"""
import ctypes
import os
import sys
import time
from typing import Dict, Any, List, Optional
import pytest
import serial as pyserial
# ------------------------------------------------------------------
# Конфигурация
# ------------------------------------------------------------------
CLI_PORT = "COM3"
DLL_DIR = r"C:\work\LTA27\lta27_configurator\code\lib"
DLL_NAME = "lta_api.dll"
# ------------------------------------------------------------------
# Исключения
# ------------------------------------------------------------------
class CLIError(Exception):
"""Ошибка протокола CLI."""
class LtaApiError(Exception):
"""Ошибка при работе с DLL."""
# ==================================================================
# 1. Обёртка DLL (lta_api.dll)
# ==================================================================
class LtaApi:
"""Обёртка DLL."""
class LtaDevRec(ctypes.Structure):
_fields_ = [
("devname", ctypes.c_char * 32),
("serial_dev", ctypes.c_char * 32),
("serial_crate", ctypes.c_char * 32),
("slot", ctypes.c_uint8),
("ip_addr", ctypes.c_uint32),
]
def __init__(self, lib_path: str):
if not os.path.exists(lib_path):
raise LtaApiError(f"Файл DLL не найден: {lib_path}")
try:
self._lib = ctypes.WinDLL(lib_path)
except OSError as error:
raise LtaApiError(f"Не удалось загрузить DLL: {error}") from error
# Прототип LTA_Search
try:
self._lib.LTA_Search.argtypes = [
ctypes.POINTER(self.LtaDevRec),
ctypes.c_uint32,
ctypes.POINTER(ctypes.c_uint32),
ctypes.c_uint32,
ctypes.c_char_p,
ctypes.c_char_p,
ctypes.c_uint8,
]
self._lib.LTA_Search.restype = ctypes.c_int32
except AttributeError:
raise LtaApiError("Функция LTA_Search не найдена в DLL")
def search(self,
max_count: int = 10,
timeout_ms: int = 5000,
name_filter: Optional[str] = None,
serial_filter: Optional[str] = None,
slot: int = 0) -> List[LtaDevRec]:
"""Поиск устройств в сети."""
dev_array = (self.LtaDevRec * max_count)()
found = ctypes.c_uint32(0)
ret = self._lib.LTA_Search(
dev_array,
max_count,
ctypes.byref(found),
timeout_ms,
name_filter.encode('utf-8') if name_filter else None,
serial_filter.encode('utf-8') if serial_filter else None,
slot
)
if ret < 0:
raise LtaApiError(f"LTA_Search вернула ошибку: {ret}")
return [dev_array[i] for i in range(found.value)]
# ==================================================================
# 2. Клиент CLI (v1.4.0)
# ==================================================================
class LTAClient:
"""
CLI‑клиент для прошивки v1.4.0.
Особенности v1.4.0:
- send_response отправляет данные в HEX (каждый байт → 2 символа).
- Ответ на ECHO: tt02<HEX‑данные> (без поля длины).
- Ответ на GET_INFO: tt03<HEX‑данные> (без поля длины).
"""
CMD_OPEN = 0x00
CMD_CLOSE = 0x01
CMD_ECHO = 0x02
CMD_GET_INFO = 0x03
# Коды ошибок
ERR_UNKNOWN_CMD = 0x01
ERR_INVALID_ARG = 0x02
ERR_INTERNAL = 0x03
ERR_NOT_CONNECTED = 0x04
_ERROR_MESSAGES = {
ERR_UNKNOWN_CMD: "неизвестная команда",
ERR_INVALID_ARG: "неверный аргумент",
ERR_INTERNAL: "внутренняя ошибка",
ERR_NOT_CONNECTED: "устройство не подключено"
}
def __init__(self, port: str = CLI_PORT, baudrate: int = 115200,
debug: bool = True): # ← debug=True по умолчанию
self.port = port
self.baudrate = baudrate
self.debug = debug
self._serial: Optional[pyserial.Serial] = None
self._connected = False
# ------------------------------------------------------------------
# Транспорт
# ------------------------------------------------------------------
def _read_response(self) -> bytes:
line = self._serial.read_until(b'\r')
if not line:
raise CLIError("Таймаут ожидания ответа")
line = line.rstrip(b'\r\n')
if self.debug:
print(f"[RX] {line.decode('ascii', errors='replace')}")
return line
def send_raw(self, raw_cmd: str) -> bytes:
"""Отправить сырую команду и получить ответ (без парсинга)."""
if self.debug:
print(f"[TX] {raw_cmd}")
self._serial.write(f"{raw_cmd}\r".encode())
self._serial.flush()
response = self._read_response()
return response
def send_command(self, cmd: int, payload: bytes = b'') -> bytes:
"""
Отправить команду с парсингом ответа.
Размер данных в команде — количество HEX-символов (len(hex_data)).
"""
hex_data = payload.hex().upper() if payload else ""
data_size = len(hex_data)
if self.debug:
print(f"[TX_LEN] {data_size} HEX-символов")
# \r внутри cmd_str
cmd_str = f"T{cmd:02X}{data_size:02X}{hex_data}\r"
resp = self.send_raw(cmd_str)
# --- Парсинг ошибки ---
if resp[:2] == b'ee':
if len(resp) >= 6:
err_code = int(resp[4:6], 16)
err_text = self._ERROR_MESSAGES.get(
err_code, f"код 0x{err_code:02X}"
)
raise CLIError(f"Ошибка команды 0x{cmd:02X}: {err_text}")
raise CLIError("Некорректный ответ об ошибке")
# --- Парсинг успеха ---
if resp[:2] != b'tt':
raise CLIError(f"Неверный префикс ответа: {resp[:4].decode('ascii', errors='replace')}")
if len(resp) < 4:
raise CLIError("Слишком короткий ответ")
resp_cmd = int(resp[2:4], 16)
if resp_cmd != cmd:
raise CLIError(f"Несовпадение кода: 0x{cmd:02X} vs 0x{resp_cmd:02X}")
# Возвращаем данные после tt<cmd>
return resp[4:]
# ------------------------------------------------------------------
# Порт
# ------------------------------------------------------------------
def open_port(self) -> None:
if self._serial and self._serial.is_open:
return
self._serial = pyserial.Serial(
port=self.port,
baudrate=self.baudrate,
bytesize=pyserial.EIGHTBITS,
parity=pyserial.PARITY_NONE,
stopbits=pyserial.STOPBITS_ONE,
timeout=1.0
)
time.sleep(0.3)
self._serial.reset_input_buffer()
self._serial.reset_output_buffer()
if self.debug:
print(f"✓ Порт {self.port} открыт")
def close_port(self) -> None:
if self._serial and self._serial.is_open:
self._serial.close()
if self.debug:
print(f"✓ Порт {self.port} закрыт")
# ------------------------------------------------------------------
# Сессия
# ------------------------------------------------------------------
def connect(self) -> bool:
"""Открыть порт и соединение."""
self.open_port()
try:
self.send_command(self.CMD_OPEN)
self._connected = True
print("✓ Соединение открыто")
return True
except CLIError:
print("✗ Ошибка открытия соединения")
return False
def close_connection(self) -> None:
"""Закрыть соединение (порт остаётся открытым)."""
self.send_command(self.CMD_CLOSE)
self._connected = False
print("✓ Соединение закрыто")
def disconnect(self) -> None:
"""Закрыть соединение и порт."""
if self._connected:
self.close_connection()
self.close_port()
# ------------------------------------------------------------------
# Команды
# ------------------------------------------------------------------
def ping(self, data: bytes = b'') -> str:
"""Проверка связи (ECHO)."""
hex_data = data.hex().upper() if data else ""
data_size = len(hex_data)
cmd_str = f"T02{data_size:02X}{hex_data}\r"
resp = self.send_raw(cmd_str)
if resp[:2] == b'ee':
if len(resp) >= 6:
err_code = int(resp[4:6], 16)
err_text = self._ERROR_MESSAGES.get(err_code, f"код 0x{err_code:02X}")
raise CLIError(f"Ошибка PING: {err_text}")
if resp[:2] != b'tt':
raise CLIError("Неверный ответ PING")
result = resp[4:].decode('ascii')
print(f"✓ Ping OK, данные: {result if result else 'пусто'}")
return result
def get_device_info(self) -> Dict[str, Any]:
"""Получить информацию об устройстве."""
resp = self.send_raw("T0300\r")
if resp[:2] == b'ee':
if len(resp) >= 6:
err_code = int(resp[4:6], 16)
err_text = self._ERROR_MESSAGES.get(err_code, f"код 0x{err_code:02X}")
raise CLIError(f"Ошибка GET_INFO: {err_text}")
if resp[:2] != b'tt':
raise CLIError("Неверный ответ GET_INFO")
hex_str = resp[4:].decode('ascii')
data = bytes.fromhex(hex_str)
idx = 0
name_len = data[idx]; idx += 1
name = data[idx:idx+name_len].decode('ascii', errors='replace')
idx += name_len
serial_len = data[idx]; idx += 1
serial_num = data[idx:idx+serial_len].decode('ascii', errors='replace')
idx += serial_len
ver_len = data[idx]; idx += 1
version = data[idx:idx+ver_len].decode('ascii', errors='replace')
return {'name': name, 'serial': serial_num, 'version': version}
# ------------------------------------------------------------------
# Контекстный менеджер
# ------------------------------------------------------------------
def __enter__(self):
self.connect()
return self
def __exit__(self, *args):
self.disconnect()
# ==================================================================
# pytest-фикстуры и тесты
# ==================================================================
@pytest.fixture(scope="session")
def cli():
try:
pyserial.Serial(CLI_PORT).close()
except (pyserial.SerialException, OSError):
pytest.skip(f"Порт {CLI_PORT} недоступен")
client = LTAClient(port=CLI_PORT)
client.open_port()
yield client
client.close_port()
@pytest.fixture(scope="session")
def lta():
dll_path = os.path.join(DLL_DIR, DLL_NAME)
if not os.path.exists(dll_path):
pytest.skip(f"DLL не найдена: {dll_path}")
return LtaApi(dll_path)
# --- Тесты ---
def test_dll_search(lta):
print("\n=== Поиск устройств через DLL ===")
devices = lta.search(max_count=5, timeout_ms=5000)
assert len(devices) > 0, "Не найдено ни одного устройства"
for device in devices:
print(f" Имя: {device.devname.decode().strip()}, "
f"SN: {device.serial_dev.decode().strip()}, "
f"Слот: {device.slot}")
def test_connect(cli):
print("\n=== Открытие соединения ===")
assert cli.connect(), "Не удалось открыть соединение"
def test_device_info(cli):
print("\n=== T0300 информация ===")
result = cli.get_device_info()
assert result['name'] and result['serial'] and result['version']
print(f"Имя: {result['name']}, Серийный: {result['serial']}, Версия: {result['version']}")
def test_echo(cli):
print("\n=== PING T020212 ===")
resp = cli.ping(b'\x12')
assert resp == '12'
print("\n=== PING T0204AABB ===")
resp = cli.ping(b'\xAA\xBB')
assert resp == 'AABB'
def test_disconnect(cli):
print("\n=== Закрытие соединения ===")
cli.close_connection()
# ==================================================================
# Ручной запуск
# ==================================================================
if __name__ == "__main__":
# 1. DLL поиск
dll_full_path = os.path.join(DLL_DIR, DLL_NAME)
if os.path.exists(dll_full_path):
try:
api = LtaApi(dll_full_path)
devs = api.search(max_count=10, timeout_ms=5000)
if devs:
print("=== Поиск устройств через DLL ===")
for dev in devs:
print(f" Имя: {dev.devname.decode().strip()}, "
f"SN: {dev.serial_dev.decode().strip()}, "
f"Слот: {dev.slot}")
except LtaApiError as err:
print(f"Ошибка DLL: {err}")
# 2. CLI сценарий
cli = LTAClient(port=CLI_PORT, debug=True)
try:
print("\n=== Открытие соединения ===")
if not cli.connect():
sys.exit(1)
print("\n=== T0300 информация ===")
info = cli.get_device_info()
print(f"Имя: {info['name']}, Серийный: {info['serial']}, Версия: {info['version']}")
print("\n=== PING T020212 ===")
ping_result = cli.ping(b'\x12')
print(f"Ответ: {ping_result}")
print("\n=== PING T0204AABB ===")
ping_result = cli.ping(b'\xAA\xBB')
print(f"Ответ: {ping_result}")
print("\n=== Закрытие соединения ===")
cli.close_connection()
except CLIError as err:
print(f"Ошибка CLI: {err}")
except KeyboardInterrupt:
print("\nПрервано")
finally:
cli.disconnect()
\ No newline at end of file
""" """
lta_cli_api.py Скрипт тестирования LTA27 по UART-CLI (v1.4.0)
Python API для CLI протокола LTA27 (v1.2.1)
С автоматической синхронизацией с платой
""" """
from lta_cli_v4 import LTAClient
import serial from lta_cli_v4 import CLIError
import time
import struct
from typing import Dict, Optional, Any, Callable, List
class LTACLIError(Exception):
"""Ошибка протокола CLI."""
pass
class LTAClient:
"""
Клиент для работы с устройством LTA по CLI протоколу.
"""
CMD_OPEN = 0x00
CMD_CLOSE = 0x01
CMD_PING = 0x02
CMD_GET_INFO = 0x0A
CMD_READ_MEM = 0x14
CMD_WRITE_MEM = 0x15
def __init__(self, port: str = "COM3", baudrate: int = 115200,
board_id: int = 0x01, password: str = "1357", debug: bool = True):
self.port = port
self.baudrate = baudrate
self.board_id = board_id
self.password = password
self.debug = debug
self._serial: Optional[serial.Serial] = None
self._connected = False
self._commands: Dict[int, tuple] = {}
# ------------------------------------------------------------------
# Регистрация команд
# ------------------------------------------------------------------
def register_command(self, code: int, name: str):
def decorator(func: Callable) -> Callable:
self._commands[code] = (name, func)
if self.debug:
print(f"[DEBUG] Команда 0x{code:02X} '{name}' зарегистрирована")
return func
return decorator
def call(self, code: int, *args, **kwargs) -> Any:
if code not in self._commands:
raise LTACLIError(f"Команда 0x{code:02X} не зарегистрирована")
_, func = self._commands[code]
return func(*args, **kwargs)
# ------------------------------------------------------------------
# Низкоуровневые методы
# ------------------------------------------------------------------
def _log(self, direction: str, data: str) -> None:
if self.debug:
print(f"[{direction}] {data}")
def _read_response(self) -> str:
"""Читает ответ до \r (с таймаутом)."""
resp = self._serial.read_until(b'\r') # стандартный метод pyserial
if not resp:
raise LTACLIError("Таймаут при чтении ответа")
resp = resp.rstrip(b'\r\n')
self._log("RX", resp.decode('ascii', errors='replace'))
return resp.decode('ascii', errors='replace')
def _wait_for_prompt(self, timeout: float = 5.0) -> None:
"""
Ожидает появления приглашения '|> '.
Если в течение timeout не получено, отправляет \r и повторяет ожидание.
"""
start = time.time()
buffer = ""
while True:
# Читаем доступные данные
if self._serial.in_waiting:
chunk = self._serial.read(self._serial.in_waiting).decode('ascii', errors='replace')
print(chunk, end='', flush=True)
buffer += chunk
if "|> " in buffer:
break
else:
time.sleep(0.01)
# Если таймаут вышел, отправляем \r для пробуждения платы
if time.time() - start > timeout:
self._serial.write(b'\r')
self._serial.flush()
print("[пробуждение платы]")
start = time.time() # сбросить таймер
buffer = "" # начать новый поиск приглашения
# Приглашение получено, дадим плате допечатать и очистим буфер (чтобы старые данные не попали в ответ)
time.sleep(0.05)
self._serial.reset_input_buffer()
def send_command(self, cmd: int, payload: bytes = b'') -> bytes:
hex_data = payload.hex().upper()
cmd_str = f"T{cmd:02X}{self.board_id:02X}{hex_data}"
self._log("TX", cmd_str)
self._serial.write(f"{cmd_str}\r".encode())
self._serial.flush()
response = self._read_response()
if response[0] == 'e':
err_code = int(response[5:7], 16) if len(response) >= 7 else 0
errors = {
0x01: "неизвестная команда",
0x02: "неверный формат или значение аргумента",
0x03: "внутренняя ошибка устройства"
}
raise LTACLIError(f"Ошибка: {errors.get(err_code, f'код 0x{err_code:02X}')}")
if response[0] != 't':
raise LTACLIError(f"Неверный ответ: {response}")
return bytes.fromhex(response[5:]) if len(response) > 5 else b''
# ------------------------------------------------------------------
# Управление соединением
# ------------------------------------------------------------------
def connect(self) -> bool:
print("\n" + "=" * 60)
print("ПОДКЛЮЧЕНИЕ К ПЛАТЕ LTA27")
print("=" * 60)
self._serial = serial.Serial(
port=self.port,
baudrate=self.baudrate,
bytesize=serial.EIGHTBITS,
parity=serial.PARITY_NONE,
stopbits=serial.STOPBITS_ONE,
timeout=1.0
)
print(f"Порт {self.port} открыт")
# Даём плате время и отправляем \r, чтобы вызвать вывод приглашения
time.sleep(0.5)
self._serial.write(b'\r')
self._serial.flush()
print("\nОжидание приглашения от платы...")
print("-" * 60)
self._wait_for_prompt()
print("-" * 60)
print("\nАутентификация...")
pwd_bytes = self.password.encode('ascii')
try:
self.send_command(self.CMD_OPEN, pwd_bytes)
self._connected = True
print("✓ Аутентификация успешна")
except LTACLIError as e:
print(f"✗ Ошибка аутентификации: {e}")
return False
# Ждём следующее приглашение (после успешного открытия)
self._wait_for_prompt()
print("✓ Плата готова к работе")
return True
def close(self) -> None:
if self._connected:
print("\nЗакрытие соединения...")
try:
self.send_command(self.CMD_CLOSE)
self._wait_for_prompt()
print("✓ Соединение закрыто")
except LTACLIError:
pass
self._connected = False
if self._serial and self._serial.is_open:
self._serial.close()
print(f"Порт {self.port} закрыт")
# ------------------------------------------------------------------
# Базовые команды
# ------------------------------------------------------------------
def ping(self) -> bool:
print("\n" + "-" * 40)
print("PING")
print("-" * 40)
self.send_command(self.CMD_PING)
self._wait_for_prompt()
print("✓ Ping OK")
return True
def get_info(self) -> Dict[str, Any]:
print("\n" + "=" * 60)
print("ИНФОРМАЦИЯ ОБ УСТРОЙСТВЕ")
print("=" * 60)
data = self.send_command(self.CMD_GET_INFO)
self._wait_for_prompt()
idx = 0
board_id = data[idx]; idx += 1
def read_str():
nonlocal idx
length = data[idx]; idx += 1
s = data[idx:idx+length].decode('ascii', errors='replace')
idx += length
return s
result = {
'board_id': board_id,
'name': read_str(),
'serial': read_str(),
'version': read_str()
}
print(f" Имя: {result['name']}")
print(f" Серийный: {result['serial']}")
print(f" Версия: {result['version']}")
print(f" Board ID: 0x{result['board_id']:02X}")
return result
def read_mem(self, address: int, size: int) -> bytes:
if not (1 <= size <= 64):
raise ValueError("Размер должен быть 1..64 байт")
print(f"\nЧтение памяти: 0x{address:08X} ({size} байт)")
payload = struct.pack('>I', address) + bytes([size])
data = self.send_command(self.CMD_READ_MEM, payload)
self._wait_for_prompt()
print(f" Данные: {data.hex().upper()}")
return data
def write_mem(self, address: int, data: bytes) -> bool:
if not (1 <= len(data) <= 64):
raise ValueError("Длина данных должна быть 1..64 байт")
print(f"\nЗапись в память: 0x{address:08X} <- {data.hex().upper()}")
payload = struct.pack('>I', address) + data
self.send_command(self.CMD_WRITE_MEM, payload)
self._wait_for_prompt()
print(" ✓ Записано")
return True
def scan_boards(self, start: int = 0, end: int = 254, timeout: float = 0.1) -> List[int]:
if not self._serial or not self._serial.is_open:
raise LTACLIError("Порт не открыт")
print("\n" + "=" * 60)
print("ПОИСК ПЛАТ НА ШИНЕ")
print("=" * 60)
found = []
original_timeout = self._serial.timeout
original_board_id = self.board_id
self._serial.timeout = timeout
for bid in range(start, end + 1):
self.board_id = bid
try:
saved_connected = self._connected
self._connected = True
self.send_command(self.CMD_PING)
found.append(bid)
print(f" Найдена плата: board_id=0x{bid:02X}")
self._connected = saved_connected
except LTACLIError:
pass
self._serial.reset_input_buffer()
self.board_id = original_board_id
self._serial.timeout = original_timeout
print(f"\nНайдено плат: {len(found)}")
return found
def __enter__(self):
self.connect()
return self
def __exit__(self, *args):
self.close()
# ==================================================================
# Пример использования
# ================================================================== # ==================================================================
if __name__ == "__main__": if __name__ == "__main__":
cli = LTAClient(port="COM3", board_id=0x01, password="1357", debug=True)
@cli.register_command(0x30, "read_fpga_reg") cli = LTAClient(port="COM6", debug=True)
def read_fpga_reg(reg_addr: int) -> int:
payload = struct.pack('>H', reg_addr)
resp = cli.send_command(0x30, payload)
return struct.unpack('>I', resp)[0]
try: try:
if not cli.connect(): if not cli.connect():
print("\nНе удалось подключиться к плате!")
exit(1) exit(1)
dev_info = cli.get_info() # Информация об устройстве
cli.ping() device_info = cli.get_device_info()
print(f"\nИмя: {device_info['name']}")
print(f"Серийный: {device_info['serial']}")
print(f"Версия: {device_info['version']}")
print("\n" + "=" * 60) # PING с данными
print("ТЕСТЫ ПАМЯТИ") cli.ping(b'\xAA\xBB')
print("=" * 60)
cli.read_mem(0x20000000, 4)
cli.read_mem(0x08000000, 16)
test_bytes = b'\xDE\xAD\xBE\xEF' # Закрытие соединения
cli.write_mem(0x20000000, test_bytes) cli.close_connection()
read_back = cli.read_mem(0x20000000, 4)
if read_back == test_bytes: # Попытка команды после закрытия
print("\n✓ Тест записи/чтения пройден успешно!") print("\n=== Попытка команды после закрытия ===")
else: try:
print(f"\n✗ Ошибка: записано {test_bytes.hex().upper()}, прочитано {read_back.hex().upper()}") cli.get_device_info()
print("ОШИБКА: команда выполнилась!")
except CLIError as err:
print(f"Ожидаемая ошибка: {err}")
print("\n" + "=" * 60) print("\n" + "=" * 60)
print("ВСЕ ТЕСТЫ ЗАВЕРШЕНЫ УСПЕШНО") print("ВСЕ ТЕСТЫ ЗАВЕРШЕНЫ")
print("=" * 60) print("=" * 60)
except LTACLIError as error: except CLIError as err:
print(f"\n[ОШИБКА ПРОТОКОЛА] {error}") print(f"\n[ОШИБКА] {err}")
except KeyboardInterrupt: except KeyboardInterrupt:
print("\n\nПрервано пользователем") print("\n\nПрервано пользователем")
except Exception as error:
print(f"\n[НЕИЗВЕСТНАЯ ОШИБКА] {error}")
finally: finally:
cli.close() cli.disconnect()
\ No newline at end of file
# -------------- EOF --------------
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment