tiny_trigger/store.py

from __future__ import annotations

import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from pydantic import BaseModel, Field

from .automation import AutomationDocument
from .models import ActionEvent


LOCAL_DIR = Path(".local")
CONFIG_PATH = LOCAL_DIR / "config.yaml"
AUTOMATIONS_PATH = LOCAL_DIR / "automations.json"
STATE_PATH = LOCAL_DIR / "state.json"
EVENTS_PATH = LOCAL_DIR / "events.jsonl"


class LocalConfig(BaseModel):
    camera_url: str | None = None
    webhook_url: str | None = None
    default_classes: str | None = None
    default_detector_model: str | None = None
    default_device: str | None = None
    default_image_size: int | None = None
    default_max_detections: int | None = None
    llm_provider: str | None = None
    replicate_api_token: str | None = None
    replicate_model: str | None = None
    replicate_reasoning_effort: str | None = None
    openai_api_key: str | None = None
    openai_model: str | None = None
    anthropic_api_key: str | None = None
    anthropic_model: str | None = None


class RuntimeState(BaseModel):
    last_fired: dict[str, float] = Field(default_factory=dict)
    snoozed_until: dict[str, float] = Field(default_factory=dict)


def load_local_config(path: Path = CONFIG_PATH) -> LocalConfig:
    if not path.exists():
        return LocalConfig()
    try:
        import yaml
    except ImportError as exc:  # pragma: no cover - dependency guard
        raise RuntimeError("Install PyYAML to load local config.") from exc
    data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    return LocalConfig.model_validate(data)


def load_saved_automations(path: Path = AUTOMATIONS_PATH) -> AutomationDocument | None:
    if not path.exists():
        return None
    return AutomationDocument.model_validate(json.loads(path.read_text(encoding="utf-8")))


def save_automations(document: AutomationDocument, path: Path = AUTOMATIONS_PATH) -> None:
    _ensure_parent(path)
    path.write_text(document.model_dump_json(by_alias=True, indent=2), encoding="utf-8")


def load_runtime_state(path: Path = STATE_PATH) -> RuntimeState:
    if not path.exists():
        return RuntimeState()
    return RuntimeState.model_validate(json.loads(path.read_text(encoding="utf-8")))


def save_runtime_state(state: RuntimeState, path: Path = STATE_PATH) -> None:
    _ensure_parent(path)
    path.write_text(state.model_dump_json(indent=2), encoding="utf-8")


def append_events(events: list[ActionEvent], path: Path = EVENTS_PATH) -> None:
    if not events:
        return
    _ensure_parent(path)
    with path.open("a", encoding="utf-8") as file:
        for event in events:
            payload: dict[str, Any] = event.model_dump(mode="json")
            payload["recorded_at"] = datetime.now(timezone.utc).isoformat()
            file.write(json.dumps(payload, ensure_ascii=False) + "\n")


def _ensure_parent(path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)