Исходный код soniks_client.antenna.rig

"""Управление приёмным устройством через Hamlib.

``Hamlib`` — системный модуль, доступный только внутри контейнера."""

from functools import lru_cache

import Hamlib

from core.configs import logger, settings
from core.exceptions import EquipmentConnectionError




class RigController:
    """Приёмное устройство через сетевой ``rigctld`` (Hamlib).

    Используется для доплеровской коррекции частоты во время прохода.
    """

    def __init__(
        self,
        rig_ip: str,
        rig_port: int,
        model=Hamlib.RIG_MODEL_NETRIGCTL,
    ) -> None:
        Hamlib.rig_set_debug(Hamlib.RIG_DEBUG_WARN)
        self.rig = Hamlib.Rig(model)
        self.rig.set_conf("rig_pathname", f"{rig_ip}:{rig_port}")
        self._vfo: int | None = None



    def connect(self) -> None:
        """Открыть соединение с ``rigctld``.

        Raises:
            EquipmentConnectionError: Hamlib вернул код ошибки.
        """

        self.rig.open()
        self._check_hamlib_error("открытие соединения с rigctld")

        # VFO читается один раз на соединение: при RIG_UPDATE_INTERVAL=0.1
        # запрос на каждую установку частоты — это лишние 10 сетевых вызовов
        # в секунду к rigctld.
        self._vfo = self.rig.get_vfo()




    def disconnect(self) -> None:
        """Закрыть соединение с ``rigctld``."""

        self._vfo = None
        self.rig.close()


    @property
    def frequency(self) -> float:
        return self.rig.get_freq()

    @frequency.setter
    def frequency(self, frequency) -> None:
        """Текущая частота приёма, Гц."""

        self.rig.set_freq(self.vfo, frequency)

    @property
    def vfo(self) -> int:
        """Текущий VFO приёмника, прочитанный при подключении."""

        if self._vfo is None:
            self._vfo = self.rig.get_vfo()

        return self._vfo

    def _check_hamlib_error(self, operation: str) -> None:
        status = getattr(self.rig, "error_status", Hamlib.RIG_OK)

        if status != Hamlib.RIG_OK:
            logger.error(
                "Hamlib: %s не удалось, код ошибки %s (%s)",
                operation,
                status,
                Hamlib.rigerror(status),
            )
            raise EquipmentConnectionError()





@lru_cache(maxsize=1)
def get_rig_controller() -> RigController:
    """Вернуть единственный на процесс контроллер приёмника.

    Раньше контроллер создавался на импорте модуля: ошибка конфигурации
    Hamlib роняла клиент на старте трейсбеком вместо записи в лог.
    """

    return RigController(
        settings.antenna.rig.IP,
        settings.antenna.rig.PORT,
    )