"""Расчёт положения спутника и радиальной скорости по TLE (Skyfield)."""
from datetime import UTC, datetime, timedelta
import numpy as np
from skyfield.api import EarthSatellite, load, wgs84
from core.configs import logger
from core.types import TLE, AzimuthsSatellitePass, StationLocation
[документация]
class SatelliteParametersCalculator:
"""Расчёт параметров пролёта по TLE и координатам станции."""
def __init__(self, station_location: StationLocation, tle: TLE) -> None:
self.ts = load.timescale()
self.satellite = EarthSatellite(tle["tle1"], tle["tle2"], tle["tle0"], self.ts)
self.station_location = wgs84.latlon(
latitude_degrees=station_location["latitude"],
longitude_degrees=station_location["longitude"],
elevation_m=station_location["elevation"],
)
[документация]
def get_position(self) -> tuple[float, float]:
"""Вернуть текущие азимут и угол места спутника в градусах."""
timestamp = datetime.now(tz=UTC)
ts = self.ts.from_datetime(timestamp)
position = (self.satellite - self.station_location).at(ts)
altitude, azimuth, _ = position.altaz()
return (
azimuth.degrees,
altitude.degrees,
)
[документация]
def get_radial_velocity(self) -> float:
"""Вернуть радиальную скорость спутника, м/с.
Положительное значение — спутник удаляется.
"""
timestamp = datetime.now(tz=UTC)
ts = self.ts.from_datetime(timestamp)
relative = self.satellite.at(ts) - self.station_location.at(ts)
pos_vector = relative.position.km
vel_vector = relative.velocity.km_per_s
if np.linalg.norm(pos_vector) == 0:
return 0
unit_direction = pos_vector / np.linalg.norm(pos_vector)
return np.dot(vel_vector, unit_direction) * 1000
[документация]
def get_azimuths_satellite_pass(
self,
start_time: datetime,
end_time: datetime,
satellite_name: str,
) -> AzimuthsSatellitePass | None:
"""Вернуть азимуты восхода, кульминации и захода для пролёта.
Окно поиска расширяется на два часа в обе стороны: сеанс мог начаться
с задержкой, и AOS уже прошло, либо LOS ещё далеко за ``end_time``.
В расширенном окне помещается несколько пролётов, поэтому берётся
целая тройка событий одного пролёта, ближайшая к окну наблюдения.
Returns:
Тройка азимутов либо ``None``, если пролёт не найден.
"""
search_start = self.ts.from_datetime(start_time - timedelta(hours=2))
search_end = self.ts.from_datetime(end_time + timedelta(hours=2))
observation_middle = start_time + (end_time - start_time) / 2
for attempt in range(3):
times, events = self.satellite.find_events(
self.station_location,
search_start,
search_end,
altitude_degrees=0,
)
try:
passes = self._group_events_into_passes(times, events)
if not passes:
raise ValueError("Не найден полный набор событий (AOS, MAX, LOS)")
# Раньше бралось первое вхождение каждого кода событий по
# отдельности: AOS мог прийти из одного пролёта, а LOS из
# следующего, и FlipStrategy получала мусорные азимуты.
aos_time, max_time, los_time = min(
passes,
key=lambda satellite_pass: abs(
(
satellite_pass[1].utc_datetime() - observation_middle
).total_seconds()
),
)
difference = self.satellite - self.station_location
aos_pos = difference.at(aos_time)
max_pos = difference.at(max_time)
los_pos = difference.at(los_time)
logger.info("Расчитаны азимуты пролёта спутника: %s", satellite_name)
return (
aos_pos.altaz()[1].degrees,
max_pos.altaz()[1].degrees,
los_pos.altaz()[1].degrees,
)
except ValueError as e:
logger.warning(
"Неполный пролёт спутника: %s, проводится новый расчет (попытка: %d), ошибка: %s",
satellite_name,
attempt + 1,
e,
)
# Если не нашли, расширяем еще на час
search_start = self.ts.from_datetime(search_start.utc_datetime() - timedelta(hours=1))
search_end = self.ts.from_datetime(search_end.utc_datetime() + timedelta(hours=1))
except Exception as e:
logger.error("Ошибка при расчете азимутов пролета: %s", e)
return None
logger.error("Не удалось определить полный пролёт спутника: %s", satellite_name)
return None
@staticmethod
def _group_events_into_passes(times, events) -> list[tuple]:
"""Собрать события Skyfield в полные пролёты «восход-кульминация-заход».
Skyfield отдаёт коды 0 (восход), 1 (кульминация) и 2 (заход) подряд по
времени. Пролёт считается полным только если все три пришли в этом
порядке; обрезанные краями окна поиска пролёты отбрасываются.
"""
passes = []
aos_time = None
max_time = None
for event_time, event in zip(times, events, strict=True):
if event == 0:
aos_time = event_time
max_time = None
elif event == 1 and aos_time is not None:
max_time = event_time
elif event == 2 and aos_time is not None and max_time is not None:
passes.append((aos_time, max_time, event_time))
aos_time = None
max_time = None
return passes