Исходный код soniks_client.post_processing

"""Постобработка прохода: водопад и метаданные сигнала.

Отправкой на портал занимается вызывающий: блок ``signal`` отсюда — только
часть метаданных наблюдения, остальное собирает ``Observation``.
"""

from pathlib import Path

from core.configs import logger, settings
from core.exceptions import (
    EmptyWaterfallError,
    TimestampError,
    WaterfallFileError,
)
from soniks_client.waterfall import Waterfall




def build_waterfall(
    raw_path: str,
    png_path: str,
    observation_id: str,
) -> dict | None:
    """Построить PNG водопада и вернуть метаданные сигнала.

    Возвращает ``None``, если водопад не построен или анализ ничего не дал.
    Сырой ``.dat`` удаляется только при успешном построении: при упавшем он
    единственное, с чем можно разбираться.
    """

    waterfall: Waterfall | None = None
    signal_metadata: dict | None = None
    waterfall_built = False

    try:
        waterfall = Waterfall(raw_path, png_path)

        # Границы шкалы не передаются: plot() сам берёт их из настроек, а
        # при WATERFALL__AUTORANGE=True подбирает по данным. Передача
        # MIN_VALUE/MAX_VALUE отсюда делала автоподбор недостижимым.
        waterfall.plot()

        waterfall_built = True

        logger.info(
            "Водопад построен [наблюдение: %s]: %s",
            observation_id,
            png_path,
        )

        try:
            signal_metadata = waterfall.get_signal_metadata()

            if signal_metadata:
                logger.info(
                    "Анализ сигнала выполнен [наблюдение: %s]: "
                    "modulation=%s, deviation=%s",
                    observation_id,
                    signal_metadata.get("modulation"),
                    signal_metadata.get("deviation_hz"),
                )
        except Exception as e:
            logger.warning(
                "Ошибка при анализе сигнала [наблюдение: %s]: %s",
                observation_id,
                e,
            )

    except (EmptyWaterfallError, TimestampError, WaterfallFileError) as e:
        # Исключения водопада без сообщения: подробности уже в логе
        # waterfall/plot.py, здесь важен сам факт и класс ошибки.
        logger.error(
            "Водопад не построен [наблюдение: %s]: %s",
            observation_id,
            type(e).__name__,
        )
    except Exception as e:
        logger.error(
            "Ошибка при построении водопада [наблюдение: %s]: %s",
            observation_id,
            e,
        )

    if settings.observation.REMOVE_WATERFALL_RAW_FILES:
        if waterfall_built:
            Path(raw_path).unlink(missing_ok=True)
        else:
            # Упавшее построение — единственный случай, когда сырой файл
            # ещё нужен: без него разбираться не с чем.
            logger.warning(
                "Сырой файл водопада сохранён для разбора [наблюдение: %s]: %s",
                observation_id,
                raw_path,
            )

    return signal_metadata