Jeroen SchweitzerandClaude Opus 4.5
381d43b60b
Build and Push / build (release) Successful in 1m28s
- Dashboard domain with Quick Links CRUD + reorder endpoints - Dashboard widgets management endpoints - Database migrations for quick_links and dashboard_widgets tables - Static file controller for Organizr widgets - Default local user when OIDC is disabled - Domain-based architecture refactor (src/domains/, src/shared/) - Test suite updated for new structure (285 tests passing) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
646 lines
21 KiB
Python
646 lines
21 KiB
Python
"""
|
|
Housekeeping Controller
|
|
|
|
Provides API endpoints for home automation via Home Assistant.
|
|
Designed for the Tatlock Housekeeper agent and other consumers.
|
|
"""
|
|
import asyncio
|
|
from fastapi import APIRouter, HTTPException, Query, Depends
|
|
from typing import List, Dict, Any, Optional
|
|
from pydantic import BaseModel, Field
|
|
|
|
from src.shared.base import BaseController
|
|
from src.shared.clients import get_homeassistant_client
|
|
from src.shared.logging import get_logger
|
|
from src.domains.auth.oidc import get_admin_user
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
# Pydantic Schemas
|
|
class Device(BaseModel):
|
|
"""Device/entity information"""
|
|
entity_id: str
|
|
name: str
|
|
domain: str
|
|
area: Optional[str] = None
|
|
state: str
|
|
attributes: Dict[str, Any] = {}
|
|
last_changed: Optional[str] = None
|
|
|
|
|
|
class DeviceListResponse(BaseModel):
|
|
"""Response for device listing"""
|
|
devices: List[Device]
|
|
|
|
|
|
class DeviceDetailResponse(Device):
|
|
"""Detailed device response"""
|
|
pass
|
|
|
|
|
|
class Area(BaseModel):
|
|
"""Area/room information"""
|
|
id: str
|
|
name: str
|
|
|
|
|
|
class AreaListResponse(BaseModel):
|
|
"""Response for area listing"""
|
|
areas: List[Area]
|
|
|
|
|
|
class DeviceControlRequest(BaseModel):
|
|
"""Request to control a device"""
|
|
action: str = Field(..., description="Action: turn_on, turn_off, or toggle")
|
|
brightness: Optional[int] = Field(None, ge=0, le=255)
|
|
color_temp: Optional[int] = None
|
|
rgb_color: Optional[List[int]] = None
|
|
|
|
class Config:
|
|
extra = "allow"
|
|
|
|
|
|
class DeviceControlResponse(BaseModel):
|
|
"""Response from device control"""
|
|
success: bool
|
|
entity_id: str
|
|
new_state: Optional[str] = None
|
|
message: str
|
|
|
|
|
|
class Scene(BaseModel):
|
|
"""Scene information"""
|
|
id: str
|
|
name: str
|
|
|
|
|
|
class SceneListResponse(BaseModel):
|
|
"""Response for scene listing"""
|
|
scenes: List[Scene]
|
|
|
|
|
|
class SceneActivateResponse(BaseModel):
|
|
"""Response from scene activation"""
|
|
success: bool
|
|
scene_id: str
|
|
message: str
|
|
|
|
|
|
class Script(BaseModel):
|
|
"""Script information"""
|
|
id: str
|
|
name: str
|
|
|
|
|
|
class ScriptListResponse(BaseModel):
|
|
"""Response for script listing"""
|
|
scripts: List[Script]
|
|
|
|
|
|
class ScriptRunRequest(BaseModel):
|
|
"""Request to run a script"""
|
|
variables: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
class ScriptRunResponse(BaseModel):
|
|
"""Response from script execution"""
|
|
success: bool
|
|
script_id: str
|
|
message: str
|
|
|
|
|
|
class Automation(BaseModel):
|
|
"""Automation information"""
|
|
id: str
|
|
name: str
|
|
enabled: bool
|
|
|
|
|
|
class AutomationListResponse(BaseModel):
|
|
"""Response for automation listing"""
|
|
automations: List[Automation]
|
|
|
|
|
|
class AutomationToggleRequest(BaseModel):
|
|
"""Request to toggle automation"""
|
|
enabled: bool
|
|
|
|
|
|
class AutomationToggleResponse(BaseModel):
|
|
"""Response from automation toggle"""
|
|
success: bool
|
|
automation_id: str
|
|
enabled: bool
|
|
message: str
|
|
|
|
|
|
class HistoryEntry(BaseModel):
|
|
"""Single history entry"""
|
|
state: str
|
|
timestamp: str
|
|
attributes: Dict[str, Any] = {}
|
|
|
|
|
|
class HistoryResponse(BaseModel):
|
|
"""Response for history query"""
|
|
entity_id: str
|
|
history: List[HistoryEntry]
|
|
|
|
|
|
class HealthResponse(BaseModel):
|
|
"""Health check response"""
|
|
status: str
|
|
connected: bool
|
|
platform: str
|
|
version: Optional[str] = None
|
|
error: Optional[str] = None
|
|
|
|
|
|
class ErrorResponse(BaseModel):
|
|
"""Standard error response"""
|
|
error: bool = True
|
|
code: str
|
|
message: str
|
|
|
|
|
|
class HousekeepingController(BaseController):
|
|
"""
|
|
Controller for home automation operations
|
|
|
|
Provides endpoints for:
|
|
- Device discovery and control
|
|
- Scene activation
|
|
- Script execution
|
|
- Automation management
|
|
- State history
|
|
"""
|
|
|
|
def __init__(self):
|
|
super().__init__(prefix="/housekeeping", tags=["Housekeeping"])
|
|
|
|
def create_router(self) -> APIRouter:
|
|
"""Create and configure the router"""
|
|
router = APIRouter(prefix=self.prefix, tags=self.tags)
|
|
|
|
@router.get(
|
|
"/health",
|
|
response_model=HealthResponse,
|
|
summary="Home automation health check"
|
|
)
|
|
async def get_health():
|
|
"""Check Home Assistant connection health"""
|
|
ha = get_homeassistant_client()
|
|
return await ha.health_check()
|
|
|
|
@router.get(
|
|
"/devices",
|
|
response_model=DeviceListResponse,
|
|
summary="List available devices"
|
|
)
|
|
async def list_devices(
|
|
domain: Optional[str] = Query(None, description="Filter by domain (light, switch, climate, etc.)"),
|
|
area: Optional[str] = Query(None, description="Filter by area/room name")
|
|
):
|
|
"""List all available devices with optional filtering"""
|
|
ha = get_homeassistant_client()
|
|
|
|
try:
|
|
states = await ha.get_states()
|
|
|
|
excluded_domains = {
|
|
"zone", "person", "device_tracker", "sun", "weather",
|
|
"persistent_notification", "update", "binary_sensor", "sensor",
|
|
"conversation", "calendar", "button", "number", "select",
|
|
"text", "time", "date", "datetime", "image", "tts", "stt"
|
|
}
|
|
|
|
devices = []
|
|
for state in states:
|
|
entity_id = state.get("entity_id", "")
|
|
entity_domain = entity_id.split(".")[0] if "." in entity_id else ""
|
|
|
|
if entity_domain in excluded_domains:
|
|
continue
|
|
|
|
if domain and entity_domain != domain:
|
|
continue
|
|
|
|
device_area = state.get("attributes", {}).get("area_id")
|
|
|
|
if area and device_area and area.lower() not in device_area.lower():
|
|
continue
|
|
|
|
device = Device(
|
|
entity_id=entity_id,
|
|
name=state.get("attributes", {}).get("friendly_name", entity_id),
|
|
domain=entity_domain,
|
|
area=device_area,
|
|
state=state.get("state", "unknown"),
|
|
attributes=state.get("attributes", {}),
|
|
last_changed=state.get("last_changed")
|
|
)
|
|
devices.append(device)
|
|
|
|
return DeviceListResponse(devices=devices)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to list devices: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error": True, "code": "CONNECTION_ERROR", "message": str(e)}
|
|
)
|
|
|
|
@router.get(
|
|
"/devices/{entity_id:path}",
|
|
response_model=DeviceDetailResponse,
|
|
responses={404: {"model": ErrorResponse}}
|
|
)
|
|
async def get_device(entity_id: str):
|
|
"""Get detailed state of a specific device"""
|
|
ha = get_homeassistant_client()
|
|
|
|
try:
|
|
state = await ha.get_state(entity_id)
|
|
|
|
if not state:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail={"error": True, "code": "DEVICE_NOT_FOUND",
|
|
"message": f"Device {entity_id} not found"}
|
|
)
|
|
|
|
entity_domain = entity_id.split(".")[0] if "." in entity_id else ""
|
|
|
|
return DeviceDetailResponse(
|
|
entity_id=entity_id,
|
|
name=state.get("attributes", {}).get("friendly_name", entity_id),
|
|
domain=entity_domain,
|
|
area=state.get("attributes", {}).get("area_id"),
|
|
state=state.get("state", "unknown"),
|
|
attributes=state.get("attributes", {}),
|
|
last_changed=state.get("last_changed")
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Failed to get device {entity_id}: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error": True, "code": "CONNECTION_ERROR", "message": str(e)}
|
|
)
|
|
|
|
@router.get(
|
|
"/areas",
|
|
response_model=AreaListResponse,
|
|
summary="List areas/rooms"
|
|
)
|
|
async def list_areas():
|
|
"""List all configured areas/rooms in Home Assistant"""
|
|
ha = get_homeassistant_client()
|
|
|
|
try:
|
|
areas = await ha.get_areas()
|
|
return AreaListResponse(
|
|
areas=[Area(id=a["id"], name=a["name"]) for a in areas]
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Failed to list areas: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error": True, "code": "CONNECTION_ERROR", "message": str(e)}
|
|
)
|
|
|
|
@router.post(
|
|
"/devices/{entity_id:path}/control",
|
|
response_model=DeviceControlResponse,
|
|
responses={404: {"model": ErrorResponse}, 400: {"model": ErrorResponse}}
|
|
)
|
|
async def control_device(
|
|
entity_id: str,
|
|
request: DeviceControlRequest,
|
|
user: Dict = Depends(get_admin_user)
|
|
):
|
|
"""Control a device (turn on, turn off, toggle, or set attributes)"""
|
|
ha = get_homeassistant_client()
|
|
|
|
valid_actions = ["turn_on", "turn_off", "toggle"]
|
|
if request.action not in valid_actions:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail={"error": True, "code": "INVALID_ACTION",
|
|
"message": f"Invalid action '{request.action}'. Must be one of: {', '.join(valid_actions)}"}
|
|
)
|
|
|
|
try:
|
|
current_state = await ha.get_state(entity_id)
|
|
if not current_state:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail={"error": True, "code": "DEVICE_NOT_FOUND",
|
|
"message": f"Device {entity_id} not found"}
|
|
)
|
|
|
|
attributes = {}
|
|
if request.brightness is not None:
|
|
attributes["brightness"] = request.brightness
|
|
if request.color_temp is not None:
|
|
attributes["color_temp"] = request.color_temp
|
|
if request.rgb_color is not None:
|
|
attributes["rgb_color"] = request.rgb_color
|
|
|
|
extra_fields = request.model_dump(exclude={"action", "brightness", "color_temp", "rgb_color"})
|
|
for key, value in extra_fields.items():
|
|
if value is not None:
|
|
attributes[key] = value
|
|
|
|
if request.action == "turn_on":
|
|
await ha.turn_on(entity_id, **attributes)
|
|
elif request.action == "turn_off":
|
|
await ha.turn_off(entity_id)
|
|
else:
|
|
await ha.toggle(entity_id)
|
|
|
|
await asyncio.sleep(0.3)
|
|
|
|
new_state = await ha.get_state(entity_id)
|
|
|
|
logger.info(f"Device {entity_id} controlled: {request.action} by {user.get('preferred_username', 'unknown')}")
|
|
|
|
return DeviceControlResponse(
|
|
success=True,
|
|
entity_id=entity_id,
|
|
new_state=new_state.get("state") if new_state else None,
|
|
message=f"Device {request.action} successful"
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Failed to control device {entity_id}: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error": True, "code": "CONNECTION_ERROR", "message": str(e)}
|
|
)
|
|
|
|
@router.get(
|
|
"/scenes",
|
|
response_model=SceneListResponse,
|
|
summary="List available scenes"
|
|
)
|
|
async def list_scenes():
|
|
"""List all available scenes in Home Assistant"""
|
|
ha = get_homeassistant_client()
|
|
|
|
try:
|
|
states = await ha.get_states()
|
|
|
|
scenes = [
|
|
Scene(
|
|
id=s["entity_id"],
|
|
name=s.get("attributes", {}).get("friendly_name", s["entity_id"])
|
|
)
|
|
for s in states
|
|
if s["entity_id"].startswith("scene.")
|
|
]
|
|
|
|
return SceneListResponse(scenes=scenes)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to list scenes: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error": True, "code": "CONNECTION_ERROR", "message": str(e)}
|
|
)
|
|
|
|
@router.post(
|
|
"/scenes/{scene_id:path}/activate",
|
|
response_model=SceneActivateResponse,
|
|
responses={404: {"model": ErrorResponse}}
|
|
)
|
|
async def activate_scene(
|
|
scene_id: str,
|
|
user: Dict = Depends(get_admin_user)
|
|
):
|
|
"""Activate a scene"""
|
|
ha = get_homeassistant_client()
|
|
|
|
try:
|
|
state = await ha.get_state(scene_id)
|
|
if not state:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail={"error": True, "code": "SCENE_NOT_FOUND",
|
|
"message": f"Scene {scene_id} not found"}
|
|
)
|
|
|
|
await ha.activate_scene(scene_id)
|
|
|
|
logger.info(f"Scene {scene_id} activated by {user.get('preferred_username', 'unknown')}")
|
|
|
|
return SceneActivateResponse(
|
|
success=True,
|
|
scene_id=scene_id,
|
|
message="Scene activated"
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Failed to activate scene {scene_id}: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error": True, "code": "CONNECTION_ERROR", "message": str(e)}
|
|
)
|
|
|
|
@router.get(
|
|
"/scripts",
|
|
response_model=ScriptListResponse,
|
|
summary="List available scripts"
|
|
)
|
|
async def list_scripts():
|
|
"""List all available scripts/sequences in Home Assistant"""
|
|
ha = get_homeassistant_client()
|
|
|
|
try:
|
|
states = await ha.get_states()
|
|
|
|
scripts = [
|
|
Script(
|
|
id=s["entity_id"],
|
|
name=s.get("attributes", {}).get("friendly_name", s["entity_id"])
|
|
)
|
|
for s in states
|
|
if s["entity_id"].startswith("script.")
|
|
]
|
|
|
|
return ScriptListResponse(scripts=scripts)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to list scripts: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error": True, "code": "CONNECTION_ERROR", "message": str(e)}
|
|
)
|
|
|
|
@router.post(
|
|
"/scripts/{script_id:path}/run",
|
|
response_model=ScriptRunResponse,
|
|
responses={404: {"model": ErrorResponse}}
|
|
)
|
|
async def run_script(
|
|
script_id: str,
|
|
request: Optional[ScriptRunRequest] = None,
|
|
user: Dict = Depends(get_admin_user)
|
|
):
|
|
"""Execute a script with optional variables"""
|
|
ha = get_homeassistant_client()
|
|
|
|
try:
|
|
state = await ha.get_state(script_id)
|
|
if not state:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail={"error": True, "code": "SCRIPT_NOT_FOUND",
|
|
"message": f"Script {script_id} not found"}
|
|
)
|
|
|
|
variables = request.variables if request else None
|
|
await ha.run_script(script_id, variables)
|
|
|
|
logger.info(f"Script {script_id} executed by {user.get('preferred_username', 'unknown')}")
|
|
|
|
return ScriptRunResponse(
|
|
success=True,
|
|
script_id=script_id,
|
|
message="Script executed"
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Failed to run script {script_id}: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error": True, "code": "CONNECTION_ERROR", "message": str(e)}
|
|
)
|
|
|
|
@router.get(
|
|
"/automations",
|
|
response_model=AutomationListResponse,
|
|
summary="List automations"
|
|
)
|
|
async def list_automations():
|
|
"""List all automations with their enabled/disabled status"""
|
|
ha = get_homeassistant_client()
|
|
|
|
try:
|
|
states = await ha.get_states()
|
|
|
|
automations = [
|
|
Automation(
|
|
id=s["entity_id"],
|
|
name=s.get("attributes", {}).get("friendly_name", s["entity_id"]),
|
|
enabled=s.get("state") == "on"
|
|
)
|
|
for s in states
|
|
if s["entity_id"].startswith("automation.")
|
|
]
|
|
|
|
return AutomationListResponse(automations=automations)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to list automations: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error": True, "code": "CONNECTION_ERROR", "message": str(e)}
|
|
)
|
|
|
|
@router.post(
|
|
"/automations/{automation_id:path}/toggle",
|
|
response_model=AutomationToggleResponse,
|
|
responses={404: {"model": ErrorResponse}}
|
|
)
|
|
async def toggle_automation(
|
|
automation_id: str,
|
|
request: AutomationToggleRequest,
|
|
user: Dict = Depends(get_admin_user)
|
|
):
|
|
"""Enable or disable an automation"""
|
|
ha = get_homeassistant_client()
|
|
|
|
try:
|
|
state = await ha.get_state(automation_id)
|
|
if not state:
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail={"error": True, "code": "AUTOMATION_NOT_FOUND",
|
|
"message": f"Automation {automation_id} not found"}
|
|
)
|
|
|
|
if request.enabled:
|
|
await ha.enable_automation(automation_id)
|
|
else:
|
|
await ha.disable_automation(automation_id)
|
|
|
|
logger.info(f"Automation {automation_id} {'enabled' if request.enabled else 'disabled'} by {user.get('preferred_username', 'unknown')}")
|
|
|
|
return AutomationToggleResponse(
|
|
success=True,
|
|
automation_id=automation_id,
|
|
enabled=request.enabled,
|
|
message=f"Automation {'enabled' if request.enabled else 'disabled'}"
|
|
)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Failed to toggle automation {automation_id}: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error": True, "code": "CONNECTION_ERROR", "message": str(e)}
|
|
)
|
|
|
|
@router.get(
|
|
"/history",
|
|
response_model=HistoryResponse,
|
|
responses={400: {"model": ErrorResponse}}
|
|
)
|
|
async def get_history(
|
|
entity_id: str = Query(..., description="Entity ID to get history for"),
|
|
hours: int = Query(24, ge=1, le=168, description="Hours of history (1-168)")
|
|
):
|
|
"""Get state history for a device"""
|
|
ha = get_homeassistant_client()
|
|
|
|
try:
|
|
history_data = await ha.get_history(entity_id, hours)
|
|
|
|
history_entries = []
|
|
if history_data and len(history_data) > 0:
|
|
for entry in history_data[0]:
|
|
history_entries.append(HistoryEntry(
|
|
state=entry.get("state", "unknown"),
|
|
timestamp=entry.get("last_changed", ""),
|
|
attributes=entry.get("attributes", {})
|
|
))
|
|
|
|
return HistoryResponse(
|
|
entity_id=entity_id,
|
|
history=history_entries
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get history for {entity_id}: {e}")
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail={"error": True, "code": "CONNECTION_ERROR", "message": str(e)}
|
|
)
|
|
|
|
return router
|
|
|
|
|
|
# Create controller instance
|
|
housekeeping_controller = HousekeepingController()
|