Hass/ha-text-ai
1
0
Fork 0
mirror of https://github.com/smkrv/ha-text-ai.git synced 2026-07-30 04:23:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

Release v2.0.0

Browse Source
This commit is contained in:
SMKRV
2024-11-24 17:27:49 +03:00
parent 5791601c7e
commit ac420b6495
2 changed files with 130 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/ha_text_ai/sensor.py
+95 -93
View File
@@ -1,4 +1,4 @@
"""Sensor platform for HA text AI."""
"""Sensor platform for HA Text AI."""
import logging
from typing import Any, Dict
@@ -13,13 +13,9 @@ from homeassistant.util import dt as dt_util
from homeassistant.util import slugify
from .const import (
# Основные константы домена
DOMAIN,
CONF_INSTANCE,
CONF_MODEL,
CONF_API_PROVIDER,
# Атрибуты сенсора
ATTR_TOTAL_RESPONSES,
ATTR_TOTAL_ERRORS,
ATTR_AVG_RESPONSE_TIME,
@@ -39,8 +35,6 @@ from .const import (
ATTR_API_STATUS,
ATTR_RESPONSE,
ATTR_QUESTION,
# Метрики
METRIC_TOTAL_TOKENS,
METRIC_PROMPT_TOKENS,
METRIC_COMPLETION_TOKENS,
@@ -49,8 +43,6 @@ from .const import (
METRIC_AVERAGE_LATENCY,
METRIC_MAX_LATENCY,
METRIC_MIN_LATENCY,
# Состояния и иконки
STATE_READY,
STATE_PROCESSING,
STATE_ERROR,
@@ -63,6 +55,8 @@ from .const import (
ENTITY_ICON_PROCESSING,
)
from .coordinator import HATextAICoordinator
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
@@ -70,42 +64,55 @@ async def async_setup_entry(
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the HA text AI sensor."""
"""Set up the HA Text AI sensor."""
coordinator = hass.data[DOMAIN][entry.entry_id]
async_add_entities([HATextAISensor(coordinator, entry)], True)
class HATextAISensor(CoordinatorEntity, SensorEntity):
"""HA text AI Sensor."""
"""HA Text AI Sensor."""
coordinator: HATextAICoordinator
def __init__(
self,
coordinator: dict,
coordinator: HATextAICoordinator,
config_entry: ConfigEntry,
) -> None:
"""Initialize the sensor."""
super().__init__(coordinator)
self._config_entry = config_entry
self._name = config_entry.title
self._attr_unique_id = f"{config_entry.entry_id}_{slugify(self._name)}"
self.entity_id = f"sensor.ha_text_ai_{slugify(self._name)}"
self._attr_name = self._name
self._attr_name = config_entry.title
self._attr_unique_id = f"{config_entry.entry_id}_{slugify(self._attr_name)}"
self.entity_id = f"sensor.ha_text_ai_{slugify(self._attr_name)}"
self._current_state = STATE_INITIALIZING
self._error_count = 0
self._last_error = None
self._last_update = None
# Получаем данные из конфигурации
model = self._config_entry.data.get(CONF_MODEL, "Unknown")
api_provider = self._config_entry.data.get(CONF_API_PROVIDER, "Unknown")
# Device info
model = config_entry.data.get(CONF_MODEL, "Unknown")
api_provider = config_entry.data.get(CONF_API_PROVIDER, "Unknown")
# Обновляем device_info с использованием DeviceInfo
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, self._attr_unique_id)},
name=self._name,
name=self._attr_name,
manufacturer="Community",
model=f"{model} ({api_provider} provider)",
sw_version="v1",
sw_version="1.0.0",
)
@property
def native_value(self) -> StateType:
"""Return the native value of the sensor."""
if self.coordinator.data is None:
return STATE_DISCONNECTED
state = self.coordinator.data.get("state", STATE_READY)
self._current_state = state
return state
@property
def icon(self) -> str:
"""Return the icon based on the current state."""
@@ -115,50 +122,21 @@ class HATextAISensor(CoordinatorEntity, SensorEntity):
return ENTITY_ICON_PROCESSING
return ENTITY_ICON
@property
def state(self) -> StateType:
"""Return the state of the sensor."""
try:
if isinstance(self.coordinator, dict):
last_update = self.coordinator.get("last_update")
if last_update:
if isinstance(last_update, str):
return dt_util.parse_datetime(last_update)
return last_update
except Exception as err:
_LOGGER.debug("Error parsing state: %s", err)
return self._current_state
@property
def extra_state_attributes(self) -> Dict[str, Any]:
"""Return entity specific state attributes."""
attributes = {
ATTR_TOTAL_RESPONSES: 0,
ATTR_MODEL: self._config_entry.data.get(CONF_MODEL, "Unknown"),
ATTR_API_PROVIDER: self._config_entry.data.get(CONF_API_PROVIDER, "Unknown"),
ATTR_API_STATUS: self._current_state,
ATTR_ERROR_COUNT: self._error_count,
ATTR_TOTAL_ERRORS: self._error_count,
ATTR_LAST_ERROR: self._last_error,
}
if isinstance(self.coordinator, dict):
# Обновляем метрики если они есть
metrics = self.coordinator.get("metrics", {})
if isinstance(metrics, dict):
attributes.update(metrics)
if self.coordinator.data:
data = self.coordinator.data
# Обновляем статус API
if "status" in self.coordinator:
attributes[ATTR_API_STATUS] = self.coordinator["status"]
# Обновляем информацию о последнем ответе
last_response = self.coordinator.get("last_response", {})
if isinstance(last_response, dict):
attributes.update({
ATTR_RESPONSE: last_response.get("response", ""),
ATTR_QUESTION: last_response.get("question", ""),
})
# Обновляем дополнительные атрибуты
# Основные атрибуты
for attr in [
ATTR_TOTAL_RESPONSES,
ATTR_AVG_RESPONSE_TIME,
@@ -171,50 +149,74 @@ class HATextAISensor(CoordinatorEntity, SensorEntity):
ATTR_PERFORMANCE_METRICS,
ATTR_HISTORY_SIZE,
ATTR_UPTIME,
ATTR_SYSTEM_PROMPT,
]:
if attr in self.coordinator:
attributes[attr] = self.coordinator[attr]
if attr in data:
attributes[attr] = data[attr]
# Метрики
if "metrics" in data:
metrics = data["metrics"]
for metric in [
METRIC_TOTAL_TOKENS,
METRIC_PROMPT_TOKENS,
METRIC_COMPLETION_TOKENS,
METRIC_SUCCESSFUL_REQUESTS,
METRIC_FAILED_REQUESTS,
METRIC_AVERAGE_LATENCY,
METRIC_MAX_LATENCY,
METRIC_MIN_LATENCY,
]:
if metric in metrics:
attributes[metric] = metrics[metric]
# Последний ответ
if "last_response" in data:
last_response = data["last_response"]
attributes[ATTR_RESPONSE] = last_response.get("response", "")
attributes[ATTR_QUESTION] = last_response.get("question", "")
return attributes
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
try:
if not isinstance(self.coordinator, dict):
self._current_state = STATE_DISCONNECTED
return
# Определяем текущее состояние
if self.coordinator.get("is_processing"):
self._current_state = STATE_PROCESSING
elif self.coordinator.get("is_rate_limited"):
self._current_state = STATE_RATE_LIMITED
elif self.coordinator.get("is_maintenance"):
self._current_state = STATE_MAINTENANCE
else:
self._current_state = STATE_READY
# Обновляем счетчик ошибок из метрик
metrics = self.coordinator.get("metrics", {})
if isinstance(metrics, dict):
self._error_count = metrics.get("total_errors", self._error_count)
except Exception as err:
_LOGGER.error("Error handling update: %s", err, exc_info=True)
self._error_count += 1
self._last_error = str(err)
self._current_state = STATE_ERROR
self.async_write_ha_state()
async def async_added_to_hass(self) -> None:
"""When entity is added to hass."""
await super().async_added_to_hass()
self._handle_coordinator_update()
self._current_state = STATE_READY
async def async_reset_error_count(self) -> None:
"""Reset the error counter."""
self._error_count = 0
self._last_error = None
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
if self.coordinator.data is None:
self._current_state = STATE_DISCONNECTED
self.async_write_ha_state()
return
try:
data = self.coordinator.data
# Обновляем состояние
if data.get("is_processing"):
self._current_state = STATE_PROCESSING
elif data.get("is_rate_limited"):
self._current_state = STATE_RATE_LIMITED
elif data.get("is_maintenance"):
self._current_state = STATE_MAINTENANCE
elif data.get("error"):
self._current_state = STATE_ERROR
self._last_error = data["error"]
self._error_count += 1
else:
self._current_state = STATE_READY
# Обновляем метрики
if "metrics" in data:
self._error_count = data["metrics"].get("total_errors", self._error_count)
self._last_update = data.get("last_update")
except Exception as err:
_LOGGER.error("Error handling update: %s", err, exc_info=True)
self._current_state = STATE_ERROR
self._last_error = str(err)
self._error_count += 1
self.async_write_ha_state()