from __future__ import annotations import asyncio import os from collections.abc import AsyncIterator from contextlib import asynccontextmanager, suppress from fastapi import FastAPI from fastapi.responses import HTMLResponse from app.core.event_core import EventCore from app.core.ha_client import FutureHaClient, HaClientConfig, service_for_state from app.core.handoff import HandoffMatrix from app.core.models import ( AuditEvent, BackupBundle, ControlProfile, ControlState, EntityState, HandoffMode, LearningState, RuntimeState, StateEvent, ) from app.core.stores import FutureStores stores = FutureStores(os.getenv("SILLYHOME_FUTURE_STORE", ".future_store")) event_core = EventCore(stores) handoff = HandoffMatrix() ha_client: FutureHaClient | None = None @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: global ha_client listener_task: asyncio.Task[None] | None = None ha_client = _ha_client_from_env() if ha_client is not None: _load_initial_ha_states(ha_client) listener_task = asyncio.create_task(_ha_listener_loop(ha_client)) try: yield finally: if listener_task is not None: listener_task.cancel() with suppress(asyncio.CancelledError): await listener_task app = FastAPI( title="SillyHome Future API", description="SillyHome v2 event-core side project.", version="2.0.0-alpha.7", lifespan=lifespan, ) @app.get("/health") def health() -> dict[str, str]: runtime = stores.runtime() return { "status": "ok", "version": app.version, "ha": runtime.websocket_status, } @app.get("/", response_class=HTMLResponse) def dashboard() -> str: return _dashboard_html() @app.get("/v2/dashboard") def dashboard_data() -> dict[str, object]: runtime = stores.runtime() learning = stores.learning() control = stores.control() latest_audit = runtime.audit[-20:] return { "websocket_status": runtime.websocket_status, "entity_count": len(runtime.entities), "learning_profiles": len(learning.profiles), "control_profiles": len(control.profiles), "rooms": list(learning.rooms.values()), "scenes": list(learning.scenes.values()), "audit": latest_audit, } @app.post("/v2/events/state", response_model=list[AuditEvent]) def ingest_state_event(event: StateEvent) -> list[AuditEvent]: return event_core.process_state_event(event, execute=_execute_ha_decision) @app.get("/v2/runtime", response_model=RuntimeState) def get_runtime() -> RuntimeState: return stores.runtime() @app.get("/v2/learning", response_model=LearningState) def get_learning() -> LearningState: return stores.learning() @app.put("/v2/learning", response_model=LearningState) def put_learning(state: LearningState) -> LearningState: return stores.save_learning(state) @app.get("/v2/control", response_model=ControlState) def get_control() -> ControlState: return stores.control() @app.put("/v2/control/{actuator_entity_id}", response_model=ControlProfile) def put_control(actuator_entity_id: str, profile: ControlProfile) -> ControlProfile: state = stores.control() state.profiles[actuator_entity_id] = profile stores.save_control(state) return profile @app.post("/v2/handoff/{actuator_entity_id}/assume", response_model=ControlProfile) def assume_control(actuator_entity_id: str) -> ControlProfile: state = stores.control() profile = state.profiles.get( actuator_entity_id, ControlProfile(actuator_entity_id=actuator_entity_id), ) updated = handoff.assume_control(profile) state.profiles[actuator_entity_id] = updated stores.save_control(state) return updated @app.post("/v2/handoff/{actuator_entity_id}/rollback", response_model=ControlProfile) def rollback_control(actuator_entity_id: str) -> ControlProfile: state = stores.control() profile = state.profiles.get( actuator_entity_id, ControlProfile(actuator_entity_id=actuator_entity_id), ) updated = handoff.rollback(profile) state.profiles[actuator_entity_id] = updated stores.save_control(state) return updated @app.get("/v2/handoff/{actuator_entity_id}", response_model=HandoffMode) def classify_handoff(actuator_entity_id: str) -> HandoffMode: profile = stores.control().profiles.get( actuator_entity_id, ControlProfile(actuator_entity_id=actuator_entity_id), ) return handoff.classify(profile) @app.get("/v2/backup/export", response_model=BackupBundle) def export_backup() -> BackupBundle: return stores.export_backup() @app.post("/v2/backup/restore") def restore_backup(bundle: BackupBundle) -> dict[str, str]: stores.restore_backup(bundle) return {"status": "restored"} def _ha_client_from_env() -> FutureHaClient | None: token = os.getenv("SUPERVISOR_TOKEN") or os.getenv("SILLYHOME_FUTURE_HA_TOKEN") if not token: return None core_url = os.getenv("SILLYHOME_FUTURE_HA_CORE_URL", "http://supervisor/core") websocket_url = os.getenv("SILLYHOME_FUTURE_HA_WS_URL") return FutureHaClient( HaClientConfig(core_url=core_url, token=token, websocket_url=websocket_url) ) def _load_initial_ha_states(client: FutureHaClient) -> None: runtime = stores.runtime() try: for entity in client.read_states(): runtime.entities[entity.entity_id] = entity runtime.websocket_status = "initial_state_loaded" except Exception as exc: runtime.websocket_status = f"initial_state_error: {exc}" stores.save_runtime(runtime) async def _ha_listener_loop(client: FutureHaClient) -> None: routed_triggers: set[str] = set() next_route_refresh = 0.0 pending_unrouted: list[StateEvent] = [] last_unrouted_flush = 0.0 while True: await asyncio.to_thread(_set_websocket_status, "connecting") try: async for event in client.listen_state_events(): loop_time = asyncio.get_running_loop().time() if loop_time >= next_route_refresh: routed_triggers = await asyncio.to_thread(_routed_trigger_ids) next_route_refresh = loop_time + 5 await asyncio.to_thread(_set_websocket_status, "connected") if event.entity_id in routed_triggers: if pending_unrouted: await asyncio.to_thread(_merge_unrouted_events, pending_unrouted) pending_unrouted = [] await asyncio.to_thread( event_core.process_state_event, event, execute=_execute_ha_decision, ) else: pending_unrouted.append(event) if len(pending_unrouted) >= 100 or loop_time - last_unrouted_flush >= 2: await asyncio.to_thread(_merge_unrouted_events, pending_unrouted) pending_unrouted = [] last_unrouted_flush = loop_time await asyncio.sleep(0) except asyncio.CancelledError: raise except Exception as exc: await asyncio.to_thread(_set_websocket_status, f"reconnecting: {exc}") await asyncio.sleep(2) def _set_websocket_status(status: str) -> None: runtime = stores.runtime() if runtime.websocket_status == status: return runtime.websocket_status = status stores.save_runtime(runtime) def _routed_trigger_ids() -> set[str]: result: set[str] = set() for profile in stores.learning().profiles.values(): for pattern in profile.patterns: if pattern.trigger_entity_id is not None: result.add(pattern.trigger_entity_id) return result def _merge_unrouted_events(events: list[StateEvent]) -> None: if not events: return runtime = stores.runtime() for event in events: runtime.entities[event.entity_id] = EntityState( entity_id=event.entity_id, domain=event.entity_id.split(".", 1)[0], state=event.new_state, changed_at=event.changed_at, area_name=event.attributes.get("area_name"), device_id=event.attributes.get("device_id"), friendly_name=event.attributes.get("friendly_name"), ) stores.save_runtime(runtime) def _execute_ha_decision(decision: object) -> bool: if ha_client is None or not hasattr(decision, "actuator_entity_id"): return False actuator_entity_id = str(decision.actuator_entity_id) target_state = getattr(decision, "target_state", None) service = service_for_state(actuator_entity_id.split(".", 1)[0], target_state) if service is None: return False domain, service_name = service ha_client.call_service(domain, service_name, {"entity_id": actuator_entity_id}) return True def _dashboard_html() -> str: return """ SillyHome Future

SillyHome Future

Audit

Lade...
"""