"""Проверка на железе, что переключение стимула внутри частотной точки работает. Зачем отдельная проверка. Число триггерных фронтов на частотную точку НЕ различает стоковый битстрим от доработанного: поле ``stages`` в ``SweepSettings`` стоковое, и сток тоже отрабатывает четыре стадии на точку. Отличие в том, КАКИЕ стадии излучают. Сток сравнивает счётчик стадий с порт-стадией точно, поэтому при ``port1Stage = 0`` и ``port2Stage = 2`` излучают только стадии 0 и 2, а 1 и 3 остаются без стимула. Доработанный битстрим сравнивает по маске, и порт излучает во всём блоке: 0 и 1 — порт 1, 2 и 3 — порт 2. Отсюда две проверки по данным: 1. **Во всех четырёх выходах есть сигнал.** Если выходы 1 и 3 на уровне шума — в приборе стоковый битстрим, заливка FPGA не удалась. 2. **Выходы попарно совпадают, когда ВЧ-ключ не подключён.** Тогда обе позиции ключа — это физически один и тот же путь, поэтому 0 ≈ 1 и 2 ≈ 3. Расхождение при отключённом ключе означает, что по стадиям разъезжается что-то ещё. Заодно проверяется и контроллер в петле триггера: при ``syncMode = 3`` мастер семплирует по вернувшемуся триггеру, так что без исправного зеркалирования в STM32 развёртка просто не пойдёт и данных не будет вовсе. Запуск (на машине, к которой подключены приборы): .venv/bin/python -m python_app.scripts.per_point_switching_check \\ --config run_config.json --collections 3 """ from __future__ import annotations import argparse import logging import statistics from pathlib import Path import numpy as np from python_app.hardware_full.librevna_multi_device_driver.models import ( build_output_channels, normalize_master_stimulus_ports, ) from python_app.hardware_full.matrix_radar_service import create_matrix_radar_service from python_app.models.dataset_model import SweepCollection from python_app.models.run_config_model import RunConfigModel logger = logging.getLogger(__name__) # Порты мастера, которые подставляет контроллер по умолчанию в # `configure_continuous_sweep`. Держится здесь, чтобы раскладка выходов в этой # проверке считалась ровно так же, как в драйвере. MASTER_STIMULUS_PORTS = (1, 2) # Ниже этого уровня считаем, что стимула в стадии не было. Порог грубый нарочно: # разница между «излучает» и «не излучает» — это десятки дБ, а не единицы. NOISE_FLOOR_DBFS = -80.0 # Насколько выходы одной пары могут расходиться при ОТКЛЮЧЁННОМ ключе. PAIR_TOLERANCE_DB = 6.0 def magnitude_db(values: np.ndarray) -> float: """Медианный модуль трассы в дБ; медиана устойчива к отдельным выбросам.""" finite = values[np.isfinite(values)] if finite.size == 0: return float("-inf") magnitude = float(np.median(np.abs(finite))) if magnitude <= 0.0: return float("-inf") return 20.0 * float(np.log10(magnitude)) def build_switch_position_pairs(config: RunConfigModel) -> list[tuple[int, int]]: """Пары выходов «один порт мастера в двух положениях ключа». Берётся из той же модели, что и раскладка стадий, поэтому не разъедется с драйвером. Соседние индексы сравнивать НЕЛЬЗЯ: при двух портах и двух положениях выход 0 — порт 1 в A, выход 1 — порт 2 в A, выход 2 — порт 1 в B, выход 3 — порт 2 в B. Сравнивать надо (0, 2) и (1, 3). """ # Те же порты мастера, что подставляет контроллер по умолчанию # (`configure_continuous_sweep(master_stimulus_ports=(1, 2))`). channels = build_output_channels( normalize_master_stimulus_ports(MASTER_STIMULUS_PORTS), config.matrix_stimulus_switch_positions, ) by_port: dict[int, list[int]] = {} for channel in channels: by_port.setdefault(channel.stimulus_port, []).append(channel.index) pairs: list[tuple[int, int]] = [] for indices in by_port.values(): ordered = sorted(indices) pairs.extend(zip(ordered, ordered[1:])) return pairs def summarize(collection: SweepCollection) -> dict[int, list[float]]: """Сгруппировать медианные уровни s21 по номеру выхода.""" by_output: dict[int, list[float]] = {} for trace in collection.traces: by_output.setdefault(trace.combo.output, []).append(magnitude_db(trace.s21)) return by_output def main() -> int: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--config", required=True, type=Path, help="Путь к run_config.json") parser.add_argument("--collections", type=int, default=3, help="Сколько коллекций снять (по умолчанию 3)") parser.add_argument( "--switch-connected", action="store_true", help="ВЧ-ключ подключён: выходы одной пары теперь ДОЛЖНЫ различаться " "(это разные антенны), поэтому расхождение — признак работы, а не ошибки", ) args = parser.parse_args() logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s") config = RunConfigModel.load_from_path(args.config) config.apply_device_model_constraints() if not config.is_matrix_radar: raise SystemExit("нужен матричный radar.model (librevna_multi)") substages = config.matrix_stimulus_switch_positions outputs = config.output_switch.positions print(f"конфиг: stimulus_switch_positions={substages}, выходов={outputs}, " f"входов={config.input_switch.positions}, combos={len(config.combos)}, " f"driver_mode={config.radar.driver_mode}") if substages <= 1: raise SystemExit("stimulus_switch_positions = 1: проверять нечего, режим выключен") radar = create_matrix_radar_service(config) radar.open() try: radar.configure(config.radar.sweep) per_output: dict[int, list[float]] = {} for index in range(max(1, args.collections)): collection = radar.acquire_collection(collection_id=index + 1) print(f" коллекция {index + 1}: трасс {len(collection.traces)}") for output, levels in summarize(collection).items(): per_output.setdefault(output, []).extend(levels) finally: radar.close() if not per_output: raise SystemExit("данных нет — развёртка не прошла (проверьте петлю триггера и контроллер)") print("\nмедианный уровень s21 по выходам:") medians: dict[int, float] = {} for output in sorted(per_output): medians[output] = statistics.median(per_output[output]) print(f" выход {output}: {medians[output]:+7.1f} дБ (трасс {len(per_output[output])})") verdict_ok = True silent = [o for o, db in medians.items() if db < NOISE_FLOOR_DBFS] print("\n1) стимул во всех стадиях:") if silent: verdict_ok = False print(f" ПРОВАЛ: без сигнала выходы {silent}. Похоже, в мастере СТОКОВЫЙ битстрим:") print(" он излучает только в стадиях, точно равных порт-стадии (0 и 2).") else: print(" ок: сигнал есть во всех выходах, значит порт излучает во всём блоке стадий") if args.switch_connected: print("2) различие выходов внутри пары (ключ ПОДКЛЮЧЁН, ждём разные антенны):") else: print("2) попарное совпадение выходов (ключ не подключён, ждём один и тот же путь):") # Пары берутся из самой модели, а не из арифметики по индексам: сравнивать # надо ОДИН порт мастера в двух положениях ключа, а соседние индексы — это # РАЗНЫЕ порты в одном положении (выход 0 = порт 1 A, выход 1 = порт 2 A, # выход 2 = порт 1 B, выход 3 = порт 2 B). pairs = build_switch_position_pairs(config) for low, high in pairs: if low not in medians or high not in medians: continue delta = abs(medians[low] - medians[high]) differs = delta > PAIR_TOLERANCE_DB # Ключ подключён -> позиции ведут на РАЗНЫЕ антенны, совпадение значило бы, # что ключ не движется. Ключ не подключён -> путь один, ждём совпадение. good = differs if args.switch_connected else not differs if not good: verdict_ok = False if args.switch_connected: status = "ок, ключ переключается" if good else "СОВПАДАЮТ — ключ, похоже, не движется" else: status = "ок" if good else "РАСХОЖДЕНИЕ" print(f" выходы {low} и {high}: расхождение {delta:4.1f} дБ — {status}") print("\nИТОГ:", "переключение стимула внутри точки работает" if verdict_ok else "проверка НЕ пройдена, см. выше") return 0 if verdict_ok else 1 if __name__ == "__main__": raise SystemExit(main())