"""Фоновые сессии связи: слежение ротатора и доплеровская коррекция частоты."""
import threading
from abc import ABC, abstractmethod
from datetime import datetime
from typing import Generic, TypeVar
from core.configs import logger, settings
from core.exceptions import NoSessionParametersError, RotatorPositionError
from core.types import TLE, StationLocation
from soniks_client.antenna.rig import RigController
from soniks_client.antenna.rotator import RotatorController
from soniks_client.antenna.satellite_parameters import SatelliteParametersCalculator
from soniks_client.antenna.tracking import create_tracking_strategy
from soniks_client.antenna.tracking.base import TrackingStrategy
T = TypeVar("T")
[документация]
class CommunicationSessionBase(Generic[T], ABC):
"""Фоновая сессия, гоняющая расчётные значения в железо во время прохода.
Потомок задаёт, что именно считать и куда отправлять; базовый класс
ведёт поток и его жизненный цикл.
"""
def __init__(self, update_interval: float) -> None:
self.update_interval = update_interval
self.station_location: StationLocation | None = None
self.tle: TLE | None = None
self.satellite_name: str | None = None
self._session_thread: threading.Thread | None = None
self._session_thread_active = threading.Event()
[документация]
def set_session_parameters(
self,
station_location: StationLocation,
tle: TLE,
) -> bool:
"""Задать координаты станции и TLE. Во время активной сессии игнорируется.
Returns:
``False``, если сессия уже активна и параметры не приняты.
Потомок обязан проверить результат: иначе он перестраивает своё
состояние на живой сессии в обход этой защиты.
"""
if self._session_thread_active.is_set():
logger.error(
"Нельзя изменить параметры наблюдения во время активного сеанса"
)
return False
self.station_location = station_location
self.tle = tle
self.satellite_name = self._get_satellite_name()
return True
[документация]
def start_session(self) -> None:
"""Запустить фоновый поток сессии.
Raises:
NoSessionParametersError: Параметры пролёта не заданы.
"""
if self._session_thread and self._session_thread.is_alive():
return
if not all([self.station_location, self.tle]):
raise NoSessionParametersError()
self._session_thread_active.clear()
self._session_thread = threading.Thread(
target=self._streaming_satellite_parameters_to_equipment,
daemon=True,
)
self._session_thread.start()
[документация]
def stop_session(self) -> None:
"""Остановить фоновый поток сессии и дождаться его завершения."""
if not self._session_thread or not self._session_thread.is_alive():
return
self._session_thread_active.set()
self._session_thread.join(timeout=5.0)
if self._session_thread.is_alive():
logger.error(
"Не удалось остановить поток сеанса связи (таймаут 5 сек). "
"Поток продолжает работать с оборудованием"
)
# Ссылка сбрасывается в любом случае: иначе объект сессии остаётся
# навсегда мёртвым — start_session() молча выходит, потому что видит
# живой поток, и доплер больше не корректируется.
self._session_thread = None
def _get_satellite_name(self) -> str:
tle0 = self.tle["tle0"]
return tle0.removeprefix("0 ")
@abstractmethod
def _streaming_satellite_parameters_to_equipment(self) -> None:
pass
@abstractmethod
def _send_message_to_equipment(self, satellite_parameters: T) -> None:
pass
[документация]
class RotatorTrackingSession(CommunicationSessionBase[tuple[float, float]]):
"""Сессия слежения поворотного устройства за спутником.
Применяет стратегию из ``ANTENNA__ROTATOR__MODE`` и двигает антенну,
только если рассогласование превысило ``ANTENNA__ROTATOR__THRESHOLD``.
По завершении всегда возвращает антенну в исходное положение.
"""
def __init__(
self,
update_interval: float,
rotator: RotatorController,
) -> None:
super().__init__(update_interval)
self._satellite_parameters: SatelliteParametersCalculator | None = None
self._strategy: TrackingStrategy = create_tracking_strategy()
self.rotator = rotator
[документация]
def set_session_parameters(
self,
station_location: StationLocation,
tle: TLE,
start_time: datetime | None = None,
end_time: datetime | None = None,
) -> bool:
if not super().set_session_parameters(station_location, tle):
return False
self._satellite_parameters = SatelliteParametersCalculator(
self.station_location, self.tle,
)
if start_time and end_time:
azimuths = self._satellite_parameters.get_azimuths_satellite_pass(
start_time, end_time, self.satellite_name,
)
if azimuths:
self._strategy.prepare_pass(*azimuths)
else:
logger.warning(
"Азимуты пролёта не получены, "
"стратегия использует значения по умолчанию"
)
self._strategy.prepare_pass(0.0, 0.0, 0.0)
else:
self._strategy.prepare_pass(0.0, 0.0, 0.0)
return True
def _is_valid_position(self, az: float, el: float) -> bool:
return (-360.0 <= az <= 360.0) and (-90.0 <= el <= 180.0)
def _streaming_satellite_parameters_to_equipment(self) -> None:
# Подключение вне try оставляло поток умирать молча, а finally с
# парковкой и disconnect не выполнялся вовсе.
try:
self.rotator.connect()
except Exception as e:
logger.error("Не удалось подключиться к ротатору: %s", e)
return
try:
logger.info("[Rotator] Начало отслеживания: %s", self.satellite_name)
# 1. Ожидание "просыпания" rotctld (Hamlib отдаёт мусор при старте)
cur_az, cur_alt = 0.0, 0.0
is_valid = False
for attempt in range(10):
if self._session_thread_active.is_set():
return
try:
cur_az, cur_alt = self.rotator.position
if self._is_valid_position(cur_az, cur_alt):
is_valid = True
break
except Exception as e:
# Молчаливый pass прятал здесь любую ошибку Hamlib
# на все 10 секунд ожидания.
logger.warning(
"Ротатор не ответил (попытка %d/10): %s", attempt + 1, e
)
logger.warning(
"Ожидание корректного ответа от ротатора (попытка %d/10)...",
attempt + 1
)
self._session_thread_active.wait(1.0)
if not is_valid:
logger.error("Ротатор не вернул корректное положение. Старт с координат 0,0.")
cur_az, cur_alt = 0.0, 0.0
self._log_starting_position(cur_az, cur_alt)
# Сохраняем последние валидные координаты
last_valid_az, last_valid_alt = cur_az, cur_alt
# 2. Основной цикл сопровождения
while not self._session_thread_active.is_set():
sat_az, sat_alt = self._satellite_parameters.get_position()
if sat_alt < 0:
self._session_thread_active.wait(self.update_interval)
continue
try:
cur_az, cur_alt = self.rotator.position
if not self._is_valid_position(cur_az, cur_alt):
logger.warning(
"Получен мусор от ротатора (AZ=%.2f, EL=%.2f). Использую последние известные координаты.",
cur_az, cur_alt
)
cur_az, cur_alt = last_valid_az, last_valid_alt
else:
last_valid_az, last_valid_alt = cur_az, cur_alt
except Exception as e:
raise RotatorPositionError() from e
target_az, target_alt = self._strategy.transform_target(
sat_az, sat_alt, cur_az, cur_alt
)
az_delta, alt_delta = self._strategy.calculate_delta(
target_az, target_alt, cur_az, cur_alt
)
if self._strategy.needs_movement(az_delta, alt_delta):
self.rotator.position = (target_az, target_alt)
logger.info(
"Rotator перемещен на AZ=%.2f°, EL=%.2f°",
target_az, target_alt
)
# Обновляем последние известные, так как мы только что отправили команду
last_valid_az, last_valid_alt = target_az, target_alt
self._session_thread_active.wait(self.update_interval)
logger.info("[Rotator] Завершение: %s", self.satellite_name)
except RotatorPositionError as e:
logger.error("Не удалось получить позицию rotator: %s", e)
except Exception as e:
logger.error("Ошибка в сеансе: %s", e)
finally:
# Обязательная парковка в позицию 0.0, 0.0 после окончания наблюдения
logger.info("Парковка ротатора в позицию (0.0, 0.0)...")
try:
self.rotator.position = (0.0, 0.0)
except Exception as e:
logger.error("Ошибка при парковке ротатора: %s", e)
self.rotator.disconnect()
def _send_message_to_equipment(self, satellite_parameters: tuple[float, float]) -> None:
try:
self.rotator.position = satellite_parameters
logger.info("Rotator AZ=%.2f°, EL=%.2f°", *satellite_parameters)
except Exception as e:
logger.exception("Ошибка отправки позиции rotator: %s", e)
def _log_starting_position(self, az: float, el: float) -> None:
logger.info(
"Начальное положение rotator AZ=%.2f°, EL=%.2f°, режим=%s",
az, el, settings.antenna.rotator.MODE.value,
)
[документация]
class RigRXDopplerCorrectedSession(CommunicationSessionBase[int]):
"""Сессия доплеровской коррекции частоты приёма.
Считает радиальную скорость спутника и отдаёт скорректированную частоту
демону ``rigctld``.
"""
_SPEED_OF_LIGHT = 299792458
def __init__(
self,
update_interval: float,
rig: RigController,
) -> None:
super().__init__(update_interval)
self._satellite_parameters: SatelliteParametersCalculator | None = None
self._frequency: int | None = None
self.rig = rig
[документация]
def set_session_parameters(
self,
station_location: StationLocation,
tle: TLE,
frequency: int,
) -> bool:
if not super().set_session_parameters(station_location, tle):
return False
self._satellite_parameters = SatelliteParametersCalculator(
self.station_location,
self.tle,
)
self._frequency = frequency
return True
def _streaming_satellite_parameters_to_equipment(self) -> None:
# Подключение вне try оставляло поток умирать молча, а finally с
# disconnect не выполнялся вовсе.
try:
self.rig.connect()
except Exception as e:
logger.error("Не удалось подключиться к rigctld: %s", e)
return
try:
logger.info("[Rig] Начало приема сигнала спутника: %s", self.satellite_name)
logger.debug("Начальная частота спутника: %d Гц", self._frequency)
while not self._session_thread_active.is_set():
radial_velocity = self._satellite_parameters.get_radial_velocity()
adjusted_freq = self._calculate_doppler_shift(radial_velocity)
self._send_message_to_equipment(adjusted_freq)
self._session_thread_active.wait(self.update_interval)
logger.info(
"[Rig] Завершение приема сигнала спутника: %s", self.satellite_name
)
except Exception as e:
logger.error("Ошибка в процессе сеанса связи: %s", e)
finally:
self.rig.disconnect()
def _send_message_to_equipment(
self,
satellite_parameters: int,
) -> None:
try:
self.rig.frequency = satellite_parameters
logger.debug("Установлена частота rig: %d Гц", satellite_parameters)
except Exception as e:
logger.exception("Ошибка при установке частоты rig: %s", e)
def _calculate_doppler_shift(self, radial_velocity: float) -> int:
return int(self._frequency * (1 - radial_velocity / self._SPEED_OF_LIGHT))