Spaces:
Sleeping
Sleeping
File size: 2,998 Bytes
d725335 a837fd9 2a9b8d1 d725335 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 | 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)
|