"""Определения спутников с портала: satyaml и ``sat.cfg`` одним архивом.
Раньше это делал ``liveupdate-satyaml.sh``: при каждом старте контейнера он
клонировал gr-satellites и soniks-satyaml из git. В закрытой сети, где открыт
только ``sonik.space``, клон падал и станция молча оставалась на определениях
из образа — то есть обновиться не могла никак. В открытой сети было не лучше:
две станции с одним тегом образа получали разный satyaml, смотря когда их
перезапускали.
Теперь источник один — портал, архив версионирован, а версия уезжает в
``client_metadata`` вместе с наблюдением. Дальше по важности:
* **ничто здесь не имеет права уронить старт.** Портал лежит, архив битый,
хеш не сошёлся — предупреждение в лог и работа на том, что уже есть;
* **скачивание и применение разделены.** Кэш живёт на постоянном томе
(``PATHS__BASE``), а каталог satyaml внутри пакета ``satellites`` — в
эфемерной файловой системе контейнера. Значит применять кэш надо на каждом
старте, а не только когда приехала новая версия.
"""
import hashlib
import re
import shutil
import tarfile
import tempfile
from pathlib import Path
import requests
from core.configs import logger, settings
# Имена внутри архива. Всё остальное игнорируется: архив приезжает по HTTPS с
# нашего же портала и сверяется по sha256, но распаковывать что попало из
# сетевого источника всё равно нельзя.
SATYAML_DIR_NAME = "satyaml"
SATCFG_NAME = "sat.cfg"
_CACHE_DIR_NAME = "sat-data"
# Единственный выбор декодера в клиенте (Фаза 3.2 дорожной карты сети):
# есть satyaml для NORAD — кадры разбирает gr-satellites, нет — только
# satnogs-граф по режиму. Оба значения уезжают в ``client_metadata`` и
# в ``SONIKS_DECODER`` скриптов прохода.
DECODER_GR_SATELLITES = "gr-satellites"
DECODER_SATNOGS = "satnogs"
_NORAD_FIELD = re.compile(r"^norad:\s*(\d+)", re.MULTILINE)
# Первый передатчик satyaml: по его модуляции клиент выбирает satnogs-граф,
# который даст gr-satellites поток IQ, когда режим портала клиенту неизвестен
# (Фаза 3.5). Regex, а не YAML: PyYAML в зависимостях клиента нет, а нужны два
# поля из сотен файлов.
_MODULATION_FIELD = re.compile(r"^\s+modulation:\s*(.+?)\s*$", re.MULTILINE)
_BAUDRATE_FIELD = re.compile(r"^\s+baudrate:\s*(\d+)", re.MULTILINE)
_CURRENT_DIR_NAME = "current"
_VERSION_FILE_NAME = "version"
# Версия, применённая на этом старте. Читается ``Flowgraph.get_metadata()`` и
# уезжает на портал: по наблюдению должно быть видно, какими определениями оно
# принято, — тем же приёмом, что и коммит флоуграфов.
_applied_version: str | None = None
[документация]
def applied_version() -> str | None:
"""Версия применённого bundle. ``None`` — станция на данных из образа."""
return _applied_version
[документация]
def sync_sat_data() -> str | None:
"""Обновить определения спутников и применить их к текущему контейнеру.
Returns:
Версия применённого bundle либо ``None``, если применять нечего и
станция работает на определениях из образа.
"""
global _applied_version
version = _cached_version()
try:
latest = _fetch_latest()
if latest and latest["version"] != version:
_download(latest)
version = latest["version"]
except Exception as e:
logger.warning("Не удалось обновить определения спутников: %s", e)
if version is None:
return None
try:
_apply(_current_path())
except Exception as e:
logger.warning("Не удалось применить определения спутников: %s", e)
return None
logger.info("Определения спутников применены, версия bundle %s", version)
_applied_version = version
return version
[документация]
def resync_sat_data() -> None:
"""Периодически подтянуть bundle, не дожидаясь рестарта контейнера.
Раньше новый спутник доезжал до работающей станции только с перезапуском:
``sync_sat_data()`` звался один раз из ``main()``. Во время прохода
раскладка откладывается — ``gr_satellites`` и выбор декодера читают тот
же каталог. Новая версия меняет список спутников станции, и статус на
портале публикуется заново.
"""
from soniks_client import api, health
try:
latest = _fetch_latest()
except Exception as e:
logger.warning("Не удалось проверить определения спутников: %s", e)
return
# Версия та же, что уже разложена в этом процессе, — делать нечего.
if latest is None or latest["version"] == _applied_version:
return
if health.seconds_to_next_observation() == 0.0:
logger.debug("Идёт проход: обновление определений спутников отложено")
return
if sync_sat_data() == latest["version"]:
api.publish_station_status()
def _cache_path() -> Path:
return settings.paths.base_path / _CACHE_DIR_NAME
def _current_path() -> Path:
return _cache_path() / _CURRENT_DIR_NAME
def _version_path() -> Path:
return _cache_path() / _VERSION_FILE_NAME
def _cached_version() -> str | None:
"""Версия bundle, лежащего в кэше, либо ``None``."""
version_file = _version_path()
if not version_file.is_file() or not _current_path().is_dir():
return None
return version_file.read_text().strip() or None
def _fetch_latest() -> dict | None:
"""Спросить портал о свежайшем bundle.
Портал без единого опубликованного bundle отвечает 404 — это не ошибка, а
состояние сети до первой публикации.
"""
# Локальный импорт: ``api`` тянет модели и настройки обмена, а этот модуль
# зовётся из ``main()`` раньше всего остального.
from soniks_client.api import _session
# Слеш на конце обязателен: без него Django отвечает редиректом.
response = _session.get(
settings.url.bundles_url + "latest/",
timeout=settings.api.TIMEOUT_IN_SECONDS_REQUEST_DATA,
)
if response.status_code == requests.codes.not_found:
logger.debug("На портале нет ни одного bundle с определениями спутников")
return None
response.raise_for_status()
return response.json()
def _download(latest: dict) -> None:
"""Скачать архив, сверить хеш и заменить им содержимое кэша."""
from soniks_client.api import _session
with tempfile.TemporaryDirectory(dir=_cache_parent()) as work_dir:
archive = Path(work_dir) / "bundle.tar.gz"
digest = hashlib.sha256()
with _session.get(
latest["url"],
timeout=settings.api.TIMEOUT_IN_SECONDS_SEND_DATA,
stream=True,
) as response:
response.raise_for_status()
with archive.open("wb") as file:
for chunk in response.iter_content(chunk_size=64 * 1024):
digest.update(chunk)
file.write(chunk)
if digest.hexdigest() != latest["sha256"]:
raise ValueError(
"хеш скачанного bundle {} не совпал с объявленным".format(
latest["version"]
)
)
unpacked = Path(work_dir) / "unpacked"
with tarfile.open(archive) as tar:
tar.extractall(unpacked, members=_wanted_members(tar))
# Замена целиком, а не по файлу: половина старого и половина нового
# bundle — состояние, которое никакая версия не описывает.
current = _current_path()
shutil.rmtree(current, ignore_errors=True)
shutil.move(str(unpacked), str(current))
_version_path().write_text(latest["version"])
logger.info("Скачан bundle с определениями спутников, версия %s", latest["version"])
def _cache_parent() -> Path:
"""Каталог кэша, гарантированно существующий.
Временная директория создаётся рядом с кэшем, а не в ``/tmp``: там tmpfs, а
``shutil.move`` через границу файловых систем перестаёт быть атомарным
переименованием и превращается в копирование сотен файлов.
"""
cache = _cache_path()
cache.mkdir(parents=True, exist_ok=True)
return cache
def _wanted_members(tar: tarfile.TarFile):
"""Только обычные файлы ``satyaml/*`` и ``sat.cfg``, без путей наружу."""
for member in tar:
parts = Path(member.name).parts
if not member.isfile() or ".." in parts or member.name.startswith("/"):
continue
if parts[0] in (SATYAML_DIR_NAME, SATCFG_NAME):
yield member
def _apply(current: Path) -> None:
"""Разложить содержимое кэша по местам, где его ищет gr-satellites."""
_apply_satyaml(current / SATYAML_DIR_NAME)
_apply_satcfg(current / SATCFG_NAME)
def _apply_satyaml(source: Path) -> None:
"""Скопировать определения в каталог satyaml пакета ``satellites``.
Путь спрашивается у самого пакета: захардкоженный
``/usr/local/lib/python3.11/dist-packages`` превращался в тихий no-op при
любой смене версии Python в базовом образе.
"""
import satellites
target = Path(satellites.__file__).parent / SATYAML_DIR_NAME
target.mkdir(parents=True, exist_ok=True)
for definition in sorted(source.glob("*")):
if definition.is_file():
shutil.copy2(definition, target / definition.name)
def _apply_satcfg(source: Path) -> None:
"""Положить ``sat.cfg`` туда, где его ищет ``gr_satellites``.
Правки оператора не затираются: файл переписывается, только если он
совпадает с тем, что доставил прошлый bundle. Иначе он остаётся как есть, а
расхождение попадает в лог — молчать здесь нельзя ни в ту, ни в другую
сторону.
"""
if not source.is_file():
return
target = Path.home() / ".gr_satellites" / SATCFG_NAME
if target.is_file() and target.read_bytes() != source.read_bytes():
logger.warning(
"%s отличается от файла из bundle и оставлен как есть; "
"удалите его, чтобы принять версию из bundle",
target,
)
return
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, target)
[документация]
def decoder_for(norad: int | None) -> str:
"""Выбрать декодер прохода по наличию satyaml для спутника.
Смотрит в каталог пакета ``satellites`` — тот же, куда ``_apply_satyaml``
кладёт bundle и где лежат штатные определения gr-satellites. Индекс
строится на каждом вызове: файлов сотни, а кэш пришлось бы сбрасывать
после каждого ``apply()``. Нет пакета — gr-satellites всё равно не
запустится, поэтому ответ ``satnogs`` здесь честный.
"""
if norad is not None and norad in known_norads():
return DECODER_GR_SATELLITES
return DECODER_SATNOGS
[документация]
def norad_of(tle: dict, norad_cat_id: int | None) -> int | None:
"""NORAD прохода: от портала (опт-ин) либо из второй строки TLE."""
if norad_cat_id is not None:
return norad_cat_id
try:
return int(tle["tle2"].split()[1])
except (KeyError, IndexError, ValueError):
return None
[документация]
def iq_mode_for(norad: int | None) -> str | None:
"""Режим satnogs-графа, который даст gr-satellites поток IQ.
Когда режим портала клиенту неизвестен, а satyaml для спутника есть,
отказывать проходу незачем: демодулятор satnogs-графа не нужен, нужен
только IQ с подходящей частотой, и её задаёт модуляция передатчика. Три
семейства покрывают gr-satellites целиком; незнакомая модуляция или
отсутствие satyaml — ``None``, и проход отказывается как раньше.
"""
definition = _definition_for(norad)
match = definition and _MODULATION_FIELD.search(definition)
if not match:
return None
modulation = match.group(1).upper()
if modulation.startswith("AFSK"):
return "AFSK"
if "BPSK" in modulation:
return "BPSK"
if "FSK" in modulation or "MSK" in modulation:
return "FSK"
return None
[документация]
def baudrate_for(norad: int | None) -> int | None:
"""Скорость первого передатчика satyaml — для задания без ``baud``."""
definition = _definition_for(norad)
match = definition and _BAUDRATE_FIELD.search(definition)
return int(match.group(1)) if match else None
[документация]
def known_norads() -> set[int]:
"""NORAD всех определений в каталоге satyaml пакета ``satellites``.
Уезжает на портал ключом ``satellites`` статуса станции: по нему портал
сможет планировать спутник с satyaml независимо от режима передатчика.
"""
return {norad for norad, _ in _definitions()}
def _definition_for(norad: int | None) -> str | None:
if norad is None:
return None
for known, definition in _definitions():
if known == norad:
return definition.read_text(errors="replace")
return None
def _definitions() -> list[tuple[int, Path]]:
"""Пары (NORAD, файл) каталога satyaml; без пакета ``satellites`` — пусто."""
try:
import satellites
except ImportError:
logger.warning("Пакет satellites не найден, gr-satellites недоступен")
return []
found: list[tuple[int, Path]] = []
for definition in (Path(satellites.__file__).parent / SATYAML_DIR_NAME).glob("*.yml"):
match = _NORAD_FIELD.search(definition.read_text(errors="replace"))
if match:
found.append((int(match.group(1)), definition))
return found