Andrey
a5e46d4627
Добавить очереди операций PBUS/SBUS, общую консоль, настройки Candle и SLCAN, декодеры DSView и обработку плотных графиков. Обновить протокол Бальзам и проверки общих компонентов.
533 lines
32 KiB
Python
533 lines
32 KiB
Python
# Проверки decoders: фиксируют ожидаемое поведение на воспроизводимых входных данных.
|
|
# Сценарии для навигации по файлу:
|
|
# test_reference_crc
|
|
# test_documented_ping
|
|
# test_documented_bridge
|
|
# test_v1_big_endian_header
|
|
# test_noise_crc_and_resync
|
|
# test_bad_version_and_length
|
|
# test_max_payload
|
|
# test_unknown_message_preserved
|
|
# Остальные сценарии продолжают эти проверки ниже в том же файле.
|
|
# Тестовые значения и подмены окружения относятся к проверке контракта; они не являются
|
|
# настройками рабочего устройства.
|
|
|
|
"""Wire vectors and DSView API boundary tests; no sigrok installation needed."""
|
|
import binascii
|
|
import importlib
|
|
from pathlib import Path
|
|
import struct
|
|
import sys
|
|
import types
|
|
import tempfile
|
|
import importlib.util
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SPEC = importlib.util.spec_from_file_location('dsview_build', ROOT / 'build.py')
|
|
BUILD = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(BUILD)
|
|
RUNTIME = tempfile.TemporaryDirectory()
|
|
BUILD.package_decoders(Path(RUNTIME.name))
|
|
(Path(RUNTIME.name) / 'decoders/common/__init__.py').write_text('')
|
|
|
|
|
|
class FakeDecoder:
|
|
def has_channel(self, channel):
|
|
return channel < 2
|
|
|
|
def register(self, kind):
|
|
return kind
|
|
|
|
def put(self, ss, es, output, data):
|
|
self.emitted.append((ss, es, output, data))
|
|
|
|
|
|
stub = types.ModuleType('sigrokdecode')
|
|
stub.Decoder = FakeDecoder
|
|
stub.OUTPUT_ANN, stub.OUTPUT_PYTHON, stub.SRD_CONF_SAMPLERATE = 0, 1, 2
|
|
previous = sys.modules.get('sigrokdecode')
|
|
sys.modules['sigrokdecode'] = stub
|
|
sys.path.insert(0, str(Path(RUNTIME.name) / 'decoders'))
|
|
try:
|
|
uart = importlib.import_module('set_uart.pd')
|
|
wire = importlib.import_module('common.setgui_decoders.set_uart')
|
|
can = importlib.import_module('set_can.pd')
|
|
cancore = importlib.import_module('common.setgui_decoders.set_can')
|
|
pm = importlib.import_module('pm35_uart.pd')
|
|
gate = importlib.import_module('gate_driver_timing.pd')
|
|
pair_decoder = importlib.import_module('transistor_pair.pd')
|
|
pmcore = importlib.import_module('common.setgui_decoders.pm35_uart')
|
|
templates_core = importlib.import_module('common.setgui_decoders.template_protocols') # Новый прикладной слой импортируется из изолированного runtime.
|
|
tms_decoder = importlib.import_module('tms_uart.pd') # Проверяет контракт точки входа нового UART-анализатора.
|
|
tms_core = importlib.import_module('common.setgui_decoders.tms_uart') # Потоковый адаптер использует поставленные общие раскладки терминала.
|
|
finally:
|
|
sys.path.pop(0)
|
|
if previous is None:
|
|
del sys.modules['sigrokdecode']
|
|
else:
|
|
sys.modules['sigrokdecode'] = previous
|
|
|
|
PING = bytes.fromhex('A5 5A 02 08 01 00 00 00 2A 00 34 12 00 00 33 EC 33 04')
|
|
BRIDGE = bytes.fromhex('AA 55 08 01 01 67 45 23 01 AA BB FE 14')
|
|
|
|
|
|
def crc32_frame(header, payload=b''):
|
|
data = header + payload
|
|
return data + struct.pack('<I', binascii.crc32(data[2:]) & 0xffffffff)
|
|
|
|
|
|
def modbus(raw):
|
|
return raw + struct.pack('<H', wire.crc16(raw, True))
|
|
|
|
|
|
def feed(parser, raw):
|
|
events = []
|
|
for i, byte in enumerate(raw):
|
|
events.extend(parser.feed(byte, i * 10, i * 10 + 9))
|
|
return events
|
|
|
|
|
|
def instance(module, **options):
|
|
decoder = module.Decoder()
|
|
decoder.options = {item['id']: item['default'] for item in decoder.options}
|
|
decoder.options.update(options)
|
|
decoder.emitted = []
|
|
decoder.metadata(2, 1000000)
|
|
decoder.start()
|
|
return decoder
|
|
|
|
|
|
def uart_bytes(decoder, raw, direction=0, base=0):
|
|
for i, byte in enumerate(raw):
|
|
ss = base + i * 100
|
|
decoder.decode(ss, ss+80, ('DATA', direction, (byte, [])))
|
|
decoder.decode(ss, ss+90, ('FRAME', direction, (byte, True)))
|
|
|
|
|
|
def frames(decoder):
|
|
return [item[3][1] for item in decoder.emitted if item[2] == 1]
|
|
|
|
|
|
class WireTests(unittest.TestCase):
|
|
def test_reference_crc(self):
|
|
self.assertEqual(wire.crc16(b'123456789'), 0x29b1)
|
|
self.assertEqual(wire.crc16(b'123456789', True), 0x4b37)
|
|
|
|
def test_documented_ping(self):
|
|
result = wire.parse_frame(PING)
|
|
self.assertEqual((result['destination'], result['sequence']), (42, 0x1234))
|
|
|
|
def test_documented_bridge(self):
|
|
result = wire.parse_frame(BRIDGE)
|
|
self.assertEqual((result['can_id'], result['payload']), (0x1234567, b'\xaa\xbb'))
|
|
|
|
def test_v1_big_endian_header(self):
|
|
frame = crc32_frame(bytes.fromhex('a55a010c12340003'), b'abc')
|
|
result = wire.parse_frame(frame)
|
|
self.assertEqual(result['sequence'], 0x1234)
|
|
self.assertIn('FIRMWARE_DATA', result['summary'])
|
|
|
|
def test_noise_crc_and_resync(self):
|
|
damaged = bytearray(PING)
|
|
damaged[-1] ^= 1
|
|
events = feed(wire.StreamParser(), b'noise\xa5' + damaged + PING + BRIDGE)
|
|
self.assertEqual(sum(e[2] is not None for e in events), 2)
|
|
self.assertTrue(any(e[3] and 'CRC' in e[3] for e in events))
|
|
|
|
def test_bad_version_and_length(self):
|
|
events = feed(wire.StreamParser(), bytes.fromhex('a55aff a55a010100000201') + PING)
|
|
self.assertEqual(sum(e[2] is not None for e in events), 1)
|
|
self.assertGreaterEqual(sum(bool(e[3]) for e in events), 2)
|
|
|
|
def test_max_payload(self):
|
|
raw = crc32_frame(bytes.fromhex('a55a020009000000000001000002'), bytes(range(256))*2)
|
|
events = feed(wire.StreamParser(), raw)
|
|
self.assertEqual(len(events), 1)
|
|
self.assertEqual(len(events[0][2]['payload']), 512)
|
|
|
|
def test_unknown_message_preserved(self):
|
|
raw = crc32_frame(bytes.fromhex('a55a020034120000000001000000'))
|
|
self.assertEqual(wire.parse_frame(raw)['message_type'], 0x1234)
|
|
|
|
def test_uart_directions_are_independent(self):
|
|
decoder = instance(uart)
|
|
for i, byte in enumerate(PING):
|
|
decoder.decode(i*10, i*10+9, ('FRAME', 0, (byte, True)))
|
|
decoder.decode(i*10, i*10+9, ('FRAME', 1, (byte, True)))
|
|
self.assertEqual([f['direction'] for f in frames(decoder)], [0, 1])
|
|
|
|
def test_uart_invalid_last_byte_not_emitted(self):
|
|
decoder = instance(uart)
|
|
uart_bytes(decoder, PING[:-1])
|
|
decoder.decode(2000, 2090, ('FRAME', 0, (PING[-1], False)))
|
|
self.assertFalse(frames(decoder))
|
|
uart_bytes(decoder, PING, base=3000)
|
|
self.assertEqual(len(frames(decoder)), 1)
|
|
|
|
def test_uart_gap_discards_partial(self):
|
|
decoder = instance(uart)
|
|
uart_bytes(decoder, PING[:10])
|
|
uart_bytes(decoder, PING, base=200000)
|
|
self.assertEqual(len(frames(decoder)), 1)
|
|
|
|
|
|
class CanTests(unittest.TestCase):
|
|
def segments(self, packet=PING):
|
|
yield bytes([0x10]) + struct.pack('<H', len(packet)) + packet[:5]
|
|
for i, offset in enumerate(range(5, len(packet), 7), 1):
|
|
yield bytes([0x20 | (i & 15)]) + packet[offset:offset+7]
|
|
|
|
def test_set_reassembly(self):
|
|
decoder = instance(can, protocol='set-v2')
|
|
for i, data in enumerate(self.segments()):
|
|
decoder.decode(i*1000, i*1000+900, ('extended', 0x122a0000, 'data', len(data), list(data)))
|
|
self.assertEqual(frames(decoder)[-1]['raw'], PING)
|
|
|
|
def test_missing_segment(self):
|
|
reader = cancore.Reassembler()
|
|
parts = list(self.segments())
|
|
reader.feed(0x122a0000, parts[0], 0, 9, 0)
|
|
events = reader.feed(0x122a0000, parts[2], 10, 19, 1)
|
|
self.assertIn('sequence', events[-1][3])
|
|
self.assertFalse(reader.pending)
|
|
|
|
def test_timeout_and_new_first(self):
|
|
reader = cancore.Reassembler()
|
|
first = next(self.segments())
|
|
reader.feed(0x122a0000, first, 0, 9, 0)
|
|
events = reader.feed(0x122a0000, first, 20, 29, 500)
|
|
self.assertIn('timeout', events[0][3])
|
|
self.assertEqual(reader.pending[0x122a0000]['start'], 20)
|
|
|
|
def test_id_mismatch(self):
|
|
reader = cancore.Reassembler()
|
|
for i, part in enumerate(self.segments()):
|
|
events = reader.feed(0x122b0000, part, i*10, i*10+9, i)
|
|
self.assertIn('disagree', events[-1][3])
|
|
|
|
def test_sequence_wrap(self):
|
|
packet = crc32_frame(bytes.fromhex('a55a0200010000002a0001008000'), bytes(128))
|
|
reader = cancore.Reassembler()
|
|
for i, part in enumerate(self.segments(packet)):
|
|
events = reader.feed(0x122a0000, part, i*10, i*10+9, i)
|
|
self.assertEqual(events[-1][2]['raw'], packet)
|
|
|
|
def test_interleaved_channels(self):
|
|
reader = cancore.Reassembler()
|
|
complete = []
|
|
for i, part in enumerate(self.segments()):
|
|
for channel in (0, 1):
|
|
complete += reader.feed(0x122a0000 | channel, part, i*10, i*10+9, i)
|
|
self.assertEqual(sum(bool(e[2] and 'raw' in e[2]) for e in complete), 2)
|
|
|
|
def test_gas(self):
|
|
result = cancore.legacy(0x1b530010, b'\x34\x12\xfe\xff')
|
|
self.assertEqual(result['registers'], [(16, 0x1234), (17, 0xfffe)])
|
|
with self.assertRaises(ValueError):
|
|
cancore.legacy(0x1b530010, b'\x00')
|
|
|
|
def test_balsam_mask_and_endianness(self):
|
|
result = cancore.legacy(0xba0010, bytes.fromhex('0000ffffa0101234'), 'balsam') # Реальная прошивка передаёт value2, value3, header, value1; отмечены первое и третье слова.
|
|
self.assertEqual(result['registers'], [(16, 0x1234), (18, 0xffff)])
|
|
self.assertIsNone(cancore.legacy(0xba000f, bytes(8), 'balsam'))
|
|
|
|
def test_ignore_standard_and_remote(self):
|
|
decoder = instance(can)
|
|
decoder.decode(0, 9, ('standard', 0x123, 'data', 0, []))
|
|
decoder.decode(10, 19, ('extended', 0x1b530010, 'remote', 0, []))
|
|
self.assertFalse(decoder.emitted)
|
|
|
|
def test_boot_vectors(self):
|
|
self.assertIn('ENTER_BOOT', cancore.legacy(0x13590702, b'')['summary'])
|
|
self.assertIn('offset=0x8', cancore.legacy(0x135b0001, bytes.fromhex('1011121314151617'))['summary'])
|
|
self.assertIn('status=0', cancore.legacy(0x1b5c0702, bytes.fromhex('00ff000000000000'))['summary'])
|
|
|
|
|
|
class TemplateProtocolTests(unittest.TestCase): # Сверяет прикладные события DSView с действующими парсерами templates.
|
|
def test_every_legacy_project_uses_its_wire_layout(self): # Все проекты проверяются настоящими кодировщиками обоих исторических форматов.
|
|
for project in templates_core.legacycan.PROJECTS: # Каталог включает Буксир, СЭДБМ, Ледокол, Бальзам, 23550, Янтарь и варианты формата.
|
|
with self.subTest(project=project.name): # Ошибка сообщает конкретный несовместимый профиль.
|
|
node = project.nodes[0] # Достаточно одного узла для проверки раскладки выбранного проекта.
|
|
packet = templates_core.legacycan.encode_write(project.format, node.rx_id, 16, (0x1234, 0xffff)) # Различающиеся байты выявляют ошибку порядка слов и байтов.
|
|
result = templates_core.decode_can(packet.can_id, packet.data, 'legacy', project=project.name) # Разбор выполняется тем же путём, что у адаптера sigrok.
|
|
self.assertEqual(result['registers'], [(16, 0x1234), (17, 0xffff)]) # Оба wire-формата должны восстанавливать одинаковые адреса и значения.
|
|
self.assertIn(node.name, result['summary']) # Вывод включает имя устройства из актуального каталога проекта.
|
|
|
|
def test_balsam_real_wire_order_and_pult(self): # Регрессия неправильного положения адресного слова в прежнем адаптере.
|
|
for ident in (0xba0004, 0xba0014, 0x745009, 0x745019): # Проверяются УМП в обоих направлениях и отдельная пара ID пульта.
|
|
result = templates_core.decode_can(ident, bytes.fromhex('11223344e07f8000'), 'balsam') # Маска 111 в header=E07F отмечает три слова начиная с командного регистра 127.
|
|
self.assertEqual(result['registers'], [(127, 0x8000), (128, 0x1122), (129, 0x3344)]) # В wire сначала идут второе и третье слова, а первое находится в конце.
|
|
self.assertIsNone(templates_core.decode_can(0xba000f, bytes(8), 'balsam')) # Отсутствующий узел не выдаётся за сообщение Balsam.
|
|
|
|
def test_ump_maps_keep_nodes_and_assigned_ids_separate(self): # Локальная карта 167/v1 и новые глобальные ID не должны смешиваться.
|
|
local = templates_core.decode_can(0x1b53007f, b'\x80\x00', 'ump-protocan', ump_node=2) # Бит 7 командного регистра означает пуск УМП в локальной карте.
|
|
self.assertIn('УМП 2', local['summary']) # Локальный адрес не теряет выбранный вручную номер устройства.
|
|
self.assertIn('cUMPstart', local['summary']) # Имена команд берутся из действующей карты, а не остаются сырым HEX.
|
|
assigned = templates_core.decode_can(0x1b530000 | 6000, b'\x01\x00', 'ump-protocan', ump_map='assigned') # Десятичный GAS ID 6000 относится к назначенной карте второго УМП.
|
|
self.assertIn('sens_error[0]', assigned['summary']) # Параметр получает имя из JSON-словаря templates.
|
|
|
|
def test_deif_standard_can_and_physical_value(self): # DEIF должен получать STD-кадры, которые прежний SET CAN отбрасывал.
|
|
decoder = instance(can, protocol='deif') # Выбор DEIF явно разрешает стандартный CANopen ID.
|
|
raw = bytes.fromhex('010130027b000000') # Объект 0x3001:02 содержит RAW=123, соответствующий углу 12.3 градуса.
|
|
decoder.decode(10, 90, ('standard', 0x201, 'data', len(raw), raw)) # Вектор проходит реальную границу адаптера sigrok.
|
|
result = frames(decoder)[0] # Результат доступен и в Python-выходе, и в аннотациях.
|
|
self.assertIn('12.3', result['summary']) # Масштаб задаёт существующий DEIF-парсер.
|
|
self.assertTrue(any(item[2] == 0 and item[3][0] == 3 for item in decoder.emitted)) # Проверяет фактическое появление подробных полей на новой дорожке.
|
|
|
|
def test_can_auto_routes_set_v2_and_balsam(self): # Авто ограничено проверяемыми сигнатурами и не угадывает исторические проекты.
|
|
decoder = instance(can, protocol='auto') # Профиль использует прежний сборщик SET v2 и общий парсер Balsam.
|
|
for index, payload in enumerate(CanTests().segments()): # Документированный SET PING разбивается штатным тестовым транспортом.
|
|
decoder.decode(index * 1000, index * 1000 + 900, ('extended', 0x122a0000, 'data', len(payload), payload)) # Номер отсчёта обеспечивает проверяемый тайм-аут сборки.
|
|
self.assertEqual(frames(decoder)[-1]['raw'], PING) # Авто должен вернуть исходный CRC-подтверждённый SET v2 пакет.
|
|
raw = bytes.fromhex('0000000080101234') # Заголовок отмечает первое слово по адресу 16, находящееся в конце CAN-данных.
|
|
decoder.decode(4000, 4900, ('extended', 0xba0014, 'data', 8, raw)) # ID относится к ответу первого УМП Balsam 167.
|
|
self.assertEqual(frames(decoder)[-1]['registers'], [(16, 0x1234)]) # Авто не интерпретирует Balsam как Boot Data ProtoCAN.
|
|
|
|
def test_uart_bridge_uses_selected_can_profile(self): # Проверяет перенос общего CAN-разбора через UART-транспорт AA55.
|
|
for profile, ident, flags, payload in (('balsam', 0xba0014, 1, bytes.fromhex('0000000080101234')), ('deif', 0x201, 0, bytes.fromhex('010130027b000000'))): # Векторы охватывают EXT Balsam и STD DEIF с разным значением IDE.
|
|
decoder = instance(uart, can_protocol=profile) # Выбор профиля моста не меняет основной потоковый формат SET UART.
|
|
body = bytes([6 + len(payload), 1, flags]) + struct.pack('<I', ident) + payload # Длина охватывает seq, flags, 32-битный ID и полезные данные моста.
|
|
raw = b'\xaa\x55' + body + struct.pack('<H', wire.crc16(body)) # CRC-16/CCITT-FALSE покрывает поле длины и весь транспортный кадр после сигнатуры.
|
|
uart_bytes(decoder, raw) # Передаёт только обычные UART-события, как настоящий базовый декодер.
|
|
result = frames(decoder)[0]['application'] # Структурированный кадр содержит результат вложенного прикладного парсера.
|
|
self.assertEqual(result['protocol'], 'Balsam' if profile == 'balsam' else 'DEIF XDi-192') # IDE и выбранный профиль должны дойти до соответствующего общего парсера.
|
|
|
|
def test_tms_noise_coordinates_and_independent_directions(self): # Проверяет потоковый поиск по CRC и сохранение координат после шума.
|
|
decoder = instance(tms_decoder) # Используются стандартная раскладка message2.h и CRC MODBUS.
|
|
tms = tms_core.tms # Тестовый кодировщик и декодер получают один канонический контракт.
|
|
layout = tms.default_layout() # Длина определяется текущими числами дискретных байтов и аналоговых слов.
|
|
raw = tms.build(2, 67, bytes([1]) + bytes(layout.digit_bytes - 1), [-123] + [0] * (layout.analog_count - 1), layout) # Ответ CMD_STD_ANS содержит первый дискрет и отрицательный аналог.
|
|
uart_bytes(decoder, b'noise' + raw) # Предшествующие байты должны быть пропущены по CRC без сдвига границ события.
|
|
uart_bytes(decoder, raw, direction=1) # Второе направление обрабатывается независимым парсером.
|
|
results = frames(decoder) # Извлекает только структурированные события готовых кадров.
|
|
self.assertEqual([item['direction'] for item in results], [0, 1]) # RX и TX не объединяются в один буфер.
|
|
self.assertEqual(results[0]['analogs'][0], -123) # Знаковый int16 сохраняется без преобразования в беззнаковое слово.
|
|
event = next(item for item in decoder.emitted if item[2] == 1) # Находит временные границы первого подтверждённого пакета.
|
|
self.assertEqual((event[0], event[1]), (500, (len(raw) + 4) * 100 + 90)) # Пять байтов шума занимают 500 отсчётов при шаге 100 в тестовом UART.
|
|
|
|
def test_tms_gap_and_crc_reject_incomplete_frames(self): # CRC и пауза должны исключать ложную выдачу неполного пакета.
|
|
decoder = instance(tms_decoder, layout='custom', digit_bytes=70, analog_count=80) # Индивидуальная раскладка поддерживает текущую посылку 22220 размером 235 байт.
|
|
tms = tms_core.tms # Источник CRC и структуры совпадает с общим парсером терминала.
|
|
layout = tms.Layout(70, 80) # 2 байта заголовка, 70 дискретов, 160 байт аналогов и 3 байта хвоста.
|
|
raw = tms.build(1, 67, bytes(70), [0] * 80, layout) # Формирует полностью корректный пакет известной длины.
|
|
damaged = bytearray(raw) # Изменение одного байта позволяет проверить обязательность CRC.
|
|
damaged[-2] ^= 1 # Повреждается старший байт CRC, а формат и длина сохраняются.
|
|
uart_bytes(decoder, damaged) # Повреждённый кадр не должен давать прикладного события.
|
|
self.assertFalse(frames(decoder)) # Подтверждает отсутствие ложного пакета при несовпадении CRC.
|
|
uart_bytes(decoder, raw[:40], base=200000) # Начало кадра поступает после паузы и сбрасывает прежний шум.
|
|
uart_bytes(decoder, raw, base=400000) # Новая пауза отбрасывает хвост и позволяет разобрать полный кадр.
|
|
self.assertEqual(len(frames(decoder)), 1) # Только последний полный пакет проходит проверку границ и CRC.
|
|
self.assertEqual(frames(decoder)[0]['raw'], raw) # Событие сохраняет точные исходные байты записи.
|
|
|
|
|
|
class PM35Tests(unittest.TestCase):
|
|
def test_single_wire_auto(self):
|
|
reader = pmcore.PM35Parser('auto')
|
|
raw = modbus(bytes.fromhex('100300000002')) + modbus(bytes.fromhex('1003041234ffff'))
|
|
events = feed(reader, raw)
|
|
self.assertEqual(len(events), 2)
|
|
self.assertTrue(all(event[2] for event in events))
|
|
|
|
def test_request_and_response(self):
|
|
request = modbus(bytes.fromhex('100300000002'))
|
|
response = modbus(bytes.fromhex('1003041234ffff'))
|
|
decoder = instance(pm)
|
|
uart_bytes(decoder, request, 1)
|
|
uart_bytes(decoder, response, 0, 10000)
|
|
self.assertEqual(len(frames(decoder)), 2)
|
|
self.assertEqual(frames(decoder)[1]['values'], [0x1234, 65535])
|
|
|
|
def test_128_word_legacy_reply(self):
|
|
events = feed(pmcore.PM35Parser('response'), modbus(b'\x10\x03\x00' + bytes(256)))
|
|
self.assertEqual(len(events[0][2]['values']), 128)
|
|
|
|
def test_write_command(self):
|
|
result = pmcore.parse_pm35(modbus(bytes.fromhex('1006007f8000')), 'request')
|
|
self.assertIn('command_bits=8000', result['summary'])
|
|
|
|
def test_exception(self):
|
|
result = pmcore.parse_pm35(modbus(bytes.fromhex('108302')), 'response')
|
|
self.assertIn('exception', result['summary'])
|
|
|
|
def test_crc_and_range_errors(self):
|
|
with self.assertRaisesRegex(ValueError, 'CRC'):
|
|
pmcore.parse_pm35(bytes.fromhex('1003000000010000'), 'request')
|
|
with self.assertRaisesRegex(ValueError, 'range'):
|
|
pmcore.parse_pm35(modbus(bytes.fromhex('1003007f0002')), 'request')
|
|
|
|
|
|
class GateAdapterTests(unittest.TestCase):
|
|
def test_wait_releases_dsview_shared_pin_tuple(self):
|
|
for channels, pair in (((1,), 0), ((0, 1), 0), ((0, 1), 1)):
|
|
with self.subTest(channels=channels, pair=pair):
|
|
decoder = instance(pair_decoder if pair else gate)
|
|
decoder.has_channel = lambda channel: channel in channels
|
|
shared_pins = tuple([0, 0])
|
|
calls = []
|
|
|
|
def wait(conditions):
|
|
# Native DSView owns one reference and PyTuple_SetItem
|
|
# rejects any reference retained by the previous caller.
|
|
self.assertEqual(sys.getrefcount(shared_pins), 2)
|
|
if len(calls) == 3:
|
|
raise StopIteration
|
|
calls.append(conditions)
|
|
decoder.samplenum = len(calls) * 10
|
|
decoder.matched = 0
|
|
return shared_pins
|
|
|
|
decoder.wait = wait
|
|
with self.assertRaises(StopIteration):
|
|
decoder.decode()
|
|
self.assertEqual(len(calls), 3)
|
|
|
|
def test_orphan_filter_boundaries_profiles_and_polarities(self):
|
|
for profile in ('1SP0635', '1SD536F2'):
|
|
for active_high in (True, False):
|
|
with self.subTest(profile=profile, active_high=active_high):
|
|
checker = gate.TimingChecker(1_000_000_000, profile=profile,
|
|
vstat_active_high=active_high, orphan_min_width_ns=50)
|
|
for start, width, expected in ((100, 20, []), (200, 49, []),
|
|
(300, 50, ['orphan']), (400, 51, ['orphan'])):
|
|
self.assertEqual(checker.on_status_edge(start, int(active_high)), [])
|
|
events = checker.on_status_edge(start + width, int(not active_high))
|
|
self.assertEqual([event['kind'] for event in events], expected)
|
|
self.assertIsNone(checker.status_start)
|
|
|
|
def test_orphan_filter_preserves_ack_errors_faults_and_missing_ack(self):
|
|
checker = gate.TimingChecker(1_000_000_000, orphan_min_width_ns=10_000)
|
|
checker.on_control_edge(0, 1)
|
|
self.assertEqual(checker.on_status_edge(250, 1)[0]['kind'], 'delay_ok')
|
|
self.assertEqual(checker.on_status_edge(950, 0)[0]['kind'], 'width_ok')
|
|
checker.on_control_edge(1000, 0)
|
|
checker.on_status_edge(1250, 1)
|
|
self.assertEqual(checker.on_status_edge(1270, 0)[0]['kind'], 'width_fail')
|
|
checker.on_status_edge(2000, 1)
|
|
self.assertEqual(checker.on_status_edge(4000, 0)[0]['kind'], 'fault')
|
|
checker.on_control_edge(5000, 1)
|
|
self.assertEqual(checker.expire(checker.next_deadline())[0]['kind'], 'missing')
|
|
|
|
def test_orphan_filter_default_disabled_and_unknown_duration_retained(self):
|
|
checker = gate.TimingChecker(1_000_000_000)
|
|
checker.on_status_edge(10, 1)
|
|
self.assertEqual(checker.on_status_edge(30, 0)[0]['short'], 'ORPHAN 20.0 ns')
|
|
checker = gate.TimingChecker(1_000_000_000, orphan_min_width_ns=50)
|
|
# A recording starting within a pulse gives no measurable pulse width.
|
|
self.assertEqual(checker.on_status_edge(10, 0)[0]['short'], 'Vstat?')
|
|
|
|
def test_orphan_filter_rejects_invalid_thresholds(self):
|
|
for value in (-1, float('nan'), float('inf')):
|
|
with self.subTest(value=value), self.assertRaisesRegex(ValueError, 'ORPHAN'):
|
|
gate.TimingChecker(100_000_000, orphan_min_width_ns=value)
|
|
|
|
def test_orphan_filter_option_through_native_edge_api(self):
|
|
decoder = instance(gate, orphan_min_width_ns=50)
|
|
decoder.metadata(2, 100_000_000)
|
|
decoder.samplenum = 0
|
|
samples = iter([(10, 1), (12, 0), (20, 1), (25, 0)])
|
|
def wait(conditions):
|
|
decoder.samplenum, level = next(samples)
|
|
decoder.matched = 2
|
|
return 0, level
|
|
decoder.wait = wait
|
|
with self.assertRaises(StopIteration):
|
|
decoder.decode()
|
|
self.assertEqual(len(decoder.emitted), 1)
|
|
self.assertEqual(decoder.emitted[0][:2], (20, 25))
|
|
self.assertEqual(decoder.emitted[0][3][1][1], 'ORPHAN 50.0 ns')
|
|
|
|
def test_profiles_through_native_edge_api(self):
|
|
for profile, active, edges, expected in [
|
|
('1SP0635', 'high', [(100, 1, 0, 1), (125, 1, 1, 2), (195, 1, 0, 2)], 700),
|
|
('1SD536F2', 'low', [(20, 1, 1, 1), (58, 1, 0, 2), (148, 1, 1, 2)], 900),
|
|
]:
|
|
with self.subTest(profile=profile):
|
|
decoder = instance(gate, profile=profile, vstat_active=active)
|
|
decoder.metadata(2, 100_000_000)
|
|
decoder.samplenum = 0
|
|
samples = iter(edges)
|
|
def wait(conditions):
|
|
decoder.samplenum, vin, vstat, decoder.matched = next(samples)
|
|
return vin, vstat
|
|
decoder.wait = wait
|
|
with self.assertRaises(StopIteration):
|
|
decoder.decode()
|
|
self.assertEqual([item[3][0] for item in decoder.emitted], [0, 1, 2])
|
|
self.assertIn(str(expected), decoder.emitted[-1][3][1][0])
|
|
|
|
def test_driver_has_no_pair_controls(self):
|
|
decoder = instance(gate)
|
|
self.assertNotIn('pair_analysis', decoder.options)
|
|
self.assertNotIn('vin2_mintime_ns', decoder.options)
|
|
self.assertEqual([c['id'] for c in gate.Decoder.optional_channels], ['vin', 'vstat'])
|
|
|
|
def test_pair_decoder_without_vstat(self):
|
|
decoder = instance(pair_decoder, vin2_mintime_ns=200, deadtime_12_ns=50)
|
|
decoder.metadata(2, 1_000_000_000)
|
|
samples = iter([(0, (0, 0)), (10, (1, 0)), (110, (0, 0)),
|
|
(140, (0, 1)), (240, (0, 0)), (280, (1, 0))])
|
|
def wait(conditions):
|
|
decoder.samplenum, pins = next(samples)
|
|
return pins
|
|
decoder.wait = wait
|
|
with self.assertRaises(StopIteration):
|
|
decoder.decode()
|
|
text = ' '.join(item[3][1][0] for item in decoder.emitted)
|
|
self.assertIn('Deadtime Vin1 -> Vin2: 30.0 ns', text)
|
|
self.assertIn('FAIL: Vin2 active', text)
|
|
self.assertIn('duty', text)
|
|
self.assertNotIn('ACK', text)
|
|
self.assertTrue(any(e[3][0] == 3 for e in decoder.emitted))
|
|
decoder.reset()
|
|
self.assertIsNone(decoder.checker)
|
|
|
|
def test_status_only_and_pair_requires_two_inputs(self):
|
|
decoder = instance(gate)
|
|
decoder.has_channel = lambda channel: channel == 1
|
|
samples = iter([(10, 1), (1010, 0)])
|
|
def wait(conditions):
|
|
self.assertEqual(conditions, [{1: 'e'}])
|
|
decoder.samplenum, status = next(samples)
|
|
decoder.matched = 1
|
|
return (255, status, 255)
|
|
decoder.wait = wait
|
|
with self.assertRaises(StopIteration):
|
|
decoder.decode()
|
|
self.assertEqual(len(decoder.emitted), 1)
|
|
decoder = instance(pair_decoder)
|
|
decoder.has_channel = lambda channel: channel == 0
|
|
with self.assertRaisesRegex(ValueError, 'Vin1 and Vin2'):
|
|
decoder.decode()
|
|
|
|
def test_missing_ack_timer_emits_error(self):
|
|
decoder = instance(gate)
|
|
decoder.metadata(2, 100_000_000)
|
|
decoder.samplenum = 0
|
|
calls = []
|
|
def wait(conditions):
|
|
calls.append(conditions)
|
|
if len(calls) == 1:
|
|
decoder.samplenum, decoder.matched = 100, 1
|
|
elif len(calls) == 2:
|
|
decoder.samplenum += conditions[2]['skip']
|
|
decoder.matched = 4
|
|
else:
|
|
raise StopIteration
|
|
return 1, 0
|
|
decoder.wait = wait
|
|
with self.assertRaises(StopIteration):
|
|
decoder.decode()
|
|
self.assertEqual([item[3][0] for item in decoder.emitted], [0, 4])
|
|
|
|
def test_packaged_core_matches_canonical_templates(self):
|
|
canonical = ROOT.parents[1] / 'python/logic_analyzer/decoders'
|
|
packaged = Path(RUNTIME.name) / 'decoders/common/setgui_decoders'
|
|
for name in ('gate_timing', 'transistor_pair', 'set_uart', 'set_can', 'pm35_uart'):
|
|
self.assertEqual((canonical / (name + '.py')).read_bytes(),
|
|
(packaged / (name + '.py')).read_bytes())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|