Исходный код core.configs.observation

"""Настройки наблюдения: префиксы, расширения и состав выгружаемых файлов."""

from pydantic import BaseModel




class ObservationSendingKey(BaseModel):
    """Имена полей формы при выгрузке на портал."""
    PAYLOAD: str = "payload"
    WATERFALL: str = "waterfall"
    DEMODDATA: str = "demoddata"
    CLIENT_METADATA: str = "client_metadata"





class ObservationPrefix(BaseModel):
    """Префиксы имён файлов наблюдения."""
    PAYLOAD: str = "payload"
    WATERFALL: str = "waterfall"
    RAW_WATERFALL: str = "raw_waterfall"
    DATA: str = "data"
    METADATA: str = "metadata"





class ObservationExtension(BaseModel):
    """Расширения файлов наблюдения."""
    PAYLOAD: str = "ogg"
    WATERFALL: str = "png"
    DATA_APT: str = "png"
    METADATA: str = "json"

    RAW_WATERFALL: str = "dat"





class ObservationUploadTypes(BaseModel):
    """Какие типы файлов выгружать на портал."""
    PAYLOAD: bool = True
    WATERFALL: bool = True





class ObservationSettings(BaseModel):
    """Параметры проведения наблюдения и состав выгружаемых файлов."""
    REMOVE_WATERFALL_RAW_FILES: bool = True
    REMOVE_OBSERVATION_DATA: bool = True

    SOAPY_RX_DEVICE: str = "driver=airspy,biastee=true"

    TIME_FORMAT_IN_FILENAME: str = "%Y-%m-%dT%H-%M-%S"
    FILENAME_TEMPLATE: str = (
        "{observation_dir}/{prefix}_{observation_id}_{timestamp}.{extension}"
    )

    RIG_UPDATE_INTERVAL: float = 0.1
    ROTATOR_UPDATE_INTERVAL: float = 3

    PRE_OBSERVATION_SCRIPT: str = "satnogs-pre"
    POST_OBSERVATION_SCRIPT: str = "satnogs-post"
    RUN_PRE_OBSERVATION_SCRIPT: bool = True
    RUN_POST_OBSERVATION_SCRIPT: bool = True

    # Потолок на выполнение satnogs-pre/satnogs-post. Декодеры и SSTV на
    # слабом железе идут минутами, поэтому значение с запасом: ограничивать
    # нужно по-настоящему зависший процесс, а не долгую постобработку.
    SCRIPT_TIMEOUT_IN_SECONDS: float = 900

    BATCH_DELAY: float = 1

    prefix: ObservationPrefix = ObservationPrefix()
    extension: ObservationExtension = ObservationExtension()
    sending_key: ObservationSendingKey = ObservationSendingKey()
    upload_types: ObservationUploadTypes = ObservationUploadTypes()



    def get_prefixes_of_files_available_for_upload(self) -> list[str]:
        """Вернуть префиксы файлов, разрешённых к выгрузке.

        Кадры и метаданные выгружаются всегда, аудио и водопад — по
        ``OBSERVATION__UPLOAD_TYPES__*``. Метаданные первыми: порядок
        префиксов задаёт порядок отправки, а до появления очереди они уходили
        на портал раньше остальных файлов.
        """
        prefixes = [self.prefix.METADATA]

        if self.upload_types.PAYLOAD:
            prefixes.append(self.prefix.PAYLOAD)
        if self.upload_types.WATERFALL:
            prefixes.append(self.prefix.WATERFALL)

        prefixes.append(self.prefix.DATA)

        return prefixes