feat(set-protocol): Раздели GAS и добавь пакетный сбор регистров
Сохрани адресацию отдельных регистров GAS 0x3 и выдели слотам тип 0x4 с маской Word0. Добавь единое C-ядро GAS_WATCH2, тонкие Python/JNI-порты и проверки общих эталонных байтов. Совместимость: тип 0x4 больше не обслуживает Modbus Coil. Потребителям PCAN_MSG_MODBUS_COIL и ProtoCanMsgToModbusCoil нужно перейти на GAS_SLOT и отдельный API слотов; регистровый GAS остаётся на 0x3.
This commit is contained in:
@@ -6,7 +6,7 @@
|
||||
from .set_uart import number, parse_frame
|
||||
|
||||
TYPES = {0: 'BROADCAST', 1: 'DISCRETE', 2: 'ANALOG', 3: 'GAS',
|
||||
4: 'MODBUS_COIL', 5: 'MODBUS_DISCRETE', 6: 'MODBUS_HOLDING',
|
||||
4: 'GAS_SLOT', 5: 'MODBUS_DISCRETE', 6: 'MODBUS_HOLDING', # Тип 0x4 теперь означает масочный слотный GAS, номера остальных Modbus-типов сохранены.
|
||||
7: 'MODBUS_INPUT', 8: 'ERROR', 9: 'BOOT_CONTROL', 10: 'BOOT_DATA_A',
|
||||
11: 'BOOT_DATA_B', 12: 'BOOT_STATUS', 13: 'BOOT_DISCOVERY', 15: 'PULSE'}
|
||||
BOOT = dict(enumerate(('IDENTIFY ENTER_BOOT BEGIN_IMAGE BEGIN_COMPAT ERASE VERIFY '
|
||||
@@ -42,7 +42,14 @@ def legacy(ident, data, profile='protocan'):
|
||||
raise ValueError('Invalid GAS register payload/range')
|
||||
result['registers'] = [(body + i//2, number(data[i:i+2])) for i in range(0, len(data), 2)]
|
||||
text += ' ' + ('READ' if not data else ' '.join('R%04X=%04X' % pair for pair in result['registers']))
|
||||
elif 4 <= msg <= 7:
|
||||
elif msg == 4: # Слотный GAS определяется номером типа, без эвристики по первым байтам данных.
|
||||
from protocan.gas_slots import GasSlots # Проверку Word0-маски, длины и порядка октетов выполняет единственный C-кодек.
|
||||
values = GasSlots().decode_address(ident, data) # Не ограничиваем анализатор адресами семейства PERIPH.
|
||||
result['slot'] = body # Адрес слота отделён от списка обычных регистров GAS 0x3.
|
||||
result['mask'] = values[0] if values else None # Пустой запрос чтения не содержит фиктивной маски.
|
||||
result['slot_words'] = [(index, values[index]) for index in range(1, 4) if values and values[0] & (1 << index)] # Публикуем только выбранные или достоверные значения, без служебного Word0.
|
||||
text += ' slot=%04X' % body + (' READ' if not values else ' mask=%04X' % values[0]) # В логе ясно различаются адрес и маска частичной записи.
|
||||
elif 5 <= msg <= 7: # Диапазон Modbus больше не включает бывший Coil 0x4.
|
||||
text += ' address=%03X count=%d' % (body >> 4, body & 15)
|
||||
elif msg == 8:
|
||||
text += ' info=%02X code=%02X' % (body >> 8, body & 255)
|
||||
|
||||
@@ -55,3 +55,7 @@ assert decode_entry(raw) == entry
|
||||
|
||||
`SETGUI` (вкладки «Шина CAN» и «Адресное пространство»), `CAN_to_RS485`.
|
||||
Там пакет лежит внутри `gui_desktop/core`, здесь — источник истины.
|
||||
|
||||
## GAS: регистры 0x3 и слоты 0x4
|
||||
|
||||
[Адресация GAS](../../doc/setcan/protocan/OAP.md): `0x3` — отдельные 16-битные регистры `0…65535`, запись 1…4 слов. `0x4` — слот с Word0-маской и тремя значениями; PERIPH сохраняет 200 слов в 67 слотах `0xMM00…0xMM42`. Формат определяется MsgType, без настройки профиля.
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
import ctypes # Передаём построение ID, проверку маски и порядок октетов единственному C-кодеку.
|
||||
from .native import get_native_core # Используем штатный поиск библиотеки без резервного Python-кодека.
|
||||
|
||||
class GasSlots: # Тип 0x4 содержит Word0 с маской и три значения в фиксированных позициях.
|
||||
def __init__(self, library=None): # Тест или приложение может передать ту же собранную библиотеку явно.
|
||||
self.lib = library or get_native_core().lib # Отсутствующее общее ядро проявляется штатной ошибкой загрузки.
|
||||
self.lib.pcan_gas_slot_request_id.argtypes = [ctypes.c_uint] * 3 # Обёртка PERIPH принимает mode, локальный слот и Route.
|
||||
self.lib.pcan_gas_slot_request_id.restype = ctypes.c_uint32 # Extended ID возвращается без знакового преобразования.
|
||||
self.lib.pcan_gas_slot_id.argtypes = [ctypes.c_uint] * 4 # Общий построитель принимает DeviceType, DeviceID, полный адрес и Route.
|
||||
self.lib.pcan_gas_slot_id.restype = ctypes.c_uint32 # Любой корректный ID типа 0x4 отличен от нулевого маркера ошибки.
|
||||
self.lib.pcan_gas_slot_encode.argtypes = [ctypes.POINTER(ctypes.c_uint16), ctypes.c_uint, ctypes.POINTER(ctypes.c_uint8)] # Четыре wire-слова включают маску, а не четвёртый регистр.
|
||||
self.lib.pcan_gas_slot_encode.restype = ctypes.c_int # Кодек возвращает длину 0/8 либо отрицательный отказ.
|
||||
self.lib.pcan_gas_slot_decode.argtypes = [ctypes.c_uint32, ctypes.c_uint, ctypes.POINTER(ctypes.c_uint8), ctypes.c_uint, ctypes.POINTER(ctypes.c_uint), ctypes.POINTER(ctypes.c_uint), ctypes.POINTER(ctypes.c_uint16)] # PERIPH-декодер дополнительно проверяет принадлежность странице и последний слот.
|
||||
self.lib.pcan_gas_slot_decode.restype = ctypes.c_int # Результат 0/4 включает маску при наличии данных.
|
||||
self.lib.pcan_gas_slot_decode_words.argtypes = [ctypes.c_uint32, ctypes.c_uint, ctypes.POINTER(ctypes.c_uint8), ctypes.c_uint, ctypes.POINTER(ctypes.c_uint16)] # Общий декодер поддерживает остальные классы узлов и все 16 бит адреса.
|
||||
self.lib.pcan_gas_slot_decode_words.restype = ctypes.c_int # Отрицательное значение означает отказ без публикации частичного результата.
|
||||
|
||||
@staticmethod # Проверка представимости Python-чисел не дублирует wire-логику.
|
||||
def _unsigned(value, maximum): # Проверяем границы до возможного сужения ctypes.
|
||||
if type(value) is not int or not 0 <= value <= maximum: # bool, дробь и отрицательное число не являются полями протокола.
|
||||
raise ValueError('Недопустимое целое значение протокола') # Не отправляем усечённое значение к другому адресу.
|
||||
|
||||
def _payload(self, words, route): # Общая упаковка используется адресным и PERIPH-построителями.
|
||||
values = None if words is None else tuple(words) # Фиксируем входной iterable до преобразования к C-массиву.
|
||||
if values is not None: # Запись или ответ содержат ровно маску и три значения.
|
||||
if len(values) != 4: # Позиции не сдвигаются при частичной записи.
|
||||
raise ValueError('Нужны Word0-маска и три значения Word1…Word3') # Не дополняем отсутствующие позиции нулями неявно.
|
||||
for value in values: # Все wire-слова имеют одинаковую ширину 16 бит.
|
||||
self._unsigned(value, 0xFFFF) # Корректные ноль и 65535 остаются допустимыми значениями.
|
||||
if values is None and route != 0: # Пустой ответ прибора не имеет определённого смысла.
|
||||
raise ValueError('Ответ должен содержать маску и значения') # Отказ возникает до отправки в транспорт.
|
||||
source = None if values is None else (ctypes.c_uint16 * 4)(*values) # Маску и значения проверяет C до кодирования.
|
||||
data = (ctypes.c_uint8 * 8)() # Буфер ограничен одним classic CAN кадром.
|
||||
length = self.lib.pcan_gas_slot_encode(source, 0 if values is None else 4, data) # Единственный кодек проверяет разрешённые биты маски и порядок октетов.
|
||||
if length < 0: # Неверная маска не заменяется другой командой.
|
||||
raise ValueError('Неверная маска GAS SLOT: допустимы непустые биты 1…3') # Бит 0 относится к служебному слову и запрещён.
|
||||
return bytes(data[:length]) # Возвращаем только готовые wire-октеты.
|
||||
|
||||
def frame(self, mode, slot, words=None, route=0): # Удобная обёртка для 200 регистров каждого mode PERIPH.
|
||||
for value in (mode, slot, route): # Исключаем переполнение host ABI до вызова C.
|
||||
self._unsigned(value, 0xFFFFFFFF) # Семантические пределы mode и local проверяет общий построитель.
|
||||
can_id = self.lib.pcan_gas_slot_request_id(mode, slot, route) # C формирует MsgType=0x4 и HEX-страницу mode.
|
||||
if can_id == 0: # Не создаём адрес для резерва страницы или неверного направления.
|
||||
raise ValueError('Неверный mode, слот или направление') # Исходная ошибка не превращается в обращение к соседнему прибору.
|
||||
payload = self._payload(words, route) # Данные одинаковы для обоих адресных API.
|
||||
self.decode(can_id, payload) # Дополнительно проверяем маску двух действующих значений последнего слота PERIPH.
|
||||
return can_id, payload # Транспорт получает готовые CAN ID и данные.
|
||||
|
||||
def frame_address(self, device_type, device_id, address, words=None, route=0): # Полный слотный API для любого узла, включая STM Invertor.
|
||||
for value in (device_type, device_id, address, route): # Python-числа не должны оборачиваться в C unsigned.
|
||||
self._unsigned(value, 0xFFFFFFFF) # Конкретную ширину полей идентификатора проверяет C.
|
||||
can_id = self.lib.pcan_gas_slot_id(device_type, device_id, address, route) # Общий построитель сохраняет все адреса 0…65535.
|
||||
if can_id == 0: # Нулевой результат означает ошибку адреса узла или слота.
|
||||
raise ValueError('Неверный адрес узла, слота или направление') # Не скрываем ошибку входа маскированием.
|
||||
return can_id, self._payload(words, route) # Не привязываем общий формат к странице PERIPH.
|
||||
|
||||
def _input(self, can_id, data, flags): # Проверяем размеры и представимость до создания ctypes-буферов.
|
||||
self._unsigned(can_id, 0xFFFFFFFF) # C затем проверит фактические 29 бит CAN ID.
|
||||
self._unsigned(flags, 0xFFFFFFFF) # Флаги IDE/RTR/ERR проверяются общим wire-кодеком.
|
||||
payload = bytes(data) # Сохраняем точную длину без усечения.
|
||||
if len(payload) > 8: # CAN FD не включается неявно.
|
||||
raise ValueError('Classic CAN допускает до восьми октетов') # Лишние октеты остаются видимой ошибкой.
|
||||
return (ctypes.c_uint8 * len(payload)).from_buffer_copy(payload), len(payload) # Нулевой запрос не требует фиктивных данных.
|
||||
|
||||
def decode(self, can_id, data, flags=1): # Возвращаем mode, локальный слот и wire-слова PERIPH с маской.
|
||||
raw, length = self._input(can_id, data, flags) # Получаем проверенный ограниченный буфер.
|
||||
mode, slot, words = ctypes.c_uint(), ctypes.c_uint(), (ctypes.c_uint16 * 4)() # Отдельные выходы сохраняют ширину host ABI.
|
||||
count = self.lib.pcan_gas_slot_decode(can_id, flags, raw, length, ctypes.byref(mode), ctypes.byref(slot), words) # C проверяет полный формат и геометрию PERIPH.
|
||||
if count < 0: # Чужая страница или несуществующая позиция не становятся регистрами другого профиля.
|
||||
raise ValueError('Неверный кадр GAS SLOT PERIPH') # Никакого автоматического перехода к типу 0x3.
|
||||
return mode.value, slot.value, tuple(words[:count]) # Пустой кортеж означает запрос чтения.
|
||||
|
||||
def decode_address(self, can_id, data, flags=1): # Декодируем маску и три значения любого адресуемого узла.
|
||||
raw, length = self._input(can_id, data, flags) # Проверяем точную длину до native-вызова.
|
||||
words = (ctypes.c_uint16 * 4)() # Четыре wire-слова включают служебную маску.
|
||||
count = self.lib.pcan_gas_slot_decode_words(can_id, flags, raw, length, words) # Общий C-кодек проверяет тип 0x4 и все транспортные условия.
|
||||
if count < 0: # Ошибка не выдаётся за допустимый пустой слот.
|
||||
raise ValueError('Неверный кадр GAS SLOT') # Сообщение пригодно для CAN-журнала.
|
||||
return tuple(words[:count]) # Адрес и класс узла уже доступны вызывающему коду в CAN ID.
|
||||
@@ -0,0 +1,115 @@
|
||||
# Тонкий порт GAS_WATCH2: форматы запросов, подтверждений и выборок проверяет общее C-ядро.
|
||||
# Python хранит неизменяемые модели и состояние подтверждённой подписки без второго кодека.
|
||||
|
||||
from __future__ import annotations # Откладываем вычисление аннотаций для совместимости с Python 3.8.
|
||||
|
||||
import ctypes # Передаём типизированные буферы единственной реализации протокола в общей библиотеке.
|
||||
from dataclasses import dataclass # Неизменяемые модели фиксируют порядок адресов и значений одной конфигурации.
|
||||
|
||||
MAX_SIGNALS = 64 # Общий контракт GAS_WATCH2 ограничивает подписку 64 регистрами по 16 бит.
|
||||
MAX_PAYLOAD_BYTES = 512 # Верхняя граница payload GUI protocol v1 позволяет C выбрать фактическую длину запроса.
|
||||
|
||||
|
||||
def _unsigned(value: int, maximum: int, name: str) -> None: # Проверяем представимость Python-числа перед сужением через ctypes.
|
||||
if type(value) is not int or not 0 <= value <= maximum: # Исключаем bool, дроби и переполнение целевого беззнакового типа.
|
||||
raise ValueError(f"{name}: требуется целое число от 0 до {maximum}") # Не допускаем незаметного усечения C-аргумента.
|
||||
|
||||
|
||||
@dataclass(frozen=True) # Изменение списка после отправки не должно менять смысл ожидаемого подтверждения.
|
||||
class GasWatchConfig: # Описываем подписку отдельно от её визуального оформления и истории графика.
|
||||
config_id: int # Ненулевой 32-битный идентификатор связывает выборки с конкретным порядком адресов.
|
||||
period_ms: int # Беззнаковый 16-битный период задаётся в миллисекундах; ноль используется для отключения.
|
||||
addresses: tuple[int, ...] # Порядок 16-битных GAS-адресов совпадает с порядком значений в выборке.
|
||||
|
||||
def __post_init__(self) -> None: # Нормализуем коллекцию и защищаем границу Python/C от сужения чисел.
|
||||
_unsigned(self.config_id, 0xFFFFFFFF, "config_id") # Диапазон соответствует uint32_t в публичном C ABI.
|
||||
if self.config_id == 0: # Нулевой идентификатор зарезервирован для состояния без установленной подписки.
|
||||
raise ValueError("config_id должен быть ненулевым") # Не создаём модель, которая не может обозначать конфигурацию.
|
||||
_unsigned(self.period_ms, 0xFFFF, "period_ms") # Диапазон соответствует uint16_t; логическое сочетание с адресами проверяет C.
|
||||
addresses = tuple(self.addresses) # Отделяем модель от изменяемого списка, переданного вызывающим приложением.
|
||||
if len(addresses) > MAX_SIGNALS: # Не передаём C массив сверх согласованного предела подписки.
|
||||
raise ValueError("GAS_WATCH2 поддерживает не более 64 адресов") # Объясняем ограничение до выделения буфера ctypes.
|
||||
for address in addresses: # Проверяем каждый элемент до записи в массив uint16_t.
|
||||
_unsigned(address, 0xFFFF, "address") # GAS использует нулевые 16-битные индексы регистров, без расширенных адресов.
|
||||
object.__setattr__(self, "addresses", addresses) # Устанавливаем неизменяемую копию внутри frozen dataclass только при создании.
|
||||
|
||||
|
||||
@dataclass(frozen=True) # Полученная выборка остаётся стабильной при передаче между рабочим потоком и GUI.
|
||||
class GasWatchSample: # Возвращаем проверенную модель, а не поля бинарного сообщения.
|
||||
config_id: int # Идентификатор принадлежит уже подтверждённой конфигурации, проверенной C-декодером.
|
||||
sequence: int # 32-битный номер выборки позволяет приложению обнаруживать пропуски доставки.
|
||||
timestamp_ms: int # 32-битная метка устройства передаётся в миллисекундах без подмены временем GUI.
|
||||
values: tuple[int | None, ...] # None обозначает ошибку источника и отличается от корректного нулевого значения.
|
||||
|
||||
def __post_init__(self) -> None: # Не допускаем сохранения изменяемого списка в публичной frozen-модели.
|
||||
_unsigned(self.config_id, 0xFFFFFFFF, "config_id") # Сохраняем представимость идентификатора в C uint32_t.
|
||||
_unsigned(self.sequence, 0xFFFFFFFF, "sequence") # Переполнение счётчика обрабатывается устройством по модулю 2^32.
|
||||
_unsigned(self.timestamp_ms, 0xFFFFFFFF, "timestamp_ms") # Ограничиваем метку диапазоном миллисекундного uint32_t.
|
||||
values = tuple(self.values) # Копируем контейнер, чтобы внешний список не изменял уже принятую выборку.
|
||||
if len(values) > MAX_SIGNALS: # Размер результата должен помещаться в общий предел числа каналов.
|
||||
raise ValueError("GAS_WATCH2 поддерживает не более 64 значений") # Отвергаем публичную модель с недопустимой геометрией.
|
||||
for value in values: # Проверяем только модель данных, не выполняя повторного разбора wire-пакета.
|
||||
if value is not None: # Недоступный источник представляется None и не требует числового преобразования.
|
||||
_unsigned(value, 0xFFFF, "value") # Достоверное значение регистра сохраняется как беззнаковое 16-битное слово.
|
||||
object.__setattr__(self, "values", values) # Закрепляем независимый кортеж в неизменяемом экземпляре.
|
||||
|
||||
|
||||
class NativeGasWatch: # Управляем ожидаемой и активной подписками поверх единственного C-кодека GAS_WATCH2.
|
||||
def __init__(self, library: ctypes.CDLL) -> None: # Принимаем уже загруженную общую библиотеку без поиска или fallback.
|
||||
self.lib = library # Сохраняем владельца C-функций на весь срок жизни адаптера.
|
||||
try: # Проверяем наличие всей новой возможности до настройки сигнатур ABI.
|
||||
self._request = library.set_gas_watch_request # C формирует запрос и проверяет допустимость полной конфигурации.
|
||||
self._ack = library.set_gas_watch_ack # C сравнивает подтверждение со всем запрошенным списком адресов.
|
||||
self._decode = library.set_gas_watch_decode # C проверяет версию, идентификатор, геометрию и маску достоверности.
|
||||
except AttributeError as error: # Старая DLL может оставаться пригодной для других возможностей протокола.
|
||||
raise RuntimeError("Пересоберите SETProtocol с set_gas_watch.c для поддержки GAS_WATCH2") from error # Явно сообщаем о необходимой сборке вместо Python-кодека.
|
||||
self._request.argtypes = [ctypes.c_uint32, ctypes.c_uint16, ctypes.POINTER(ctypes.c_uint16), ctypes.c_size_t, ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t] # Передаём id, период, адреса и явные размеры входа/выхода по C ABI.
|
||||
self._request.restype = ctypes.c_size_t # Ноль означает отказ C-валидатора, положительное значение — длину payload.
|
||||
self._ack.argtypes = [ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t, ctypes.c_uint32, ctypes.c_uint16, ctypes.POINTER(ctypes.c_uint16), ctypes.c_size_t] # Сопоставляем байты ACK с полным содержимым ожидаемой конфигурации.
|
||||
self._ack.restype = ctypes.c_int # Ненулевой результат подтверждает точное принятие подписки устройством.
|
||||
self._decode.argtypes = [ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t, ctypes.c_uint32, ctypes.c_size_t, ctypes.POINTER(ctypes.c_uint32), ctypes.POINTER(ctypes.c_uint32), ctypes.POINTER(ctypes.c_uint16), ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t] # C получает ожидаемую конфигурацию и отдельные буферы значений и достоверности.
|
||||
self._decode.restype = ctypes.c_int # Отрицательный результат означает устаревшее либо повреждённое сообщение.
|
||||
self.reset() # Новый экземпляр не принимает DATA до подтверждения своей первой подписки.
|
||||
|
||||
def reset(self) -> None: # Сбрасываем состояние соединения при переподключении без отправки wire-команд.
|
||||
self._pending: GasWatchConfig | None = None # Позднее подтверждение предыдущего соединения больше не ожидается.
|
||||
self._active: GasWatchConfig | None = None # Ранее подтверждённые DATA нельзя привязать к новому соединению автоматически.
|
||||
|
||||
def request(self, config: GasWatchConfig) -> bytes: # Подготавливаем запрос, оставляя старую активную подписку до нового ACK.
|
||||
if not isinstance(config, GasWatchConfig): # Тонкий порт требует заранее нормализованную модель вместо произвольного объекта.
|
||||
raise TypeError("Ожидается GasWatchConfig") # Не допускаем обхода проверок представимости аргументов ctypes.
|
||||
for known in (self._active, self._pending): # Сравниваем идентификатор с ещё значимыми конфигурациями текущего соединения.
|
||||
if known is not None and known.config_id == config.config_id and known != config: # Один id не может обозначать другой порядок адресов или период.
|
||||
raise ValueError("Новая конфигурация требует нового config_id") # Иначе C не сможет отличить старые DATA с тем же id и количеством.
|
||||
addresses = (ctypes.c_uint16 * len(config.addresses))(*config.addresses) # Передаём C непрерывный массив слов без собственного кодирования endian.
|
||||
output = (ctypes.c_uint8 * MAX_PAYLOAD_BYTES)() # Резервируем транспортный предел; реальную длину определяет C-builder.
|
||||
size = self._request(config.config_id, config.period_ms, addresses, len(addresses), output, len(output)) # Единственный C-валидатор проверяет дубликаты и сочетание периода с количеством.
|
||||
if not size: # При ошибке запроса сохраняем предыдущие ожидаемую и активную подписки.
|
||||
raise ValueError("Недопустимая конфигурация GAS_WATCH2") # Не публикуем частично сформированный payload.
|
||||
self._pending = config # После успешной сборки ожидаем ACK именно последнего отправляемого запроса.
|
||||
return bytes(output[:size]) # Копируем готовые байты C без разбора или пересборки полей в Python.
|
||||
|
||||
def validate_ack(self, payload: bytes, config: GasWatchConfig) -> None: # Активируем подписку только после полного подтверждения её содержимого.
|
||||
if self._pending is None or config != self._pending: # Поздний ACK предыдущего запроса не должен отменять более новую ожидаемую конфигурацию.
|
||||
raise ValueError("Подтверждение не соответствует ожидаемой конфигурации GAS_WATCH2") # Сохраняем активную подписку при гонке ответов.
|
||||
data = (ctypes.c_uint8 * len(payload)).from_buffer_copy(payload) # Предоставляем C входные байты без знания их внутреннего расположения.
|
||||
addresses = (ctypes.c_uint16 * len(config.addresses))(*config.addresses) # Сохраняем исходный порядок для строгой проверки эха в C.
|
||||
if not self._ack(data, len(data), config.config_id, config.period_ms, addresses, len(addresses)): # Требуем совпадения всей конфигурации, а не одного количества каналов.
|
||||
raise ValueError("Устройство не подтвердило полную конфигурацию GAS_WATCH2") # Повреждённый или частичный ACK не изменяет состояния подписок.
|
||||
self._active = config # Только подтверждённая конфигурация определяет допустимые DATA и порядок возвращаемых значений.
|
||||
self._pending = None # Повторный либо поздний ACK больше не считается ответом на ожидаемую команду.
|
||||
|
||||
def sample(self, payload: bytes) -> GasWatchSample: # Преобразуем проверенную C выборку в неизменяемую модель языка GUI.
|
||||
config = self._active # Один вызов использует один согласованный снимок ссылки на подтверждённую конфигурацию.
|
||||
if config is None: # До первого ACK и после reset принадлежность значений неизвестна.
|
||||
raise RuntimeError("Сначала подтвердите конфигурацию GAS_WATCH2") # Не принимаем неподтверждённые или оставшиеся от прошлого соединения DATA.
|
||||
data = (ctypes.c_uint8 * len(payload)).from_buffer_copy(payload) # Буфер ctypes передаёт готовый пакет C-декодеру без Python-парсинга.
|
||||
sequence = ctypes.c_uint32() # C запишет номер выборки только после полной проверки пакета.
|
||||
timestamp = ctypes.c_uint32() # C вернёт исходную миллисекундную метку устройства.
|
||||
values = (ctypes.c_uint16 * MAX_SIGNALS)() # Буфер рассчитан на максимальные 64 регистра, согласованные с C-контрактом.
|
||||
valid = (ctypes.c_uint8 * MAX_SIGNALS)() # C разворачивает wire-маску в отдельный признак достоверности каждого значения.
|
||||
count = self._decode(data, len(data), config.config_id, len(config.addresses), ctypes.byref(sequence), ctypes.byref(timestamp), values, valid, MAX_SIGNALS) # Проверку id, длины и маски полностью выполняет общее ядро.
|
||||
if count < 0: # Устаревшие и повреждённые DATA не меняют конфигурацию и не попадают на график.
|
||||
raise ValueError("Устаревшие или повреждённые данные GAS_WATCH2") # Не возвращаем нулевые значения из неуспешного C-вызова.
|
||||
words = tuple(values[index] if valid[index] else None for index in range(count)) # Преобразуем результат C в модель, сохраняя различие отсутствующего значения и нуля.
|
||||
return GasWatchSample(config.config_id, sequence.value, timestamp.value, words) # Закрепляем идентификатор именно той конфигурации, которую подтвердил C-декодер.
|
||||
@@ -46,7 +46,7 @@ class MsgType(IntEnum):
|
||||
DISCRETE = 0b0001
|
||||
ANALOG = 0b0010
|
||||
GENERAL_ADDRESS_SPACE = 0b0011
|
||||
MODBUS_COIL = 0b0100
|
||||
GAS_SLOT = 0b0100 # Тип 0x4 однозначно выбирает слотный GAS с маской вместо прежнего Modbus Coil.
|
||||
MODBUS_DISCRETE = 0b0101
|
||||
MODBUS_HOLDING = 0b0110
|
||||
MODBUS_INPUT = 0b0111
|
||||
@@ -115,7 +115,7 @@ MSGTYPE_RU = {
|
||||
MsgType.DISCRETE: "Discrete",
|
||||
MsgType.ANALOG: "Analog",
|
||||
MsgType.GENERAL_ADDRESS_SPACE: "General Address Space",
|
||||
MsgType.MODBUS_COIL: "Modbus Coil",
|
||||
MsgType.GAS_SLOT: "GAS Slot", # Отдельное имя показывает, что адрес в Body относится к слоту, а не к Modbus-регистру.
|
||||
MsgType.MODBUS_DISCRETE: "Modbus Discrete",
|
||||
MsgType.MODBUS_HOLDING: "Modbus Holding",
|
||||
MsgType.MODBUS_INPUT: "Modbus Input",
|
||||
@@ -134,14 +134,26 @@ MSGTYPE_RU = {
|
||||
#: но не присваивает приборам имён.
|
||||
DEVICE_TYPE_NAMES = {
|
||||
0x0: "Верхний уровень",
|
||||
0x1: "PERIPH 0–15", # Excel «Интерфейс протокола», C6/F6: первый тип содержит младшие 16 номеров периферии.
|
||||
0x2: "PERIPH 16–31", # Excel C7/F7: второй тип продолжает нумерацию, сохраняя четырёхбитный Device ID.
|
||||
0x6: "Логические анализаторы",
|
||||
}
|
||||
|
||||
DEVICE_ADDRESS_NAMES = {
|
||||
(0x1, 0x5): "УМП1", # Excel B50/C50/F50 закрепляет отдельный адрес первого модуля нагрузки.
|
||||
(0x1, 0x6): "УМП2", # Excel B51/C51/F51 отделяет второй модуль от первого и от задатчика.
|
||||
(0x1, 0x8): "Задатчик", # Excel B53/C53/F53 задаёт адрес задатчика без сдвига старого CAN Mode на единицу.
|
||||
(0x6, 0xE): "Altera Logic",
|
||||
(0x7, 0xD): "configurator",
|
||||
(0x7, 0xE): "STM Invertor", # Отдельный ПЧ STM32G474VET6: адрес 7/E проверен по портам Templates; 7/D и 7/F заняты.
|
||||
}
|
||||
|
||||
|
||||
def periph_address(periph_id: int) -> tuple[int, int]: # Возвращает адрес ProtoCAN для номера из 32 строк обновлённой таблицы PERIPH.
|
||||
if not 0 <= periph_id < 32: # Число 32 задаёт полный диапазон двух типов по 16 экземпляров; за его пределами адрес не назначается.
|
||||
raise ValueError("ID PERIPH должен быть в диапазоне 0…31") # Запрещает незаметное оборачивание ошибочного номера на другую плату.
|
||||
return 1 + periph_id // 16, periph_id % 16 # Старшая группа выбирает Type 1/2, младшие четыре бита — Device ID 0…15.
|
||||
|
||||
SETTINGS_RESULT_RU = {
|
||||
SettingsResult.OK: "успех",
|
||||
SettingsResult.INVALID_DLC: "неверный DLC",
|
||||
@@ -408,9 +420,10 @@ def _ascii(data: bytes) -> str:
|
||||
return "".join(chr(b) if 32 <= b < 127 else "." for b in data)
|
||||
|
||||
|
||||
def decode(raw_id: int, data: bytes, ide: bool = True, rtr: bool = False) -> Decoded:
|
||||
def decode(raw_id: int, data: bytes, ide: bool = True, rtr: bool = False) -> Decoded: # Тип 0x3 всегда адресует регистры, тип 0x4 всегда адресует слоты; настройка профиля не требуется.
|
||||
"""Разбирает кадр ProtoCAN в структуру Decoded."""
|
||||
pid = ProtoCanId.parse(raw_id)
|
||||
source_data = bytes(data) # Сохраняем фактическую длину кадра до отображения ограниченного classic CAN payload.
|
||||
data = bytes(data[:8])
|
||||
res = Decoded(id=pid, dlc=len(data), data=data, ide=ide, rtr=rtr)
|
||||
|
||||
@@ -431,12 +444,27 @@ def decode(raw_id: int, data: bytes, ide: bool = True, rtr: bool = False) -> Dec
|
||||
if rtr:
|
||||
res.warnings.append("Remote frame: полезная нагрузка отсутствует")
|
||||
|
||||
if pid.msg_type == MsgType.GAS_SLOT: # Слотный кодек вызывается только для своего типа 0x4, без угадывания по содержимому.
|
||||
from .gas_slots import GasSlots # Тонкая модель вызывает единственный C-кодек формата и проверок.
|
||||
try: # Ошибочный слот остаётся видимым в журнале, не превращаясь в регистровый кадр.
|
||||
values = GasSlots().decode_address(raw_id, source_data, (1 if ide else 0) | (2 if rtr else 0)) # Общий C-кодек проверяет маску и точную длину слота любого узла, включая отдельный инвертор.
|
||||
except ValueError as error: # Отказ протокола отображаем как ошибку кадра, а не подменяем legacy-декодированием.
|
||||
res.summary = "GAS SLOT: неверный кадр" # Короткий текст подходит для строки CAN-лога.
|
||||
res.warnings.append(str(error)) # Подробности доступны в панели выбранного сообщения.
|
||||
return res # Отвергнутый слот не получает вымышленных адресов регистров.
|
||||
res.fields.append(("GAS SLOT", "MsgType 0x4")) # Отдельный номер сообщения определяет формат без настройки профиля соединения.
|
||||
if values: # Запрос чтения не содержит маски или значений.
|
||||
res.fields.append(("Word0: маска", "0x%04X" % values[0])) # Нулевая позиция отображается как служебная маска, а не как регистр.
|
||||
for index in range(1, 4): # Три значения имеют фиксированные позиции Word1…Word3 в восьмибайтовом кадре.
|
||||
if values[0] & (1 << index): # Невыбранные значения не отображаются как записанные или достоверные измерения.
|
||||
res.fields.append(("Слот 0x%04X, Word%d" % (pid.body, index), "0x%04X (%d)" % (values[index], values[index]))) # Полная координата сохраняет адрес слота и выбранную позицию.
|
||||
res.summary = "GAS SLOT slot=0x%04X%s" % (pid.body, " mask=0x%04X" % values[0] if values else " (запрос)") # CAN-журнал явно показывает маску частичного изменения.
|
||||
return res # Не передаём четыре слова старому обработчику адреса первого регистра.
|
||||
handler = {
|
||||
MsgType.BROADCAST: _decode_broadcast,
|
||||
MsgType.DISCRETE: _decode_discrete,
|
||||
MsgType.ANALOG: _decode_analog,
|
||||
MsgType.GENERAL_ADDRESS_SPACE: _decode_gas,
|
||||
MsgType.MODBUS_COIL: _decode_modbus_bit,
|
||||
MsgType.MODBUS_DISCRETE: _decode_modbus_bit,
|
||||
MsgType.MODBUS_HOLDING: _decode_modbus_reg,
|
||||
MsgType.MODBUS_INPUT: _decode_modbus_reg,
|
||||
@@ -611,6 +639,12 @@ def _decode_gas(res: Decoded) -> None:
|
||||
res.fields.append((" Адрес первого регистра", "0x%04X (%d)" % (start, start)))
|
||||
if res.dlc % 2:
|
||||
res.warnings.append("DLC = %d нечётный: регистры GAS передаются парами байт" % res.dlc)
|
||||
res.summary = "GAS: неверная длина данных" # Усечённое слово не должно становиться достоверным значением или уставкой в GUI.
|
||||
return # Не публикуем частичный список регистров повреждённого кадра.
|
||||
if start + res.dlc // 2 > 0x10000: # Адрес FFFF допустим для одного слова, следующий адрес уже выходит за 16-битное пространство.
|
||||
res.warnings.append("Диапазон GAS выходит за адрес 0xFFFF") # Не создаём вымышленные адреса 0x10000 и выше.
|
||||
res.summary = "GAS: неверный диапазон регистров" # Ошибка остаётся видимой в CAN-журнале.
|
||||
return # Не передаём некорректные значения потребителям трендов или карты данных.
|
||||
vals = regs_le(res.data)
|
||||
if len(vals) > 4:
|
||||
res.warnings.append("В одном кадре GAS не более 4 регистров, получено %d" % len(vals))
|
||||
@@ -631,7 +665,7 @@ def _decode_gas(res: Decoded) -> None:
|
||||
|
||||
def _decode_modbus_bit(res: Decoded) -> None:
|
||||
str_adr, reg_count = split_modbus(res.id.body)
|
||||
kind = "COIL" if res.id.msg_type == MsgType.MODBUS_COIL else "DISCRETE"
|
||||
kind = "DISCRETE" # После назначения 0x4 слотному GAS битовым Modbus-сообщением остаётся только тип 0x5.
|
||||
res.fields.append((" Modbus.StrAdr", "0x%03X (%d)" % (str_adr, str_adr)))
|
||||
res.fields.append((" Modbus.RegCount", "%d" % reg_count))
|
||||
expected = reg_count % 8 + 1
|
||||
|
||||
@@ -51,6 +51,8 @@ class MessageType(IntEnum):
|
||||
GAS_CATALOG = 0x11
|
||||
GAS_WATCH_SET = 0x12
|
||||
GAS_WATCH_DATA = 0x13
|
||||
GAS_WATCH2_SET = 0x14 # Новый тип подписки передаёт идентификатор и полный упорядоченный список GAS-регистров, сохраняя старый 0x12.
|
||||
GAS_WATCH2_DATA = 0x15 # Новый тип выборки связывает значения с конфигурацией и признаками достоверности, сохраняя старый 0x13.
|
||||
SENSOR_SCAN = 0x20
|
||||
SENSOR_LIST = 0x21
|
||||
SENSOR_READ = 0x22
|
||||
|
||||
@@ -128,10 +128,12 @@ class CanFrameFields(QWidget): # Компонент содержит тольк
|
||||
else: # Типы без отдельного подтипа используют Body по собственному назначению.
|
||||
sub_label = "—" # Подпись явно показывает отсутствие подтипа.
|
||||
arg_label = "Body (обычно 0)" # Для PULSE сохраняем исходное описание аргумента.
|
||||
if msg_type in (pc.MsgType.MODBUS_COIL, pc.MsgType.MODBUS_DISCRETE, pc.MsgType.MODBUS_HOLDING, pc.MsgType.MODBUS_INPUT): # Все четыре вида Modbus задают начальный адрес.
|
||||
if msg_type in (pc.MsgType.MODBUS_DISCRETE, pc.MsgType.MODBUS_HOLDING, pc.MsgType.MODBUS_INPUT): # Только сохранённые типы 0x5…0x7 используют Modbus-адрес и количество.
|
||||
arg_label = "StrAdr (0..4095)" # Двенадцатибитный адрес показан в прежних единицах регистров.
|
||||
elif msg_type == pc.MsgType.GENERAL_ADDRESS_SPACE: # GAS использует всё поле Body под адрес.
|
||||
arg_label = "Адрес регистра (0..65535)" # Полный 16-битный диапазон указан в подписи.
|
||||
elif msg_type == pc.MsgType.GAS_SLOT: # Новый тип 0x4 использует весь Body под номер слота, а не StrAdr/RegCount.
|
||||
arg_label = "Адрес слота (0..65535), Data: маска + 3 слова" # Напоминаем оператору о Word0 с маской и фиксированных позициях трёх значений.
|
||||
elif msg_type == pc.MsgType.ERROR: # Код ошибки вводится в отдельный байт Body.
|
||||
arg_label = "ErrorCode (0..255)" # Показываем допустимый однобайтовый диапазон.
|
||||
elif settings: # SETTINGS редактируется двухбайтовым компонентом.
|
||||
@@ -146,7 +148,7 @@ class CanFrameFields(QWidget): # Компонент содержит тольк
|
||||
return pc.merge_discrete(pc.DiscreteType[self.sub.currentText()], arg) # Поле собирается общим протокольным модулем.
|
||||
if msg_type == pc.MsgType.ANALOG: # Аналоговое сообщение передаёт SensorID в Body.
|
||||
return pc.merge_analog(pc.AnalogType[self.sub.currentText()], arg) # Существующий кодек задаёт расположение подтипа и SensorID.
|
||||
if msg_type in (pc.MsgType.MODBUS_COIL, pc.MsgType.MODBUS_DISCRETE, pc.MsgType.MODBUS_HOLDING, pc.MsgType.MODBUS_INPUT): # У четырёх видов Modbus одинаковый формат Body.
|
||||
if msg_type in (pc.MsgType.MODBUS_DISCRETE, pc.MsgType.MODBUS_HOLDING, pc.MsgType.MODBUS_INPUT): # Формат Modbus применяется только к трём оставшимся типам 0x5…0x7.
|
||||
return pc.merge_modbus(arg, count) # Адрес и количество объединяются общей функцией протокола.
|
||||
if msg_type == pc.MsgType.ERROR: # Сохраняем существующую категорию ошибки с нулевым кодом источника.
|
||||
return pc.merge_error(0, arg) # Размещение кода ошибки остаётся ответственностью общего ядра.
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
import json # Общие машинные эталоны независимы от клиентского построителя кадра.
|
||||
from pathlib import Path # Находим спецификацию относительно репозитория.
|
||||
import unittest # Проверяем реальные ctypes-привязки без заглушки протокольного ядра.
|
||||
from protocan.gas_slots import GasSlots # Все правила wire-формата выполняет C, а не тестовая модель.
|
||||
from protocan import protocan as pc # Проверяем штатный CAN-декодер, который вызывают GUI и журналы.
|
||||
|
||||
class GasSlotTests(unittest.TestCase): # Интеграционные проверки обоих типов GAS и границ адресного API.
|
||||
def setUp(self): # Каждая проверка использует настоящую собранную shared-библиотеку.
|
||||
self.codec = GasSlots() # Отсутствующая библиотека не заменяется Python fallback.
|
||||
|
||||
def test_vectors(self): # Проверяем ID, маску и три значения по нормативным независимым байтам.
|
||||
path = Path(__file__).resolve().parents[2] / 'doc/setcan/protocan/examples/gas-slot-vectors.json' # Те же примеры опубликованы в документации.
|
||||
vectors = json.loads(path.read_text(encoding='utf-8')) # UTF-8 сохраняет названия и описание профиля.
|
||||
for item in vectors['vectors']: # Охватываем обе группы адресов и неполный последний слот PERIPH.
|
||||
words = item.get('words', vectors['words']) # Последний слот имеет только два действующих значения.
|
||||
expected = item.get('payload_hex', vectors['payload_hex']) # Его отдельный эталон не включает несуществующее третье значение.
|
||||
can_id, data = self.codec.frame(item['mode'], item['local_slot']) # Пустой запрос не несёт служебной маски.
|
||||
self.assertEqual((can_id, data), (int(item['request_id'], 16), b'')) # Тип 0x4 и адрес должны совпасть с фиксированным эталоном.
|
||||
reply, payload = self.codec.frame(item['mode'], item['local_slot'], words, route=1) # Ответ явно содержит маску достоверности.
|
||||
self.assertEqual(reply, int(item['response_id'], 16)) # Направление не меняет узел или слот.
|
||||
self.assertEqual(payload.hex().upper(), expected) # Проверяем все восемь октетов, включая расположение Word0.
|
||||
self.assertEqual(self.codec.decode(reply, payload), (item['mode'], item['local_slot'], tuple(words))) # Полный round-trip сохраняет маску отдельно от трёх значений.
|
||||
|
||||
def test_invalid(self): # Неверные координаты, маски и длины не должны стать другой командой.
|
||||
for mode, slot in ((32, 0), (5, 67), (-1, 0), (2**32, 0)): # Верхняя граница страницы после маски — LocalSlot 66.
|
||||
with self.assertRaises(ValueError): self.codec.frame(mode, slot) # Ошибка не маскируется до допустимого адреса.
|
||||
for payload in (b'\0', b'\0' * 6, b'\0' * 9, b'\0' * 8): # В том числе запрещена полная запись с пустой маской.
|
||||
with self.assertRaises(ValueError): self.codec.decode(0x19540500, payload) # Точный размер и Word0 проверяет C.
|
||||
for mask in (0, 1, 0x10, 0xFFFF): # Бит 0 и старшие биты не адресуют данные слота.
|
||||
with self.assertRaises(ValueError): self.codec.frame(5, 0, [mask, 1, 2, 3]) # Не заменяем отвергнутую маску полной записью.
|
||||
with self.assertRaises(ValueError): self.codec.frame(5, 66, [14, 1, 2, 3]) # Третье значение неполного слота не существует.
|
||||
with self.assertRaises(ValueError): self.codec.frame(5, 0, [2, 65536, 0, 0]) # Переполнение u16 не превращается в допустимый ноль.
|
||||
can_id, payload = self.codec.frame(5, 0, [4, 0, 0x1234, 0], route=1) # Корректный частичный ответ для проверки транспортных флагов.
|
||||
with self.assertRaises(ValueError): self.codec.decode(can_id, payload, flags=3) # RTR отвергается до обращения к provider.
|
||||
|
||||
def test_types_are_unambiguous(self): # Один и тот же payload получает разный смысл только по номеру типа.
|
||||
can_id, payload = self.codec.frame(5, 0, [4, 0, 0x1234, 0], route=1) # Word2 выбран, Word1/Word3 передаются в фиксированных позициях и игнорируются.
|
||||
slot = pc.decode(can_id, payload) # Никакого параметра gas_profile для типа 0x4 больше нет.
|
||||
self.assertIn('slot=0x0500', slot.summary) # Адрес показывается в единицах слота.
|
||||
self.assertIn('mask=0x0004', slot.summary) # В CAN-журнале видна частичная запись.
|
||||
self.assertIsNone(slot.registers) # Старые потребители не получают фиктивный последовательный список регистров.
|
||||
self.assertTrue(any('Word2' in name for name, value in slot.fields)) # Только выбранная позиция отображается как достоверное значение.
|
||||
self.assertFalse(any('Word1' in name or 'Word3' in name for name, value in slot.fields)) # Неотмеченные поля не публикуются как измерения.
|
||||
register = pc.decode(0x19530500, payload) # Тип 0x3 всегда интерпретируется как обычные последовательные слова.
|
||||
self.assertEqual(register.registers, [(0x0500, 4), (0x0501, 0), (0x0502, 0x1234), (0x0503, 0)]) # Нулевое слово payload не является маской регистрового GAS.
|
||||
self.assertTrue(pc.decode(can_id, payload + b'\0').warnings) # Девятый октет не теряется до проверки слотного формата.
|
||||
overflow = pc.decode(0x1953FFFF, b'\x01\x00\x02\x00') # По максимальному адресу нельзя разместить два последовательных регистра.
|
||||
self.assertTrue(overflow.warnings) # Ошибка диапазона должна быть видна в журнале.
|
||||
self.assertIsNone(overflow.registers) # Некорректный адрес 0x10000 не становится источником тренда.
|
||||
|
||||
def test_generic_address_and_analyzer(self): # Тип 0x4 должен работать также за пределами PERIPH и его HEX-страниц.
|
||||
can_id, payload = self.codec.frame_address(7, 14, 0xFFFF, [2, 0x5678, 0xDEAD, 0], route=1) # Неотмеченное значение сохраняется как обычный заполнитель и не является командой записи.
|
||||
self.assertEqual(can_id, 0x1FE4FFFF) # Полный адрес FFFF принадлежит отдельному инвертору 7/E.
|
||||
self.assertEqual(self.codec.decode_address(can_id, payload), (2, 0x5678, 0xDEAD, 0)) # Общий декодер проверяет маску, не навязывая страницу mode.
|
||||
self.assertIn('slot=0xFFFF', pc.decode(can_id, payload).summary) # Штатный GUI-декодер использует общий адресный API.
|
||||
with self.assertRaises(ValueError): self.codec.frame_address(7, 14, 0x10000) # Не допускаем оборачивания в слот ноль.
|
||||
from logic_analyzer.decoders.set_can import legacy # Проверяем тот же публичный путь анализа аппаратного захвата.
|
||||
decoded = legacy(can_id, payload) # Анализатор вызывает общий C-кодек слотного формата.
|
||||
self.assertEqual(decoded['slot_words'], [(1, 0x5678)]) # Служебная маска и невыбранные слова не становятся данными анализатора.
|
||||
self.assertNotIn('registers', decoded) # Не смешиваем координаты слота с плоским регистровым адресом.
|
||||
|
||||
if __name__ == '__main__': # Поддерживаем прямой запуск без pytest.
|
||||
unittest.main() # Любой отказ native-кодека отражается стандартным статусом тестов.
|
||||
@@ -0,0 +1,169 @@
|
||||
# Проверяем неизменяемые модели и настоящий C ABI GAS_WATCH2 на общих эталонных байтах.
|
||||
# Python не подменяет C-кодек: проверки wire-пакетов выполняются только с указанной SETPROTOCOL_LIBRARY.
|
||||
|
||||
import ctypes # Загружаем настоящую общую библиотеку для проверки границы Python/C.
|
||||
import json # Читаем общий JSON-fixture, используемый также C- и JVM-проверками.
|
||||
import os # Путь к проверяемой сборке получает тестовый процесс через SETPROTOCOL_LIBRARY.
|
||||
import unittest # Запускаем воспроизводимые проверки модели, подписок и ошибочных данных.
|
||||
from dataclasses import FrozenInstanceError # Проверяем запрет изменения уже созданных моделей.
|
||||
from pathlib import Path # Находим эталон относительно исходников независимо от текущего каталога.
|
||||
|
||||
from protocan.gas_watch import GasWatchConfig, GasWatchSample, NativeGasWatch # Проверяем публичный тонкий порт без доступа к другому Python-кодеку.
|
||||
from protocan.protocol import MessageType # Проверяем выделение новых типов сообщений без изменения старых значений.
|
||||
|
||||
FIXTURE_PATH = Path(__file__).resolve().parents[2] / "c/set-protocol/tests/fixtures/gas-watch-v2.json" # Общие байты находятся рядом с C-тестами как единый контракт портов.
|
||||
|
||||
|
||||
class GasWatchModelTests(unittest.TestCase): # Эти проверки не требуют DLL, поскольку относятся к представлению моделей Python.
|
||||
def test_models_copy_sequences_and_are_frozen(self): # Изменение исходных списков не должно менять ожидаемый ACK и принятую выборку.
|
||||
addresses = [0, 1, 0xFF01] # Повторяем три разрозненных адреса из пользовательского примера.
|
||||
config = GasWatchConfig(1, 100, addresses) # Переданный список должен быть независимо сохранён как кортеж.
|
||||
addresses[0] = 99 # Моделируем последующее редактирование таблицы сигналов GUI.
|
||||
self.assertEqual((0, 1, 0xFF01), config.addresses) # Проверяем сохранение исходного порядка подписки.
|
||||
with self.assertRaises(FrozenInstanceError): # Идентификатор уже сформированной модели нельзя менять на месте.
|
||||
config.config_id = 2 # Присваивание проверяет защиту frozen dataclass, а не изменение состояния устройства.
|
||||
values = [0, None, 65535] # Корректный ноль и недостоверное значение должны оставаться различимыми.
|
||||
sample = GasWatchSample(1, 2, 3, values) # Публичная модель также должна копировать изменяемый контейнер.
|
||||
values[0] = 5 # Моделируем повторное использование входного списка вызывающим приложением.
|
||||
self.assertEqual((0, None, 65535), sample.values) # Ранее созданная выборка не должна изменяться вслед за списком.
|
||||
with self.assertRaises(FrozenInstanceError): # Метка времени проверенной выборки остаётся неизменяемой.
|
||||
sample.timestamp_ms = 4 # Присваивание должно быть отклонено механизмом dataclass.
|
||||
|
||||
def test_integer_ranges_prevent_ctypes_narrowing(self): # Python обязан отвергать значения, которые ctypes незаметно усекло бы до C-типа.
|
||||
for config_id in (0, -1, 0x100000000, True, 1.5): # Проверяем нулевой, отрицательный, переполненный и нецелые идентификаторы.
|
||||
with self.subTest(config_id=config_id), self.assertRaises(ValueError): # Каждый ошибочный id должен отказать до вызова C.
|
||||
GasWatchConfig(config_id, 100, (0,)) # Остальные поля допустимы и не маскируют проверяемую ошибку.
|
||||
for period in (-1, 65536, True, 1.5): # Период представляет uint16_t в миллисекундах без преобразования bool и float.
|
||||
with self.subTest(period=period), self.assertRaises(ValueError): # Проверяем все источники возможного сужения периода.
|
||||
GasWatchConfig(1, period, (0,)) # Валидный идентификатор и адрес изолируют неверный период.
|
||||
for address in (-1, 65536, True, 1.5): # GAS-регистр задаётся uint16_t, а не произвольным Python-числом.
|
||||
with self.subTest(address=address), self.assertRaises(ValueError): # Любой недопустимый элемент массива должен отказать.
|
||||
GasWatchConfig(1, 100, (address,)) # Один адрес позволяет проверить границу независимо от числа каналов.
|
||||
with self.assertRaises(ValueError): # Подписка не должна превышать общий предел 64 каналов.
|
||||
GasWatchConfig(1, 100, tuple(range(65))) # 65 уникальных допустимых адресов изолируют ограничение размера.
|
||||
with self.assertRaises(ValueError): # Значения публичной модели также не должны выходить за 16 бит.
|
||||
GasWatchSample(1, 2, 3, (65536,)) # Отдельная ошибочная величина проверяет защиту модели результата.
|
||||
|
||||
def test_old_library_has_explicit_rebuild_error(self): # Устаревшая библиотека должна дать понятную ошибку без Python fallback.
|
||||
with self.assertRaisesRegex(RuntimeError, "set_gas_watch.c"): # Сообщение должно назвать недостающую реализацию для пересборки.
|
||||
NativeGasWatch(object()) # Объект без новых ABI-символов воспроизводит отсутствие функции в старой DLL.
|
||||
|
||||
def test_message_types_preserve_legacy_values(self): # Новые wire-типы не должны переопределять существующий GAS_WATCH.
|
||||
self.assertEqual((0x12, 0x13), (MessageType.GAS_WATCH_SET, MessageType.GAS_WATCH_DATA)) # Сохраняем совместимость старого профиля v1.
|
||||
self.assertEqual((0x14, 0x15), (MessageType.GAS_WATCH2_SET, MessageType.GAS_WATCH2_DATA)) # Выделенные типы однозначно выбирают новый формат.
|
||||
|
||||
|
||||
@unittest.skipUnless(os.environ.get("SETPROTOCOL_LIBRARY"), "Укажите SETPROTOCOL_LIBRARY для проверки настоящего C-ядра") # Без явного пути не подменяем C тестовым Python-кодеком.
|
||||
class NativeGasWatchTests(unittest.TestCase): # Проверяем реальные builder, ACK-валидатор и DATA-декодер общей библиотеки.
|
||||
@classmethod # Одну DLL и общий fixture безопасно переиспользуют отдельные экземпляры адаптера каждого теста.
|
||||
def setUpClass(cls): # Загружаем внешний контракт один раз до тестов C ABI.
|
||||
cls.library = ctypes.CDLL(os.environ["SETPROTOCOL_LIBRARY"]) # Указанный путь должен вести к сборке с set_gas_watch.c.
|
||||
cls.fixture = json.loads(FIXTURE_PATH.read_text(encoding="utf-8")) # Эталонные данные читаются в обязательной кодировке UTF-8.
|
||||
cls.request_bytes = bytes.fromhex(cls.fixture["request_hex"]) # Фиксированные байты отделяют ожидаемый wire-вектор от реализации C.
|
||||
cls.sample_bytes = bytes.fromhex(cls.fixture["sample_hex"]) # DATA проверяется теми же байтами, что и другие языковые порты.
|
||||
|
||||
def setUp(self): # Изолируем pending и active между независимыми сценариями.
|
||||
self.core = NativeGasWatch(self.library) # Каждый тест начинает с отсутствующей подтверждённой конфигурацией.
|
||||
self.config = GasWatchConfig(self.fixture["config_id"], self.fixture["period_ms"], self.fixture["addresses"]) # Сохраняем порядок общих трёх адресов.
|
||||
|
||||
def activate(self): # Повторяем нормальный цикл отправки конфигурации и её точного подтверждения.
|
||||
payload = self.core.request(self.config) # Запрос формирует настоящий C-builder.
|
||||
self.core.validate_ack(payload, self.config) # Полное эхо устройства проверяет настоящий C ACK-валидатор.
|
||||
|
||||
def test_shared_fixture_request_ack_and_sample(self): # Основной сценарий фиксирует совместимость байтов и семантику признаков достоверности.
|
||||
self.assertEqual(self.request_bytes, self.core.request(self.config)) # Сопоставляем результат C-builder с общим независимым эталоном.
|
||||
with self.assertRaises(RuntimeError): # DATA до ACK не имеет подтверждённого соответствия каналам.
|
||||
self.core.sample(self.sample_bytes) # Даже корректный пакет должен быть отвергнут до активации конфигурации.
|
||||
self.core.validate_ack(self.request_bytes, self.config) # Активируем подписку только по полному точному эху.
|
||||
sample = self.core.sample(self.sample_bytes) # C проверяет config_id, количество, длину, padding и значения invalid-каналов.
|
||||
expected_values = tuple(value if valid else None for value, valid in zip(self.fixture["values"], self.fixture["valid"])) # Преобразуем семантику fixture, не декодируя wire-пакет.
|
||||
self.assertEqual(GasWatchSample(self.config.config_id, self.fixture["sequence"], self.fixture["timestamp_ms"], expected_values), sample) # Сверяем всю неизменяемую модель результата.
|
||||
zero_sample = bytes.fromhex("0200030078563412040302014433221105000000003412") # Фиксированный вектор содержит достоверный ноль, invalid-ноль и слово 0x1234.
|
||||
self.assertEqual((0, None, 0x1234), self.core.sample(zero_sample).values) # Недостоверное значение не должно превращаться в отображаемый ноль.
|
||||
|
||||
def test_native_request_rejects_duplicates_and_inconsistent_stop(self): # Логические ограничения протокола должны проверяться C, а не вторым валидатором Python.
|
||||
self.activate() # Действующая подписка нужна для проверки сохранения состояния при отказе builder.
|
||||
invalid = (GasWatchConfig(2, 100, (0, 0)), GasWatchConfig(2, 0, (0,)), GasWatchConfig(2, 100, ())) # Проверяем дубликаты и оба несовместимых сочетания периода с количеством.
|
||||
for config in invalid: # Каждая ошибочная конфигурация должна быть отвергнута независимо от предыдущей попытки.
|
||||
with self.subTest(config=config), self.assertRaises(ValueError): # Отказ C-builder преобразуется в явную ошибку Python.
|
||||
self.core.request(config) # Никакой неправильный запрос не должен менять активную подписку.
|
||||
self.assertEqual(self.config.config_id, self.core.sample(self.sample_bytes).config_id) # После отказа прежние подтверждённые DATA продолжают приниматься.
|
||||
stop = GasWatchConfig(self.config.config_id + 1, 0, ()) # Пустой список с нулевым периодом представляет правильную команду отключения.
|
||||
self.assertEqual(bytes.fromhex("020000007956341200000000"), self.core.request(stop)) # C-builder должен поддерживать отдельный идентификатор остановки.
|
||||
self.core.validate_ack(bytes.fromhex("020000007956341200000000"), stop) # Полный ACK пустой конфигурации также должен проходить C-проверку.
|
||||
|
||||
def test_ack_checks_full_configuration_and_preserves_active(self): # Равного количества каналов недостаточно для принятия подтверждения.
|
||||
self.activate() # Создаём активную подписку, которая должна пережить ошибочный ACK следующей команды.
|
||||
changed = GasWatchConfig(self.config.config_id + 1, 200, (0xFF01, 1, 0)) # Новый порядок того же числа адресов получает новый идентификатор.
|
||||
changed_request = self.core.request(changed) # Pending меняется, а действующий профиль остаётся активным до ACK.
|
||||
self.assertEqual(self.config.config_id, self.core.sample(self.sample_bytes).config_id) # Старые DATA разрешены, пока устройство не подтвердило переключение.
|
||||
for supplied in (self.config, changed): # Отдельно проверяем поздний ACK и пакет, не соответствующий новой конфигурации.
|
||||
with self.subTest(supplied=supplied), self.assertRaises(ValueError): # Ни один вариант не должен активировать неподтверждённую подписку.
|
||||
self.core.validate_ack(self.request_bytes, supplied) # Вариант old отвергает pending-проверка, вариант new — C-сравнение эха.
|
||||
altered = bytearray(changed_request) # Моделируем ACK с тем же id, периодом и count, но другим порядком адресов.
|
||||
altered[-2:] = b"\x02\x00" # Меняем один адрес в конце вектора, не затрагивая геометрию ответа.
|
||||
with self.assertRaises(ValueError): # Полное сравнение адресов должно обнаружить подмену при прежнем количестве каналов.
|
||||
self.core.validate_ack(bytes(altered), changed) # C проверяет весь список и не разрешает частичную подписку.
|
||||
self.core.validate_ack(changed_request, changed) # Корректное полное эхо атомарно переводит pending в active.
|
||||
with self.assertRaises(ValueError): # Пакет прежнего id после успешного переключения становится устаревшим.
|
||||
self.core.sample(self.sample_bytes) # Количество осталось тем же, поэтому проверка должна опираться на идентификатор.
|
||||
new_sample = bytes.fromhex("0200030079563412fffffffff0ffffff037b00feff0000") # Фиксированный DATA-вектор относится к новому config_id 0x12345679.
|
||||
self.assertEqual(changed.config_id, self.core.sample(new_sample).config_id) # Новый подтверждённый id разрешает данные именно новой конфигурации.
|
||||
|
||||
def test_malformed_ack_and_sample_do_not_poison_subscription(self): # Ошибки отдельных пакетов не должны разрушать рабочую подписку.
|
||||
self.core.request(self.config) # Pending необходим для проверки точного формата первого подтверждения.
|
||||
for payload in (self.request_bytes[:-1], self.request_bytes + b"\x00"): # Отсечённые и дополненные ACK имеют неправильную длину.
|
||||
with self.subTest(ack=payload), self.assertRaises(ValueError): # C должен отвергнуть и неполное, и нестрогое эхо.
|
||||
self.core.validate_ack(payload, self.config) # Pending сохраняется, позволяя принять следующий корректный ACK.
|
||||
self.core.validate_ack(self.request_bytes, self.config) # Корректное подтверждение после ошибок всё ещё активирует исходную подписку.
|
||||
malformed = ( # Список фиксированных искажений покрывает геометрию, достоверность и привязку к конфигурации.
|
||||
self.sample_bytes[:-1], # Отбрасываем последнее слово частично: общая длина больше не соответствует count.
|
||||
self.sample_bytes + b"\x00", # Лишний байт также нарушает точную геометрию пакета.
|
||||
bytes.fromhex("0100030078563412fffffffff0ffffff037b00feff0000"), # Версия 1 не должна интерпретироваться как новый формат версии 2.
|
||||
bytes.fromhex("0200030079563412fffffffff0ffffff037b00feff0000"), # Другой id с прежним count моделирует поздние DATA другой подписки.
|
||||
bytes.fromhex("0200020078563412fffffffff0ffffff037b00feff"), # Корректная геометрия двух значений не совпадает с тремя ожидаемыми адресами.
|
||||
bytes.fromhex("0200030078563412fffffffff0ffffff837b00feff0000"), # Старший padding-бит bitmap для трёх каналов обязан быть нулём.
|
||||
bytes.fromhex("0200030078563412fffffffff0ffffff037b00feff0100"), # Значение invalid-канала обязано быть нулём, иначе данные неоднозначны.
|
||||
) # Закрываем перечень независимых отрицательных векторов C-декодера.
|
||||
for payload in malformed: # Проверяем каждый отказ с тем же неизменным состоянием подписки.
|
||||
with self.subTest(data=payload), self.assertRaises(ValueError): # Ни один некорректный пакет не должен превращаться в модель.
|
||||
self.core.sample(payload) # Все проверки структуры пакета выполняются общим C-декодером.
|
||||
self.assertEqual(self.config.config_id, self.core.sample(self.sample_bytes).config_id) # Следующий корректный пакет доказывает сохранение active после ошибки.
|
||||
|
||||
def test_failed_decode_keeps_all_c_outputs_unchanged(self): # Проверяем обещание ABI не публиковать частичные результаты при ошибке пакета.
|
||||
data = (ctypes.c_uint8 * len(self.sample_bytes)).from_buffer_copy(self.sample_bytes) # Передаём эталонные байты, изменяя только ожидаемый id C-вызова.
|
||||
sequence = ctypes.c_uint32(11) # Отличимый sentinel обнаруживает запись счётчика до окончания проверки.
|
||||
timestamp = ctypes.c_uint32(22) # Независимый sentinel защищает проверку метки времени.
|
||||
values = (ctypes.c_uint16 * 3)(31, 32, 33) # Отличимые слова обнаруживают частичное заполнение выходного массива.
|
||||
valid = (ctypes.c_uint8 * 3)(41, 42, 43) # Признаки также не должны перезаписываться при устаревшем id.
|
||||
result = self.core.lib.set_gas_watch_decode(data, len(data), self.config.config_id + 1, 3, ctypes.byref(sequence), ctypes.byref(timestamp), values, valid, 3) # Проверяем настоящий C ABI без обёртки создания модели.
|
||||
self.assertEqual(-1, result) # Несовпадающий expected_config_id обязан отвергнуть даже корректно сформированный пакет.
|
||||
self.assertEqual((11, 22, (31, 32, 33), (41, 42, 43)), (sequence.value, timestamp.value, tuple(values), tuple(valid))) # Каждый выход должен сохранить исходный sentinel.
|
||||
|
||||
def test_pending_replacement_id_reuse_and_reset(self): # Проверяем гонки нескольких запросов и очистку состояния при переподключении.
|
||||
self.activate() # Ранее подтверждённая конфигурация задаёт исходный active.
|
||||
changed = GasWatchConfig(self.config.config_id + 1, 100, (1, 0, 0xFF01)) # Первый новый порядок начинает ожидать подтверждение.
|
||||
self.core.request(changed) # Старый active остаётся доступным, пока новый pending не подтверждён.
|
||||
newest = GasWatchConfig(self.config.config_id + 2, 100, (0xFF01, 0, 1)) # Следующая команда заменяет ожидаемый список ещё до первого ACK.
|
||||
newest_request = self.core.request(newest) # Актуальным pending становится только последний запрос.
|
||||
with self.assertRaises(ValueError): # ACK уже заменённого pending не должен возвращать старое ожидание.
|
||||
self.core.validate_ack(self.request_bytes, changed) # Проверка pending срабатывает до анализа неподходящего payload.
|
||||
for reused in (GasWatchConfig(self.config.config_id, 200, self.config.addresses), GasWatchConfig(newest.config_id, 200, newest.addresses)): # Active и pending id нельзя переиспользовать с другим содержимым.
|
||||
with self.subTest(reused=reused), self.assertRaises(ValueError): # Такая подмена лишила бы DATA возможности определить порядок значений.
|
||||
self.core.request(reused) # Ошибка не должна уничтожать последний ожидаемый запрос.
|
||||
self.core.validate_ack(newest_request, newest) # Последний корректный pending всё ещё подтверждается после отвергнутых попыток.
|
||||
self.core.reset() # Переподключение удаляет как ожидаемую, так и подтверждённую подписку.
|
||||
with self.assertRaises(RuntimeError): # Старая выборка после reset требует нового цикла подтверждения.
|
||||
self.core.sample(self.sample_bytes) # Данные прошлого соединения не должны отображаться автоматически.
|
||||
with self.assertRaises(ValueError): # Ни один ACK прошлого соединения больше не ожидается.
|
||||
self.core.validate_ack(newest_request, newest) # Reset запрещает позднее восстановление прежнего active.
|
||||
|
||||
def test_maximum_subscription_uses_real_c_builder(self): # Проверяем безопасную передачу полного массива из 64 каналов через ctypes.
|
||||
config = GasWatchConfig(0xFFFFFFFF, 65535, tuple(range(64))) # Используем максимальные id и период при допустимом числе уникальных адресов.
|
||||
payload = self.core.request(config) # Размер payload определяет настоящий C-builder в общем транспортном буфере.
|
||||
self.assertEqual(140, len(payload)) # Версия 2 задаёт 12 байт заголовка и 64 двухбайтных адреса, всего 140 байт.
|
||||
self.core.validate_ack(payload, config) # Полное эхо максимального массива проверяется без усечения и частичного принятия.
|
||||
|
||||
|
||||
if __name__ == "__main__": # Разрешаем запуск этого набора проверок напрямую наряду с unittest discovery.
|
||||
unittest.main() # Стандартный runner сообщает результаты, ошибки DLL и пропущенные C-проверки.
|
||||
Reference in New Issue
Block a user