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

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

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

from functools import lru_cache

import Hamlib

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




class RotatorController:
    """Поворотное устройство через Hamlib.

    Создаётся только при ``ANTENNA__ROTATOR__ENABLED``; иначе
    :func:`get_rotator_controller` возвращает ``None``.
    """

    def __init__(self, model: str, baud: int, port: str):
        self.rot_port = port
        self.rot_baud = baud
        self.rot_name = getattr(Hamlib, model)
        self.rot = Hamlib.Rot(self.rot_name)
        self.rot.state.rotport.pathname = self.rot_port
        self.rot.state.rotport.parm.serial.rate = self.rot_baud



    def connect(self) -> None:
        """Открыть соединение с поворотным устройством.

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

        self.rot.open()

        status = getattr(self.rot, "error_status", Hamlib.RIG_OK)

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




    def disconnect(self) -> None:
        """Закрыть соединение с поворотным устройством."""

        self.rot.close()


    @property
    def position(self) -> list[float]:
        return self.rot.get_position()

    @position.setter
    def position(self, pos: tuple[float, float]) -> None:
        """Положение антенны как ``[азимут, угол места]`` в градусах."""

        self.rot.set_position(*pos)





@lru_cache(maxsize=1)
def get_rotator_controller() -> RotatorController | None:
    """Вернуть единственный на процесс контроллер ротатора либо ``None``.

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

    if not settings.antenna.rotator.ENABLED:
        return None

    rotator = settings.antenna.rotator

    try:
        return RotatorController(rotator.MODEL, rotator.BAUD, rotator.PORT)
    except AttributeError:
        logger.error(
            "Hamlib не знает модели ротатора %r, слежение отключено. "
            "Проверьте ANTENNA__ROTATOR__MODEL",
            rotator.MODEL,
        )

        return None