jpmschweitzerandClaude Opus 4.5
c45fdcb528
Introduces home automation endpoints for the Tatlock Housekeeper agent: - Device discovery and control (turn_on, turn_off, toggle, set_brightness) - Scene activation and script execution - Automation management (enable/disable) - State history queries and area discovery Includes Home Assistant REST client and configuration. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
410 lines
12 KiB
Python
410 lines
12 KiB
Python
"""
|
|
Home Assistant REST API Client
|
|
|
|
Provides interface to Home Assistant REST API for home automation control.
|
|
Uses long-lived access token authentication.
|
|
API Reference: https://developers.home-assistant.io/docs/api/rest/
|
|
"""
|
|
import httpx
|
|
import json
|
|
from typing import Optional, Dict, List, Any
|
|
from datetime import datetime, timedelta, timezone
|
|
from src.logging_config import get_logger
|
|
from src.config import get_settings
|
|
|
|
logger = get_logger(__name__)
|
|
settings = get_settings()
|
|
|
|
|
|
class HomeAssistantClient:
|
|
"""
|
|
HTTP client for Home Assistant REST API
|
|
|
|
Uses long-lived access token authentication via Bearer token.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: Optional[str] = None,
|
|
token: Optional[str] = None,
|
|
timeout: int = 30
|
|
):
|
|
"""
|
|
Initialize Home Assistant client
|
|
|
|
Args:
|
|
base_url: Home Assistant base URL (default from settings)
|
|
token: Long-lived access token (default from settings)
|
|
timeout: Request timeout in seconds
|
|
"""
|
|
self.base_url = (base_url or settings.homeassistant_url).rstrip("/")
|
|
self.token = token or settings.homeassistant_token
|
|
self.timeout = timeout
|
|
|
|
if not self.token:
|
|
logger.warning("Home Assistant token not configured")
|
|
|
|
def _get_headers(self) -> Dict[str, str]:
|
|
"""Get request headers with Bearer token authentication"""
|
|
return {
|
|
"Authorization": f"Bearer {self.token}",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
# ========================================================================
|
|
# Health & Discovery
|
|
# ========================================================================
|
|
|
|
async def health_check(self) -> Dict[str, Any]:
|
|
"""
|
|
Check Home Assistant API connectivity and get version info
|
|
|
|
HA Endpoint: GET /api/
|
|
|
|
Returns:
|
|
Dict with connected status, platform name, and version
|
|
"""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/",
|
|
headers=self._get_headers()
|
|
)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
return {
|
|
"status": "healthy",
|
|
"connected": True,
|
|
"platform": "home_assistant",
|
|
"version": data.get("version", "unknown")
|
|
}
|
|
return {
|
|
"status": "unhealthy",
|
|
"connected": False,
|
|
"platform": "home_assistant",
|
|
"error": f"HTTP {response.status_code}"
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Home Assistant health check failed: {e}")
|
|
return {
|
|
"status": "unhealthy",
|
|
"connected": False,
|
|
"platform": "home_assistant",
|
|
"error": str(e)
|
|
}
|
|
|
|
async def get_states(self) -> List[Dict[str, Any]]:
|
|
"""
|
|
Get all entity states
|
|
|
|
HA Endpoint: GET /api/states
|
|
|
|
Returns:
|
|
List of all entity states
|
|
"""
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/states",
|
|
headers=self._get_headers()
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def get_state(self, entity_id: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get state of a specific entity
|
|
|
|
HA Endpoint: GET /api/states/<entity_id>
|
|
|
|
Args:
|
|
entity_id: Entity ID (e.g., "light.living_room")
|
|
|
|
Returns:
|
|
Entity state dict or None if not found
|
|
"""
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/states/{entity_id}",
|
|
headers=self._get_headers()
|
|
)
|
|
if response.status_code == 404:
|
|
return None
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def get_config(self) -> Dict[str, Any]:
|
|
"""
|
|
Get Home Assistant configuration (includes areas)
|
|
|
|
HA Endpoint: GET /api/config
|
|
|
|
Returns:
|
|
Configuration dict including components, location, etc.
|
|
"""
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/config",
|
|
headers=self._get_headers()
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
# ========================================================================
|
|
# Device Control
|
|
# ========================================================================
|
|
|
|
async def call_service(
|
|
self,
|
|
domain: str,
|
|
service: str,
|
|
entity_id: Optional[str] = None,
|
|
service_data: Optional[Dict[str, Any]] = None
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Call a Home Assistant service
|
|
|
|
HA Endpoint: POST /api/services/<domain>/<service>
|
|
|
|
Args:
|
|
domain: Service domain (e.g., "light", "switch", "scene")
|
|
service: Service name (e.g., "turn_on", "turn_off", "toggle")
|
|
entity_id: Target entity ID (optional for some services)
|
|
service_data: Additional service data/attributes
|
|
|
|
Returns:
|
|
List of changed states
|
|
"""
|
|
payload = service_data.copy() if service_data else {}
|
|
if entity_id:
|
|
payload["entity_id"] = entity_id
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.post(
|
|
f"{self.base_url}/api/services/{domain}/{service}",
|
|
headers=self._get_headers(),
|
|
json=payload
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def turn_on(
|
|
self,
|
|
entity_id: str,
|
|
**attributes
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Turn on an entity with optional attributes
|
|
|
|
Args:
|
|
entity_id: Entity ID (e.g., "light.living_room")
|
|
**attributes: Additional attributes (brightness, color_temp, etc.)
|
|
|
|
Returns:
|
|
List of changed states
|
|
"""
|
|
domain = entity_id.split(".")[0]
|
|
return await self.call_service(
|
|
domain=domain,
|
|
service="turn_on",
|
|
entity_id=entity_id,
|
|
service_data=attributes if attributes else None
|
|
)
|
|
|
|
async def turn_off(self, entity_id: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
Turn off an entity
|
|
|
|
Args:
|
|
entity_id: Entity ID
|
|
|
|
Returns:
|
|
List of changed states
|
|
"""
|
|
domain = entity_id.split(".")[0]
|
|
return await self.call_service(
|
|
domain=domain,
|
|
service="turn_off",
|
|
entity_id=entity_id
|
|
)
|
|
|
|
async def toggle(self, entity_id: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
Toggle an entity
|
|
|
|
Args:
|
|
entity_id: Entity ID
|
|
|
|
Returns:
|
|
List of changed states
|
|
"""
|
|
domain = entity_id.split(".")[0]
|
|
return await self.call_service(
|
|
domain=domain,
|
|
service="toggle",
|
|
entity_id=entity_id
|
|
)
|
|
|
|
# ========================================================================
|
|
# Scenes
|
|
# ========================================================================
|
|
|
|
async def activate_scene(self, scene_id: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
Activate a scene
|
|
|
|
Args:
|
|
scene_id: Scene entity ID (e.g., "scene.movie_night")
|
|
|
|
Returns:
|
|
List of changed states
|
|
"""
|
|
return await self.call_service(
|
|
domain="scene",
|
|
service="turn_on",
|
|
entity_id=scene_id
|
|
)
|
|
|
|
# ========================================================================
|
|
# Scripts
|
|
# ========================================================================
|
|
|
|
async def run_script(
|
|
self,
|
|
script_id: str,
|
|
variables: Optional[Dict[str, Any]] = None
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Execute a script with optional variables
|
|
|
|
Args:
|
|
script_id: Script entity ID (e.g., "script.bedtime_routine")
|
|
variables: Script variables
|
|
|
|
Returns:
|
|
List of changed states
|
|
"""
|
|
service_data = {"variables": variables} if variables else None
|
|
return await self.call_service(
|
|
domain="script",
|
|
service="turn_on",
|
|
entity_id=script_id,
|
|
service_data=service_data
|
|
)
|
|
|
|
# ========================================================================
|
|
# Automations
|
|
# ========================================================================
|
|
|
|
async def enable_automation(self, automation_id: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
Enable an automation
|
|
|
|
Args:
|
|
automation_id: Automation entity ID
|
|
|
|
Returns:
|
|
List of changed states
|
|
"""
|
|
return await self.call_service(
|
|
domain="automation",
|
|
service="turn_on",
|
|
entity_id=automation_id
|
|
)
|
|
|
|
async def disable_automation(self, automation_id: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
Disable an automation
|
|
|
|
Args:
|
|
automation_id: Automation entity ID
|
|
|
|
Returns:
|
|
List of changed states
|
|
"""
|
|
return await self.call_service(
|
|
domain="automation",
|
|
service="turn_off",
|
|
entity_id=automation_id
|
|
)
|
|
|
|
# ========================================================================
|
|
# History
|
|
# ========================================================================
|
|
|
|
async def get_history(
|
|
self,
|
|
entity_id: str,
|
|
hours: int = 24
|
|
) -> List[List[Dict[str, Any]]]:
|
|
"""
|
|
Get state history for an entity
|
|
|
|
HA Endpoint: GET /api/history/period/<timestamp>
|
|
|
|
Args:
|
|
entity_id: Entity ID to get history for
|
|
hours: Number of hours of history (default 24)
|
|
|
|
Returns:
|
|
List of state history entries
|
|
"""
|
|
start_time = datetime.now(timezone.utc) - timedelta(hours=hours)
|
|
timestamp = start_time.isoformat()
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/api/history/period/{timestamp}",
|
|
headers=self._get_headers(),
|
|
params={
|
|
"filter_entity_id": entity_id,
|
|
"minimal_response": "true"
|
|
}
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
# ========================================================================
|
|
# Areas (via template API)
|
|
# ========================================================================
|
|
|
|
async def get_areas(self) -> List[Dict[str, str]]:
|
|
"""
|
|
Get all areas/rooms
|
|
|
|
Note: The REST API doesn't have a direct areas endpoint.
|
|
This uses the template API to render area data.
|
|
|
|
HA Endpoint: POST /api/template
|
|
|
|
Returns:
|
|
List of area dicts with id and name
|
|
"""
|
|
template = """
|
|
{% set areas_list = [] %}
|
|
{% for area in areas() %}
|
|
{% set areas_list = areas_list + [{"id": area, "name": area_name(area)}] %}
|
|
{% endfor %}
|
|
{{ areas_list | tojson }}
|
|
"""
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.post(
|
|
f"{self.base_url}/api/template",
|
|
headers=self._get_headers(),
|
|
json={"template": template}
|
|
)
|
|
response.raise_for_status()
|
|
# Response is rendered template as string
|
|
return json.loads(response.text)
|
|
|
|
|
|
# Singleton instance
|
|
_homeassistant_client: Optional[HomeAssistantClient] = None
|
|
|
|
|
|
def get_homeassistant_client() -> HomeAssistantClient:
|
|
"""Get singleton Home Assistant client instance"""
|
|
global _homeassistant_client
|
|
if _homeassistant_client is None:
|
|
_homeassistant_client = HomeAssistantClient()
|
|
return _homeassistant_client
|