Исходный код soniks_client.sdr_check

"""Проверка приёмника перед применением конфигурации с портала.

Открывает устройство python-биндингом SoapySDR и выставляет частоту
дискретизации, антенный вход и усиление — то, что в реальном проходе делает
граф. Работает в подпроцессе с таймаутом: зависший драйвер не должен вешать
поток планировщика. Запускать только вне прохода — приёмник занят графом.

Без биндинга (старая база образа) проверка пропускается с предупреждением:
иначе на такой станции не применился бы ни один конфиг.
"""

import json
import subprocess
import sys

TIMEOUT_SECONDS = 20
NO_BINDING = 3




def check(
    device: str,
    sample_rate: float | None,
    antenna: str | None,
    gain: float | None,
    timeout: float = TIMEOUT_SECONDS,
) -> str | None:
    """Открыть приёмник с заданными параметрами.

    Returns:
        ``None``, если устройство открылось (или проверить нечем), иначе
        текст ошибки для лога и портала.
    """

    # Здесь, а не на уровне модуля: тот же файл запускается дочерним процессом,
    # которому настройки станции ни к чему.
    from core.configs import logger

    args = json.dumps(
        {"device": device, "sample_rate": sample_rate, "antenna": antenna, "gain": gain}
    )

    try:
        proc = subprocess.run(
            [sys.executable, __file__, args],
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired:
        return f"приёмник {device} не ответил за {timeout:g} с"
    except OSError as e:
        return f"проверка приёмника не запустилась: {e}"

    if proc.returncode == 0:
        return None

    if proc.returncode == NO_BINDING:
        logger.warning("Биндинг SoapySDR недоступен, проверка приёмника пропущена")
        return None

    lines = proc.stderr.strip().splitlines()

    return f"приёмник {device}: {lines[-1] if lines else f'код {proc.returncode}'}"



def _probe(args: dict) -> None:
    try:
        import SoapySDR
    except ImportError:
        sys.exit(NO_BINDING)

    try:
        device = SoapySDR.Device(args["device"])
        rx = SoapySDR.SOAPY_SDR_RX

        if args.get("sample_rate") is not None:
            device.setSampleRate(rx, 0, float(args["sample_rate"]))
        if args.get("antenna"):
            device.setAntenna(rx, 0, args["antenna"])
        if args.get("gain") is not None:
            device.setGain(rx, 0, float(args["gain"]))
    except Exception as e:
        print(e, file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    _probe(json.loads(sys.argv[1]))