pigway-frame-flow/pigway_frame_flow/models.py

from datetime import datetime
from enum import StrEnum
from typing import Annotated, Any, Literal

from pydantic import BaseModel, ConfigDict, Field, model_validator

from pigway_frame_flow.errors import ErrorCode


class StrictModel(BaseModel):
    model_config = ConfigDict(extra="forbid")


class Fit(StrEnum):
    contain = "contain"
    cover = "cover"
    smart = "smart"


class Profile(StrEnum):
    color = "color"
    grayscale = "grayscale"


class ImageContent(StrictModel):
    type: Literal["image"]
    image_id: int | None = Field(default=None, gt=0)
    image_name: str | None = Field(default=None, min_length=1, max_length=120)
    asset_id: str | None = Field(default=None, pattern=r"^[a-f0-9]{32}$")
    base64: str | None = Field(default=None, max_length=28_000_000)
    name: str = Field(default="image.jpg", min_length=1, max_length=120)
    fit: Fit = Fit.contain
    profile: Profile = Profile.color

    @model_validator(mode="after")
    def one_source(self):
        if (
            sum(
                x is not None
                for x in [self.image_id, self.image_name, self.asset_id, self.base64]
            )
            != 1
        ):
            raise ValueError("Specify exactly one image_id, asset_id, or base64")
        return self


class URLContent(StrictModel):
    type: Literal["image_url"]
    url: str = Field(max_length=2048)
    name: str = Field(default="image.jpg", max_length=120)
    fit: Fit = Fit.contain
    profile: Profile = Profile.color


class TemplateContent(StrictModel):
    type: Literal["template"]
    template: str = Field(pattern=r"^[a-z0-9-]{1,60}$")
    data: dict[str, Any] = Field(default_factory=dict)
    profile: Profile = Profile.color


class FutureContent(StrictModel):
    type: Literal["html", "markdown"]
    data: str = ""


Content = Annotated[
    ImageContent | URLContent | TemplateContent | FutureContent,
    Field(discriminator="type"),
]


class Options(StrictModel):
    duration: int = Field(default=0, ge=0, le=604800)
    restore_previous: bool = False
    priority: int = Field(default=50, ge=0, le=100)

    @model_validator(mode="after")
    def restoration(self):
        if self.restore_previous and self.duration == 0:
            raise ValueError("restore_previous requires a positive duration")
        return self


class Source(StrictModel):
    client: str | None = Field(default=None, max_length=80)
    request_id: str | None = Field(default=None, max_length=120)


class DisplayRequest(StrictModel):
    device_id: str = "living-room"
    content: Content
    options: Options = Field(default_factory=Options)
    source: Source = Field(default_factory=Source)


class Accepted(BaseModel):
    operation_id: str
    job_id: str
    status: Literal["accepted"] = "accepted"


class OperationStatus(StrEnum):
    QUEUED = "QUEUED"
    RENDERING = "RENDERING"
    UPLOADING = "UPLOADING"
    SYNCING = "SYNCING"
    COMPLETED = "COMPLETED"
    FAILED = "FAILED"
    UNKNOWN = "UNKNOWN"


class ErrorBody(BaseModel):
    code: ErrorCode
    message: str
    request_id: str
    details: dict[str, Any] = Field(default_factory=dict)


class ErrorResponse(BaseModel):
    error: ErrorBody


class OperationResult(BaseModel):
    image_id: int | None = None
    asset_id: str | None = None
    rendered_asset_id: str | None = None
    url: str | None = None
    deferred: bool = False


class Operation(BaseModel):
    operation_id: str
    status: OperationStatus
    device_id: str
    created_at: str
    updated_at: str
    result: OperationResult | None = None
    error: ErrorBody | None = None
    request_id: str
    client_id: str
    action: str


class Capabilities(BaseModel):
    upload: bool = True
    gallery: bool = True
    delete: bool = True
    switch: bool = True
    remote_refresh: bool = True
    battery: bool = True
    color: bool = True
    cloud_upload_without_display: bool = False


class DisplayGeometry(BaseModel):
    width: int
    height: int
    rotation: int
    color_mode: Profile = Profile.color


class Power(BaseModel):
    battery_percent: int | None
    charging: bool


class CurrentContent(BaseModel):
    type: Literal["image"] = "image"
    id: int | None
    cover_id: int | None
    synced: bool


class Device(BaseModel):
    id: str
    name: str
    driver: Literal["kokonna"] = "kokonna"
    status: Literal["online", "offline"]
    model: str
    model_code: str
    firmware: str | None
    display: DisplayGeometry
    power: Power
    current_content: CurrentContent
    last_seen_at: str | None
    timezone: str
    capabilities: Capabilities = Field(default_factory=Capabilities)
    slideshow: dict[str, Any] = Field(default_factory=dict)
    storage_used_bytes: int | None = None
    model_verified: bool = False
    can_restore: bool = False


class GalleryImage(BaseModel):
    width: int | None = None
    height: int | None = None
    size_bytes: int | None = None
    media_type: str | None = None
    image_id: int
    name: str
    created_at: str | None
    current: bool
    url: str


class Schedule(StrictModel):
    type: Literal["cron", "daily", "weekdays", "weekends", "date", "temporary"]
    cron: str | None = None
    time: str | None = Field(default=None, pattern=r"^([01]\d|2[0-3]):[0-5]\d$")
    at: datetime | None = None


class RestoreContent(StrictModel):
    type: Literal["restore"]


class JobInput(StrictModel):
    enabled: bool = True
    device_id: str = "living-room"
    name: str = Field(min_length=1, max_length=120)
    schedule: Schedule
    source: Content | RestoreContent
    display_duration: int = Field(default=0, ge=0, le=604800)
    restore_previous: bool = False
    priority: int = Field(default=50, ge=0, le=100)

    @model_validator(mode="after")
    def validate_options(self):
        Options(
            duration=self.display_duration,
            restore_previous=self.restore_previous,
            priority=self.priority,
        )
        return self


class Job(JobInput):
    id: str
    created_at: str
    updated_at: str
    last_run: str | None = None
    last_result: str | None = None
    next_run: str | None = None


class TemplateInfo(BaseModel):
    id: str
    name: str
    width: int = 1200
    height: int = 1600
    fields: list[str]
    defaults: dict[str, Any]


class TemplateConfig(StrictModel):
    data: dict[str, Any]


class History(BaseModel):
    id: int
    client_id: str
    request_id: str
    operation_id: str | None
    device_id: str
    action: str
    result: str
    timestamp: str


class SettingsView(BaseModel):
    version: str
    mock: bool
    write_enabled: bool
    api_key_configured: bool
    api_token_configured: bool
    timezone: str
    device_id: str
    weather_provider: str


class Health(BaseModel):
    status: str
    version: str = "0.1.0"


class Readiness(BaseModel):
    status: str
    checks: dict[str, bool]


class Asset(BaseModel):
    source: Literal["uploaded", "generated"] = "uploaded"
    asset_id: str
    name: str
    url: str
    created_at: str


class DeleteRequest(StrictModel):
    confirm_current: bool = False


class RestoreRequest(StrictModel):
    device_id: str = "living-room"


class Version(BaseModel):
    product: str = "Pigway Frame Flow"
    version: str = "0.1.0"
    api_version: str = "v1"
    commit: str = "development"
    build_time: str = "unknown"


class PlaylistInput(StrictModel):
    device_id: str = "living-room"
    name: str = Field(min_length=1, max_length=120)
    image_ids: list[int] = Field(min_length=1, max_length=100)
    interval_minutes: int = Field(default=60, ge=5, le=10080)
    mode: Literal["queue", "random"] = "queue"
    enabled: bool = False

    @model_validator(mode="after")
    def ids(self):
        if any(x <= 0 for x in self.image_ids) or len(set(self.image_ids)) != len(
            self.image_ids
        ):
            raise ValueError("Image IDs must be positive and unique")
        return self


class Playlist(PlaylistInput):
    id: str
    position: int
    next_run: str | None


class ReconcileRequest(StrictModel):
    acknowledge: bool = False


class Reconciliation(BaseModel):
    operation_id: str
    current_image_id: int | None
    expected_image_id: int | None
    cloud_images: list[GalleryImage]
    resolved: bool


class CurrentDisplay(BaseModel):
    device_id: str
    image_id: int | None
    synced: bool
    source_known: bool
    source: dict[str, Any] | None
    asset_id: str | None
    operation_id: str | None
    preview_url: str | None
    previous_image_id: int | None
    overrides: list[dict[str, Any]]


class WeatherConfig(StrictModel):
    provider: Literal["mock", "open-meteo"] = "mock"
    location: str = Field(
        default="示例地点 / Shanghai Jinshan", min_length=1, max_length=120
    )
    latitude: float = Field(default=30.74, ge=-90, le=90)
    longitude: float = Field(default=121.34, ge=-180, le=180)


class WeatherView(BaseModel):
    config: WeatherConfig
    current: dict[str, Any]
    forecast: list[dict[str, Any]]


class ImageURLConfig(StrictModel):
    allowed_hosts: list[str] = Field(default_factory=list, max_length=30)

    @model_validator(mode="after")
    def hosts(self):
        import re

        if any(
            not re.fullmatch(r"[a-zA-Z0-9](?:[a-zA-Z0-9.-]{0,251}[a-zA-Z0-9])?", h)
            for h in self.allowed_hosts
        ):
            raise ValueError("Use hostnames only, without URLs or wildcards")
        self.allowed_hosts = sorted(set(h.lower() for h in self.allowed_hosts))
        return self


class Widget(StrictModel):
    id: int
    name: str
    tags: list[str] = Field(default_factory=list)
    arg_keys: list[str] = Field(default_factory=list)


class WidgetState(StrictModel):
    source: Literal["live", "snapshot", "mock"]
    configured: bool
    status: str = "OK"
    observed_at: str | None = None
    catalog: list[Widget]
    current: Widget | None = None
    current_verified: bool = False
    writable: Literal[False] = False