sintonia/shared/libretime_shared/config/_models.py

import sys
from enum import Enum
from typing import List, Literal, Optional, Union

from pydantic import BaseModel, Field, field_validator
from typing_extensions import Annotated

from ._fields import AnyHttpUrlStr, AnyUrlStr, StrNoLeadingSlash, StrNoTrailingSlash

if sys.version_info < (3, 9):
    from backports.zoneinfo import ZoneInfo, ZoneInfoNotFoundError
else:
    from zoneinfo import ZoneInfo, ZoneInfoNotFoundError


# GeneralConfig
########################################################################################


# pylint: disable=too-few-public-methods
class GeneralConfig(BaseModel):
    public_url: AnyHttpUrlStr
    api_key: str
    secret_key: str

    timezone: str = "UTC"

    allowed_cors_origins: List[AnyHttpUrlStr] = []

    @field_validator("timezone")
    @classmethod
    def _validate_timezone(cls, value: str) -> str:
        try:
            ZoneInfo(value)
        except ZoneInfoNotFoundError as exception:
            raise ValueError(f"invalid timezone '{value}'") from exception

        return value


# StorageConfig
########################################################################################


# pylint: disable=too-few-public-methods
class StorageConfig(BaseModel):
    path: StrNoTrailingSlash = "/srv/libretime"


# DatabaseConfig
########################################################################################


# pylint: disable=too-few-public-methods
class DatabaseConfig(BaseModel):
    host: str = "localhost"
    port: int = 5432
    name: str = "libretime"
    user: str = "libretime"
    password: str = "libretime"

    @property
    def url(self) -> str:
        return (
            f"postgresql://{self.user}:{self.password}"
            f"@{self.host}:{self.port}/{self.name}"
        )


# RabbitMQConfig
########################################################################################


# pylint: disable=too-few-public-methods
class RabbitMQConfig(BaseModel):
    host: str = "localhost"
    port: int = 5672
    user: str = "libretime"
    password: str = "libretime"
    vhost: str = "/libretime"

    @property
    def url(self) -> str:
        return (
            f"amqp://{self.user}:{self.password}"
            f"@{self.host}:{self.port}/{self.vhost}"
        )


# StreamConfig
########################################################################################


class BaseInput(BaseModel):
    enabled: bool = True
    public_url: Optional[AnyUrlStr] = None


class InputKind(str, Enum):
    HARBOR = "harbor"


class HarborInput(BaseInput):
    kind: Literal[InputKind.HARBOR] = InputKind.HARBOR
    mount: StrNoLeadingSlash
    port: int
    secure: bool = False


class MainHarborInput(HarborInput):
    mount: str = "main"
    port: int = 8001


class ShowHarborInput(HarborInput):
    mount: str = "show"
    port: int = 8002


class Inputs(BaseModel):
    main: HarborInput = MainHarborInput()
    show: HarborInput = ShowHarborInput()


class AudioChannels(str, Enum):
    STEREO = "stereo"
    MONO = "mono"


class BaseAudio(BaseModel):
    channels: AudioChannels = AudioChannels.STEREO
    bitrate: int

    @field_validator("bitrate")
    @classmethod
    def _validate_bitrate(cls, value: int) -> int:
        # Once the liquidsoap script generation supports it, fine tune
        # the bitrate validation for each format
        bitrates = (32, 48, 64, 96, 128, 160, 192, 224, 256, 320)
        if value not in bitrates:
            raise ValueError(f"invalid bitrate {value}, must be one of {bitrates}")
        return value


class AudioFormat(str, Enum):
    AAC = "aac"
    MP3 = "mp3"
    OGG = "ogg"
    OPUS = "opus"


class AudioAAC(BaseAudio):
    format: Literal[AudioFormat.AAC] = AudioFormat.AAC


class AudioMP3(BaseAudio):
    format: Literal[AudioFormat.MP3] = AudioFormat.MP3


class AudioOGG(BaseAudio):
    format: Literal[AudioFormat.OGG] = AudioFormat.OGG
    enable_metadata: Optional[bool] = False


class AudioOpus(BaseAudio):
    format: Literal[AudioFormat.OPUS] = AudioFormat.OPUS


class IcecastOutput(BaseModel):
    kind: Literal["icecast"] = "icecast"
    enabled: bool = False
    public_url: Optional[AnyUrlStr] = None

    host: str = "localhost"
    port: int = 8000
    mount: StrNoLeadingSlash
    source_user: str = "source"
    source_password: str
    admin_user: str = "admin"
    admin_password: Optional[str] = None

    audio: Annotated[
        Union[AudioAAC, AudioMP3, AudioOGG, AudioOpus],
        Field(discriminator="format"),
    ]

    name: Optional[str] = None
    description: Optional[str] = None
    website: Optional[str] = None
    genre: Optional[str] = None

    mobile: bool = False


class ShoutcastOutput(BaseModel):
    kind: Literal["shoutcast"] = "shoutcast"
    enabled: bool = False
    public_url: Optional[AnyUrlStr] = None

    host: str = "localhost"
    port: int = 8000
    source_user: str = "source"
    source_password: str
    admin_user: str = "admin"
    admin_password: Optional[str] = None

    audio: Annotated[
        Union[AudioAAC, AudioMP3],
        Field(discriminator="format"),
    ]

    name: Optional[str] = None
    website: Optional[str] = None
    genre: Optional[str] = None

    mobile: bool = False


class SystemOutput(str, Enum):
    ALSA = "alsa"
    AO = "ao"
    OSS = "oss"
    PORTAUDIO = "portaudio"
    PULSEAUDIO = "pulseaudio"


class BaseSystemOutput(BaseModel):
    enabled: bool = False


class ALSASystemOutput(BaseSystemOutput):
    kind: Literal[SystemOutput.ALSA] = SystemOutput.ALSA
    device: Optional[str] = None


class AOSystemOutput(BaseSystemOutput):
    kind: Literal[SystemOutput.AO] = SystemOutput.AO


class OSSSystemOutput(BaseSystemOutput):
    kind: Literal[SystemOutput.OSS] = SystemOutput.OSS


class PortAudioSystemOutput(BaseSystemOutput):
    kind: Literal[SystemOutput.PORTAUDIO] = SystemOutput.PORTAUDIO


class PulseAudioSystemOutput(BaseSystemOutput):
    kind: Literal[SystemOutput.PULSEAUDIO] = SystemOutput.PULSEAUDIO
    device: Optional[str] = None


AnySystemOutput = Annotated[
    Union[
        ALSASystemOutput,
        AOSystemOutput,
        OSSSystemOutput,
        PortAudioSystemOutput,
        PulseAudioSystemOutput,
    ],
    Field(discriminator="kind", default=SystemOutput.PULSEAUDIO),
]


# pylint: disable=too-few-public-methods
class Outputs(BaseModel):
    icecast: List[IcecastOutput] = Field([], max_length=3)
    shoutcast: List[ShoutcastOutput] = Field([], max_length=1)
    system: List[AnySystemOutput] = Field([], max_length=1)

    @property
    def merged(self) -> List[Union[IcecastOutput, ShoutcastOutput]]:
        return self.icecast + self.shoutcast


# pylint: disable=too-few-public-methods
class StreamConfig(BaseModel):
    """Stream configuration model."""

    inputs: Inputs = Inputs()
    outputs: Outputs = Outputs()  # type: ignore[call-arg]