218 lines
32 KiB
Python
218 lines
32 KiB
Python
"""Контрольные потоки, независимая FIR-модель и проверка поставки сигма-дельта декодера.""" # Проверки не требуют настоящего прибора и установленного DSView.
|
|
import ast # Совместимость синтаксиса проверяется для Python 3.6 старых установок DSView.
|
|
import importlib.util # Ядро и адаптер загружаются непосредственно из проверяемых исходников.
|
|
import random # Закреплённое зерно формирует воспроизводимый поток для независимой свёртки.
|
|
from pathlib import Path # Пути вычисляются относительно templates, независимо от текущего каталога.
|
|
import subprocess # Изолированный Python проверяет переносимость упакованного декодера.
|
|
import sys # Исполняемый файл тестового Python используется для проверки чистого импорта.
|
|
import tempfile # Упаковка выполняется в отдельный каталог без изменения установленного DSView.
|
|
import types # Минимальный модуль sigrokdecode подменяет только границу API приложения.
|
|
import unittest # Тесты доступны через стандартный unittest discovery.
|
|
import weakref # Модель движка проверяет освобождение внутреннего буфера уровней перед следующим wait.
|
|
from unittest.mock import patch # Подмена импортов действует только во время загрузки адаптера.
|
|
|
|
ROOT = Path(__file__).resolve().parents[3] # Общий корень содержит python и tools/dsview одного checkout templates.
|
|
|
|
|
|
def load(name, path): # Общая загрузка не добавляет исходники в глобальный путь поиска Python.
|
|
spec = importlib.util.spec_from_file_location(name, ROOT / path) # Явный путь исключает случайный импорт установленной старой версии.
|
|
module = importlib.util.module_from_spec(spec) # Каждый вызов создаёт самостоятельное пространство имён.
|
|
spec.loader.exec_module(module) # Выполняем настоящий исходник, а не его копию в тесте.
|
|
return module # Тесты получают и фильтр, и рецепт упаковки без зависимостей от GUI.
|
|
|
|
|
|
CORE = load('sigma_delta_core', 'python/logic_analyzer/decoders/sigma_delta.py') # Одна каноническая реализация используется всеми сценариями.
|
|
|
|
|
|
class CaptureEnd(Exception): # Исключение воспроизводит остановку wait на конце конечного цифрового захвата.
|
|
pass # Дополнительные данные не нужны: неполный блок остаётся только во внутреннем состоянии фильтра.
|
|
|
|
|
|
class PinBuffer: # Возвращаемый движком буфер нельзя удерживать между вызовами wait в libsigrokdecode4DSL.
|
|
def __init__(self, pins): # Обёртка позволяет обнаружить оставшуюся ссылку без зависимости от счётчика ссылок Python.
|
|
self.pins = pins # Храним уровни только одного выбранного фронта захвата.
|
|
|
|
def __iter__(self): # Копирование в список должно сразу освободить временный объект результата wait.
|
|
return iter(self.pins) # Новый список получает значения уровней без ссылки на внутренний буфер движка.
|
|
|
|
def __getitem__(self, index): # Прямое чтение допустимо, но сохранение этого объекта до следующего wait недопустимо.
|
|
return self.pins[index] # Так модель воспроизводит обычный доступ к tuple возвращённых уровней.
|
|
|
|
|
|
class FakeDecoder: # Модель границы sigrok принимает настоящие переходы CLK и уровни DATA.
|
|
def has_channel(self, channel): # Отсутствующий обязательный канал должен обнаруживаться до ожидания фронтов.
|
|
return channel in self.connected # Сценарии отдельно исключают DATA и CLK.
|
|
|
|
def register(self, kind): # Различаем аннотации и структурированные Python-пакеты.
|
|
return kind # Нативные дескрипторы для проверки содержимого выходов не требуются.
|
|
|
|
def put(self, start, end, kind, data): # Записываем фактические границы и полезную нагрузку каждого результата.
|
|
self.emitted.append((start, end, kind, data)) # Поздние проверки могут обнаружить лишний бит или неполный выходной блок.
|
|
|
|
def wait(self, condition): # Сканируем только переходы CLK требуемой полярности.
|
|
if getattr(self, 'last_buffer', lambda: None)() is not None: # Настоящий DSView заполняет прежний tuple через PyTuple_SetItem, требующий отсутствия внешних ссылок.
|
|
raise RuntimeError('DSView pin buffer retained across wait calls.') # Регрессия обнаруживается до следующего выхода, как ошибка SystemError нативного движка.
|
|
self.conditions.append(condition) # Проверяем, что адаптер запрашивает фронт нужной линии.
|
|
for sample, pins in self.trace: # Изменения DATA между фронтами остаются видны, но не запускают приём бита.
|
|
previous, self.last_pins = self.last_pins, pins # Храним предыдущий уровень CLK, включая начальный уровень записи.
|
|
if previous[1] != pins[1] and pins[1] == (condition == {1: 'r'}): # Нарастающий фронт завершается единицей, спадающий нулём.
|
|
self.samplenum = sample # Номер реально выбранного фронта задаёт время результата.
|
|
buffer = PinBuffer(pins) # Временная обёртка моделирует принадлежащий движку буфер pinvalues.
|
|
self.last_buffer = weakref.ref(buffer) # Слабая ссылка обнаруживает удержание результата адаптером и сама не продлевает его жизнь.
|
|
return buffer # DATA берётся в момент CLK, а уровни нужно скопировать до следующего вызова wait.
|
|
raise CaptureEnd() # Конец записи не добавляет отсутствующие фронты и не вызывает искусственный flush.
|
|
|
|
|
|
def capture(bits, falling_bits=None, initial_clock=0): # Генерируем физическую трассу с десятью отсчётами на такт.
|
|
trace = [] # Начальное состояние задаётся отдельно и не считается событием фронта.
|
|
for index, bit in enumerate(bits): # Каждый бит имеет подготовку DATA, нарастающий и спадающий фронты CLK.
|
|
trace.append((index * 10 + 1, (bit, 0))) # Установка DATA до такта проверяет отсутствие приёма по его собственному фронту.
|
|
trace.append((index * 10 + 3, (bit, 1))) # Через два отсчёта после подготовки приходит нарастающий фронт CLK.
|
|
trace.append((index * 10 + 8, (falling_bits[index] if falling_bits is not None else bit, 0))) # Спадающий фронт может захватывать независимое тестовое значение DATA.
|
|
return ((0, initial_clock), trace) # Начальное состояние CLK позволяет проверять захват, начавшийся с высокого уровня.
|
|
|
|
|
|
def adapter(options=None, bits=(), falling_bits=None, samplerate=1000000, connected=(0, 1), initial_clock=0): # Все настройки сценария передаются до вызова decode, как в sigrok.
|
|
stub = types.ModuleType('sigrokdecode') # Тест не использует установленный на компьютере libsigrokdecode.
|
|
stub.Decoder, stub.OUTPUT_ANN, stub.OUTPUT_PYTHON, stub.SRD_CONF_SAMPLERATE = FakeDecoder, 0, 1, 2 # Числа лишь различают типы выходов и сообщение о частоте.
|
|
with patch.dict(sys.modules, {'sigrokdecode': stub, 'common.setgui_decoders.sigma_delta': CORE}): # Явная подмена ядра не затрагивает модули других тестов DSView.
|
|
module = load('sigma_delta_adapter', 'tools/dsview/decoders/sigma_delta/pd.py') # Загружаем настоящий адаптер API версии 3.
|
|
decoder = module.Decoder() # Конструктор сбрасывает метаданные перед новым захватом.
|
|
decoder.options = {item['id']: item['default'] for item in decoder.options} # Начальные настройки берутся из декларации адаптера.
|
|
decoder.options.update(options or {}) # Сценарий переопределяет только проверяемые параметры.
|
|
decoder.connected = connected # Подключённые каналы учитываются штатным методом has_channel.
|
|
initial, trace = capture(bits, falling_bits, initial_clock) # Преобразуем биты в последовательность физических состояний входов.
|
|
decoder.last_pins, decoder.trace = initial, iter(trace) # Начальный уровень не возвращается как выбранный тактовый фронт.
|
|
decoder.emitted, decoder.conditions = [], [] # Каждый захват получает отдельные журналы результатов и ожиданий.
|
|
if samplerate is not None: # В отдельном сценарии частота захвата намеренно неизвестна.
|
|
decoder.metadata(2, samplerate) # При десяти отсчётах на такт CLK равен samplerate / 10.
|
|
decoder.start() # Регистрируем оба выхода до передачи первых фронтов.
|
|
return decoder # Сценарий сам выбирает ожидаемое завершение или ошибку конфигурации.
|
|
|
|
|
|
def samples(decoder): # Выделяем только структурированные выходные отсчёты среди текстовых аннотаций.
|
|
return [event for event in decoder.emitted if event[2] == 1] # Дескриптор один соответствует OUTPUT_PYTHON тестового движка.
|
|
|
|
|
|
class SigmaDeltaTests(unittest.TestCase): # Независимые ожидаемые результаты выявляют ошибки порядка каскадов и децимации.
|
|
def test_cic_matches_independent_fir_convolution(self): # Каскад скользящих сумм задаёт эталон без интеграторов и гребенчатых разностей.
|
|
rng = random.Random(271828) # Постоянное зерно делает ошибку воспроизводимой на любой платформе.
|
|
bits = [rng.randrange(2) for _ in range(5000)] # Длинный поток многократно переполняет ограниченные регистры CIC.
|
|
for order in (1, 2, 3): # Проверяются все поддерживаемые sinc-фильтры.
|
|
for ratio in (1, 2, 3, 7, 16, 256): # Нечётные OSR выявляют скрытую зависимость от степени двойки.
|
|
with self.subTest(order=order, ratio=ratio): # Диагностика указывает режим несовпавшего фильтра.
|
|
taps = [1] # Импульсная характеристика до первого каскада является единичным импульсом.
|
|
for _ in range(order): # Сворачиваем N прямоугольных окон длины OSR.
|
|
taps = [sum(taps[max(0, index - ratio + 1):min(len(taps), index + 1)]) for index in range(len(taps) + ratio - 1)] # Коэффициенты строятся прямой FIR-свёрткой без алгоритма проверяемого ядра.
|
|
state = CORE.SigmaDeltaFilter(ratio, order) # Каждый режим начинает с чистых регистров.
|
|
for index, bit in enumerate(bits): # Фильтру передаются те же биты, что использует независимый эталон.
|
|
event = state.feed(bit) # До границы выходного блока результат отсутствует.
|
|
ready = (index + 1) % ratio == 0 and index + 1 >= order * ratio # Ожидаемые границы учитывают децимацию и начальное установление.
|
|
self.assertEqual(event is not None, ready) # Проверяем также точное число пропущенных и выданных результатов.
|
|
if ready: # Прямую свёртку вычисляем только на установившихся выходных отсчётах.
|
|
weighted = sum(tap * bits[index - lag] for lag, tap in enumerate(taps)) # FIR использует входную историю и не зависит от внутренних регистров CIC.
|
|
self.assertEqual(event['code'], 2 * weighted - sum(taps)) # Целочисленный код обязан совпадать точно, без допуска округления.
|
|
self.assertEqual(event['density'], weighted / ratio ** order) # Проверяется нормировка фильтрованной плотности единиц.
|
|
self.assertEqual(event['value'], event['code'] / ratio ** order) # Нормированная амплитуда должна сохранять тот же масштаб.
|
|
|
|
def test_dc_extremes_zero_and_inversion(self): # Контрольные плотности проверяют знак и физически понятные крайние значения.
|
|
for order in (1, 2, 3): # Нулевая предыстория не должна смещать постоянный сигнал после установления.
|
|
for pattern, expected in (([0], -1), ([1], 1), ([0, 1], 0), ([0, 1, 1, 1], 0.5)): # Плотности 0, 100, 50 и 75 процентов имеют известную нормированную амплитуду.
|
|
for invert in (False, True): # Инверсия должна менять знак при любом порядке фильтра.
|
|
state = CORE.SigmaDeltaFilter(8, order, invert) # OSR восемь содержит целое число периодов всех контрольных шаблонов.
|
|
results = [event['value'] for bit in pattern * 64 for event in [state.feed(bit)] if event is not None] # Собираем только завершённые и установившиеся отсчёты.
|
|
self.assertTrue(results) # Пустой выход не может считаться успешной проверкой постоянного сигнала.
|
|
self.assertEqual(set(results), {expected * (-1 if invert else 1)}) # Каждый выходной отсчёт обязан иметь правильный знак и масштаб.
|
|
|
|
def test_invalid_filter_settings_and_logic_levels(self): # Ошибочные настройки должны отклоняться до появления числового результата.
|
|
for ratio in (0, -1, 1.5, True, 65537): # Проверяются обе границы диапазона и недопустимые типы OSR.
|
|
with self.assertRaises(ValueError): # Тихое округление OSR меняло бы частоту и усиление фильтра.
|
|
CORE.SigmaDeltaFilter(ratio) # Используется реальный конструктор ядра.
|
|
for order in (0, 4, 1.5, True): # Разрешены только целые порядки один, два и три.
|
|
with self.assertRaises(ValueError): # Неизвестный порядок не должен заменяться начальным значением.
|
|
CORE.SigmaDeltaFilter(order=order) # Проверка независима от адаптера DSView.
|
|
for bit in (-1, 2, 255, None): # Значение 255 моделирует неподключённый цифровой вход sigrok.
|
|
with self.assertRaises(ValueError): # Невалидный вход нельзя включать в статистику единиц.
|
|
CORE.SigmaDeltaFilter().feed(bit) # Ошибка обнаруживается на первом неправильном уровне.
|
|
|
|
def test_adapter_edges_rates_voltage_and_partial_tail(self): # Проверяется полный путь от переходов CLK до аннотаций и Python-пакетов.
|
|
for edge in ('rising', 'falling'): # Разные значения на двух фронтах выявляют выбор неверной полярности CLK.
|
|
decoder = adapter({'filter': 'sinc1', 'decimation': 4, 'clock_edge': edge, 'full_scale': 2.5}, [1] * 11, [0] * 11) # Одиннадцать битов содержат два полных блока и неполный хвост из трёх битов.
|
|
with self.assertRaises(CaptureEnd): # Реальный decode завершается только по концу трассы.
|
|
decoder.decode() # Все числовые результаты формирует адаптер, а не тестовый фильтр.
|
|
output = samples(decoder) # Текстовые строки анализируются отдельно от структурированного канала.
|
|
self.assertEqual(len(output), 2) # Последние три бита не должны создавать третий отсчёт.
|
|
for start, end, _, packet in output: # Проверяем каждый выданный блок, включая первый после начала записи.
|
|
self.assertEqual(end - start, 30) # Четыре фронта CLK охватывают три периода по десять отсчётов.
|
|
self.assertEqual(packet[0], 'SAMPLE') # Тип пакета стабилен для последующего стекового декодера.
|
|
self.assertEqual(packet[1]['value'], 1 if edge == 'rising' else -1) # DATA на спаде намеренно противоположна уровню на нарастающем фронте.
|
|
self.assertEqual(packet[1]['voltage'], 2.5 if edge == 'rising' else -2.5) # Физический масштаб должен умножать уже восстановленную амплитуду.
|
|
self.assertEqual(packet[1]['clock_hz'], 100000) # Мегагерц захвата и десять отсчётов на период означают CLK 100 кГц.
|
|
self.assertEqual(packet[1]['output_hz'], 25000) # OSR четыре уменьшает частоту результата до 25 кГц.
|
|
self.assertEqual({tuple(condition.items()) for condition in decoder.conditions}, {((1, 'r' if edge == 'rising' else 'f'),)}) # Адаптер должен ждать только выбранный фронт второго входа.
|
|
self.assertEqual(len(decoder.emitted), 10) # Два результата содержат по четыре аннотации и одному Python-пакету.
|
|
|
|
def test_adapter_warmup_missing_samplerate_and_initial_high_clock(self): # Начальный высокий CLK не считается фронтом, а отсутствие герц не мешает фильтрации.
|
|
decoder = adapter({'decimation': 4}, [1] * 15, samplerate=None, initial_clock=1) # sinc3 пропускает два первых блока и имеет один полный установленный результат.
|
|
with self.assertRaises(CaptureEnd): # Пятнадцатый бит оставляет незавершённый четвёртый блок.
|
|
decoder.decode() # Конец захвата не должен дополнять его шестнадцатым битом.
|
|
output = samples(decoder) # Результаты должны существовать без метаданных частоты.
|
|
self.assertEqual(len(output), 1) # Выдаётся только третий полный блок после двух блоков установления.
|
|
self.assertEqual(output[0][:2], (83, 113)) # Третий блок содержит реальные нарастающие фронты девятого–двенадцатого битов.
|
|
self.assertEqual(output[0][3][1]['value'], 1) # Начальное установление не должно создавать отрицательное смещение постоянного DATA.
|
|
self.assertIsNone(output[0][3][1]['clock_hz']) # Без частоты захвата нельзя присваивать CLK вымышленное значение.
|
|
self.assertEqual({event[3][0] for event in decoder.emitted if event[2] == 0}, {0, 1, 2}) # Три амплитудные строки остаются, строка частоты отсутствует.
|
|
|
|
def test_single_bit_decimation_and_reset(self): # OSR один проверяет отсутствие деления на ноль и границы точечных аннотаций.
|
|
decoder = adapter({'filter': 'sinc1', 'decimation': 1, 'invert': 'yes'}, [0, 1]) # Инверсия должна действовать до формирования нормированного кода.
|
|
with self.assertRaises(CaptureEnd): # Два фронта дают два результата и затем штатный конец захвата.
|
|
decoder.decode() # Первая частота неизвестна, вторая измеряется по предыдущему фронту.
|
|
output = samples(decoder) # В режиме OSR один каждое принятое значение создаёт пакет.
|
|
self.assertEqual([event[3][1]['value'] for event in output], [1, -1]) # Получаем инвертированные уровни в биполярном диапазоне.
|
|
self.assertIsNone(output[0][3][1]['clock_hz']) # Один фронт не позволяет измерить период.
|
|
self.assertEqual(output[1][3][1]['output_hz'], 100000) # При OSR один выходная частота совпадает с CLK.
|
|
self.assertTrue(all(start == end for start, end, _, _ in output)) # Точечный блок не получает искусственно удлинённый временной интервал.
|
|
decoder.reset() # Повторное использование объекта начинает новый захват без старых метаданных.
|
|
self.assertIsNone(decoder.samplerate) # Частота предыдущего прибора не должна переноситься в новый захват.
|
|
|
|
def test_adapter_rejects_missing_channels_and_invalid_options(self): # Некорректная конфигурация не должна дожидаться первого такта.
|
|
for connected in ((0,), (1,), ()): # Поочерёдно отсутствуют CLK, DATA и оба входа.
|
|
with self.assertRaises(ValueError): # Требование обязательных каналов проверяет сам адаптер.
|
|
adapter(connected=connected).decode() # Пустая трасса отличает раннюю проверку от случайного успешного чтения.
|
|
for options in ({'full_scale': 0}, {'full_scale': -1}, {'full_scale': float('nan')}, {'full_scale': float('inf')}, {'filter': 'sinc4'}, {'clock_edge': 'both'}, {'invert': 'maybe'}, {'decimation': 2.5}): # Ошибки охватывают масштаб, перечисления и тип OSR.
|
|
with self.assertRaises(ValueError): # Ни одна ошибка не должна превращаться в CaptureEnd или числовой отсчёт.
|
|
adapter(options).decode() # Проверяются фактические параметры нового декодера.
|
|
for rate in (0, -1, float('nan'), float('inf')): # Частота захвата задаётся только положительным конечным числом.
|
|
with self.assertRaises(ValueError): # Метаданные проверяются до расчёта частоты CLK.
|
|
adapter(samplerate=rate) # Неверная частота не должна попадать в выходной пакет.
|
|
|
|
def test_packaged_adapter_imports_in_isolated_python(self): # Ведомость должна включать адаптер и каноническое ядро без установленного SETGUI.
|
|
build = load('sigma_delta_packager', 'tools/dsview/build.py') # Используем штатную упаковку встроенного DSView.
|
|
with tempfile.TemporaryDirectory() as temporary: # Каталог содержит только подготовленную поставку, без внешних зависимостей.
|
|
root = Path(temporary) # Все файлы проверки ограничены временной директорией.
|
|
build.package_decoders(root) # Общий рецепт копирует адаптеры и ядра из protocol-files.json.
|
|
script = '\n'.join(( # Изолированный процесс не наследует sys.path и подмены модулей текущих тестов.
|
|
'import importlib, pathlib, sys, types # Для импорта нужны только стандартные модули Python.', # Каждая строка дочернего сценария также документирована.
|
|
'root = pathlib.Path(sys.argv[1]) # Корень упакованных декодеров передаётся отдельным аргументом.', # Аргумент исключает зависимость от рабочего каталога.
|
|
'sys.path.insert(0, str(root)) # Разрешаем импорт только из подготовленной поставки.', # Исходники templates не добавляются в поиск.
|
|
"stub = types.ModuleType('sigrokdecode') # Моделируем только доступную при импорте базу API.", # Настоящий DSView не запускается.
|
|
'stub.Decoder = object # Методы базы при проверке импорта не вызываются.', # Выходные константы нужны только после start.
|
|
"sys.modules['sigrokdecode'] = stub # Подмена ограничена дочерним процессом.", # Глобальное окружение пользователя не изменяется.
|
|
"adapter = importlib.import_module('sigma_delta') # Проверяем стандартную точку входа пакета.", # Импорт обязан транзитивно загрузить ядро из common.
|
|
"core = importlib.import_module('common.setgui_decoders.sigma_delta') # Ядро должно входить в ту же поставку.", # Проверяем отсутствие скрытой зависимости от рабочего checkout.
|
|
'pathlib.Path(core.__file__).resolve().relative_to(root.resolve()) # Местоположение ядра обязано принадлежать пакету.', # Неправильное происхождение вызывает ошибку.
|
|
"assert adapter.Decoder.id == 'sigma_delta' # Идентификатор совпадает с меню приложения.", # Сохранённые сессии используют тот же идентификатор.
|
|
"assert core.SigmaDeltaFilter(1, 1).feed(1)['value'] == 1 # Упакованное ядро выполняет действительное восстановление.", # Проверяется и работоспособность, и полнота импорта.
|
|
)) # В сценарии используются настоящие переводы строк, а не экранированные команды оболочки.
|
|
result = subprocess.run([sys.executable, '-I', '-c', script, str(root / 'decoders')], cwd=temporary, capture_output=True, text=True) # Отдельные аргументы сохраняют пути с пробелами и не запускают оболочку.
|
|
self.assertEqual(result.returncode, 0, result.stderr) # Отсутствующий модуль или неверное расположение ядра делают проверку неуспешной.
|
|
|
|
def test_portable_python_syntax(self): # Старые версии DSView используют Python 3.6, хотя сами тесты могут выполняться новым Python.
|
|
paths = [ROOT / 'python/logic_analyzer/decoders/sigma_delta.py'] + list((ROOT / 'tools/dsview/decoders/sigma_delta').glob('*.py')) # Проверяются оба устанавливаемых слоя и точка входа пакета.
|
|
for path in paths: # Ошибка указывает точное имя файла поставки.
|
|
ast.parse(path.read_text(encoding='utf-8'), filename=str(path), feature_version=(3, 6)) # Одновременно проверяем UTF-8 и совместимость синтаксиса.
|
|
|
|
|
|
if __name__ == '__main__': # Разрешаем запуск отдельного тестового файла без pytest и GUI.
|
|
unittest.main() # Стандартный runner выдаёт количество проверок и описание несоответствий.
|