diff --git a/README.md b/README.md index c9638cb..453d5ca 100644 --- a/README.md +++ b/README.md @@ -318,7 +318,7 @@ automation: - 🤖 **Model and Provider Information**: Tracking current AI model and service provider - 🚦 **System Status**: Real-time API and processing readiness - 📊 **Performance Metrics**: Request success rates and response times -- 💬 **Conversation Tracking**: Token usage and interaction history +- 💬 **Conversation Tracking**: Token usage and interaction history (for more precise token counting, install `tiktoken`; otherwise, a fallback estimation method is automatically used) - 🕒 **Last Interaction Details**: Recent query and response tracking - ❤️ **System Health**: Error monitoring and service uptime diff --git a/custom_components/ha_text_ai/__init__.py b/custom_components/ha_text_ai/__init__.py index 2aefca7..ef824f7 100644 --- a/custom_components/ha_text_ai/__init__.py +++ b/custom_components/ha_text_ai/__init__.py @@ -11,8 +11,9 @@ from __future__ import annotations import logging import os import shutil +import hashlib from datetime import datetime, timedelta -from typing import Any, Dict +from typing import Any, Dict, TypeVar import voluptuous as vol from async_timeout import timeout @@ -52,11 +53,13 @@ from .const import ( SERVICE_SET_SYSTEM_PROMPT, DEFAULT_MAX_HISTORY, CONF_MAX_HISTORY_SIZE, + ICONS_SUBDOMAIN, ) _LOGGER = logging.getLogger(__name__) CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN) +ConfigType = TypeVar("ConfigType", bound=Dict[str, Any]) SERVICE_SCHEMA_ASK_QUESTION = vol.Schema({ vol.Required("instance"): cv.string, @@ -90,19 +93,18 @@ def get_coordinator_by_instance(hass: HomeAssistant, instance: str) -> HATextAIC raise HomeAssistantError(f"Instance {instance} not found") -async def async_setup(hass: HomeAssistant, config: Dict[str, Any]) -> bool: - """Set up the HA Text AI component.""" - hass.data.setdefault(DOMAIN, {}) +def get_file_hash(file_path: str) -> str: + """Calculate SHA256 hash of file.""" + sha256_hash = hashlib.sha256() + with open(file_path, "rb") as f: + for byte_block in iter(lambda: f.read(4096), b""): + sha256_hash.update(byte_block) + return sha256_hash.hexdigest() - try: - source = os.path.join(os.path.dirname(__file__), 'icons', 'icon.svg') - dest_dir = os.path.join(hass.config.path('www'), 'icons') - os.makedirs(dest_dir, exist_ok=True) - dest = os.path.join(dest_dir, 'icon.png') - if not os.path.exists(dest): - shutil.copyfile(source, dest) - except Exception as ex: - _LOGGER.warning("Failed to copy custom icon: %s", str(ex)) +async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: + """Set up the Home Assistant Text AI component.""" + # Initialize domain data storage + hass.data.setdefault(DOMAIN, {}) async def async_ask_question(call: ServiceCall) -> None: """Handle ask_question service.""" @@ -150,6 +152,7 @@ async def async_setup(hass: HomeAssistant, config: Dict[str, Any]) -> bool: _LOGGER.error("Error setting system prompt: %s", str(err)) raise HomeAssistantError(f"Failed to set system prompt: {str(err)}") + # Register services hass.services.async_register( DOMAIN, SERVICE_ASK_QUESTION, @@ -178,6 +181,53 @@ async def async_setup(hass: HomeAssistant, config: Dict[str, Any]) -> bool: schema=SERVICE_SCHEMA_SET_SYSTEM_PROMPT ) + # Handle icons + try: + source_icon_path = os.path.join( + os.path.dirname(__file__), + ICONS_SUBDOMAIN, + 'icon@2x.png' + ) + + destination_directory = os.path.join( + hass.config.path('www'), + DOMAIN, + ICONS_SUBDOMAIN + ) + + destination_icon_path = os.path.join( + destination_directory, + 'icon.png' + ) + + if not os.path.exists(source_icon_path): + _LOGGER.error("Source icon not found: %s", source_icon_path) + return True + + def create_directory(): + os.makedirs(destination_directory, exist_ok=True) + + await hass.async_add_executor_job(create_directory) + + should_copy = True + + if os.path.exists(destination_icon_path): + source_hash = await hass.async_add_executor_job(get_file_hash, source_icon_path) + dest_hash = await hass.async_add_executor_job(get_file_hash, destination_icon_path) + should_copy = source_hash != dest_hash + + if should_copy: + def copy_file(): + shutil.copyfile(source_icon_path, destination_icon_path) + + await hass.async_add_executor_job(copy_file) + _LOGGER.debug("Icon updated: %s", destination_icon_path) + + except PermissionError as e: + _LOGGER.error("Permission denied when managing icons: %s", str(e)) + except Exception as e: + _LOGGER.error("Failed to manage icons: %s", str(e)) + return True async def async_check_api(session, endpoint: str, headers: dict, provider: str) -> bool: diff --git a/custom_components/ha_text_ai/const.py b/custom_components/ha_text_ai/const.py index 0d1a300..aa897bd 100644 --- a/custom_components/ha_text_ai/const.py +++ b/custom_components/ha_text_ai/const.py @@ -41,7 +41,9 @@ CONF_IS_ANTHROPIC: Final = "is_anthropic" CONF_CONTEXT_MESSAGES: Final = "context_messages" ABSOLUTE_MAX_HISTORY_SIZE = 500 -MAX_ENTRY_SIZE = 1 * 1024 * 1024 +MAX_ATTRIBUTE_SIZE = 4 * 1024 +MAX_HISTORY_FILE_SIZE = 1 * 1024 * 1024 +ICONS_SUBDOMAIN = "icons" # Default values DEFAULT_MODEL: Final = "gpt-4o-mini" diff --git a/custom_components/ha_text_ai/coordinator.py b/custom_components/ha_text_ai/coordinator.py index bc2c815..8751fec 100644 --- a/custom_components/ha_text_ai/coordinator.py +++ b/custom_components/ha_text_ai/coordinator.py @@ -12,9 +12,13 @@ import logging import traceback import aiofiles import os +import json import asyncio +import psutil +import re +import math from datetime import datetime, timedelta -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Union from homeassistant.core import HomeAssistant from homeassistant.helpers.update_coordinator import DataUpdateCoordinator @@ -35,11 +39,26 @@ from .const import ( DEFAULT_MAX_HISTORY, DEFAULT_CONTEXT_MESSAGES, ABSOLUTE_MAX_HISTORY_SIZE, - MAX_ENTRY_SIZE, + MAX_ATTRIBUTE_SIZE, + MAX_HISTORY_FILE_SIZE, ) _LOGGER = logging.getLogger(__name__) +def _check_memory_available(self): + """Check if enough memory is available.""" + memory = psutil.virtual_memory() + + # Log the total and available memory + _LOGGER.debug("Total memory: %s, Available memory: %s", memory.total, memory.available) + + if memory.available > 1024 * 1024 * 100: # 100MB + _LOGGER.debug("Sufficient memory available: %s bytes", memory.available) + return True + else: + _LOGGER.warning("Insufficient memory available: %s bytes", memory.available) + return False + class AsyncFileHandler: """Async context manager for file operations.""" @@ -62,7 +81,7 @@ class HATextAICoordinator(DataUpdateCoordinator): hass: HomeAssistant, client: Any, model: str, - update_interval: int, # Moved up + update_interval: int, instance_name: str, max_tokens: int = DEFAULT_MAX_TOKENS, temperature: float = DEFAULT_TEMPERATURE, @@ -78,26 +97,13 @@ class HATextAICoordinator(DataUpdateCoordinator): from .config_flow import normalize_name self.normalized_name = normalize_name(instance_name) - - self.hass = hass - self.client = client - self.model = model - self.temperature = temperature - self.max_tokens = max_tokens - self.max_history_size = min( - max(1, max_history_size), - ABSOLUTE_MAX_HISTORY_SIZE + self._metrics_file = os.path.join( + hass.config.path(".storage"), + "ha_text_ai_history", + f"ha_text_ai_metrics_{normalize_name(instance_name)}.json" ) - self.is_anthropic = is_anthropic - - # Initialize essential attributes first - self._is_processing = False - self._is_rate_limited = False - self._is_maintenance = False - self.endpoint_status = "ready" - self._system_prompt = None - self._conversation_history = [] + # Initialize performance metrics first self._performance_metrics = { "total_tokens": 0, "prompt_tokens": 0, @@ -110,6 +116,26 @@ class HATextAICoordinator(DataUpdateCoordinator): "min_latency": float("inf"), } + # Continue with other initializations + self.hass = hass + self.client = client + self.model = model + self.temperature = temperature + self.max_tokens = max_tokens + self.max_history_size = min( + max(1, max_history_size), + ABSOLUTE_MAX_HISTORY_SIZE + ) + self.is_anthropic = is_anthropic + + # Initialize essential attributes + self._is_processing = False + self._is_rate_limited = False + self._is_maintenance = False + self.endpoint_status = "ready" + self._system_prompt = None + self._conversation_history = [] + self._last_response = { "timestamp": dt_util.utcnow().isoformat(), "question": "", @@ -122,7 +148,7 @@ class HATextAICoordinator(DataUpdateCoordinator): update_interval_td = timedelta(seconds=update_interval) - # Call super().__init__ BEFORE other property access + # Call super().__init__ super().__init__( hass, _LOGGER, @@ -130,11 +156,11 @@ class HATextAICoordinator(DataUpdateCoordinator): update_interval=update_interval_td, ) - # Now initialize _initial_state (after super().__init__) + # Now initialize _initial_state self._initial_state = { "state": STATE_READY, "metrics": self._performance_metrics.copy(), - "last_response": self.last_response.copy(), # Accessing here + "last_response": self.last_response.copy(), "is_processing": self._is_processing, "is_rate_limited": self._is_rate_limited, "is_maintenance": self._is_maintenance, @@ -156,17 +182,22 @@ class HATextAICoordinator(DataUpdateCoordinator): hass.config.path(".storage"), "ha_text_ai_history" ) - os.makedirs(self._history_dir, exist_ok=True) + hass.async_create_task(self._create_history_dir()) hass.async_create_task(self._check_history_directory()) + hass.async_create_task(self._migrate_history_from_txt_to_json()) + + hass.async_create_task(self._initialize_metrics()) + # History file path using instance name self._history_file = os.path.join( self._history_dir, - f"{self.normalized_name}_history.txt" + f"{self.normalized_name}_history.json" ) - # Maximum history file size (10 MB) - self._max_history_file_size = 10 * 1024 * 1024 + + # Maximum history file size (1 MB) from const.py + self._max_history_file_size = MAX_HISTORY_FILE_SIZE # Asynchronous file initialization hass.async_create_task(self.async_initialize_history_file()) @@ -221,46 +252,206 @@ class HATextAICoordinator(DataUpdateCoordinator): """ self._last_response = value + async def _file_exists(self, path: str) -> bool: + """ + Check if file exists asynchronously. + + Args: + path (str): Full path to the file to check + + Returns: + bool: True if file exists, False otherwise + """ + try: + result = await self.hass.async_add_executor_job(os.path.exists, path) + + _LOGGER.debug(f"File existence check: {path} - {'Exists' if result else 'Not Found'}") + + return result + + except Exception as e: + _LOGGER.error(f"Error checking file existence for {path}: {e}") + return False + + async def _create_history_dir(self): + """ + Asynchronously create history directory. + + Creates the directory for storing history files + without blocking the event loop. + """ + try: + await self.hass.async_add_executor_job( + lambda: os.makedirs(self._history_dir, exist_ok=True) + ) + _LOGGER.debug(f"Directory creation details: exist_ok=True") + except PermissionError: + _LOGGER.error(f"Permission denied when creating history directory: {self._history_dir}") + raise + except OSError as e: + _LOGGER.error(f"Error creating history directory {self._history_dir}: {e}") + raise + + async def _initialize_metrics(self) -> None: + self._performance_metrics = await self._load_metrics() or { + "total_tokens": 0, + "prompt_tokens": 0, + "completion_tokens": 0, + "successful_requests": 0, + "failed_requests": 0, + "total_errors": 0, + "average_latency": 0, + "max_latency": 0, + "min_latency": float("inf"), + } + + async def _load_metrics(self) -> Dict[str, Any]: + try: + if await self._file_exists(self._metrics_file): + def read_metrics(): + with open(self._metrics_file, 'r') as f: + try: + return json.load(f) + except json.JSONDecodeError: + _LOGGER.warning("Metrics file corrupted, creating new") + return None + return await self.hass.async_add_executor_job(read_metrics) + except Exception as e: + _LOGGER.warning(f"Failed to load metrics: {e}") + return None + + async def _save_metrics(self) -> None: + try: + def write_metrics(): + with open(self._metrics_file, 'w') as f: + json.dump(self._performance_metrics, f) + await self.hass.async_add_executor_job(write_metrics) + except Exception as e: + _LOGGER.warning(f"Failed to save metrics: {e}") + + async def _update_metrics(self, latency: float, response: dict) -> None: + """Update performance metrics and save to storage.""" + metrics = self._performance_metrics + tokens = response.get("tokens", {}) + + metrics["total_tokens"] += tokens.get("total", 0) + metrics["prompt_tokens"] += tokens.get("prompt", 0) + metrics["completion_tokens"] += tokens.get("completion", 0) + metrics["successful_requests"] += 1 + + metrics["average_latency"] = ( + (metrics["average_latency"] * (metrics["successful_requests"] - 1) + latency) + / metrics["successful_requests"] + ) + metrics["max_latency"] = max(metrics["max_latency"], latency) + metrics["min_latency"] = min(metrics["min_latency"], latency) + + # Save metrics after update + await self._save_metrics() + + async def _get_current_metrics(self) -> Dict[str, Any]: + """Get current performance metrics.""" + metrics = self._performance_metrics.copy() + _LOGGER.debug(f"Current performance metrics: {metrics}") + return metrics + + async def _handle_error(self, error: Exception) -> None: + """Enhanced error handling with metric persistence.""" + self._performance_metrics["total_errors"] += 1 + self._performance_metrics["failed_requests"] += 1 + + # Save metrics after error + await self._save_metrics() + + error_details = { + "timestamp": dt_util.utcnow().isoformat(), + "model": self.model, + "instance": self.instance_name, + "error_message": str(error), + "error_type": type(error).__name__, + "traceback": traceback.format_exc() if _LOGGER.isEnabledFor(logging.DEBUG) else None, + } + + # Specific error type handling + error_mapping = { + HomeAssistantError: {"is_ha_error": True}, + ConnectionError: { + "is_connection_error": True, + "is_rate_limited": True + }, + TimeoutError: {"is_timeout": True}, + PermissionError: {"is_permission_denied": True}, + ValueError: {"is_validation_error": True} + } + + for error_type, error_flags in error_mapping.items(): + if isinstance(error, error_type): + error_details.update(error_flags) + break + + # Update system state based on error type + if error_details.get("is_rate_limited"): + self._is_rate_limited = True + _LOGGER.warning(f"Rate limit detected for {self.instance_name}") + + if error_details.get("is_connection_error"): + self.endpoint_status = "unavailable" + + self.last_response = error_details + _LOGGER.error(f"AI Processing Error: {error_details}") + + if _LOGGER.isEnabledFor(logging.DEBUG): + _LOGGER.debug(f"Full Error Traceback: {error_details['traceback']}") + async def async_initialize_history_file(self) -> None: """ Asynchronously initialize history file. Creates the file and writes an initialization timestamp - without blocking the event loop. + without blocking the event loop using Home Assistant's executor. """ try: - # Use asyncio.to_thread to perform file I/O in a separate thread - await asyncio.to_thread(self._sync_initialize_history_file) + await self.hass.async_add_executor_job(self._sync_initialize_history_file) except Exception as e: _LOGGER.error(f"Could not initialize history file: {e}") _LOGGER.debug(traceback.format_exc()) - def _sync_initialize_history_file(self) -> None: - """ - Synchronous method to create and initialize history file. - - Runs in a separate thread to avoid blocking the event loop. - """ + async def _sync_initialize_history_file(self) -> None: try: - with open(self._history_file, 'a') as f: - f.write(f"History initialized at: {dt_util.utcnow().isoformat()}\n") + if not await self._file_exists(self._history_file): + def write_empty_history(): + with open(self._history_file, 'w') as f: + json.dump([], f) + await self.hass.async_add_executor_job(write_empty_history) except Exception as e: - _LOGGER.error(f"Synchronous history file initialization failed: {e}") + _LOGGER.error(f"History file initialization failed: {e}") - # Add size check to _update_history method + # Size check to _update_history method async def _update_history(self, question: str, response: dict) -> None: - """Update conversation history with size limits.""" + """Update conversation history with size validation.""" try: - # Limit entry size - question = question[:MAX_ENTRY_SIZE] - response_content = response.get("content", "")[:MAX_ENTRY_SIZE] - history_entry = { "timestamp": dt_util.utcnow().isoformat(), - "question": question, - "response": response_content, + "question": self._truncate_text(question, MAX_ATTRIBUTE_SIZE), + "response": self._truncate_text( + response.get("content", ""), + MAX_ATTRIBUTE_SIZE + ), } + # Check approximate entry size + entry_size = len(json.dumps(history_entry).encode('utf-8')) + current_size = await self._check_file_size(self._history_file) + + if current_size + entry_size > MAX_HISTORY_FILE_SIZE: + _LOGGER.warning( + f"History size limit approaching. " + f"Current: {current_size}, " + f"Entry: {entry_size}, " + f"Max: {MAX_HISTORY_FILE_SIZE}" + ) + await self._rotate_history() + self._conversation_history.append(history_entry) # Enforce size limit @@ -269,48 +460,93 @@ class HATextAICoordinator(DataUpdateCoordinator): await self._write_history_entry(history_entry) - await self._rotate_history() - except Exception as e: _LOGGER.error(f"Error updating history: {e}") _LOGGER.debug(traceback.format_exc()) - # Update _write_history_entry method async def _write_history_entry(self, entry: dict) -> None: - """Write a single history entry to file asynchronously.""" + """Write history entry with file size checks.""" try: - if not os.path.exists(self._history_dir): - os.makedirs(self._history_dir, exist_ok=True) + if not await self._file_exists(self._history_dir): + await self._create_history_dir() - _LOGGER.debug(f"Writing history entry to {self._history_file}") - - async with AsyncFileHandler(self._history_file) as f: - await f.write( - f"{entry['timestamp']}: " - f"Question: {entry['question']} - " - f"Response: {entry['response']}\n" + # Check current file size + current_size = 0 + if await self._file_exists(self._history_file): + current_size = await self.hass.async_add_executor_job( + os.path.getsize, + self._history_file ) - await f.flush() - _LOGGER.debug(f"Successfully wrote history entry") + # Calculate approximate entry size + entry_size = len(json.dumps(entry).encode('utf-8')) + + if current_size + entry_size > MAX_HISTORY_FILE_SIZE: + _LOGGER.warning( + f"History file size limit reached. Current: {current_size}, " + f"Entry size: {entry_size}, Max: {MAX_HISTORY_FILE_SIZE}" + ) + # Trigger rotation before writing + await self._rotate_history() + + # Continue with writing + history = [] + if await self._file_exists(self._history_file): + async with AsyncFileHandler(self._history_file, 'r') as f: + content = await f.read() + if content: + history = json.loads(content) + + history.append(entry) + + if len(history) > self.max_history_size: + history = history[-self.max_history_size:] + + async with AsyncFileHandler(self._history_file, 'w') as f: + await f.write(json.dumps(history, indent=2)) except Exception as e: _LOGGER.error(f"Error writing history entry: {e}") _LOGGER.debug(traceback.format_exc()) - def _sync_write_history_entry(self, entry: dict) -> None: + async def _check_file_size(self, file_path: str) -> int: """ - Synchronous method to write history entry. + Check file size asynchronously. - Runs in a separate thread to avoid blocking the event loop. + Args: + file_path: Path to file to check + + Returns: + int: File size in bytes """ try: - with open(self._history_file, 'a') as f: - f.write( - f"{entry['timestamp']}: " - f"Question: {entry['question']} - " - f"Response: {entry['response']}\n" - ) + if await self._file_exists(file_path): + size = await self.hass.async_add_executor_job(os.path.getsize, file_path) + _LOGGER.debug(f"File size check for {file_path}: {size} bytes") + return size + return 0 + except Exception as e: + _LOGGER.error(f"Error checking file size for {file_path}: {e}") + return 0 + + def _sync_write_history_entry(self, entry: dict) -> None: + """Synchronous method to write history entry.""" + try: + history = [] + if os.path.exists(self._history_file): + with open(self._history_file, 'r') as f: + content = f.read() + if content: + history = json.loads(content) + + history.append(entry) + + if len(history) > self.max_history_size: + history = history[-self.max_history_size:] + + with open(self._history_file, 'w') as f: + json.dump(history, f, indent=2) + except Exception as e: _LOGGER.error(f"Synchronous history entry writing failed: {e}") @@ -318,9 +554,8 @@ class HATextAICoordinator(DataUpdateCoordinator): """Rotate conversation history with file management.""" try: _LOGGER.debug(f"Starting history rotation for {self._history_file}") - await asyncio.to_thread(self._sync_rotate_history) + await self.hass.async_add_executor_job(self._rotate_history_files) _LOGGER.debug(f"Completed history rotation") - except Exception as e: _LOGGER.error(f"Error rotating history: {e}") _LOGGER.debug(traceback.format_exc()) @@ -332,7 +567,7 @@ class HATextAICoordinator(DataUpdateCoordinator): try: # Test write permission in a separate thread test_file_path = os.path.join(self._history_dir, ".write_test") - await asyncio.to_thread(self._sync_test_directory_write, test_file_path) + await self.hass.async_add_executor_job(self._sync_test_directory_write, test_file_path) except PermissionError: _LOGGER.error(f"No write permissions for history directory: {self._history_dir}") @@ -350,92 +585,128 @@ class HATextAICoordinator(DataUpdateCoordinator): except Exception as e: _LOGGER.error(f"Directory write test failed: {e}") - def _sync_rotate_history(self) -> None: - """ - Synchronous method to rotate history files. - - Runs in a separate thread to avoid blocking the event loop. - """ + async def _rotate_history_files(self) -> None: + """Rotate history files with size validation.""" try: - # Check and manage file size - if os.path.exists(self._history_file): - file_size = os.path.getsize(self._history_file) - if file_size > self._max_history_file_size: - # Create timestamped archive + if await self._file_exists(self._history_file): + current_size = await self._check_file_size(self._history_file) + + if current_size > MAX_HISTORY_FILE_SIZE: + _LOGGER.info( + f"Rotating history file. " + f"Current size: {current_size}, " + f"Max allowed: {MAX_HISTORY_FILE_SIZE}" + ) + + # Create archive filename with timestamp archive_file = os.path.join( self._history_dir, - f"{self.normalized_name}_history_{dt_util.utcnow().strftime('%Y%m%d_%H%M%S')}.txt" + f"{self.normalized_name}_history_{dt_util.utcnow().strftime('%Y%m%d_%H%M%S')}.json" ) - os.rename(self._history_file, archive_file) - # Optional: Log rotation event - _LOGGER.info(f"History file rotated: {archive_file}") - - # Ensure history doesn't exceed max size - if len(self._conversation_history) > self.max_history_size: - # Trim in-memory history - self._conversation_history = self._conversation_history[-self.max_history_size:] - - # Write history entries - with open(self._history_file, 'a') as f: - for entry in self._conversation_history: - f.write( - f"{entry['timestamp']}: {entry['question']} - {entry['response']}\n" + # Rotate files + await self.hass.async_add_executor_job( + os.rename, + self._history_file, + archive_file ) + # Create new history file with recent entries + async with AsyncFileHandler(self._history_file, 'w') as f: + await f.write(json.dumps( + self._conversation_history[-self.max_history_size:], + indent=2 + )) + + _LOGGER.info(f"History file rotated to: {archive_file}") + except Exception as e: - _LOGGER.error(f"Synchronous history rotation failed: {e}") + _LOGGER.error(f"History rotation failed: {e}") _LOGGER.debug(traceback.format_exc()) async def _async_update_data(self) -> Dict[str, Any]: - """Update data via library.""" + """Update coordinator data with improved error handling and performance.""" + try: - current_state = self._get_current_state() - _LOGGER.debug( - f"Updating data for {self.instance_name}, current state: {current_state}" - ) + async with asyncio.Semaphore(1): + current_state = self._get_current_state() - limited_history = [] - if self._conversation_history: - for entry in self._conversation_history[-3:]: - limited_entry = { - "timestamp": entry["timestamp"], - "question": entry["question"][:4096], - "response": entry["response"][:4096], - } - limited_history.append(limited_entry) + limited_history = await self._get_limited_history() - limited_last_response = self.last_response.copy() - if "response" in limited_last_response: - limited_last_response["response"] = limited_last_response["response"][:4096] - if "question" in limited_last_response: - limited_last_response["question"] = limited_last_response["question"][:4096] + metrics = await self._get_current_metrics() - data = { - "state": current_state, - "metrics": self._performance_metrics, - "last_response": limited_last_response, - "is_processing": self._is_processing, - "is_rate_limited": self._is_rate_limited, - "is_maintenance": self._is_maintenance, - "endpoint_status": self.endpoint_status, - "uptime": (dt_util.utcnow() - self._start_time).total_seconds(), - "system_prompt": self._system_prompt[:4096] if self._system_prompt else None, - "history_size": len(self._conversation_history), - "conversation_history": limited_history, - "normalized_name": self.normalized_name, - } + if metrics is None: + metrics = {} - # Validate data - if not isinstance(data, dict): - raise ValueError("Invalid data format") + data = { + "state": current_state, + "metrics": metrics, + "last_response": await self._get_sanitized_last_response(), + "is_processing": self._is_processing, + "is_rate_limited": self._is_rate_limited, + "is_maintenance": self._is_maintenance, + "endpoint_status": self.endpoint_status, + "uptime": self._calculate_uptime(), + "system_prompt": self._get_truncated_system_prompt(), + "history_size": len(self._conversation_history), + "conversation_history": limited_history, + "normalized_name": self.normalized_name, + } - _LOGGER.debug(f"Updated data for {self.instance_name}: {data}") - return data + await self._validate_update_data(data) + return data + + except asyncio.CancelledError: + _LOGGER.warning("Update was cancelled") + raise except Exception as err: - _LOGGER.error(f"Error updating data for {self.instance_name}: {err}") - return self._initial_state + _LOGGER.error(f"Error updating data: {err}", exc_info=True) + return self._get_safe_initial_state() + + async def _get_limited_history(self) -> List[Dict[str, Any]]: + """Get limited conversation history with size constraints.""" + limited_history = [] + if self._conversation_history: + for entry in self._conversation_history[-3:]: + limited_entry = { + "timestamp": entry["timestamp"], + "question": self._truncate_text(entry["question"]), + "response": self._truncate_text(entry["response"]), + } + limited_history.append(limited_entry) + return limited_history + + def _truncate_text(self, text: str, max_length: int = MAX_ATTRIBUTE_SIZE) -> str: + """Safely truncate text to maximum length.""" + return text[:max_length] if text else "" + + async def _get_sanitized_last_response(self) -> Dict[str, Any]: + """Get sanitized version of last response.""" + response = self.last_response.copy() + if "response" in response: + response["response"] = self._truncate_text(response["response"]) + if "question" in response: + response["question"] = self._truncate_text(response["question"]) + return response + + def _calculate_uptime(self) -> float: + """Calculate current uptime in seconds.""" + return (dt_util.utcnow() - self._start_time).total_seconds() + + def _get_truncated_system_prompt(self) -> Optional[str]: + """Get truncated system prompt.""" + return self._truncate_text(self._system_prompt, 4096) if self._system_prompt else None + + async def _validate_update_data(self, data: Dict[str, Any]) -> None: + """Validate update data structure and content.""" + required_keys = ["state", "metrics", "last_response"] + for key in required_keys: + if key not in data: + raise ValueError(f"Missing required key: {key}") + + if not isinstance(data["metrics"], dict): + raise ValueError("Invalid metrics format") async def async_update_ha_state(self) -> None: """Update Home Assistant state.""" @@ -466,66 +737,61 @@ class HATextAICoordinator(DataUpdateCoordinator): return STATE_ERROR return STATE_READY - def _calculate_context_tokens(self, messages: List[Dict[str, str]], model: str = None) -> int: - """ - Estimate tokens for conversation context. + def _get_safe_initial_state(self) -> Dict[str, Any]: + """Return safe initial state when update fails.""" + return { + "state": STATE_ERROR, + "metrics": {}, + "last_response": self.last_response, + "is_processing": False, + "is_rate_limited": False, + "is_maintenance": False, + "endpoint_status": "error", + "uptime": self._calculate_uptime(), + "system_prompt": None, + "history_size": 0, + "conversation_history": [], + "normalized_name": self.normalized_name, + } - Args: - messages: List of message dictionaries - model: Optional model name for provider-specific estimation + def _calculate_context_tokens(self, messages: List[Dict[str, str]]) -> int: + total_tokens = 0 + + # Compile regular expressions for performance + number_pattern = re.compile(r'[0-9]') + special_char_pattern = re.compile(r'[^\w\s]') + whitespace_pattern = re.compile(r'\s+') - Returns: - Estimated number of tokens - """ try: - # Anthropic specific token counting - if self.is_anthropic and hasattr(self.client, 'count_tokens'): - return sum(self.client.count_tokens(msg['content']) for msg in messages) + for message in messages: + text = message.get('content', '') + if text: + # Normalize whitespace + text = whitespace_pattern.sub(' ', text.strip()) - def estimate_tokens(text: str) -> int: - """ - Flexible token estimation algorithm. + # Advanced token estimation heuristics + words = text.split() + for word in words: + # Complexity-based token calculation + if len(word) > 8: # Long words + total_tokens += 2 + elif number_pattern.search(word): # Words with numbers + total_tokens += 1.5 + elif special_char_pattern.search(word): # Words with special characters + total_tokens += 1.5 + elif word.isupper(): # Acronyms and technical terms + total_tokens += 1.5 + else: # Regular words + total_tokens += 1 - Heuristics: - - Count words - - Estimate special characters - - Fallback to character-based estimation - """ - # Word-based estimation - words = len(text.split()) + # Additional correction for technical texts + total_tokens = math.ceil(total_tokens * 1.2) - # Special character handling - special_chars = sum(1 for char in text if not char.isalnum()) - - # Character-based fallback - char_tokens = len(text) // 4 - - # Combine estimations with bias towards words - total_tokens = (words * 1.5) + (special_chars * 0.5) + char_tokens - - return max(int(total_tokens), words) - - # Calculate total tokens across all messages - total_tokens = sum(estimate_tokens(msg['content']) for msg in messages) - - # Logging for debugging - _LOGGER.debug( - f"Token Estimation: " - f"Messages: {len(messages)}, " - f"Estimated Tokens: {total_tokens}" - ) - - return total_tokens + return int(total_tokens) except Exception as e: - # Safe fallback with detailed logging - _LOGGER.warning( - f"Token estimation failed. " - f"Error: {e}. " - f"Using conservative estimation." - ) - - # Conservative token estimation + _LOGGER.error(f"Token calculation error: {e}. Processed {len(messages)} messages.") + # Safe fallback with minimal token estimation return len(messages) * 100 async def async_ask_question( @@ -571,6 +837,9 @@ class HATextAICoordinator(DataUpdateCoordinator): """ Enhanced question processing with intelligent token management. """ + if self.client is None: + raise HomeAssistantError("AI client not initialized") + try: self._is_processing = True await self.async_update_ha_state() @@ -650,7 +919,7 @@ class HATextAICoordinator(DataUpdateCoordinator): # Update metrics end_time = dt_util.utcnow() latency = (end_time - start_time).total_seconds() - self._update_metrics(latency, response) + await self._update_metrics(latency, response) await self._update_history(question, response) @@ -667,10 +936,11 @@ class HATextAICoordinator(DataUpdateCoordinator): async def async_process_message(self, question: str, **kwargs) -> dict: """Process message using the AI client.""" try: - if self.is_anthropic: - response = await self._process_anthropic_message(question, **kwargs) - else: - response = await self._process_openai_message(question, **kwargs) + async with asyncio.timeout(60): # 60 second timeout + if self.is_anthropic: + response = await self._process_anthropic_message(question, **kwargs) + else: + response = await self._process_openai_message(question, **kwargs) self.last_response = { "timestamp": dt_util.utcnow().isoformat(), @@ -678,11 +948,15 @@ class HATextAICoordinator(DataUpdateCoordinator): "response": response["content"], "model": kwargs.get("model", self.model), "instance": self.instance_name, + "normalized_name": self.normalized_name, "error": None, } return response + except asyncio.TimeoutError: + raise HomeAssistantError("Request timed out") + except Exception as err: self._handle_error(err) raise @@ -732,73 +1006,73 @@ class HATextAICoordinator(DataUpdateCoordinator): _LOGGER.error(f"Error in OpenAI API call: {str(e)}") raise - def _update_metrics(self, latency: float, response: dict) -> None: - """Update performance metrics.""" - metrics = self._performance_metrics - tokens = response.get("tokens", {}) + async def _migrate_history_from_txt_to_json(self) -> None: + try: + old_history_file = os.path.join( + self._history_dir, + f"{self.normalized_name}_history.txt" + ) - metrics["total_tokens"] += tokens.get("total", 0) - metrics["prompt_tokens"] += tokens.get("prompt", 0) - metrics["completion_tokens"] += tokens.get("completion", 0) - metrics["successful_requests"] += 1 + if not await self._file_exists(old_history_file): + return - metrics["average_latency"] = ( - (metrics["average_latency"] * (metrics["successful_requests"] - 1) + latency) - / metrics["successful_requests"] - ) - metrics["max_latency"] = max(metrics["max_latency"], latency) - metrics["min_latency"] = min(metrics["min_latency"], latency) + _LOGGER.info(f"Found old history file for {self.instance_name}, starting migration to JSON format") - def _handle_error(self, error: Exception) -> None: - """ - Enhanced error handling with comprehensive diagnostics. + # Read old history + history_entries = [] + async with AsyncFileHandler(old_history_file, 'r') as f: + content = await f.read() - Captures detailed error information, tracks error metrics, - and provides context for troubleshooting AI processing issues. - """ - self._performance_metrics["total_errors"] += 1 - self._performance_metrics["failed_requests"] += 1 + # Parse txt content + for line in content.split('\n'): + if not line or line.startswith("History initialized at:"): + continue - error_details = { - "timestamp": dt_util.utcnow().isoformat(), - "model": self.model, - "instance": self.instance_name, - "error_message": str(error), - "error_type": type(error).__name__, - "traceback": traceback.format_exc() if _LOGGER.isEnabledFor(logging.DEBUG) else None, - } + try: + # Parse the old format: "timestamp: Question: question - Response: response" + parts = line.split(': ', 1) + if len(parts) != 2: + continue - # Specific error type handling - error_mapping = { - HomeAssistantError: {"is_ha_error": True}, - ConnectionError: { - "is_connection_error": True, - "is_rate_limited": True - }, - TimeoutError: {"is_timeout": True}, - PermissionError: {"is_permission_denied": True}, - ValueError: {"is_validation_error": True} - } + timestamp = parts[0] + content_parts = parts[1].split(' - ') + if len(content_parts) != 2: + continue - for error_type, error_flags in error_mapping.items(): - if isinstance(error, error_type): - error_details.update(error_flags) - break + question = content_parts[0].replace('Question: ', '') + response = content_parts[1].replace('Response: ', '') - # Update system state based on error type - if error_details.get("is_rate_limited"): - self._is_rate_limited = True - _LOGGER.warning(f"Rate limit detected for {self.instance_name}") + history_entries.append({ + "timestamp": timestamp, + "question": question, + "response": response + }) - if error_details.get("is_connection_error"): - self.endpoint_status = "unavailable" + except Exception as e: + _LOGGER.warning(f"Error parsing history line: {line}. Error: {e}") + continue - self.last_response = error_details - _LOGGER.error(f"AI Processing Error: {error_details}") + if history_entries: + # Write to new JSON file + async with AsyncFileHandler(self._history_file, 'w') as f: + await f.write(json.dumps(history_entries, indent=2)) - # Optional: Add more sophisticated error tracking or notification logic - if _LOGGER.isEnabledFor(logging.DEBUG): - _LOGGER.debug(f"Full Error Traceback: {error_details['traceback']}") + # Backup old file + backup_file = old_history_file + '.backup' + os.rename(old_history_file, backup_file) + + _LOGGER.info( + f"Successfully migrated {len(history_entries)} entries " + f"from txt to JSON for {self.instance_name}. " + f"Old file backed up as {backup_file}" + ) + + # Update in-memory history + self._conversation_history = history_entries + + except Exception as e: + _LOGGER.error(f"Error during history migration for {self.instance_name}: {e}") + _LOGGER.debug(traceback.format_exc()) async def async_clear_history(self) -> None: """ @@ -807,16 +1081,11 @@ class HATextAICoordinator(DataUpdateCoordinator): Removes in-memory history and deletes history file. """ try: - # Clear in-memory history self._conversation_history = [] - - # Safely remove history file - if os.path.exists(self._history_file): - await asyncio.to_thread(os.remove, self._history_file) - + if await self._file_exists(self._history_file): + await self.hass.async_add_executor_job(os.remove, self._history_file) await self.async_update_ha_state() _LOGGER.info(f"History for {self.instance_name} cleared successfully") - except Exception as e: _LOGGER.error(f"Error clearing history: {e}") _LOGGER.debug(traceback.format_exc()) diff --git a/custom_components/ha_text_ai/manifest.json b/custom_components/ha_text_ai/manifest.json index 7fe691a..51cca0a 100644 --- a/custom_components/ha_text_ai/manifest.json +++ b/custom_components/ha_text_ai/manifest.json @@ -13,16 +13,16 @@ "loggers": ["custom_components.ha_text_ai"], "mqtt": [], "quality_scale": "silver", - "requirements": [ - "openai>=1.12.0", - "anthropic>=0.8.0", - "aiohttp>=3.8.0", - "async-timeout>=4.0.0", - "certifi>=2024.2.2" - ], + "requirements": [ + "openai>=1.12.0", + "anthropic>=0.8.0", + "aiohttp>=3.8.0", + "async-timeout>=4.0.0", + "certifi>=2024.2.2" + ], "single_config_entry": false, "ssdp": [], "usb": [], - "version": "2.0.5-beta", + "version": "2.0.6-beta", "zeroconf": [] } diff --git a/custom_components/ha_text_ai/sensor.py b/custom_components/ha_text_ai/sensor.py index 52f952b..c8380fc 100644 --- a/custom_components/ha_text_ai/sensor.py +++ b/custom_components/ha_text_ai/sensor.py @@ -67,6 +67,7 @@ from .const import ( ENTITY_ICON_PROCESSING, DEFAULT_NAME_PREFIX, CONF_MAX_HISTORY_SIZE, + MAX_ATTRIBUTE_SIZE, ) from .coordinator import HATextAICoordinator @@ -178,12 +179,29 @@ class HATextAISensor(CoordinatorEntity, SensorEntity): def _sanitize_attributes(self, attributes: Dict[str, Any]) -> Dict[str, Any]: """Sanitize all attributes for JSON serialization.""" - return { + sanitized = { key: self._sanitize_value(value) for key, value in attributes.items() if value is not None } + # Log metrics for debugging + metrics_keys = [ + METRIC_TOTAL_TOKENS, + METRIC_PROMPT_TOKENS, + METRIC_COMPLETION_TOKENS, + METRIC_SUCCESSFUL_REQUESTS, + METRIC_FAILED_REQUESTS, + METRIC_AVERAGE_LATENCY, + METRIC_MAX_LATENCY, + METRIC_MIN_LATENCY, + ] + + metrics_values = {k: sanitized.get(k) for k in metrics_keys if k in sanitized} + _LOGGER.debug(f"Metrics for {self.entity_id}: {metrics_values}") + + return sanitized + @property def native_value(self) -> StateType: """Return the native value of the sensor.""" @@ -212,68 +230,65 @@ class HATextAISensor(CoordinatorEntity, SensorEntity): try: data = self.coordinator.data + metrics = data.get("metrics", {}) + + # Base attributes attributes = { ATTR_MODEL: self._config_entry.data.get(CONF_MODEL, "Unknown"), - ATTR_API_PROVIDER: self._config_entry.data.get( - CONF_API_PROVIDER, "Unknown" - ), + ATTR_API_PROVIDER: self._config_entry.data.get(CONF_API_PROVIDER, "Unknown"), ATTR_API_STATUS: self._current_state, - ATTR_TOTAL_ERRORS: self._error_count, - ATTR_LAST_ERROR: self._last_error, + ATTR_TOTAL_ERRORS: metrics.get("total_errors", 0), "instance_name": self._instance_name, "normalized_name": self._normalized_name, - ATTR_SYSTEM_PROMPT: data.get("system_prompt"), + ATTR_SYSTEM_PROMPT: (data.get("system_prompt", "")[:MAX_ATTRIBUTE_SIZE] + if data.get("system_prompt") else None), ATTR_IS_PROCESSING: data.get("is_processing", False), ATTR_IS_RATE_LIMITED: data.get("is_rate_limited", False), ATTR_IS_MAINTENANCE: data.get("is_maintenance", False), ATTR_ENDPOINT_STATUS: data.get("endpoint_status", "unknown"), - ATTR_UPTIME: data.get("uptime", 0), + ATTR_UPTIME: round(data.get("uptime", 0), 2), ATTR_HISTORY_SIZE: data.get("history_size", 0), - ATTR_CONVERSATION_HISTORY: data.get("conversation_history", []), } - # Add metrics - metrics = data.get("metrics", {}) - if isinstance(metrics, dict): - self._metrics = metrics - attributes.update( - { - METRIC_TOTAL_TOKENS: metrics.get("total_tokens", 0), - METRIC_PROMPT_TOKENS: metrics.get("prompt_tokens", 0), - METRIC_COMPLETION_TOKENS: metrics.get("completion_tokens", 0), - METRIC_SUCCESSFUL_REQUESTS: metrics.get( - "successful_requests", 0 - ), - METRIC_FAILED_REQUESTS: metrics.get("failed_requests", 0), - METRIC_AVERAGE_LATENCY: metrics.get("average_latency", 0), - METRIC_MAX_LATENCY: metrics.get("max_latency", 0), - METRIC_MIN_LATENCY: metrics.get("min_latency", float("inf")), + # History limit + conversation_history = data.get("conversation_history", []) + if conversation_history: + limited_history = [] + for entry in conversation_history: + limited_entry = { + "timestamp": entry["timestamp"], + "question": entry["question"][:MAX_ATTRIBUTE_SIZE], + "response": entry["response"][:MAX_ATTRIBUTE_SIZE] } - ) + limited_history.append(limited_entry) + attributes[ATTR_CONVERSATION_HISTORY] = limited_history - # Add last response + # Metrics + if isinstance(metrics, dict): + attributes.update({ + METRIC_TOTAL_TOKENS: metrics.get("total_tokens", 0), + METRIC_PROMPT_TOKENS: metrics.get("prompt_tokens", 0), + METRIC_COMPLETION_TOKENS: metrics.get("completion_tokens", 0), + METRIC_SUCCESSFUL_REQUESTS: metrics.get("successful_requests", 0), + METRIC_FAILED_REQUESTS: metrics.get("failed_requests", 0), + METRIC_AVERAGE_LATENCY: round(metrics.get("average_latency", 0), 2), + METRIC_MAX_LATENCY: round(metrics.get("max_latency", 0), 2), + METRIC_MIN_LATENCY: (metrics.get("min_latency") + if metrics.get("min_latency") != float("inf") + else None), + }) + + # Last response handling last_response = data.get("last_response", {}) if isinstance(last_response, dict): - self._last_response = last_response - attributes.update( - { - ATTR_RESPONSE: last_response.get("response", ""), - ATTR_QUESTION: last_response.get("question", ""), - "last_model": last_response.get("model", ""), - "last_timestamp": last_response.get("timestamp", ""), - "last_error": last_response.get("error"), - } - ) - - # Add performance metrics if available - if ATTR_PERFORMANCE_METRICS in data: - attributes[ATTR_PERFORMANCE_METRICS] = data[ - ATTR_PERFORMANCE_METRICS - ] - - # Add API version if available - if ATTR_API_VERSION in data: - attributes[ATTR_API_VERSION] = data[ATTR_API_VERSION] + attributes.update({ + ATTR_RESPONSE: last_response.get("response", "")[:MAX_ATTRIBUTE_SIZE], + ATTR_QUESTION: last_response.get("question", "")[:MAX_ATTRIBUTE_SIZE], + "last_model": last_response.get("model", ""), + "last_timestamp": last_response.get("timestamp", ""), + "last_error": (last_response.get("error", "")[:MAX_ATTRIBUTE_SIZE] + if last_response.get("error") else None), + }) return self._sanitize_attributes(attributes) @@ -299,6 +314,12 @@ class HATextAISensor(CoordinatorEntity, SensorEntity): self._is_processing = data.get("is_processing", False) + # Update metrics + metrics = data.get("metrics", {}) + if isinstance(metrics, dict): + self._metrics.update(metrics) + _LOGGER.debug(f"Updated metrics for {self.entity_id}: {self._metrics}") + # Update conversation history and system prompt self._conversation_history = data.get("conversation_history", []) self._system_prompt = data.get("system_prompt")