"""Скрипты вокруг прохода: ``satnogs-pre`` и ``satnogs-post``.
Оба получают одинаковый набор аргументов и запускаются одинаково, поэтому
сборка командной строки и запуск разделены: первая — чистая функция, её
видно из теста без подпроцесса.
"""
import json
import subprocess
from core.configs import logger, raw_logger, settings
from core.configs.flowgraph import MODES
from core.types import TLE
from soniks_client.subprocess_log import start_logged_process, stop_logging
[документация]
def build_script_argv(
script_type: str,
observation_id: str,
frequency: int,
tle: TLE,
timestamp: str,
baud: int,
mode: str,
) -> list[str]:
"""Собрать командную строку скрипта прохода.
Режим, которого нет в таблице, заменяется на ``FLOWGRAPH__DEFAULT_MODE``:
скрипту нужно имя потокового графа, а не сам режим. Подмена пишется в лог
предупреждением, потому что тихой она уже стоила двух обходов бага
``SSTV_PD120``: это имя уходит шестым аргументом в ``satnogs-pre`` и
``satnogs-post``, оттуда в ``scripts/find_samp_rate.py``, а тот определяет
по подстрокам в нём частоту дискретизации IQ. Имя из чужого режима даёт
неверную частоту и UDP-потоку в gr-satellites, и имени IQ-дампа — без
единого признака отказа.
Подмена доживает сюда, только пока портал не знает режимов станции: после
их публикации неизвестный режим отказывается раньше, в
``execute_observation``, и проход не начинается.
"""
known_mode = mode if mode in MODES else settings.flowgraph.DEFAULT_MODE
if known_mode != mode:
logger.warning(
"Режим %s порталу известен, а таблице клиента нет: скриптам прохода"
" уходит имя графа режима %s, частота дискретизации будет от него",
mode,
known_mode,
)
return [
script_type,
observation_id,
str(frequency),
json.dumps(tle),
timestamp,
str(baud),
MODES[known_mode]["script_filename"],
]
[документация]
def run_script(
argv: list[str],
timeout: float,
env: dict[str, str] | None = None,
) -> None:
"""Выполнить скрипт прохода, слив его вывод в лог.
Ошибки гасятся: ни отсутствующий скрипт, ни зависший не должны отменять
проход. Зависший добивается по таймауту.
"""
script_type = argv[0]
logger.info("Запуск скрипта %s", script_type)
try:
process, log_thread = start_logged_process(argv, _log_line, env)
except OSError as e:
logger.error("Ошибка при выполнении скрипта %s: %s", script_type, e)
return
try:
process.wait(timeout=timeout)
except subprocess.TimeoutExpired:
logger.warning(
"Принудительное завершение %s: превышен таймаут %.0f с",
script_type,
timeout,
)
process.kill()
process.wait()
finally:
stop_logging(log_thread, f"скрипта {script_type}")
def _log_line(line: str) -> None:
raw_logger.log(settings.log.script_log_level, line)