Andrey
4d101302be
Подключить профили прошивки из JSON и подготовку Python для CCS/Keil. Добавить SCI-программатор F28335 и поддержку STM32 через J-Link. Обновить Linux-консоль, CAN485 и Modbus и проверки общих портов.
55 lines
11 KiB
Python
55 lines
11 KiB
Python
"""Тонкий ctypes-порт общего кодека Modbus-шлюза; USB и UI не повторяют wire-алгоритмы.""" # Один C-источник собирается в ARM-службу и библиотеку ПК.
|
||
import ctypes as ct # Native ABI передаёт структуры отдельно от wire-пакетов.
|
||
import os # Разрешает явно выбрать проверенную библиотеку в standalone-сборке и тестах.
|
||
from pathlib import Path # Ищет библиотеку в пакете потребителя и локальном результате сборки.
|
||
import sys # Установленный SETGUI и frozen runtime имеют разные исходные каталоги.
|
||
from ipaddress import IPv4Address # Преобразует адрес редактора в числовое поле модели без разбора Modbus.
|
||
class Request(ct.Structure): # Структура должна соответствовать set_modbus_gateway.h, а не wire-layout.
|
||
_fields_ = [('transaction', ct.c_uint16), ('length', ct.c_uint16), ('unit', ct.c_uint8), ('pdu', ct.c_uint8 * 253)] # Размер PDU ограничен общим контрактом Modbus.
|
||
class Route(ct.Structure): # Поля выравниваются компилятором C; по проводу их сериализует ядро.
|
||
_fields_ = [('source', ct.c_uint8), ('target', ct.c_uint8), ('transport', ct.c_uint8), ('interface_id', ct.c_uint8), ('ipv4', ct.c_uint32), ('port', ct.c_uint16)] # Numeric IPv4 не зависит от endian процесса.
|
||
class Config(ct.Structure): # Хранит до двадцати маршрутов и отдельные параметры физической RS485.
|
||
_fields_ = [('baud', ct.c_uint32), ('timeout_ms', ct.c_uint16), ('parity', ct.c_uint8), ('stop_bits', ct.c_uint8), ('default_serial', ct.c_uint8), ('count', ct.c_uint8), ('routes', Route * 20)] # Модель повторяет только ABI, а не формат USB-сообщения.
|
||
class GatewayCodec: # Отсутствие C-библиотеки сообщается явно без алгоритмического Python fallback.
|
||
def __init__(self, library=None): # Можно подставить конкретную DLL из изолированной проверенной сборки.
|
||
candidates = [library or os.environ.get('ZYNQ_MODBUS_LIBRARY', '')] # Явный путь имеет приоритет над установленными копиями.
|
||
for base in (Path(sys.prefix), Path(getattr(sys, '_MEIPASS', sys.prefix)), *Path(__file__).resolve().parents): # Frozen и исходный SETGUI используют одинаковые назначения native-библиотеки.
|
||
candidates.extend((base / 'gui_desktop/native/zynq_modbus.dll', base / 'src/gui_desktop/native/zynq_modbus.dll', base / 'build/modbus_gateway/zynq_modbus.dll')) # Выбираются только реальные файловые кандидаты без загрузки случайного имени из PATH.
|
||
path = next((Path(p) for p in candidates if p and Path(p).is_file()), None) # Поиск не создаёт файлов и не меняет среду пользователя.
|
||
if path is None: raise OSError('Не найдена библиотека zynq_modbus.dll; выполните сборку Modbus-шлюза') # Не предлагает устаревший Python-кодек как незаметную замену.
|
||
self.lib = ct.CDLL(str(path)) # Библиотека содержит только общий переносимый C-код templates.
|
||
signatures = {'smg_tcp_size': ([ct.c_void_p, ct.c_size_t], ct.c_int), 'smg_rtu_response_size': ([ct.c_void_p, ct.c_size_t], ct.c_int), 'smg_decode_tcp': ([ct.c_void_p, ct.c_size_t, ct.POINTER(Request)], ct.c_int), 'smg_decode_rtu': ([ct.c_void_p, ct.c_size_t, ct.POINTER(Request)], ct.c_int), 'smg_check': ([ct.POINTER(Request)], ct.c_uint), 'smg_config_default': ([ct.POINTER(Config)], None), 'smg_config_valid': ([ct.POINTER(Config)], ct.c_int), 'smg_config_pack': ([ct.c_uint, ct.POINTER(Config), ct.c_void_p, ct.c_size_t], ct.c_size_t), 'smg_config_unpack': ([ct.c_void_p, ct.c_size_t, ct.POINTER(ct.c_uint), ct.POINTER(Config)], ct.c_int), 'smg_rtu': ([ct.c_uint8, ct.c_void_p, ct.c_size_t, ct.c_void_p, ct.c_size_t], ct.c_size_t), 'smg_tcp': ([ct.POINTER(Request), ct.c_uint8, ct.c_void_p, ct.c_size_t, ct.c_void_p, ct.c_size_t], ct.c_size_t), 'smg_accept_rtu': ([ct.POINTER(Request), ct.c_uint8, ct.c_void_p, ct.c_size_t, ct.c_void_p, ct.c_size_t], ct.c_size_t), 'smg_exception': ([ct.POINTER(Request), ct.c_uint, ct.c_void_p], ct.c_size_t)} # Явные типы исключают усечение указателей и size_t на Windows x64.
|
||
for name, (arguments, result) in signatures.items(): fn = getattr(self.lib, name); fn.argtypes = arguments; fn.restype = result # Каждая экспортируемая функция имеет проверяемый ABI.
|
||
self.lib.smg_accept_tcp.argtypes=[ct.POINTER(Request),ct.c_uint8,ct.c_void_p,ct.c_size_t,ct.c_void_p,ct.c_size_t]; self.lib.smg_accept_tcp.restype=ct.c_size_t # Общий FTDI переносит MBAP/TCP через loopback службы без собственной реализации корреляции в Python.
|
||
def default(self): # Начальные значения получает из единственного общего ядра.
|
||
result = Config(); self.lib.smg_config_default(ct.byref(result)); return result # Python не задаёт другую скорость либо таблицу по умолчанию.
|
||
def config(self, baud, timeout, parity, stops, default_serial, routes): # Преобразует только значения визуальной модели в структуру ABI.
|
||
result = self.default(); result.baud = int(baud); result.timeout_ms = int(timeout); result.parity = int(parity); result.stop_bits = int(stops); result.default_serial = bool(default_serial) # Значения ограничиваются UI и повторно проверяются C.
|
||
if len(routes) > len(result.routes): raise ValueError('Допустимо не более 20 маршрутов') # Проверка ёмкости ABI предшествует присваиванию индексов.
|
||
result.count = len(routes) # Количество адресует только заполненную часть фиксированного массива.
|
||
for i, row in enumerate(routes): result.routes[i] = Route(int(row['source']), int(row['target']), int(row['transport']), int(row['interface_id']), int(IPv4Address(row['host'])) if row['transport'] else 0, int(row['port']) if row['transport'] else 0) # Преобразование адреса не содержит алгоритма Modbus.
|
||
if not self.lib.smg_config_valid(ct.byref(result)): raise ValueError('Проверьте скорости, адреса 1…247, уникальность адресов ПК и параметры маршрутов') # Семантические ограничения находятся в templates C.
|
||
return result # Возвращённая модель ещё не применена к плате.
|
||
def control(self, config=None): # Создаёт чтение либо полную транзакционную запись настроек.
|
||
request = Request(); request.unit = 248; request.length = self.lib.smg_config_pack(1 if config is None else 2, ct.byref(config) if config is not None else None, request.pdu, len(request.pdu)) # Адрес 248 выделен контрактом шлюза для управления.
|
||
if not request.length: raise ValueError('Некорректная конфигурация шлюза') # Не отправляет пустую либо частично собранную таблицу.
|
||
return request # Транспорт самостоятельно выбирает USB или TCP для этого PDU.
|
||
def decode_tcp(self, wire): # Порт передаёт ровно один ADU после получения его размера от C.
|
||
result = Request() # Native-структура не ссылается на временный входной bytes.
|
||
if not self.lib.smg_decode_tcp(bytes(wire), len(wire), ct.byref(result)): raise ValueError('Повреждённый MBAP') # Фрагментацию разрешает потоковый порт, а не повторный разбор в Python.
|
||
return result # Сохраняет исходные Transaction ID и Unit ID локального клиента.
|
||
def rtu(self, request): # Формирует один USB RTU запрос посредством C-кодека.
|
||
output = ct.create_string_buffer(260); n = self.lib.smg_rtu(request.unit, request.pdu, request.length, output, len(output)); return output.raw[:n] # CRC и endian определяются только templates.
|
||
def tcp_reply(self, request, pdu): # Возвращает проверенный PDU исходному локальному TCP-клиенту.
|
||
output = ct.create_string_buffer(260); n = self.lib.smg_tcp(ct.byref(request), request.unit, bytes(pdu), len(pdu), output, len(output)); return output.raw[:n] # Корреляция не вычисляется вручную в Qt.
|
||
def accept(self, request, wire, tcp=False): # Проверяет CRC либо MBAP, slave, функцию и форму ответа единым C-ядром.
|
||
output = ct.create_string_buffer(253); accept=self.lib.smg_accept_tcp if tcp else self.lib.smg_accept_rtu; n = accept(ct.byref(request), request.unit, bytes(wire), len(wire), output, len(output)) # Способ транспорта не ослабляет проверку Transaction ID и формы ответа.
|
||
if not n: raise ValueError('Повреждённый ответ USB: CRC, адрес или функция не совпадают') # Не скрывает ошибку связи последним отображённым значением.
|
||
return output.raw[:n] # PDU готов для локального TCP либо загрузки модели настроек.
|
||
def read_config(self, pdu): # Декодирует только подтверждённый ответ управления.
|
||
result = Config(); operation = ct.c_uint() # Память модели принадлежит Python и не сохраняется внутри C.
|
||
if not self.lib.smg_config_unpack(bytes(pdu), len(pdu), ct.byref(operation), ct.byref(result)) or operation.value != 3: raise ValueError('Плата отклонила настройки либо вернула неизвестную версию шлюза') # Исключение Modbus не принимается за конфигурацию.
|
||
return result # Полученные значения подтверждены платой, а не взяты из локального черновика.
|
||
def exception(self, request, code): # Ошибки локального USB также оборачиваются через общее ядро.
|
||
output = ct.create_string_buffer(2); n = self.lib.smg_exception(ct.byref(request), code, output); return output.raw[:n] # Формат ошибки не повторяется в Python.
|