sintonia/playout/libretime_playout/player/events.py

from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import Dict, Literal, Optional, Union

from dateutil.parser import isoparse
from pydantic import BaseModel, BeforeValidator, Field, parse_obj_as
from typing_extensions import Annotated

from ..config import CACHE_DIR
from ..utils import mime_guess_extension

EVENT_KEY_FORMAT = "%Y-%m-%d-%H-%M-%S"


def event_key_to_datetime(value: Union[str, datetime]) -> datetime:
    if isinstance(value, str):
        value = datetime.strptime(value, EVENT_KEY_FORMAT)
    return value


def datetime_to_event_key(value: Union[str, datetime]) -> str:
    if isinstance(value, datetime):
        value = value.strftime(EVENT_KEY_FORMAT)
    return value


def event_isoparse(value: str) -> datetime:
    return isoparse(value).replace(tzinfo=None).replace(microsecond=0)


class EventKind(str, Enum):
    FILE = "file"
    ACTION = "event"
    WEB_STREAM_BUFFER_START = "stream_buffer_start"
    WEB_STREAM_OUTPUT_START = "stream_output_start"
    WEB_STREAM_BUFFER_END = "stream_buffer_end"
    WEB_STREAM_OUTPUT_END = "stream_output_end"


EventKeyDatetime = Annotated[datetime, BeforeValidator(event_key_to_datetime)]


class BaseEvent(BaseModel):
    start: EventKeyDatetime
    end: EventKeyDatetime

    @property
    def start_key(self) -> str:
        return datetime_to_event_key(self.start)

    @property
    def end_key(self) -> str:
        return datetime_to_event_key(self.end)

    def ended(self) -> bool:
        return datetime.utcnow() > self.end


class FileEvent(BaseEvent):
    type: Literal[EventKind.FILE]

    # Schedule
    row_id: int
    uri: Optional[str] = None
    id: int

    # Show data
    show_name: str

    # File
    fade_in: float
    fade_out: float
    cue_in: float
    cue_out: float

    track_title: Optional[str] = None
    artist_name: Optional[str] = None

    mime: str
    replay_gain: Optional[float] = None
    filesize: int

    file_ready: bool = False

    @property
    def file_ext(self) -> str:
        return mime_guess_extension(self.mime)

    @property
    def local_filepath(self) -> Path:
        return CACHE_DIR / f"{self.id}{self.file_ext}"


class WebStreamEvent(BaseEvent):
    type: Literal[
        EventKind.WEB_STREAM_BUFFER_START,
        EventKind.WEB_STREAM_OUTPUT_START,
        EventKind.WEB_STREAM_BUFFER_END,
        EventKind.WEB_STREAM_OUTPUT_END,
    ]

    # Schedule
    row_id: int
    uri: str
    id: int

    # Show data
    show_name: str


class ActionEventKind(str, Enum):
    SWITCH_OFF = "switch_off"
    KICK_OUT = "kick_out"


class ActionEvent(BaseEvent):
    type: Literal[EventKind.ACTION]
    # TODO: user ActionEventKind enum
    event_type: str


AnyEvent = Annotated[
    Union[FileEvent, WebStreamEvent, ActionEvent],
    Field(discriminator="type"),
]


def parse_any_event(value: dict) -> AnyEvent:
    return parse_obj_as(AnyEvent, value)  # type: ignore


FileEvents = Dict[str, FileEvent]
Events = Dict[str, AnyEvent]