Home Assistant Core 源码深度解析 - 实战经验与最佳实践

"""示例：创建一个简单的传感器集成"""

from __future__ import annotations

import logging
from datetime import timedelta
from typing import Any

import voluptuous as vol

from homeassistant.components.sensor import (
    PLATFORM_SCHEMA,
    SensorEntity,
    SensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    CONF_NAME,
    CONF_SCAN_INTERVAL,
    UnitOfTemperature,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.config_validation import (
    PLATFORM_SCHEMA_BASE,
    string,
)
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.update_coordinator import (
    CoordinatorEntity,
    DataUpdateCoordinator,
    UpdateFailed,
)

_LOGGER = logging.getLogger(__name__)

# 配置模式定义
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
    vol.Required(CONF_NAME): string,
    vol.Optional(CONF_SCAN_INTERVAL, default=timedelta(seconds=30)): vol.All(
        cv.time_period, cv.positive_timedelta
    ),
})

async def async_setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    async_add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """设置传感器平台 - 平台设置的标准入口点
    
    参数:
        hass: Home Assistant核心实例
        config: 平台配置字典
        async_add_entities: 添加实体的回调函数
        discovery_info: 发现信息（可选）
        
    功能说明:
        1. 解析和验证配置参数
        2. 创建数据更新协调器
        3. 初始化传感器实体
        4. 注册实体到平台
        
    最佳实践:
        - 使用异步方法进行非阻塞初始化
        - 通过DataUpdateCoordinator管理数据更新
        - 实现完整的错误处理和日志记录
        - 遵循Home Assistant的配置和实体模式
    """
    name = config[CONF_NAME]
    scan_interval = config[CONF_SCAN_INTERVAL]
    
    # 创建数据更新协调器
    coordinator = MyDataUpdateCoordinator(
        hass,
        _LOGGER,
        name=f"{name} coordinator",
        update_interval=scan_interval,
    )
    
    # 首次数据获取
    await coordinator.async_config_entry_first_refresh()
    
    # 创建传感器实体列表
    entities = [
        MyTemperatureSensor(coordinator, name, "temperature"),
        MyHumiditySensor(coordinator, name, "humidity"),
    ]
    
    # 添加实体到Home Assistant
    async_add_entities(entities)

class MyDataUpdateCoordinator(DataUpdateCoordinator):
    """自定义数据更新协调器 - 管理传感器数据的获取和更新
    
    职责:
        1. 定期从数据源获取最新数据
        2. 处理数据获取异常和重试逻辑
        3. 通知所有相关实体进行状态更新
        4. 优化API调用频率和资源使用
        
    设计特点:
        - 集中化的数据管理减少重复API调用
        - 自动处理更新失败和网络异常
        - 支持多个实体共享同一数据源
        - 内置背压控制和频率限制
    """
    
    def __init__(
        self,
        hass: HomeAssistant,
        logger: logging.Logger,
        *,
        name: str,
        update_interval: timedelta,
    ) -> None:
        """初始化数据更新协调器
        
        参数:
            hass: Home Assistant实例
            logger: 日志记录器
            name: 协调器名称
            update_interval: 更新间隔
        """
        super().__init__(
            hass,
            logger,
            name=name,
            update_interval=update_interval,
        )
        
        # 初始化数据源连接参数
        self.api_client = None  # 在实际应用中初始化API客户端
        
    async def _async_update_data(self) -> dict[str, Any]:
        """异步获取数据 - 协调器的核心方法
        
        返回值:
            包含所有传感器数据的字典
            
        异常处理:
            - 网络连接异常
            - API响应异常  
            - 数据格式异常
            
        返回格式:
        {
            "temperature": 23.5,
            "humidity": 65.2,
            "last_update": "2024-01-01T00:00:00Z"
        }
        """
        try:
            # 模拟数据获取（实际应用中调用真实API）
            data = await self._fetch_sensor_data()
            
            if not data:
                raise UpdateFailed("No data received from sensor")
            
            # 数据验证和处理
            processed_data = self._process_raw_data(data)
            
            self.logger.debug("Successfully updated sensor data: %s", processed_data)
            return processed_data
            
        except Exception as err:
            # 统一异常处理
            self.logger.error("Error fetching sensor data: %s", err)
            raise UpdateFailed(f"Error communicating with sensor: {err}") from err
    
    async def _fetch_sensor_data(self) -> dict[str, Any]:
        """获取原始传感器数据 - 与外部API交互的具体实现
        
        返回值:
            从传感器API获取的原始数据
            
        实现要点:
            - 使用aiohttp进行异步HTTP请求
            - 实现请求重试和超时控制
            - 处理认证和API密钥
            - 解析不同格式的响应数据
        """
        # 示例：异步HTTP请求
        # async with aiohttp.ClientSession() as session:
        #     async with session.get(
        #         "https://api.example.com/sensors/data",
        #         headers={"Authorization": f"Bearer {self.api_key}"},
        #         timeout=aiohttp.ClientTimeout(total=10)
        #     ) as response:
        #         response.raise_for_status()
        #         return await response.json()
        
        # 模拟数据
        import random
        import time
        
        await asyncio.sleep(0.1)  # 模拟网络延迟
        return {
            "temperature": round(random.uniform(15.0, 30.0), 1),
            "humidity": round(random.uniform(30.0, 80.0), 1),
            "timestamp": time.time(),
        }
    
    def _process_raw_data(self, raw_data: dict[str, Any]) -> dict[str, Any]:
        """处理原始数据 - 数据清理和转换
        
        参数:
            raw_data: 从API获取的原始数据
            
        返回值:
            处理后的标准化数据
            
        处理内容:
            - 单位转换（华氏度转摄氏度等）
            - 数值范围验证
            - 数据格式标准化
            - 异常值过滤
        """
        temperature = raw_data.get("temperature")
        humidity = raw_data.get("humidity")
        
        # 数据验证
        if temperature is None or not (-40 <= temperature <= 80):
            raise UpdateFailed(f"Invalid temperature value: {temperature}")
        
        if humidity is None or not (0 <= humidity <= 100):
            raise UpdateFailed(f"Invalid humidity value: {humidity}")
        
        return {
            "temperature": temperature,
            "humidity": humidity,
            "last_update": dt_util.utcnow().isoformat(),
        }

class MyTemperatureSensor(CoordinatorEntity, SensorEntity):
    """温度传感器实体 - 展示传感器实体的标准实现
    
    继承关系:
        - CoordinatorEntity: 提供数据协调器集成
        - SensorEntity: 提供传感器特定功能
        
    核心特性:
        - 自动数据更新通过协调器
        - 标准化的实体属性和方法
        - 设备信息和诊断支持
        - 状态恢复和持久化
    """
    
    def __init__(
        self,
        coordinator: MyDataUpdateCoordinator,
        name: str,
        sensor_type: str,
    ) -> None:
        """初始化温度传感器
        
        参数:
            coordinator: 数据更新协调器
            name: 传感器名称
            sensor_type: 传感器类型标识
        """
        super().__init__(coordinator)
        
        # 实体基本属性
        self._attr_name = f"{name} Temperature"
        self._attr_unique_id = f"{name}_temperature"
        
        # 传感器特定属性
        self._attr_native_unit_of_measurement = UnitOfTemperature.CELSIUS
        self._attr_device_class = SensorDeviceClass.TEMPERATURE
        self._attr_state_class = SensorStateClass.MEASUREMENT
        
        # 设备信息
        self._attr_device_info = DeviceInfo(
            identifiers={(DOMAIN, name)},
            name=name,
            manufacturer="Custom Sensors",
            model="Temperature Monitor",
            sw_version="1.0",
        )
    
    @property
    def native_value(self) -> float | None:
        """返回传感器的原始数值
        
        返回值:
            温度数值，数据不可用时返回None
            
        注意事项:
            - 返回原始数值，单位转换由Home Assistant处理
            - None值表示传感器不可用或数据无效
            - 避免在此方法中进行复杂计算
        """
        if self.coordinator.data:
            return self.coordinator.data.get("temperature")
        return None
    
    @property
    def extra_state_attributes(self) -> dict[str, Any] | None:
        """返回额外的状态属性
        
        返回值:
            包含额外信息的字典
            
        常用属性:
            - 数据源信息
            - 更新时间戳
            - 传感器特定参数
            - 诊断和调试信息
        """
        if not self.coordinator.data:
            return None
            
        return {
            "last_update": self.coordinator.data.get("last_update"),
            "data_source": "Custom API",
            "sensor_type": "temperature",
        }
    
    @property
    def available(self) -> bool:
        """返回实体是否可用
        
        返回值:
            实体可用状态
            
        可用性判断:
            - 协调器最后更新是否成功
            - 数据是否在有效范围内
            - 外部服务是否正常
        """
        return (
            self.coordinator.last_update_success 
            and self.coordinator.data is not None
            and "temperature" in self.coordinator.data
        )

# 配置入口点设置
async def async_setup_entry(
    hass: HomeAssistant, 
    entry: ConfigEntry
) -> bool:
    """配置条目设置 - 现代Home Assistant集成的标准入口
    
    参数:
        hass: Home Assistant实例
        entry: 配置条目对象
        
    返回值:
        设置是否成功
        
    功能:
        - 解析配置条目数据
        - 初始化集成组件
        - 设置平台（传感器、开关等）
        - 注册服务和事件监听器
    """
    # 存储配置数据
    hass.data.setdefault(DOMAIN, {})
    hass.data[DOMAIN][entry.entry_id] = entry.data
    
    # 设置平台
    await hass.config_entries.async_forward_entry_setups(
        entry, ["sensor"]
    )
    
    return True

async def async_unload_entry(
    hass: HomeAssistant, 
    entry: ConfigEntry
) -> bool:
    """卸载配置条目 - 清理集成资源
    
    参数:
        hass: Home Assistant实例  
        entry: 配置条目对象
        
    返回值:
        卸载是否成功
        
    清理内容:
        - 停止数据更新协调器
        - 移除实体和平台
        - 清理存储的数据
        - 取消事件监听器
    """
    # 卸载平台
    unload_ok = await hass.config_entries.async_unload_platforms(
        entry, ["sensor"]
    )
    
    if unload_ok:
        # 清理数据
        hass.data[DOMAIN].pop(entry.entry_id)
    
    return unload_ok

"""示例：创建带有响应数据的服务"""

import voluptuous as vol
from homeassistant.core import HomeAssistant, ServiceCall, ServiceResponse
from homeassistant.helpers import config_validation as cv

# 服务参数验证模式
SERVICE_ANALYZE_DATA_SCHEMA = vol.Schema({
    vol.Required("entity_id"): cv.entity_ids,
    vol.Optional("period", default=24): cv.positive_int,
    vol.Optional("analysis_type", default="basic"): vol.In(
        ["basic", "detailed", "statistical"]
    ),
})

async def async_setup_services(hass: HomeAssistant) -> None:
    """注册自定义服务 - 服务注册的最佳实践示例
    
    服务设计原则:
        1. 明确的参数验证和文档
        2. 丰富的响应数据
        3. 完整的错误处理
        4. 异步执行支持
        5. 权限和安全检查
    """
    
    async def async_analyze_sensor_data(call: ServiceCall) -> ServiceResponse:
        """分析传感器数据服务 - 展示复杂服务实现
        
        参数:
            call: 服务调用对象，包含参数和上下文
            
        返回值:
            服务响应数据字典
            
        功能特性:
            - 多实体数据分析
            - 时间序列处理  
            - 统计计算和趋势分析
            - 结构化响应数据
            
        响应格式:
        {
            "analysis_result": {
                "entity_id": "sensor.temperature",
                "period_hours": 24,
                "statistics": {
                    "min": 18.5,
                    "max": 25.3,
                    "average": 22.1,
                    "trend": "stable"
                },
                "data_points": 144,
                "timestamp": "2024-01-01T00:00:00Z"
            }
        }
        """
        entity_ids = call.data["entity_id"]
        period_hours = call.data["period"]
        analysis_type = call.data["analysis_type"]
        
        _LOGGER.info(
            "Analyzing sensor data for entities %s over %d hours",
            entity_ids, period_hours
        )
        
        results = {}
        
        for entity_id in entity_ids:
            try:
                # 获取历史数据
                history_data = await _get_entity_history(
                    hass, entity_id, period_hours
                )
                
                if not history_data:
                    _LOGGER.warning("No history data found for %s", entity_id)
                    continue
                
                # 执行数据分析
                analysis_result = await _perform_data_analysis(
                    history_data, analysis_type
                )
                
                results[entity_id] = {
                    "period_hours": period_hours,
                    "analysis_type": analysis_type,
                    **analysis_result,
                    "timestamp": dt_util.utcnow().isoformat(),
                }
                
            except Exception as err:
                _LOGGER.error("Error analyzing data for %s: %s", entity_id, err)
                results[entity_id] = {
                    "error": str(err),
                    "timestamp": dt_util.utcnow().isoformat(),
                }
        
        return {"analysis_results": results}
    
    # 注册服务
    hass.services.async_register(
        DOMAIN,
        "analyze_data", 
        async_analyze_sensor_data,
        schema=SERVICE_ANALYZE_DATA_SCHEMA,
        supports_response=SupportsResponse.ONLY,  # 仅支持响应模式
    )

async def _get_entity_history(
    hass: HomeAssistant, 
    entity_id: str, 
    period_hours: int
) -> list[State]:
    """获取实体历史数据 - 历史数据查询的高效实现
    
    参数:
        hass: Home Assistant实例
        entity_id: 目标实体ID  
        period_hours: 历史数据时间范围（小时）
        
    返回值:
        状态历史列表
        
    优化策略:
        - 限制查询时间范围避免性能问题
        - 使用recorder组件的高效查询API
        - 实现数据采样减少内存使用
        - 缓存常用查询结果
    """
    from homeassistant.components import recorder
    from homeassistant.components.recorder import history
    
    # 计算查询时间范围
    end_time = dt_util.utcnow()
    start_time = end_time - timedelta(hours=period_hours)
    
    # 查询历史数据
    with recorder.session_scope(hass=hass) as session:
        history_data = history.get_states(
            hass,
            start_time,
            end_time,
            [entity_id],
            session=session,
        )
    
    return history_data.get(entity_id, [])

async def _perform_data_analysis(
    history_data: list[State], 
    analysis_type: str
) -> dict[str, Any]:
    """执行数据分析 - 统计计算和趋势分析
    
    参数:
        history_data: 历史状态数据列表
        analysis_type: 分析类型
        
    返回值:
        分析结果字典
        
    支持的分析类型:
        - basic: 基本统计（最大值、最小值、平均值）
        - detailed: 详细统计（标准差、分位数、异常值）
        - statistical: 高级统计（趋势分析、相关性、预测）
    """
    if not history_data:
        return {"error": "No data available for analysis"}
    
    # 提取数值数据
    numeric_values = []
    timestamps = []
    
    for state in history_data:
        try:
            value = float(state.state)
            numeric_values.append(value)
            timestamps.append(state.last_changed)
        except (ValueError, TypeError):
            continue  # 跳过非数值状态
    
    if not numeric_values:
        return {"error": "No numeric data found"}
    
    # 基本统计计算
    result = {
        "data_points": len(numeric_values),
        "min_value": min(numeric_values),
        "max_value": max(numeric_values),
        "average": sum(numeric_values) / len(numeric_values),
        "first_timestamp": timestamps[0].isoformat() if timestamps else None,
        "last_timestamp": timestamps[-1].isoformat() if timestamps else None,
    }
    
    if analysis_type in ["detailed", "statistical"]:
        # 详细统计
        import statistics
        
        result.update({
            "median": statistics.median(numeric_values),
            "standard_deviation": statistics.stdev(numeric_values) if len(numeric_values) > 1 else 0,
            "variance": statistics.variance(numeric_values) if len(numeric_values) > 1 else 0,
        })
    
    if analysis_type == "statistical":
        # 高级统计和趋势分析
        trend = _calculate_trend(numeric_values, timestamps)
        result.update({
            "trend": trend,
            "volatility": _calculate_volatility(numeric_values),
            "anomalies": _detect_anomalies(numeric_values),
        })
    
    return result

"""异步编程最佳实践示例"""

import asyncio
from typing import Any
import aiohttp
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator

class OptimizedDataCoordinator(DataUpdateCoordinator):
    """性能优化的数据协调器实现"""
    
    def __init__(self, hass: HomeAssistant) -> None:
        super().__init__(
            hass,
            _LOGGER,
            name="optimized_coordinator",
            update_interval=timedelta(minutes=5),
        )
        
        # 连接池配置 - 复用HTTP连接提升性能
        self._session = aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(
                limit=10,              # 总连接池大小
                limit_per_host=5,      # 每个主机的最大连接数
                ttl_dns_cache=300,     # DNS缓存时间
                use_dns_cache=True,    # 启用DNS缓存
                keepalive_timeout=60,  # Keep-alive超时
                enable_cleanup_closed=True,  # 自动清理关闭的连接
            ),
            timeout=aiohttp.ClientTimeout(total=30),  # 总超时时间
        )
    
    async def _async_update_data(self) -> dict[str, Any]:
        """高效的数据更新实现
        
        性能优化策略:
            1. 并发API调用 - 同时请求多个数据源
            2. 连接池复用 - 避免重复建立TCP连接
            3. 合理的超时设置 - 平衡响应时间和可靠性
            4. 数据缓存 - 减少不必要的网络请求
            5. 异常处理优化 - 快速失败和恢复
        """
        try:
            # 并发执行多个API调用
            tasks = [
                self._fetch_sensor_data("temperature"),
                self._fetch_sensor_data("humidity"), 
                self._fetch_sensor_data("pressure"),
                self._fetch_weather_data(),
            ]
            
            # 等待所有任务完成，使用as_completed处理部分失败
            results = {}
            async for task in asyncio.as_completed(tasks):
                try:
                    result = await task
                    results.update(result)
                except Exception as err:
                    _LOGGER.warning("Partial data fetch failed: %s", err)
                    # 继续处理其他成功的任务
            
            if not results:
                raise UpdateFailed("All data sources failed")
            
            return results
            
        except Exception as err:
            _LOGGER.error("Data update failed: %s", err)
            raise UpdateFailed(f"Update error: {err}") from err
    
    async def _fetch_sensor_data(self, sensor_type: str) -> dict[str, Any]:
        """单个传感器数据获取 - 带重试机制的API调用"""
        url = f"https://api.example.com/sensors/{sensor_type}"
        
        # 实现指数退避重试
        for attempt in range(3):
            try:
                async with self._session.get(
                    url,
                    headers={"Authorization": f"Bearer {self.api_token}"}
                ) as response:
                    response.raise_for_status()
                    data = await response.json()
                    return {sensor_type: data["value"]}
                    
            except asyncio.TimeoutError:
                wait_time = 2 ** attempt  # 指数退避：2, 4, 8秒
                _LOGGER.warning(
                    "Timeout fetching %s data, retrying in %ds (attempt %d/3)",
                    sensor_type, wait_time, attempt + 1
                )
                if attempt < 2:  # 不在最后一次尝试时等待
                    await asyncio.sleep(wait_time)
            except Exception as err:
                _LOGGER.error("Error fetching %s data: %s", sensor_type, err)
                if attempt == 2:  # 最后一次尝试
                    raise
                await asyncio.sleep(1)
        
        raise UpdateFailed(f"Failed to fetch {sensor_type} after 3 attempts")
    
    async def async_shutdown(self) -> None:
        """协调器关闭时的清理工作"""
        await self._session.close()
        await super().async_shutdown()

# 高效的批量操作示例
async def async_batch_update_entities(
    hass: HomeAssistant, 
    entity_updates: dict[str, dict[str, Any]]
) -> None:
    """批量更新实体状态 - 减少事件总线负载
    
    参数:
        hass: Home Assistant实例
        entity_updates: 实体更新数据字典
        
    优化策略:
        1. 批量状态设置减少事件数量
        2. 使用单个上下文避免重复创建
        3. 预验证数据减少运行时错误
        4. 异步执行避免阻塞主循环
    """
    if not entity_updates:
        return
    
    # 创建统一的更新上下文
    context = Context(id=ulid_now())
    
    # 预验证所有更新数据
    validated_updates = {}
    for entity_id, update_data in entity_updates.items():
        try:
            # 验证实体ID格式
            if not valid_entity_id(entity_id):
                _LOGGER.warning("Invalid entity ID: %s", entity_id)
                continue
            
            # 验证状态数据
            state = update_data.get("state")
            if state is None:
                _LOGGER.warning("No state provided for %s", entity_id)
                continue
            
            # 验证状态值长度
            state_str = str(state)
            if len(state_str) > MAX_LENGTH_STATE_STATE:
                _LOGGER.warning("State too long for %s", entity_id)
                continue
            
            validated_updates[entity_id] = {
                "state": state_str,
                "attributes": update_data.get("attributes", {}),
                "force_update": update_data.get("force_update", False),
            }
            
        except Exception as err:
            _LOGGER.error("Validation error for %s: %s", entity_id, err)
    
    # 批量执行状态更新
    update_tasks = []
    for entity_id, update_data in validated_updates.items():
        task = hass.async_create_task(
            hass.states.async_set(
                entity_id,
                update_data["state"],
                update_data["attributes"],
                update_data["force_update"],
                context,
            )
        )
        update_tasks.append(task)
    
    # 等待所有更新完成
    if update_tasks:
        try:
            await asyncio.gather(*update_tasks, return_exceptions=True)
            _LOGGER.debug("Batch updated %d entities", len(update_tasks))
        except Exception as err:
            _LOGGER.error("Batch update failed: %s", err)

"""内存和资源管理最佳实践"""

from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from weakref import WeakSet, WeakValueDictionary
import gc

@dataclass(slots=True)
class OptimizedEntity:
    """内存优化的实体类实现
    
    优化技术:
        1. __slots__ 减少内存占用
        2. 弱引用避免循环引用
        3. 延迟计算减少初始化开销
        4. 对象池减少GC压力
    """
    
    # 基本属性
    entity_id: str
    name: str | None = None
    
    # 使用弱引用避免循环引用
    _listeners: WeakSet = field(default_factory=WeakSet, init=False)
    _state_cache: dict[str, Any] = field(default_factory=dict, init=False)
    
    # 延迟计算的属性
    _computed_attributes: dict[str, Any] | None = field(default=None, init=False)
    
    def add_listener(self, listener: Any) -> None:
        """添加监听器 - 使用弱引用避免内存泄漏"""
        self._listeners.add(listener)
    
    def remove_listener(self, listener: Any) -> None:
        """移除监听器"""
        self._listeners.discard(listener)  # discard不会抛出异常
    
    @property  
    def computed_attributes(self) -> dict[str, Any]:
        """延迟计算的属性 - 仅在需要时计算"""
        if self._computed_attributes is None:
            self._computed_attributes = self._calculate_attributes()
        return self._computed_attributes
    
    def _calculate_attributes(self) -> dict[str, Any]:
        """计算属性的具体实现"""
        # 执行耗时的属性计算
        return {"calculated_at": time.time()}
    
    def clear_cache(self) -> None:
        """清理缓存数据"""
        self._state_cache.clear()
        self._computed_attributes = None

class EntityPool:
    """实体对象池 - 减少对象创建和销毁开销
    
    设计特点:
        1. 对象复用减少内存分配
        2. 类型安全的对象管理
        3. 自动清理机制
        4. 统计和监控支持
    """
    
    def __init__(self, max_size: int = 1000) -> None:
        self.max_size = max_size
        self._pool: list[OptimizedEntity] = []
        self._active_entities: WeakValueDictionary[str, OptimizedEntity] = WeakValueDictionary()
        self._stats = {
            "created": 0,
            "reused": 0,
            "disposed": 0,
        }
    
    def get_entity(self, entity_id: str, name: str | None = None) -> OptimizedEntity:
        """获取或创建实体对象"""
        # 检查是否已存在活跃实体
        if existing := self._active_entities.get(entity_id):
            return existing
        
        # 尝试从对象池获取
        if self._pool:
            entity = self._pool.pop()
            entity.entity_id = entity_id
            entity.name = name
            entity.clear_cache()  # 清理之前的状态
            self._stats["reused"] += 1
        else:
            # 创建新实体
            entity = OptimizedEntity(entity_id=entity_id, name=name)
            self._stats["created"] += 1
        
        # 注册为活跃实体
        self._active_entities[entity_id] = entity
        return entity
    
    def return_entity(self, entity: OptimizedEntity) -> None:
        """归还实体到对象池"""
        if len(self._pool) < self.max_size:
            # 清理实体状态
            entity.clear_cache()
            entity._listeners.clear()
            
            # 归还到池中
            self._pool.append(entity)
            self._stats["disposed"] += 1
        
        # 从活跃实体中移除
        self._active_entities.pop(entity.entity_id, None)
    
    def get_stats(self) -> dict[str, int]:
        """获取对象池统计信息"""
        return {
            **self._stats,
            "pool_size": len(self._pool),
            "active_entities": len(self._active_entities),
        }
    
    def cleanup(self) -> None:
        """清理对象池"""
        self._pool.clear()
        self._active_entities.clear()
        
        # 强制垃圾回收
        gc.collect()

# 内存监控工具
class MemoryMonitor:
    """内存使用监控器 - 帮助识别内存泄漏和优化点"""
    
    def __init__(self, hass: HomeAssistant) -> None:
        self.hass = hass
        self._last_snapshot = self._take_snapshot()
        
        # 定期监控内存使用
        async def _periodic_monitor():
            while True:
                await asyncio.sleep(300)  # 每5分钟检查一次
                await self._check_memory_usage()
        
        hass.async_create_background_task(
            _periodic_monitor(), 
            "memory_monitor"
        )
    
    def _take_snapshot(self) -> dict[str, Any]:
        """获取内存使用快照"""
        import psutil
        import tracemalloc
        
        process = psutil.Process()
        memory_info = process.memory_info()
        
        snapshot = {
            "timestamp": time.time(),
            "rss": memory_info.rss,  # 物理内存
            "vms": memory_info.vms,  # 虚拟内存
            "entities": len(self.hass.states.async_entity_ids()),
            "events": len(self.hass.bus.async_listeners()),
            "services": sum(len(services) for services in self.hass.services.async_services().values()),
        }
        
        # 如果启用了tracemalloc，添加详细信息
        if tracemalloc.is_tracing():
            current, peak = tracemalloc.get_traced_memory()
            snapshot.update({
                "traced_current": current,
                "traced_peak": peak,
            })
        
        return snapshot
    
    async def _check_memory_usage(self) -> None:
        """检查内存使用情况"""
        current_snapshot = self._take_snapshot()
        
        # 计算内存增长
        rss_growth = current_snapshot["rss"] - self._last_snapshot["rss"]
        entity_growth = current_snapshot["entities"] - self._last_snapshot["entities"]
        
        # 记录内存使用情况
        _LOGGER.debug(
            "Memory usage: RSS=%d MB (+%d MB), Entities=%d (+%d), Events=%d, Services=%d",
            current_snapshot["rss"] // 1024 // 1024,
            rss_growth // 1024 // 1024,
            current_snapshot["entities"],
            entity_growth,
            current_snapshot["events"],
            current_snapshot["services"],
        )
        
        # 检查是否存在内存泄漏
        if rss_growth > 50 * 1024 * 1024:  # 增长超过50MB
            _LOGGER.warning(
                "Significant memory growth detected: %d MB in 5 minutes",
                rss_growth // 1024 // 1024
            )
            
            # 触发垃圾回收
            collected = gc.collect()
            _LOGGER.info("Garbage collection freed %d objects", collected)
        
        self._last_snapshot = current_snapshot

"""错误处理和系统恢复最佳实践"""

import functools
from typing import Callable, TypeVar, ParamSpec
from homeassistant.exceptions import HomeAssistantError

P = ParamSpec('P')
T = TypeVar('T')

def robust_operation(
    retries: int = 3,
    delay: float = 1.0,
    exponential_backoff: bool = True,
    exceptions: tuple[type[Exception], ...] = (Exception,)
):
    """健壮操作装饰器 - 自动重试和错误恢复
    
    参数:
        retries: 最大重试次数
        delay: 基础延迟时间（秒）
        exponential_backoff: 是否使用指数退避
        exceptions: 需要重试的异常类型
        
    特性:
        - 智能重试策略
        - 异常分类处理
        - 详细的错误日志
        - 性能监控集成
    """
    def decorator(func: Callable[P, T]) -> Callable[P, T]:
        @functools.wraps(func)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
            last_exception = None
            
            for attempt in range(retries + 1):
                try:
                    # 执行原始函数
                    result = await func(*args, **kwargs)
                    
                    # 记录成功恢复
                    if attempt > 0:
                        _LOGGER.info(
                            "%s succeeded after %d retries",
                            func.__name__, attempt
                        )
                    
                    return result
                    
                except exceptions as err:
                    last_exception = err
                    
                    # 最后一次尝试，不再重试
                    if attempt == retries:
                        _LOGGER.error(
                            "%s failed after %d retries: %s",
                            func.__name__, retries, err
                        )
                        break
                    
                    # 计算延迟时间
                    if exponential_backoff:
                        sleep_time = delay * (2 ** attempt)
                    else:
                        sleep_time = delay
                    
                    _LOGGER.warning(
                        "%s failed (attempt %d/%d): %s, retrying in %.1fs",
                        func.__name__, attempt + 1, retries + 1, err, sleep_time
                    )
                    
                    await asyncio.sleep(sleep_time)
                
                except Exception as err:
                    # 不可重试的异常
                    _LOGGER.error(
                        "%s failed with non-retryable exception: %s",
                        func.__name__, err
                    )
                    raise
            
            # 所有重试都失败
            raise last_exception
        
        return wrapper
    return decorator

class CircuitBreaker:
    """断路器模式实现 - 防止级联故障
    
    状态转换:
        - CLOSED: 正常工作状态，允许请求通过
        - OPEN: 故障状态，直接拒绝请求  
        - HALF_OPEN: 测试状态，允许少量请求测试恢复
        
    应用场景:
        - 外部API调用保护
        - 数据库连接故障恢复
        - 网络服务降级处理
    """
    
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 60,
        expected_exception: type[Exception] = Exception,
    ) -> None:
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.expected_exception = expected_exception
        
        # 状态管理
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
    
    async def call(self, func: Callable, *args, **kwargs):
        """通过断路器调用函数"""
        if self.state == "OPEN":
            # 检查是否可以进入半开状态
            if (
                self.last_failure_time and
                time.time() - self.last_failure_time > self.recovery_timeout
            ):
                self.state = "HALF_OPEN"
                _LOGGER.info("Circuit breaker entering HALF_OPEN state")
            else:
                raise HomeAssistantError("Circuit breaker is OPEN")
        
        try:
            # 执行函数调用
            result = await func(*args, **kwargs)
            
            # 调用成功，重置故障计数
            if self.state in ("HALF_OPEN", "CLOSED"):
                self.failure_count = 0
                self.state = "CLOSED"
                if self.state == "HALF_OPEN":
                    _LOGGER.info("Circuit breaker recovered, state: CLOSED")
            
            return result
            
        except self.expected_exception as err:
            # 记录失败
            self.failure_count += 1
            self.last_failure_time = time.time()
            
            # 检查是否需要打开断路器
            if self.failure_count >= self.failure_threshold:
                self.state = "OPEN"
                _LOGGER.warning(
                    "Circuit breaker opened after %d failures",
                    self.failure_count
                )
            
            raise err

# 健壮的集成实现示例
class ResilientIntegration:
    """弹性集成实现 - 综合错误处理和恢复策略"""
    
    def __init__(self, hass: HomeAssistant, config: dict) -> None:
        self.hass = hass
        self.config = config
        
        # 断路器保护外部API调用
        self.api_circuit_breaker = CircuitBreaker(
            failure_threshold=3,
            recovery_timeout=30,
            expected_exception=(aiohttp.ClientError, asyncio.TimeoutError),
        )
        
        # 健康状态监控
        self.health_status = {
            "api_available": True,
            "last_successful_update": None,
            "consecutive_failures": 0,
            "degraded_mode": False,
        }
        
        # 降级模式配置
        self.degraded_config = {
            "update_interval_multiplier": 3,  # 降级时增加更新间隔
            "disable_non_critical_features": True,
            "use_cached_data": True,
        }
    
    @robust_operation(retries=2, delay=5.0)
    async def async_update_data(self) -> dict[str, Any]:
        """弹性数据更新实现"""
        try:
            # 通过断路器调用API
            data = await self.api_circuit_breaker.call(
                self._fetch_api_data
            )
            
            # 更新健康状态
            self.health_status.update({
                "api_available": True,
                "last_successful_update": dt_util.utcnow(),
                "consecutive_failures": 0,
                "degraded_mode": False,
            })
            
            return data
            
        except Exception as err:
            # 更新失败计数
            self.health_status["consecutive_failures"] += 1
            
            # 检查是否需要进入降级模式
            if self.health_status["consecutive_failures"] >= 3:
                return await self._handle_degraded_mode(err)
            
            raise err
    
    async def _fetch_api_data(self) -> dict[str, Any]:
        """实际的API数据获取"""
        async with aiohttp.ClientSession() as session:
            async with session.get(
                self.config["api_url"],
                timeout=aiohttp.ClientTimeout(total=10)
            ) as response:
                response.raise_for_status()
                return await response.json()
    
    async def _handle_degraded_mode(self, error: Exception) -> dict[str, Any]:
        """处理降级模式 - 系统故障时的降级策略
        
        降级策略:
            1. 使用缓存数据
            2. 返回默认/安全值
            3. 禁用非关键功能
            4. 增加更新间隔
        """
        _LOGGER.warning(
            "Entering degraded mode due to repeated failures: %s", error
        )
        
        self.health_status["degraded_mode"] = True
        
        # 尝试使用缓存数据
        cached_data = await self._get_cached_data()
        if cached_data:
            _LOGGER.info("Using cached data in degraded mode")
            return cached_data
        
        # 返回安全的默认值
        _LOGGER.info("Using default values in degraded mode")
        return self._get_safe_defaults()
    
    async def _get_cached_data(self) -> dict[str, Any] | None:
        """获取缓存数据"""
        # 实现数据缓存逻辑
        cache_key = f"{DOMAIN}_data_cache"
        return self.hass.data.get(cache_key)
    
    def _get_safe_defaults(self) -> dict[str, Any]:
        """获取安全的默认值"""
        return {
            "temperature": 20.0,
            "humidity": 50.0,
            "status": "unknown",
            "last_update": dt_util.utcnow().isoformat(),
            "degraded_mode": True,
        }
    
    async def async_health_check(self) -> dict[str, Any]:
        """健康检查接口 - 提供系统状态信息"""
        return {
            "integration": DOMAIN,
            "status": "healthy" if not self.health_status["degraded_mode"] else "degraded",
            "api_available": self.health_status["api_available"],
            "last_successful_update": self.health_status["last_successful_update"],
            "consecutive_failures": self.health_status["consecutive_failures"],
            "circuit_breaker_state": self.api_circuit_breaker.state,
        }

"""数据验证和安全处理最佳实践"""

import re
from typing import Any
import voluptuous as vol
from homeassistant.helpers import config_validation as cv
from homeassistant.exceptions import HomeAssistantError

# 安全的数据验证模式
SECURE_SENSOR_SCHEMA = vol.Schema({
    vol.Required("name"): vol.All(cv.string, vol.Length(min=1, max=100)),
    vol.Required("value"): vol.All(
        vol.Coerce(float),
        vol.Range(min=-1000, max=1000)  # 合理的数值范围
    ),
    vol.Optional("unit"): vol.All(cv.string, vol.Length(max=20)),
    vol.Optional("attributes", default={}): vol.Schema({
        vol.Extra: vol.All(
            vol.Any(str, int, float, bool, None),
            vol.Length(max=255) if isinstance(val, str) else val  # 字符串长度限制
        )
    }),
    vol.Optional("device_class"): vol.In([
        "temperature", "humidity", "pressure", "battery", "energy"
    ]),
})

class DataSanitizer:
    """数据清理和安全处理工具类
    
    功能:
        1. 输入数据验证和清理
        2. XSS和注入攻击防护  
        3. 数据格式标准化
        4. 恶意内容过滤
    """
    
    # 危险字符模式
    DANGEROUS_PATTERNS = [
        re.compile(r'<script[^>]*>.*?</script>', re.IGNORECASE | re.DOTALL),
        re.compile(r'javascript:', re.IGNORECASE),
        re.compile(r'on\w+\s*=', re.IGNORECASE),
        re.compile(r'<iframe[^>]*>.*?</iframe>', re.IGNORECASE | re.DOTALL),
    ]
    
    # 允许的HTML标签（如果需要支持富文本）
    ALLOWED_HTML_TAGS = {'b', 'i', 'u', 'em', 'strong', 'br', 'p'}
    
    @classmethod
    def sanitize_string(cls, value: str, max_length: int = 255) -> str:
        """清理字符串输入
        
        参数:
            value: 输入字符串
            max_length: 最大长度限制
            
        返回值:
            清理后的安全字符串
            
        清理步骤:
            1. 长度截断
            2. 危险模式过滤
            3. 特殊字符转义
            4. Unicode规范化
        """
        if not isinstance(value, str):
            value = str(value)
        
        # 长度限制
        if len(value) > max_length:
            value = value[:max_length]
        
        # 移除危险模式
        for pattern in cls.DANGEROUS_PATTERNS:
            value = pattern.sub('', value)
        
        # HTML实体编码
        import html
        value = html.escape(value)
        
        # Unicode规范化
        import unicodedata
        value = unicodedata.normalize('NFKC', value)
        
        # 移除控制字符
        value = ''.join(char for char in value if not unicodedata.category(char).startswith('C'))
        
        return value.strip()
    
    @classmethod
    def sanitize_dict(cls, data: dict[str, Any], max_depth: int = 10) -> dict[str, Any]:
        """递归清理字典数据
        
        参数:
            data: 输入字典
            max_depth: 最大递归深度
            
        返回值:
            清理后的字典
            
        安全检查:
            - 递归深度限制
            - 键名安全检查
            - 值类型验证
            - 大小限制
        """
        if max_depth <= 0:
            raise ValueError("Dictionary nesting too deep")
        
        if not isinstance(data, dict):
            raise TypeError(f"Expected dict, got {type(data)}")
        
        if len(data) > 100:
            raise ValueError("Dictionary too large")
        
        sanitized = {}
        for key, value in data.items():
            # 清理键名
            if not isinstance(key, str):
                key = str(key)
            
            clean_key = cls.sanitize_string(key, max_length=50)
            if not clean_key or len(clean_key.strip()) == 0:
                continue  # 跳过无效键名
            
            # 清理值
            if isinstance(value, str):
                clean_value = cls.sanitize_string(value)
            elif isinstance(value, dict):
                clean_value = cls.sanitize_dict(value, max_depth - 1)
            elif isinstance(value, list):
                clean_value = cls.sanitize_list(value, max_depth - 1)
            elif isinstance(value, (int, float, bool)) or value is None:
                clean_value = value
            else:
                # 不支持的类型转为字符串
                clean_value = cls.sanitize_string(str(value))
            
            sanitized[clean_key] = clean_value
        
        return sanitized
    
    @classmethod 
    def sanitize_list(cls, data: list[Any], max_depth: int = 10) -> list[Any]:
        """清理列表数据"""
        if max_depth <= 0:
            raise ValueError("List nesting too deep")
        
        if len(data) > 1000:
            raise ValueError("List too large")
        
        sanitized = []
        for item in data:
            if isinstance(item, str):
                clean_item = cls.sanitize_string(item)
            elif isinstance(item, dict):
                clean_item = cls.sanitize_dict(item, max_depth - 1)
            elif isinstance(item, list):
                clean_item = cls.sanitize_list(item, max_depth - 1)
            elif isinstance(item, (int, float, bool)) or item is None:
                clean_item = item
            else:
                clean_item = cls.sanitize_string(str(item))
            
            sanitized.append(clean_item)
        
        return sanitized

class SecureEntityValidator:
    """安全的实体验证器 - 确保实体数据的安全性和一致性"""
    
    @staticmethod
    def validate_entity_id(entity_id: str) -> str:
        """验证实体ID的安全性和格式
        
        验证项目:
            1. 格式正确性（domain.object_id）
            2. 长度限制
            3. 字符安全性
            4. 域名白名单
        """
        if not isinstance(entity_id, str):
            raise ValueError("Entity ID must be a string")
        
        # 长度检查
        if len(entity_id) > MAX_LENGTH_STATE_ENTITY_ID:
            raise ValueError(f"Entity ID too long: {len(entity_id)} > {MAX_LENGTH_STATE_ENTITY_ID}")
        
        # 格式验证
        if not VALID_ENTITY_ID.match(entity_id):
            raise ValueError(f"Invalid entity ID format: {entity_id}")
        
        # 提取并验证域名
        domain, object_id = entity_id.split('.', 1)
        
        # 域名白名单检查（可选）
        ALLOWED_DOMAINS = {
            'sensor', 'binary_sensor', 'switch', 'light', 'cover',
            'climate', 'fan', 'lock', 'alarm_control_panel', 'camera'
        }
        
        if domain not in ALLOWED_DOMAINS:
            _LOGGER.warning("Potentially unsafe domain in entity ID: %s", domain)
        
        return entity_id.lower()
    
    @staticmethod
    def validate_state_value(state: Any) -> str:
        """验证状态值的安全性"""
        if state is None:
            return "unknown"
        
        # 转换为字符串
        state_str = str(state)
        
        # 长度检查
        if len(state_str) > MAX_LENGTH_STATE_STATE:
            raise ValueError(f"State value too long: {len(state_str)} > {MAX_LENGTH_STATE_STATE}")
        
        # 安全性清理
        clean_state = DataSanitizer.sanitize_string(state_str, MAX_LENGTH_STATE_STATE)
        
        # 禁止某些特殊状态值
        FORBIDDEN_STATES = {'<script>', 'javascript:', 'data:', 'vbscript:'}
        if any(forbidden in clean_state.lower() for forbidden in FORBIDDEN_STATES):
            raise ValueError(f"Forbidden content in state value: {state}")
        
        return clean_state
    
    @staticmethod
    def validate_attributes(attributes: dict[str, Any]) -> dict[str, Any]:
        """验证实体属性的安全性"""
        if not isinstance(attributes, dict):
            raise ValueError("Attributes must be a dictionary")
        
        # 清理属性数据
        clean_attributes = DataSanitizer.sanitize_dict(attributes, max_depth=5)
        
        # 检查属性数量
        if len(clean_attributes) > 50:
            _LOGGER.warning("Entity has too many attributes: %d", len(clean_attributes))
            # 截断到前50个属性
            clean_attributes = dict(list(clean_attributes.items())[:50])
        
        # 计算总大小
        import json
        attrs_json = json.dumps(clean_attributes)
        if len(attrs_json) > 16384:  # 16KB限制
            raise ValueError("Attributes data too large")
        
        return clean_attributes

# 实际应用示例
async def secure_entity_update(
    hass: HomeAssistant,
    entity_id: str, 
    state: Any,
    attributes: dict[str, Any] | None = None
) -> None:
    """安全的实体更新实现
    
    集成所有安全验证和错误处理
    """
    try:
        # 验证和清理输入数据
        clean_entity_id = SecureEntityValidator.validate_entity_id(entity_id)
        clean_state = SecureEntityValidator.validate_state_value(state)
        clean_attributes = SecureEntityValidator.validate_attributes(attributes or {})
        
        # 执行状态更新
        hass.states.async_set(
            clean_entity_id,
            clean_state, 
            clean_attributes,
            force_update=False,  # 避免不必要的事件
        )
        
        _LOGGER.debug(
            "Securely updated entity %s to state %s", 
            clean_entity_id, clean_state
        )
        
    except Exception as err:
        _LOGGER.error(
            "Secure entity update failed for %s: %s", 
            entity_id, err
        )
        
        # 记录安全事件（可选）
        hass.bus.async_fire("security_event", {
            "type": "invalid_entity_update",
            "entity_id": entity_id,
            "error": str(err),
            "timestamp": dt_util.utcnow().isoformat(),
        })
        
        raise HomeAssistantError(f"Entity update validation failed: {err}") from err