jpmschweitzerandClaude Opus 4.5
e5d50dda77
Build and Push / build (release) Successful in 56s
- Update all client endpoints to use /housekeeping/ prefix - Add critical rule requiring list_devices() before control actions - Add housekeeping API spec documentation 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
556 lines
15 KiB
Python
556 lines
15 KiB
Python
"""
|
|
HTTP client for the Core-API service.
|
|
|
|
Provides async methods for home automation operations via Home Assistant.
|
|
Core-API is a separate service that wraps the Home Assistant REST API
|
|
into LLM-friendly endpoints.
|
|
"""
|
|
from typing import Any, Optional
|
|
|
|
import httpx
|
|
from pydantic import BaseModel, Field
|
|
|
|
from src.core.config import config
|
|
from src.core.logging_config import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
# ============================================================================
|
|
# Response Models
|
|
# ============================================================================
|
|
|
|
|
|
class Device(BaseModel):
|
|
"""Device from Home Assistant."""
|
|
|
|
entity_id: str
|
|
name: str
|
|
state: str
|
|
domain: str
|
|
area: Optional[str] = None
|
|
attributes: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class DeviceState(BaseModel):
|
|
"""Detailed state of a device."""
|
|
|
|
entity_id: str
|
|
state: str
|
|
attributes: dict[str, Any] = Field(default_factory=dict)
|
|
last_changed: Optional[str] = None
|
|
last_updated: Optional[str] = None
|
|
|
|
|
|
class Scene(BaseModel):
|
|
"""Scene from Home Assistant."""
|
|
|
|
entity_id: str
|
|
name: str
|
|
friendly_name: Optional[str] = None
|
|
|
|
|
|
class Script(BaseModel):
|
|
"""Script from Home Assistant."""
|
|
|
|
entity_id: str
|
|
name: str
|
|
description: Optional[str] = None
|
|
last_triggered: Optional[str] = None
|
|
|
|
|
|
class Automation(BaseModel):
|
|
"""Automation from Home Assistant."""
|
|
|
|
entity_id: str
|
|
name: str
|
|
state: str = "on"
|
|
last_triggered: Optional[str] = None
|
|
|
|
|
|
class HistoryEntry(BaseModel):
|
|
"""History entry for an entity."""
|
|
|
|
state: str
|
|
timestamp: str
|
|
attributes: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class ControlResult(BaseModel):
|
|
"""Result of a device control operation."""
|
|
|
|
success: bool
|
|
entity_id: str
|
|
action: str
|
|
message: str = ""
|
|
|
|
|
|
class Area(BaseModel):
|
|
"""Area/room from Home Assistant."""
|
|
|
|
area_id: str
|
|
name: str
|
|
device_count: int = 0
|
|
|
|
|
|
# ============================================================================
|
|
# Client
|
|
# ============================================================================
|
|
|
|
|
|
class CoreAPIClient:
|
|
"""
|
|
Async HTTP client for Core-API (Home Assistant wrapper).
|
|
|
|
Usage:
|
|
async with CoreAPIClient() as client:
|
|
devices = await client.list_devices()
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: Optional[str] = None,
|
|
api_key: Optional[str] = None,
|
|
timeout: int = 30,
|
|
):
|
|
"""
|
|
Initialize the client.
|
|
|
|
Args:
|
|
base_url: Core-API URL (defaults to config)
|
|
api_key: API key for authentication (defaults to config)
|
|
timeout: Request timeout in seconds
|
|
"""
|
|
self.base_url = base_url or str(config.CORE_API_HOST)
|
|
self.api_key = api_key or config.CORE_API_KEY
|
|
self.timeout = timeout
|
|
self._client: Optional[httpx.AsyncClient] = None
|
|
|
|
async def __aenter__(self) -> "CoreAPIClient":
|
|
"""Create HTTP client on context entry."""
|
|
headers = {}
|
|
if self.api_key:
|
|
headers["Authorization"] = f"Bearer {self.api_key}"
|
|
|
|
self._client = httpx.AsyncClient(
|
|
base_url=self.base_url,
|
|
headers=headers,
|
|
timeout=self.timeout,
|
|
)
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
|
|
"""Close HTTP client on context exit."""
|
|
if self._client:
|
|
await self._client.aclose()
|
|
self._client = None
|
|
|
|
def _ensure_client(self) -> httpx.AsyncClient:
|
|
"""Ensure client is initialized."""
|
|
if self._client is None:
|
|
raise RuntimeError(
|
|
"Client not initialized. Use 'async with CoreAPIClient() as client:'"
|
|
)
|
|
return self._client
|
|
|
|
# ========================================================================
|
|
# Device Discovery
|
|
# ========================================================================
|
|
|
|
async def list_devices(
|
|
self,
|
|
domain: Optional[str] = None,
|
|
area: Optional[str] = None,
|
|
) -> list[Device]:
|
|
"""
|
|
List devices, optionally filtered by domain or area.
|
|
|
|
Args:
|
|
domain: Filter by domain (light, switch, climate, etc.)
|
|
area: Filter by area (living_room, bedroom, etc.)
|
|
|
|
Returns:
|
|
List of devices matching filters
|
|
"""
|
|
client = self._ensure_client()
|
|
|
|
params: dict[str, str] = {}
|
|
if domain:
|
|
params["domain"] = domain
|
|
if area:
|
|
params["area"] = area
|
|
|
|
logger.debug("core_api_list_devices", domain=domain, area=area)
|
|
|
|
response = await client.get("/housekeeping/devices", params=params or None)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
return [Device(**d) for d in data.get("devices", [])]
|
|
|
|
async def list_areas(self) -> list[Area]:
|
|
"""
|
|
List all areas/rooms in Home Assistant.
|
|
|
|
Returns:
|
|
List of areas with device counts
|
|
"""
|
|
client = self._ensure_client()
|
|
|
|
logger.debug("core_api_list_areas")
|
|
|
|
response = await client.get("/housekeeping/areas")
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
return [Area(**a) for a in data.get("areas", [])]
|
|
|
|
async def get_device_state(self, entity_id: str) -> DeviceState:
|
|
"""
|
|
Get the current state of a specific device.
|
|
|
|
Args:
|
|
entity_id: Home Assistant entity ID (e.g., light.living_room)
|
|
|
|
Returns:
|
|
Current device state with attributes
|
|
"""
|
|
client = self._ensure_client()
|
|
|
|
logger.debug("core_api_get_state", entity_id=entity_id)
|
|
|
|
response = await client.get(f"/housekeeping/devices/{entity_id}")
|
|
response.raise_for_status()
|
|
|
|
return DeviceState(**response.json())
|
|
|
|
# ========================================================================
|
|
# Device Control
|
|
# ========================================================================
|
|
|
|
async def turn_on(
|
|
self,
|
|
entity_id: str,
|
|
brightness: Optional[int] = None,
|
|
color_temp: Optional[int] = None,
|
|
rgb_color: Optional[tuple[int, int, int]] = None,
|
|
) -> ControlResult:
|
|
"""
|
|
Turn on a device.
|
|
|
|
Args:
|
|
entity_id: Device to turn on
|
|
brightness: Optional brightness (0-255) for lights
|
|
color_temp: Optional color temperature in Kelvin for lights
|
|
rgb_color: Optional RGB color tuple for lights
|
|
|
|
Returns:
|
|
Result of the operation
|
|
"""
|
|
client = self._ensure_client()
|
|
|
|
payload: dict[str, Any] = {"action": "turn_on"}
|
|
if brightness is not None:
|
|
payload["brightness"] = brightness
|
|
if color_temp is not None:
|
|
payload["color_temp"] = color_temp
|
|
if rgb_color is not None:
|
|
payload["rgb_color"] = list(rgb_color)
|
|
|
|
logger.info("core_api_turn_on", entity_id=entity_id, payload=payload)
|
|
|
|
response = await client.post(
|
|
f"/housekeeping/devices/{entity_id}/control",
|
|
json=payload,
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
return ControlResult(
|
|
success=data.get("success", True),
|
|
entity_id=entity_id,
|
|
action="turn_on",
|
|
message=data.get("message", ""),
|
|
)
|
|
|
|
async def turn_off(self, entity_id: str) -> ControlResult:
|
|
"""
|
|
Turn off a device.
|
|
|
|
Args:
|
|
entity_id: Device to turn off
|
|
|
|
Returns:
|
|
Result of the operation
|
|
"""
|
|
client = self._ensure_client()
|
|
|
|
logger.info("core_api_turn_off", entity_id=entity_id)
|
|
|
|
response = await client.post(
|
|
f"/housekeeping/devices/{entity_id}/control",
|
|
json={"action": "turn_off"},
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
return ControlResult(
|
|
success=data.get("success", True),
|
|
entity_id=entity_id,
|
|
action="turn_off",
|
|
message=data.get("message", ""),
|
|
)
|
|
|
|
async def toggle(self, entity_id: str) -> ControlResult:
|
|
"""
|
|
Toggle a device's state.
|
|
|
|
Args:
|
|
entity_id: Device to toggle
|
|
|
|
Returns:
|
|
Result of the operation
|
|
"""
|
|
client = self._ensure_client()
|
|
|
|
logger.info("core_api_toggle", entity_id=entity_id)
|
|
|
|
response = await client.post(
|
|
f"/housekeeping/devices/{entity_id}/control",
|
|
json={"action": "toggle"},
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
return ControlResult(
|
|
success=data.get("success", True),
|
|
entity_id=entity_id,
|
|
action="toggle",
|
|
message=data.get("message", ""),
|
|
)
|
|
|
|
# ========================================================================
|
|
# Scenes
|
|
# ========================================================================
|
|
|
|
async def list_scenes(self) -> list[Scene]:
|
|
"""
|
|
List all available scenes.
|
|
|
|
Returns:
|
|
List of scenes
|
|
"""
|
|
client = self._ensure_client()
|
|
|
|
logger.debug("core_api_list_scenes")
|
|
|
|
response = await client.get("/housekeeping/scenes")
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
return [Scene(**s) for s in data.get("scenes", [])]
|
|
|
|
async def activate_scene(self, scene_id: str) -> ControlResult:
|
|
"""
|
|
Activate a scene.
|
|
|
|
Args:
|
|
scene_id: Scene entity ID (e.g., scene.movie_night)
|
|
|
|
Returns:
|
|
Result of the operation
|
|
"""
|
|
client = self._ensure_client()
|
|
|
|
logger.info("core_api_activate_scene", scene_id=scene_id)
|
|
|
|
response = await client.post(f"/housekeeping/scenes/{scene_id}/activate")
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
return ControlResult(
|
|
success=data.get("success", True),
|
|
entity_id=scene_id,
|
|
action="activate",
|
|
message=data.get("message", ""),
|
|
)
|
|
|
|
# ========================================================================
|
|
# Scripts
|
|
# ========================================================================
|
|
|
|
async def list_scripts(self) -> list[Script]:
|
|
"""
|
|
List all available scripts.
|
|
|
|
Returns:
|
|
List of scripts
|
|
"""
|
|
client = self._ensure_client()
|
|
|
|
logger.debug("core_api_list_scripts")
|
|
|
|
response = await client.get("/housekeeping/scripts")
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
return [Script(**s) for s in data.get("scripts", [])]
|
|
|
|
async def run_script(
|
|
self,
|
|
script_id: str,
|
|
variables: Optional[dict[str, Any]] = None,
|
|
) -> ControlResult:
|
|
"""
|
|
Run a script.
|
|
|
|
Args:
|
|
script_id: Script entity ID (e.g., script.good_morning)
|
|
variables: Optional variables to pass to the script
|
|
|
|
Returns:
|
|
Result of the operation
|
|
"""
|
|
client = self._ensure_client()
|
|
|
|
payload: dict[str, Any] = {}
|
|
if variables:
|
|
payload["variables"] = variables
|
|
|
|
logger.info("core_api_run_script", script_id=script_id)
|
|
|
|
response = await client.post(
|
|
f"/housekeeping/scripts/{script_id}/run",
|
|
json=payload or None,
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
return ControlResult(
|
|
success=data.get("success", True),
|
|
entity_id=script_id,
|
|
action="run",
|
|
message=data.get("message", ""),
|
|
)
|
|
|
|
# ========================================================================
|
|
# Automations
|
|
# ========================================================================
|
|
|
|
async def list_automations(self) -> list[Automation]:
|
|
"""
|
|
List all automations.
|
|
|
|
Returns:
|
|
List of automations with their states
|
|
"""
|
|
client = self._ensure_client()
|
|
|
|
logger.debug("core_api_list_automations")
|
|
|
|
response = await client.get("/housekeeping/automations")
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
return [Automation(**a) for a in data.get("automations", [])]
|
|
|
|
async def toggle_automation(
|
|
self,
|
|
automation_id: str,
|
|
enable: bool,
|
|
) -> ControlResult:
|
|
"""
|
|
Enable or disable an automation.
|
|
|
|
Args:
|
|
automation_id: Automation entity ID
|
|
enable: True to enable, False to disable
|
|
|
|
Returns:
|
|
Result of the operation
|
|
"""
|
|
client = self._ensure_client()
|
|
|
|
logger.info(
|
|
"core_api_toggle_automation",
|
|
automation_id=automation_id,
|
|
enable=enable,
|
|
)
|
|
|
|
response = await client.post(
|
|
f"/housekeeping/automations/{automation_id}/toggle",
|
|
json={"enable": enable},
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
return ControlResult(
|
|
success=data.get("success", True),
|
|
entity_id=automation_id,
|
|
action="enable" if enable else "disable",
|
|
message=data.get("message", ""),
|
|
)
|
|
|
|
# ========================================================================
|
|
# History
|
|
# ========================================================================
|
|
|
|
async def get_history(
|
|
self,
|
|
entity_id: str,
|
|
hours: int = 24,
|
|
) -> list[HistoryEntry]:
|
|
"""
|
|
Get history for an entity.
|
|
|
|
Args:
|
|
entity_id: Entity to get history for
|
|
hours: Number of hours of history (default: 24)
|
|
|
|
Returns:
|
|
List of historical state entries
|
|
"""
|
|
client = self._ensure_client()
|
|
|
|
logger.debug("core_api_get_history", entity_id=entity_id, hours=hours)
|
|
|
|
response = await client.get(
|
|
"/housekeeping/history",
|
|
params={"entity_id": entity_id, "hours": hours},
|
|
)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
return [HistoryEntry(**h) for h in data.get("history", [])]
|
|
|
|
# ========================================================================
|
|
# Health Check
|
|
# ========================================================================
|
|
|
|
async def health_check(self) -> bool:
|
|
"""
|
|
Check if core-api and Home Assistant are healthy.
|
|
|
|
Returns:
|
|
True if healthy, False otherwise
|
|
"""
|
|
try:
|
|
client = self._ensure_client()
|
|
response = await client.get("/housekeeping/health")
|
|
return response.status_code == 200
|
|
except Exception as e:
|
|
logger.warning("core_api_health_check_failed", error=str(e))
|
|
return False
|
|
|
|
|
|
# Global client factory
|
|
async def get_core_api_client() -> CoreAPIClient:
|
|
"""
|
|
Get a core-api client instance.
|
|
|
|
Usage:
|
|
async with get_core_api_client() as client:
|
|
devices = await client.list_devices()
|
|
"""
|
|
return CoreAPIClient()
|