142 lines
3.2 KiB
Python
142 lines
3.2 KiB
Python
from datetime import datetime
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Dict, Literal, Optional, Union
|
|
|
|
from dateutil.parser import isoparse
|
|
from pydantic import BaseModel, Field, parse_obj_as, validator
|
|
from typing_extensions import Annotated
|
|
|
|
from ..config import CACHE_DIR
|
|
from ..utils import mime_guess_extension
|
|
|
|
if TYPE_CHECKING:
|
|
from pydantic.typing import AnyClassMethod
|
|
|
|
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"
|
|
|
|
|
|
def event_datetime_validator(prop: str) -> "AnyClassMethod":
|
|
return validator(prop, pre=True, allow_reuse=True)(event_key_to_datetime)
|
|
|
|
|
|
class BaseEvent(BaseModel):
|
|
start: datetime
|
|
end: datetime
|
|
|
|
_start_validator = event_datetime_validator("start")
|
|
_end_validator = event_datetime_validator("end")
|
|
|
|
@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]
|