2022-12-07 14:39:23 +01:00
|
|
|
import sys
|
2022-08-22 16:57:27 +02:00
|
|
|
from enum import Enum
|
2023-03-01 19:58:16 +01:00
|
|
|
from typing import TYPE_CHECKING, Any, List, Literal, Optional, Union
|
2022-07-25 21:30:15 +02:00
|
|
|
|
|
|
|
# pylint: disable=no-name-in-module
|
2022-08-22 16:57:27 +02:00
|
|
|
from pydantic import AnyHttpUrl, AnyUrl, BaseModel, Field, validator
|
2022-09-09 20:21:59 +02:00
|
|
|
from typing_extensions import Annotated
|
2022-07-25 21:30:15 +02:00
|
|
|
|
2022-12-07 14:39:23 +01:00
|
|
|
if sys.version_info < (3, 9):
|
|
|
|
from backports.zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
|
else:
|
2022-09-14 12:48:08 +02:00
|
|
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
2022-12-07 14:39:23 +01:00
|
|
|
|
2022-09-14 12:48:08 +02:00
|
|
|
|
2022-07-25 21:30:15 +02:00
|
|
|
if TYPE_CHECKING:
|
|
|
|
from pydantic.typing import AnyClassMethod
|
|
|
|
|
|
|
|
|
|
|
|
def no_trailing_slash_validator(key: str) -> "AnyClassMethod":
|
|
|
|
# pylint: disable=unused-argument
|
2022-07-30 21:11:45 +02:00
|
|
|
def strip_trailing_slash(cls: Any, value: Any) -> Any:
|
|
|
|
if isinstance(value, str):
|
|
|
|
return value.rstrip("/")
|
|
|
|
return value
|
2022-07-25 21:30:15 +02:00
|
|
|
|
|
|
|
return validator(key, pre=True, allow_reuse=True)(strip_trailing_slash)
|
|
|
|
|
|
|
|
|
2022-08-22 16:57:27 +02:00
|
|
|
def no_leading_slash_validator(key: str) -> "AnyClassMethod":
|
|
|
|
# pylint: disable=unused-argument
|
|
|
|
def strip_leading_slash(cls: Any, value: Any) -> Any:
|
|
|
|
if isinstance(value, str):
|
|
|
|
return value.lstrip("/")
|
|
|
|
return value
|
|
|
|
|
|
|
|
return validator(key, pre=True, allow_reuse=True)(strip_leading_slash)
|
|
|
|
|
|
|
|
|
|
|
|
# GeneralConfig
|
|
|
|
########################################################################################
|
|
|
|
|
|
|
|
|
2022-07-25 21:30:15 +02:00
|
|
|
# pylint: disable=too-few-public-methods
|
|
|
|
class GeneralConfig(BaseModel):
|
|
|
|
public_url: AnyHttpUrl
|
|
|
|
api_key: str
|
2023-03-22 10:14:11 +01:00
|
|
|
secret_key: Optional[str] = None
|
2022-07-25 21:30:15 +02:00
|
|
|
|
2022-09-14 12:48:08 +02:00
|
|
|
timezone: str = "UTC"
|
|
|
|
|
2023-03-23 15:40:30 +01:00
|
|
|
allowed_cors_origins: List[AnyHttpUrl] = []
|
|
|
|
|
2022-07-25 21:30:15 +02:00
|
|
|
# Validators
|
|
|
|
_public_url_no_trailing_slash = no_trailing_slash_validator("public_url")
|
|
|
|
|
2022-09-14 12:48:08 +02:00
|
|
|
@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
|
|
|
|
|
2022-07-25 21:30:15 +02:00
|
|
|
|
2022-08-22 16:57:27 +02:00
|
|
|
# StorageConfig
|
|
|
|
########################################################################################
|
|
|
|
|
2023-02-01 11:14:08 +01:00
|
|
|
|
2022-07-25 21:30:15 +02:00
|
|
|
# pylint: disable=too-few-public-methods
|
|
|
|
class StorageConfig(BaseModel):
|
|
|
|
path: str = "/srv/libretime"
|
|
|
|
|
|
|
|
# Validators
|
|
|
|
_path_no_trailing_slash = no_trailing_slash_validator("path")
|
|
|
|
|
|
|
|
|
2022-08-22 16:57:27 +02:00
|
|
|
# DatabaseConfig
|
|
|
|
########################################################################################
|
|
|
|
|
2023-02-01 11:14:08 +01:00
|
|
|
|
2022-07-25 21:30:15 +02:00
|
|
|
# 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}"
|
|
|
|
)
|
|
|
|
|
|
|
|
|
2022-08-22 16:57:27 +02:00
|
|
|
# RabbitMQConfig
|
|
|
|
########################################################################################
|
|
|
|
|
2023-02-01 11:14:08 +01:00
|
|
|
|
2022-07-25 21:30:15 +02:00
|
|
|
# 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}"
|
|
|
|
)
|
2022-08-22 16:57:27 +02:00
|
|
|
|
|
|
|
|
|
|
|
# StreamConfig
|
|
|
|
########################################################################################
|
|
|
|
|
|
|
|
|
|
|
|
class BaseInput(BaseModel):
|
|
|
|
enabled: bool = True
|
|
|
|
public_url: Optional[AnyUrl] = None
|
|
|
|
|
|
|
|
|
|
|
|
class InputKind(str, Enum):
|
|
|
|
HARBOR = "harbor"
|
|
|
|
|
|
|
|
|
|
|
|
class HarborInput(BaseInput):
|
|
|
|
kind: Literal[InputKind.HARBOR] = InputKind.HARBOR
|
|
|
|
mount: str
|
|
|
|
port: int
|
2023-03-30 20:39:02 +02:00
|
|
|
secure: bool = False
|
2022-08-22 16:57:27 +02:00
|
|
|
|
|
|
|
_mount_no_leading_slash = no_leading_slash_validator("mount")
|
|
|
|
|
|
|
|
|
|
|
|
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
|
|
|
|
|
|
|
|
@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[AnyUrl] = None
|
|
|
|
|
|
|
|
host: str = "localhost"
|
|
|
|
port: int = 8000
|
|
|
|
mount: str
|
|
|
|
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
|
|
|
|
|
2023-10-14 09:13:04 +02:00
|
|
|
mobile: bool = False
|
|
|
|
|
2022-08-22 16:57:27 +02:00
|
|
|
_mount_no_leading_slash = no_leading_slash_validator("mount")
|
|
|
|
|
|
|
|
|
|
|
|
class ShoutcastOutput(BaseModel):
|
|
|
|
kind: Literal["shoutcast"] = "shoutcast"
|
|
|
|
enabled: bool = False
|
|
|
|
public_url: Optional[AnyUrl] = 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
|
|
|
|
|
2023-10-14 09:13:04 +02:00
|
|
|
mobile: bool = False
|
|
|
|
|
2022-08-22 16:57:27 +02:00
|
|
|
|
|
|
|
class SystemOutputKind(str, Enum):
|
|
|
|
ALSA = "alsa"
|
|
|
|
AO = "ao"
|
|
|
|
OSS = "oss"
|
|
|
|
PORTAUDIO = "portaudio"
|
|
|
|
PULSEAUDIO = "pulseaudio"
|
|
|
|
|
|
|
|
|
|
|
|
class SystemOutput(BaseModel):
|
|
|
|
enabled: bool = False
|
|
|
|
kind: SystemOutputKind = SystemOutputKind.ALSA
|
|
|
|
|
|
|
|
|
|
|
|
# pylint: disable=too-few-public-methods
|
|
|
|
class Outputs(BaseModel):
|
|
|
|
icecast: List[IcecastOutput] = Field([], max_items=3)
|
|
|
|
shoutcast: List[ShoutcastOutput] = Field([], max_items=1)
|
|
|
|
system: List[SystemOutput] = Field([], max_items=1)
|
|
|
|
|
|
|
|
@property
|
2023-03-01 19:58:16 +01:00
|
|
|
def merged(self) -> List[Union[IcecastOutput, ShoutcastOutput]]:
|
|
|
|
return self.icecast + self.shoutcast
|
2022-08-22 16:57:27 +02:00
|
|
|
|
|
|
|
|
|
|
|
# pylint: disable=too-few-public-methods
|
|
|
|
class StreamConfig(BaseModel):
|
|
|
|
"""Stream configuration model."""
|
|
|
|
|
|
|
|
inputs: Inputs = Inputs()
|
2023-03-14 11:56:18 +01:00
|
|
|
outputs: Outputs = Outputs() # type: ignore[call-arg]
|