diff --git a/surfsense_backend/app/agents/researcher/nodes.py b/surfsense_backend/app/agents/researcher/nodes.py index 5d01eecd9..d1d7272ab 100644 --- a/surfsense_backend/app/agents/researcher/nodes.py +++ b/surfsense_backend/app/agents/researcher/nodes.py @@ -208,6 +208,14 @@ async def search_single_connector( top_k=top_k, search_mode=search_mode, ) + elif connector == "HOME_ASSISTANT_CONNECTOR": + source_object, chunks = await connector_service.search_home_assistant( + user_query=reformulated_query, + user_id=user_id, + search_space_id=search_space_id, + top_k=top_k, + search_mode=search_mode, + ) except Exception as e: logging.error("Error searching connector %s: %s", connector, traceback.format_exc()) return (connector, None, [], str(e)) @@ -683,6 +691,7 @@ async def fetch_documents_by_ids( "CLICKUP_CONNECTOR": "ClickUp (Selected)", "AIRTABLE_CONNECTOR": "Airtable (Selected)", "LUMA_CONNECTOR": "Luma Events (Selected)", + "HOME_ASSISTANT_CONNECTOR": "Home Assistant (Selected)", } source_object = { diff --git a/surfsense_backend/app/agents/researcher/utils.py b/surfsense_backend/app/agents/researcher/utils.py index 16ff0b0a7..4b74866a0 100644 --- a/surfsense_backend/app/agents/researcher/utils.py +++ b/surfsense_backend/app/agents/researcher/utils.py @@ -46,6 +46,7 @@ CONNECTOR_METADATA: dict[str, ConnectorMetadata] = { "CLICKUP_CONNECTOR": ConnectorMetadata("📋", "ClickUp tasks", "ClickUp"), "LUMA_CONNECTOR": ConnectorMetadata("🎯", "Luma events", "Luma"), "ELASTICSEARCH_CONNECTOR": ConnectorMetadata("🔎", "Elasticsearch chunks", "Elasticsearch"), + "HOME_ASSISTANT_CONNECTOR": ConnectorMetadata("🏠", "Home Assistant items", "Home Assistant"), } # Default metadata for unknown connectors diff --git a/surfsense_backend/app/connectors/home_assistant_connector.py b/surfsense_backend/app/connectors/home_assistant_connector.py new file mode 100644 index 000000000..c2d4f1349 --- /dev/null +++ b/surfsense_backend/app/connectors/home_assistant_connector.py @@ -0,0 +1,426 @@ +""" +Home Assistant connector for fetching smart home data. + +This connector interfaces with the Home Assistant REST API to retrieve: +- Entity states and history +- Automations and scripts +- Logbook events +- Device and area information +""" + +import logging +from datetime import datetime, timedelta +from typing import Any + +import aiohttp + +logger = logging.getLogger(__name__) + + +class HomeAssistantConnector: + """Connector for Home Assistant smart home platform.""" + + def __init__(self, ha_url: str, access_token: str): + """ + Initialize the Home Assistant connector. + + Args: + ha_url: Base URL of Home Assistant instance (e.g., http://homeassistant.local:8123) + access_token: Long-lived access token for authentication + """ + self.ha_url = ha_url.rstrip("/") + self.access_token = access_token + self.headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json", + } + + async def _make_request( + self, endpoint: str, method: str = "GET", data: dict | None = None + ) -> tuple[Any | None, str | None]: + """ + Make an authenticated request to Home Assistant API. + + Args: + endpoint: API endpoint (e.g., /api/states) + method: HTTP method + data: Optional request body + + Returns: + Tuple of (response_data, error_message) + """ + url = f"{self.ha_url}{endpoint}" + + try: + async with aiohttp.ClientSession() as session: + async with session.request( + method, url, headers=self.headers, json=data, timeout=30 + ) as response: + if response.status == 200: + return await response.json(), None + elif response.status == 401: + return None, "Invalid or expired access token" + elif response.status == 404: + return None, f"Endpoint not found: {endpoint}" + else: + error_text = await response.text() + return None, f"API error {response.status}: {error_text}" + except aiohttp.ClientConnectorError as e: + return None, f"Connection error: Unable to reach Home Assistant at {self.ha_url}. {e!s}" + except TimeoutError: + return None, f"Request timeout connecting to {self.ha_url}" + except Exception as e: + return None, f"Unexpected error: {e!s}" + + async def test_connection(self) -> tuple[bool, str | None]: + """ + Test the connection to Home Assistant. + + Returns: + Tuple of (success, error_message) + """ + result, error = await self._make_request("/api/") + if error: + return False, error + return True, None + + async def get_states(self) -> tuple[list[dict], str | None]: + """ + Get all entity states. + + Returns: + Tuple of (states_list, error_message) + """ + return await self._make_request("/api/states") + + async def get_entity_history( + self, + entity_id: str, + start_time: datetime | None = None, + end_time: datetime | None = None, + ) -> tuple[list[dict], str | None]: + """ + Get history for a specific entity. + + Args: + entity_id: Entity ID (e.g., sensor.temperature) + start_time: Start of history period + end_time: End of history period + + Returns: + Tuple of (history_list, error_message) + """ + if start_time is None: + start_time = datetime.now() - timedelta(days=1) + if end_time is None: + end_time = datetime.now() + + start_str = start_time.isoformat() + end_str = end_time.isoformat() + + endpoint = f"/api/history/period/{start_str}?end_time={end_str}&filter_entity_id={entity_id}" + result, error = await self._make_request(endpoint) + + if error: + return [], error + # History returns nested list, flatten it + if result and len(result) > 0: + return result[0], None + return [], None + + async def get_logbook( + self, + start_time: datetime | None = None, + end_time: datetime | None = None, + entity_id: str | None = None, + ) -> tuple[list[dict], str | None]: + """ + Get logbook entries. + + Args: + start_time: Start of period + end_time: End of period + entity_id: Optional entity to filter by + + Returns: + Tuple of (logbook_entries, error_message) + """ + if start_time is None: + start_time = datetime.now() - timedelta(days=1) + + start_str = start_time.isoformat() + endpoint = f"/api/logbook/{start_str}" + + if end_time: + endpoint += f"?end_time={end_time.isoformat()}" + if entity_id: + separator = "&" if "?" in endpoint else "?" + endpoint += f"{separator}entity={entity_id}" + + return await self._make_request(endpoint) + + async def get_services(self) -> tuple[dict, str | None]: + """ + Get all available services. + + Returns: + Tuple of (services_dict, error_message) + """ + return await self._make_request("/api/services") + + async def get_config(self) -> tuple[dict, str | None]: + """ + Get Home Assistant configuration. + + Returns: + Tuple of (config_dict, error_message) + """ + return await self._make_request("/api/config") + + async def get_events(self) -> tuple[list[dict], str | None]: + """ + Get list of available event types. + + Returns: + Tuple of (events_list, error_message) + """ + return await self._make_request("/api/events") + + async def get_automations(self) -> tuple[list[dict], str | None]: + """ + Get all automation entities. + + Returns: + Tuple of (automations_list, error_message) + """ + states, error = await self.get_states() + if error: + return [], error + + automations = [ + state for state in states if state.get("entity_id", "").startswith("automation.") + ] + return automations, None + + async def get_scripts(self) -> tuple[list[dict], str | None]: + """ + Get all script entities. + + Returns: + Tuple of (scripts_list, error_message) + """ + states, error = await self.get_states() + if error: + return [], error + + scripts = [state for state in states if state.get("entity_id", "").startswith("script.")] + return scripts, None + + async def get_scenes(self) -> tuple[list[dict], str | None]: + """ + Get all scene entities. + + Returns: + Tuple of (scenes_list, error_message) + """ + states, error = await self.get_states() + if error: + return [], error + + scenes = [state for state in states if state.get("entity_id", "").startswith("scene.")] + return scenes, None + + async def get_sensors(self) -> tuple[list[dict], str | None]: + """ + Get all sensor entities. + + Returns: + Tuple of (sensors_list, error_message) + """ + states, error = await self.get_states() + if error: + return [], error + + sensors = [state for state in states if state.get("entity_id", "").startswith("sensor.")] + return sensors, None + + async def get_all_indexable_data( + self, + start_date: datetime | None = None, + end_date: datetime | None = None, + ) -> tuple[list[dict], str | None]: + """ + Get all data suitable for indexing. + + This includes: + - Automations with their configurations + - Scripts with their configurations + - Scenes + - Logbook events for the specified period + + Args: + start_date: Start of period for logbook events + end_date: End of period for logbook events + + Returns: + Tuple of (indexable_items, error_message) + """ + items = [] + errors = [] + + # Get automations + automations, error = await self.get_automations() + if error: + errors.append(f"Automations: {error}") + else: + for auto in automations: + items.append({ + "type": "automation", + "entity_id": auto.get("entity_id"), + "name": auto.get("attributes", {}).get("friendly_name", auto.get("entity_id")), + "state": auto.get("state"), + "last_triggered": auto.get("attributes", {}).get("last_triggered"), + "attributes": auto.get("attributes", {}), + }) + + # Get scripts + scripts, error = await self.get_scripts() + if error: + errors.append(f"Scripts: {error}") + else: + for script in scripts: + items.append({ + "type": "script", + "entity_id": script.get("entity_id"), + "name": script.get("attributes", {}).get("friendly_name", script.get("entity_id")), + "state": script.get("state"), + "last_triggered": script.get("attributes", {}).get("last_triggered"), + "attributes": script.get("attributes", {}), + }) + + # Get scenes + scenes, error = await self.get_scenes() + if error: + errors.append(f"Scenes: {error}") + else: + for scene in scenes: + items.append({ + "type": "scene", + "entity_id": scene.get("entity_id"), + "name": scene.get("attributes", {}).get("friendly_name", scene.get("entity_id")), + "state": scene.get("state"), + "attributes": scene.get("attributes", {}), + }) + + # Get logbook events + logbook, error = await self.get_logbook(start_time=start_date, end_time=end_date) + if error: + errors.append(f"Logbook: {error}") + else: + for entry in logbook: + items.append({ + "type": "logbook_event", + "entity_id": entry.get("entity_id"), + "name": entry.get("name", "Unknown"), + "message": entry.get("message", ""), + "when": entry.get("when"), + "state": entry.get("state"), + "domain": entry.get("domain"), + }) + + error_msg = "; ".join(errors) if errors else None + return items, error_msg + + def format_automation_to_markdown(self, automation: dict) -> str: + """Format an automation to markdown.""" + name = automation.get("name", "Unknown Automation") + entity_id = automation.get("entity_id", "") + state = automation.get("state", "unknown") + last_triggered = automation.get("last_triggered", "Never") + attributes = automation.get("attributes", {}) + + md = f"# Automation: {name}\n\n" + md += f"**Entity ID:** {entity_id}\n" + md += f"**State:** {state}\n" + md += f"**Last Triggered:** {last_triggered}\n\n" + + if attributes.get("id"): + md += f"**Automation ID:** {attributes.get('id')}\n" + if attributes.get("mode"): + md += f"**Mode:** {attributes.get('mode')}\n" + if attributes.get("current"): + md += f"**Current Running:** {attributes.get('current')}\n" + + return md + + def format_script_to_markdown(self, script: dict) -> str: + """Format a script to markdown.""" + name = script.get("name", "Unknown Script") + entity_id = script.get("entity_id", "") + state = script.get("state", "unknown") + last_triggered = script.get("last_triggered", "Never") + attributes = script.get("attributes", {}) + + md = f"# Script: {name}\n\n" + md += f"**Entity ID:** {entity_id}\n" + md += f"**State:** {state}\n" + md += f"**Last Triggered:** {last_triggered}\n\n" + + if attributes.get("mode"): + md += f"**Mode:** {attributes.get('mode')}\n" + + return md + + def format_scene_to_markdown(self, scene: dict) -> str: + """Format a scene to markdown.""" + name = scene.get("name", "Unknown Scene") + entity_id = scene.get("entity_id", "") + attributes = scene.get("attributes", {}) + + md = f"# Scene: {name}\n\n" + md += f"**Entity ID:** {entity_id}\n" + + if attributes.get("entity_id"): + md += f"\n**Controlled Entities:**\n" + for eid in attributes.get("entity_id", []): + md += f"- {eid}\n" + + return md + + def format_logbook_event_to_markdown(self, event: dict) -> str: + """Format a logbook event to markdown.""" + name = event.get("name", "Unknown") + entity_id = event.get("entity_id", "") + message = event.get("message", "") + when = event.get("when", "") + state = event.get("state", "") + domain = event.get("domain", "") + + md = f"# Event: {name}\n\n" + md += f"**Time:** {when}\n" + if entity_id: + md += f"**Entity ID:** {entity_id}\n" + if domain: + md += f"**Domain:** {domain}\n" + if state: + md += f"**State:** {state}\n" + if message: + md += f"\n**Message:** {message}\n" + + return md + + def format_item_to_markdown(self, item: dict) -> str: + """Format any item to markdown based on its type.""" + item_type = item.get("type", "") + + if item_type == "automation": + return self.format_automation_to_markdown(item) + elif item_type == "script": + return self.format_script_to_markdown(item) + elif item_type == "scene": + return self.format_scene_to_markdown(item) + elif item_type == "logbook_event": + return self.format_logbook_event_to_markdown(item) + else: + # Generic formatting + return f"# {item.get('name', 'Unknown')}\n\n{item}" diff --git a/surfsense_backend/app/db.py b/surfsense_backend/app/db.py index f67d06215..f376afb17 100644 --- a/surfsense_backend/app/db.py +++ b/surfsense_backend/app/db.py @@ -50,6 +50,7 @@ class DocumentType(str, Enum): AIRTABLE_CONNECTOR = "AIRTABLE_CONNECTOR" LUMA_CONNECTOR = "LUMA_CONNECTOR" ELASTICSEARCH_CONNECTOR = "ELASTICSEARCH_CONNECTOR" + HOME_ASSISTANT_CONNECTOR = "HOME_ASSISTANT_CONNECTOR" class SearchSourceConnectorType(str, Enum): @@ -71,6 +72,7 @@ class SearchSourceConnectorType(str, Enum): AIRTABLE_CONNECTOR = "AIRTABLE_CONNECTOR" LUMA_CONNECTOR = "LUMA_CONNECTOR" ELASTICSEARCH_CONNECTOR = "ELASTICSEARCH_CONNECTOR" + HOME_ASSISTANT_CONNECTOR = "HOME_ASSISTANT_CONNECTOR" class ChatType(str, Enum): diff --git a/surfsense_backend/app/routes/__init__.py b/surfsense_backend/app/routes/__init__.py index dc1492324..b329491ee 100644 --- a/surfsense_backend/app/routes/__init__.py +++ b/surfsense_backend/app/routes/__init__.py @@ -11,6 +11,9 @@ from .google_calendar_add_connector_route import ( from .google_gmail_add_connector_route import ( router as google_gmail_add_connector_router, ) +from .home_assistant_add_connector_route import ( + router as home_assistant_add_connector_router, +) from .llm_config_routes import router as llm_config_router from .logs_routes import router as logs_router from .luma_add_connector_route import router as luma_add_connector_router @@ -31,6 +34,7 @@ router.include_router(google_calendar_add_connector_router) router.include_router(google_gmail_add_connector_router) router.include_router(airtable_add_connector_router) router.include_router(luma_add_connector_router) +router.include_router(home_assistant_add_connector_router) router.include_router(llm_config_router) router.include_router(logs_router) router.include_router(site_configuration_router) diff --git a/surfsense_backend/app/routes/home_assistant_add_connector_route.py b/surfsense_backend/app/routes/home_assistant_add_connector_route.py new file mode 100644 index 000000000..0ecaf7fcf --- /dev/null +++ b/surfsense_backend/app/routes/home_assistant_add_connector_route.py @@ -0,0 +1,163 @@ +""" +Home Assistant connector authentication route. + +This route allows users to add a Home Assistant connector by providing +their Home Assistant URL and Long-Lived Access Token. +""" + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel, HttpUrl +from sqlalchemy.ext.asyncio import AsyncSession + +from app.connectors.home_assistant_connector import HomeAssistantConnector +from app.db import SearchSourceConnector, SearchSourceConnectorType, User, get_async_session +from app.users import current_active_user + +router = APIRouter() + + +class HomeAssistantConnectorRequest(BaseModel): + """Request model for adding a Home Assistant connector.""" + + search_space_id: int + ha_url: str # Using str instead of HttpUrl for flexibility with local addresses + ha_access_token: str + name: str = "Home Assistant" + + +class HomeAssistantConnectorResponse(BaseModel): + """Response model for Home Assistant connector creation.""" + + id: int + name: str + connector_type: str + message: str + + +@router.post("/auth/home-assistant/add", response_model=HomeAssistantConnectorResponse) +async def add_home_assistant_connector( + request: HomeAssistantConnectorRequest, + user: User = Depends(current_active_user), + session: AsyncSession = Depends(get_async_session), +): + """ + Add a Home Assistant connector. + + This endpoint allows users to connect their Home Assistant instance + by providing the URL and a Long-Lived Access Token. + + To create a Long-Lived Access Token in Home Assistant: + 1. Go to your Home Assistant instance + 2. Click on your profile (bottom left) + 3. Scroll down to "Long-Lived Access Tokens" + 4. Create a new token and copy it + + Args: + request: The connector configuration + user: Current authenticated user + session: Database session + + Returns: + HomeAssistantConnectorResponse with connector details + """ + # Validate and normalize URL + ha_url = request.ha_url.rstrip("/") + + # Test connection to Home Assistant + ha_client = HomeAssistantConnector( + ha_url=ha_url, + access_token=request.ha_access_token, + ) + + connected, error = await ha_client.test_connection() + if not connected: + raise HTTPException( + status_code=400, + detail=f"Failed to connect to Home Assistant: {error}", + ) + + # Get Home Assistant config to verify and get instance name + config, config_error = await ha_client.get_config() + if config_error: + # Non-fatal, just use provided name + instance_name = request.name + else: + # Use location name from config if available + instance_name = config.get("location_name", request.name) + + # Create connector config + connector_config = { + "HA_URL": ha_url, + "HA_ACCESS_TOKEN": request.ha_access_token, + } + + # Create the connector in the database + db_connector = SearchSourceConnector( + name=instance_name, + connector_type=SearchSourceConnectorType.HOME_ASSISTANT_CONNECTOR, + config=connector_config, + search_space_id=request.search_space_id, + user_id=str(user.id), + is_indexable=True, + ) + + session.add(db_connector) + await session.commit() + await session.refresh(db_connector) + + return HomeAssistantConnectorResponse( + id=db_connector.id, + name=db_connector.name, + connector_type=db_connector.connector_type.value, + message=f"Successfully connected to Home Assistant at {ha_url}", + ) + + +@router.post("/auth/home-assistant/test") +async def test_home_assistant_connection( + ha_url: str, + ha_access_token: str, + user: User = Depends(current_active_user), +): + """ + Test connection to a Home Assistant instance. + + This endpoint allows users to verify their credentials before + creating a connector. + + Args: + ha_url: Home Assistant URL + ha_access_token: Long-Lived Access Token + user: Current authenticated user + + Returns: + Connection status and instance info + """ + # Normalize URL + ha_url = ha_url.rstrip("/") + + # Test connection + ha_client = HomeAssistantConnector( + ha_url=ha_url, + access_token=ha_access_token, + ) + + connected, error = await ha_client.test_connection() + if not connected: + raise HTTPException( + status_code=400, + detail=f"Connection failed: {error}", + ) + + # Get config for additional info + config, _ = await ha_client.get_config() + + return { + "status": "success", + "message": "Successfully connected to Home Assistant", + "instance_info": { + "location_name": config.get("location_name", "Unknown") if config else "Unknown", + "version": config.get("version", "Unknown") if config else "Unknown", + "time_zone": config.get("time_zone", "Unknown") if config else "Unknown", + }, + } diff --git a/surfsense_backend/app/routes/search_source_connectors_routes.py b/surfsense_backend/app/routes/search_source_connectors_routes.py index 4e62035ff..5d2ea3387 100644 --- a/surfsense_backend/app/routes/search_source_connectors_routes.py +++ b/surfsense_backend/app/routes/search_source_connectors_routes.py @@ -44,6 +44,7 @@ from app.tasks.connector_indexers import ( index_github_repos, index_google_calendar_events, index_google_gmail_messages, + index_home_assistant_data, index_jira_issues, index_linear_issues, index_luma_events, @@ -688,6 +689,22 @@ async def index_connector_content( ) response_message = "Elasticsearch indexing started in the background." + elif ( + connector.connector_type + == SearchSourceConnectorType.HOME_ASSISTANT_CONNECTOR + ): + from app.tasks.celery_tasks.connector_tasks import ( + index_home_assistant_data_task, + ) + + logger.info( + f"Triggering Home Assistant indexing for connector {connector_id} into search space {search_space_id}" + ) + index_home_assistant_data_task.delay( + connector_id, search_space_id, str(user.id), indexing_from, indexing_to + ) + response_message = "Home Assistant indexing started in the background." + else: raise HTTPException( status_code=400, @@ -1523,3 +1540,41 @@ async def run_elasticsearch_indexing( f"Critical error in run_elasticsearch_indexing for connector {connector_id}: {e}", exc_info=True, ) + + +async def run_home_assistant_indexing( + session: AsyncSession, + connector_id: int, + search_space_id: int, + user_id: str, + start_date: str, + end_date: str, +): + """Runs the Home Assistant indexing task and updates the timestamp.""" + try: + indexed_count, error_message = await index_home_assistant_data( + session, + connector_id, + search_space_id, + user_id, + start_date, + end_date, + update_last_indexed=False, + ) + if error_message: + logger.error( + f"Home Assistant indexing failed for connector {connector_id}: {error_message}" + ) + else: + logger.info( + f"Home Assistant indexing successful for connector {connector_id}. Indexed {indexed_count} items." + ) + # Update the last indexed timestamp only on success + await update_connector_last_indexed(session, connector_id) + await session.commit() + except Exception as e: + await session.rollback() + logger.error( + f"Critical error in run_home_assistant_indexing for connector {connector_id}: {e}", + exc_info=True, + ) diff --git a/surfsense_backend/app/services/connector_service.py b/surfsense_backend/app/services/connector_service.py index 28f70d285..ef1ca1758 100644 --- a/surfsense_backend/app/services/connector_service.py +++ b/surfsense_backend/app/services/connector_service.py @@ -2541,3 +2541,107 @@ class ConnectorService: } return result_object, elasticsearch_chunks + + async def search_home_assistant( + self, + user_query: str, + user_id: str, + search_space_id: int, + top_k: int = 20, + search_mode: SearchMode = SearchMode.CHUNKS, + ) -> tuple: + """ + Search for Home Assistant documents and return both the source information and langchain documents + + Args: + user_query: The user's query + user_id: The user's ID + search_space_id: The search space ID to search in + top_k: Maximum number of results to return + search_mode: Search mode (CHUNKS or DOCUMENTS) + + Returns: + tuple: (sources_info, langchain_documents) + """ + if search_mode == SearchMode.CHUNKS: + ha_chunks = await self.chunk_retriever.hybrid_search( + query_text=user_query, + top_k=top_k, + user_id=user_id, + search_space_id=search_space_id, + document_type="HOME_ASSISTANT_CONNECTOR", + ) + elif search_mode == SearchMode.DOCUMENTS: + ha_chunks = await self.document_retriever.hybrid_search( + query_text=user_query, + top_k=top_k, + user_id=user_id, + search_space_id=search_space_id, + document_type="HOME_ASSISTANT_CONNECTOR", + ) + # Transform document retriever results to match expected format + ha_chunks = self._transform_document_results(ha_chunks) + + # Early return if no results + if not ha_chunks: + return { + "id": 35, + "name": "Home Assistant", + "type": "HOME_ASSISTANT_CONNECTOR", + "sources": [], + }, [] + + # Process each chunk and create sources + sources_list = [] + async with self.counter_lock: + for _i, chunk in enumerate(ha_chunks): + # Extract document metadata + document = chunk.get("document", {}) + metadata = document.get("metadata", {}) + + # Extract Home Assistant-specific metadata + entity_id = metadata.get("entity_id", "") + item_type = metadata.get("item_type", "") + state = metadata.get("state", "") + ha_url = metadata.get("ha_url", "") + + # Create a descriptive title + title = document.get("title", "Home Assistant Item") + + # Create description from content + description = chunk.get("content", "")[:150] + if len(description) == 150: + description += "..." + + # Add state info to description if available + if state: + description = f"[{state}] {description}" + + # Construct URL to entity in Home Assistant + url = "" + if ha_url and entity_id: + # Link to entity in Home Assistant UI + url = f"{ha_url}/config/entities/entity-{entity_id}" + + source = { + "id": chunk.get("chunk_id", self.source_id_counter), + "title": title, + "description": description, + "url": url, + "entity_id": entity_id, + "item_type": item_type, + "state": state, + } + + self.source_id_counter += 1 + sources_list.append(source) + + # Create result object + result_object = { + "id": 35, # Assign a unique ID for the Home Assistant connector + "name": "Home Assistant", + "type": "HOME_ASSISTANT_CONNECTOR", + "sources": sources_list, + } + + return result_object, ha_chunks diff --git a/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py b/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py index 5e6907499..072e1ce1c 100644 --- a/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py +++ b/surfsense_backend/app/tasks/celery_tasks/connector_tasks.py @@ -600,3 +600,46 @@ async def _index_elasticsearch_documents( await run_elasticsearch_indexing( session, connector_id, search_space_id, user_id, start_date, end_date ) + + +@celery_app.task(name="index_home_assistant_data", bind=True) +def index_home_assistant_data_task( + self, + connector_id: int, + search_space_id: int, + user_id: str, + start_date: str, + end_date: str, +): + """Celery task to index Home Assistant data.""" + import asyncio + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + + try: + loop.run_until_complete( + _index_home_assistant_data( + connector_id, search_space_id, user_id, start_date, end_date + ) + ) + finally: + loop.close() + + +async def _index_home_assistant_data( + connector_id: int, + search_space_id: int, + user_id: str, + start_date: str, + end_date: str, +): + """Index Home Assistant data with new session.""" + from app.routes.search_source_connectors_routes import ( + run_home_assistant_indexing, + ) + + async with get_celery_session_maker()() as session: + await run_home_assistant_indexing( + session, connector_id, search_space_id, user_id, start_date, end_date + ) diff --git a/surfsense_backend/app/tasks/connector_indexers/__init__.py b/surfsense_backend/app/tasks/connector_indexers/__init__.py index 766506f70..f202b4262 100644 --- a/surfsense_backend/app/tasks/connector_indexers/__init__.py +++ b/surfsense_backend/app/tasks/connector_indexers/__init__.py @@ -18,6 +18,7 @@ Available indexers: - Google Calendar: Index events from Google Calendar - Luma: Index events from Luma - Elasticsearch: Index documents from Elasticsearch instances +- Home Assistant: Index automations, scripts, and events from Home Assistant """ # Communication platforms @@ -32,6 +33,7 @@ from .elasticsearch_indexer import index_elasticsearch_documents from .github_indexer import index_github_repos from .google_calendar_indexer import index_google_calendar_events from .google_gmail_indexer import index_google_gmail_messages +from .home_assistant_indexer import index_home_assistant_data from .jira_indexer import index_jira_issues # Issue tracking and project management @@ -61,4 +63,6 @@ __all__ = [ # noqa: RUF022 # Communication platforms "index_slack_messages", "index_google_gmail_messages", + # Smart home + "index_home_assistant_data", ] diff --git a/surfsense_backend/app/tasks/connector_indexers/home_assistant_indexer.py b/surfsense_backend/app/tasks/connector_indexers/home_assistant_indexer.py new file mode 100644 index 000000000..eaaa010da --- /dev/null +++ b/surfsense_backend/app/tasks/connector_indexers/home_assistant_indexer.py @@ -0,0 +1,332 @@ +""" +Home Assistant connector indexer. +""" + +from datetime import datetime, timedelta + +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import config +from app.connectors.home_assistant_connector import HomeAssistantConnector +from app.db import Document, DocumentType, SearchSourceConnectorType +from app.services.task_logging_service import TaskLoggingService +from app.utils.document_converters import ( + create_document_chunks, + generate_content_hash, + generate_unique_identifier_hash, +) + +from .base import ( + check_document_by_unique_identifier, + get_connector_by_id, + logger, + update_connector_last_indexed, +) + + +async def index_home_assistant_data( + session: AsyncSession, + connector_id: int, + search_space_id: int, + user_id: str, + start_date: str | None = None, + end_date: str | None = None, + update_last_indexed: bool = True, +) -> tuple[int, str | None]: + """ + Index Home Assistant data including automations, scripts, scenes, and logbook events. + + Args: + session: Database session + connector_id: ID of the Home Assistant connector + search_space_id: ID of the search space to store documents in + user_id: User ID + start_date: Start date for indexing logbook events (YYYY-MM-DD format) + end_date: End date for indexing logbook events (YYYY-MM-DD format) + update_last_indexed: Whether to update the last_indexed_at timestamp (default: True) + + Returns: + Tuple containing (number of documents indexed, error message or None) + """ + task_logger = TaskLoggingService(session, search_space_id) + + # Log task start + log_entry = await task_logger.log_task_start( + task_name="home_assistant_indexing", + source="connector_indexing_task", + message=f"Starting Home Assistant indexing for connector {connector_id}", + metadata={ + "connector_id": connector_id, + "user_id": str(user_id), + "start_date": start_date, + "end_date": end_date, + }, + ) + + try: + # Get the connector from the database + connector = await get_connector_by_id( + session, connector_id, SearchSourceConnectorType.HOME_ASSISTANT_CONNECTOR + ) + + if not connector: + await task_logger.log_task_failure( + log_entry, + f"Connector with ID {connector_id} not found", + "Connector not found", + {"error_type": "ConnectorNotFound"}, + ) + return 0, f"Connector with ID {connector_id} not found" + + # Get Home Assistant credentials from the connector config + ha_url = connector.config.get("HA_URL") + ha_token = connector.config.get("HA_ACCESS_TOKEN") + + if not ha_url or not ha_token: + await task_logger.log_task_failure( + log_entry, + f"Home Assistant credentials not found in connector config for connector {connector_id}", + "Missing Home Assistant credentials", + {"error_type": "MissingCredentials"}, + ) + return 0, "Home Assistant credentials not found in connector config" + + # Initialize Home Assistant client + await task_logger.log_task_progress( + log_entry, + f"Initializing Home Assistant client for connector {connector_id}", + {"stage": "client_initialization"}, + ) + + ha_client = HomeAssistantConnector(ha_url=ha_url, access_token=ha_token) + + # Test connection + connected, connection_error = await ha_client.test_connection() + if not connected: + await task_logger.log_task_failure( + log_entry, + f"Failed to connect to Home Assistant: {connection_error}", + "Connection failed", + {"error_type": "ConnectionError"}, + ) + return 0, f"Failed to connect to Home Assistant: {connection_error}" + + # Calculate date range for logbook events + calculated_end_date = datetime.now() + + if connector.last_indexed_at: + last_indexed_naive = ( + connector.last_indexed_at.replace(tzinfo=None) + if connector.last_indexed_at.tzinfo + else connector.last_indexed_at + ) + + if last_indexed_naive > calculated_end_date: + logger.warning( + f"Last indexed date ({last_indexed_naive.strftime('%Y-%m-%d')}) is in the future. Using 7 days ago instead." + ) + calculated_start_date = calculated_end_date - timedelta(days=7) + else: + calculated_start_date = last_indexed_naive + logger.info( + f"Using last_indexed_at ({calculated_start_date.strftime('%Y-%m-%d')}) as start date" + ) + else: + # Default to 7 days of logbook events for initial indexing + calculated_start_date = calculated_end_date - timedelta(days=7) + logger.info( + f"No last_indexed_at found, using {calculated_start_date.strftime('%Y-%m-%d')} (7 days ago) as start date" + ) + + # Parse provided dates if available + if start_date: + calculated_start_date = datetime.strptime(start_date, "%Y-%m-%d") + if end_date: + calculated_end_date = datetime.strptime(end_date, "%Y-%m-%d") + + await task_logger.log_task_progress( + log_entry, + f"Fetching Home Assistant data from {calculated_start_date.strftime('%Y-%m-%d')} to {calculated_end_date.strftime('%Y-%m-%d')}", + { + "stage": "fetching_data", + "start_date": calculated_start_date.strftime("%Y-%m-%d"), + "end_date": calculated_end_date.strftime("%Y-%m-%d"), + }, + ) + + # Get all indexable data + items, fetch_error = await ha_client.get_all_indexable_data( + start_date=calculated_start_date, + end_date=calculated_end_date, + ) + + if fetch_error: + logger.warning(f"Some data fetch errors occurred: {fetch_error}") + + if not items: + logger.info("No items found to index from Home Assistant") + if update_last_indexed: + await update_connector_last_indexed( + session, connector, update_last_indexed + ) + await session.commit() + + await task_logger.log_task_success( + log_entry, + "No new items found to index", + {"items_indexed": 0}, + ) + return 0, None + + await task_logger.log_task_progress( + log_entry, + f"Processing {len(items)} items from Home Assistant", + {"stage": "processing_items", "total_items": len(items)}, + ) + + # Process each item + documents_indexed = 0 + + for item in items: + try: + item_type = item.get("type", "unknown") + entity_id = item.get("entity_id", "") + item_name = item.get("name", "Unknown") + + # Generate unique identifier based on item type and entity_id + if item_type == "logbook_event": + # For logbook events, include timestamp in unique ID + unique_id = f"{entity_id}_{item.get('when', '')}" + else: + # For automations, scripts, scenes - use entity_id + unique_id = entity_id + + unique_identifier_hash = generate_unique_identifier_hash(unique_id) + + # Check for existing document + existing_doc = await check_document_by_unique_identifier( + session, unique_identifier_hash + ) + + # Format content to markdown + content = ha_client.format_item_to_markdown(item) + content_hash = generate_content_hash(content) + + # Build metadata + metadata = { + "entity_id": entity_id, + "item_type": item_type, + "state": item.get("state", ""), + "ha_url": ha_url, + } + + if item_type in ["automation", "script"]: + metadata["last_triggered"] = item.get("last_triggered", "") + elif item_type == "logbook_event": + metadata["when"] = item.get("when", "") + metadata["domain"] = item.get("domain", "") + metadata["message"] = item.get("message", "") + + # Generate title + title = f"{item_type.title()}: {item_name}" + + if existing_doc: + # Check if content has changed + if existing_doc.content_hash == content_hash: + logger.debug(f"Skipping unchanged item: {title}") + continue + + # Update existing document + existing_doc.title = title + existing_doc.content = content + existing_doc.content_hash = content_hash + existing_doc.document_metadata = metadata + + # Delete old chunks and create new ones + existing_doc.chunks.clear() + + # Create new chunks + chunks = create_document_chunks( + content, + existing_doc, + config.chunker_instance, + config.embedding_model_instance, + ) + for chunk in chunks: + existing_doc.chunks.append(chunk) + + documents_indexed += 1 + logger.info(f"Updated existing Home Assistant item: {title}") + + else: + # Create new document + document = Document( + title=title, + document_type=DocumentType.HOME_ASSISTANT_CONNECTOR, + document_metadata=metadata, + content=content, + content_hash=content_hash, + unique_identifier_hash=unique_identifier_hash, + search_space_id=search_space_id, + ) + + # Create chunks + chunks = create_document_chunks( + content, + document, + config.chunker_instance, + config.embedding_model_instance, + ) + for chunk in chunks: + document.chunks.append(chunk) + + session.add(document) + documents_indexed += 1 + logger.info(f"Indexed new Home Assistant item: {title}") + + except Exception as e: + logger.error(f"Error processing item {item.get('name', 'unknown')}: {e!s}") + continue + + # Update last indexed timestamp + if update_last_indexed: + await update_connector_last_indexed(session, connector, update_last_indexed) + + # Commit all changes + await session.commit() + + await task_logger.log_task_success( + log_entry, + f"Successfully indexed {documents_indexed} Home Assistant items", + {"items_indexed": documents_indexed}, + ) + + logger.info(f"Successfully indexed {documents_indexed} Home Assistant items") + return documents_indexed, None + + except SQLAlchemyError as e: + await session.rollback() + error_message = f"Database error during Home Assistant indexing: {e!s}" + logger.error(error_message) + + await task_logger.log_task_failure( + log_entry, + error_message, + "Database error", + {"error_type": "DatabaseError"}, + ) + return 0, error_message + + except Exception as e: + await session.rollback() + error_message = f"Unexpected error during Home Assistant indexing: {e!s}" + logger.error(error_message) + + await task_logger.log_task_failure( + log_entry, + error_message, + "Unexpected error", + {"error_type": "UnexpectedError"}, + ) + return 0, error_message