From be06fddce1e1ae5d15837a385f7cf1de13674391 Mon Sep 17 00:00:00 2001 From: SMKRV Date: Tue, 10 Dec 2024 16:19:17 +0300 Subject: [PATCH] fix: Display only last Q&A in sensor state to prevent data truncation - Show only the latest question and answer in sensor state - Keep full conversation history in attributes - Fix truncation issues in Home Assistant UI - Maintain backwards compatibility - No configuration changes required --- custom_components/ha_text_ai/const.py | 2 + custom_components/ha_text_ai/coordinator.py | 69 ++++++++++++++------- 2 files changed, 47 insertions(+), 24 deletions(-) diff --git a/custom_components/ha_text_ai/const.py b/custom_components/ha_text_ai/const.py index aa897bd..add50f2 100644 --- a/custom_components/ha_text_ai/const.py +++ b/custom_components/ha_text_ai/const.py @@ -56,6 +56,8 @@ DEFAULT_NAME_PREFIX = "ha_text_ai" DEFAULT_CONTEXT_MESSAGES: Final = 5 +TRUNCATION_INDICATOR = " … " + # Parameter constraints MIN_TEMPERATURE: Final = 0.0 MAX_TEMPERATURE: Final = 2.0 diff --git a/custom_components/ha_text_ai/coordinator.py b/custom_components/ha_text_ai/coordinator.py index 498ec74..c2f34e4 100644 --- a/custom_components/ha_text_ai/coordinator.py +++ b/custom_components/ha_text_ai/coordinator.py @@ -41,6 +41,7 @@ ABSOLUTE_MAX_HISTORY_SIZE, MAX_ATTRIBUTE_SIZE, MAX_HISTORY_FILE_SIZE, + TRUNCATION_INDICATOR, ) _LOGGER = logging.getLogger(__name__) @@ -647,15 +648,14 @@ async def _rotate_history_files(self) -> None: async def _async_update_data(self) -> Dict[str, Any]: """Update coordinator data with improved error handling and performance.""" - try: async with asyncio.Semaphore(1): current_state = self._get_current_state() - limited_history = await self._get_limited_history() + # Get limited history with info + history_data = await self._get_limited_history() metrics = await self._get_current_metrics() - if metrics is None: metrics = {} @@ -670,47 +670,68 @@ async def _async_update_data(self) -> Dict[str, Any]: "uptime": self._calculate_uptime(), "system_prompt": self._get_truncated_system_prompt(), "history_size": len(self._conversation_history), - "conversation_history": limited_history, + "conversation_history": history_data["entries"], + "history_info": history_data["info"], "normalized_name": self.normalized_name, } 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: {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.""" + async def _get_limited_history(self) -> Dict[str, Any]: + """Get limited conversation history showing only last Q&A.""" 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 + last_entry = self._conversation_history[-1] + limited_entry = { + "timestamp": last_entry["timestamp"], + "question": last_entry["question"][:4096] + (TRUNCATION_INDICATOR if len(last_entry["question"]) > 4096 else ""), + "response": last_entry["response"][:4096] + (TRUNCATION_INDICATOR if len(last_entry["response"]) > 4096 else ""), + } + limited_history.append(limited_entry) - 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 "" + history_info = { + "total_entries": len(self._conversation_history), + "displayed_entries": len(limited_history), + "full_history_available": True, + "history_path": self._history_file, + } + + return { + "entries": limited_history, + "full_history": self._conversation_history, + "info": history_info + } async def _get_sanitized_last_response(self) -> Dict[str, Any]: - """Get sanitized version of last response.""" + """Get sanitized version of last response with truncation indicators.""" response = self.last_response.copy() + if "response" in response: - response["response"] = self._truncate_text(response["response"]) + original_response = response["response"] + is_response_truncated = len(original_response) > 4096 + response["response"] = original_response[:4096] + (TRUNCATION_INDICATOR if is_response_truncated else "") + response["is_response_truncated"] = is_response_truncated + response["full_response_length"] = len(original_response) + if "question" in response: - response["question"] = self._truncate_text(response["question"]) + original_question = response["question"] + is_question_truncated = len(original_question) > 4096 + response["question"] = original_question[:4096] + (TRUNCATION_INDICATOR if is_question_truncated else "") + response["is_question_truncated"] = is_question_truncated + response["full_question_length"] = len(original_question) + return response + 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 "" + def _calculate_uptime(self) -> float: """Calculate current uptime in seconds.""" return (dt_util.utcnow() - self._start_time).total_seconds()