Add housekeeping API for Home Assistant integration
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>
This commit is contained in:
Claude Opus 4.5
@@ -0,0 +1,409 @@
|
||||
"""
|
||||
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
|
||||
Reference in New Issue
Block a user