sintonia/playout/libretime_playout/player/events.py

135 lines
2.9 KiB
Python
Raw Permalink Normal View History

from datetime import datetime
2022-07-01 12:40:24 +02:00
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)
2022-07-01 12:40:24 +02:00
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]