radar_system/python_app/scripts/per_point_switching_check.py

"""Проверка на железе, что переключение стимула внутри частотной точки работает.

Зачем отдельная проверка. Число триггерных фронтов на частотную точку НЕ
различает стоковый битстрим от доработанного: поле ``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())