tv_show_v1/hardware/wled_controller_module.py

import yaml
import os
import requests
import time
from typing import Optional, Dict, Any
from .wled_module import WledState, WledSegment
from utils.logger_config import logger

class WledController:
    """
    WLED控制器类
    负责从配置文件读取设置，生成WledState对象，并提供设备控制功能
    """
    
    def __init__(self, config_path: str = "led_config.yaml"):
        """
        初始化WLED控制器
        
        Args:
            config_path: 配置文件路径，默认为led_config.yaml
            
        Raises:
            FileNotFoundError: 配置文件不存在
            KeyError: 必需的配置项缺失
            ValueError: 配置值无效
        """
        self.config_path = config_path
        self.config = self._load_config()
        
        # 验证必需的配置项
        if 'wled_ip' not in self.config:
            error_msg = "配置文件中缺少必需的 'wled_ip' 配置项"
            logger.error(error_msg)
            raise KeyError(error_msg)
        
        if 'led_group_count' not in self.config:
            error_msg = "配置文件中缺少必需的 'led_group_count' 配置项"
            logger.error(error_msg)
            raise KeyError(error_msg)
        
        if 'segments' not in self.config:
            error_msg = "配置文件中缺少必需的 'segments' 配置项"
            logger.error(error_msg)
            raise KeyError(error_msg)
        
        # 验证配置值
        self.wled_ip = self.config['wled_ip']
        if not isinstance(self.wled_ip, str) or not self.wled_ip.strip():
            error_msg = f"无效的 wled_ip 配置值: {self.wled_ip}"
            logger.error(error_msg)
            raise ValueError(error_msg)
        
        self.led_group_count = self.config['led_group_count']
        if not isinstance(self.led_group_count, int) or self.led_group_count <= 0:
            error_msg = f"无效的 led_group_count 配置值: {self.led_group_count}，必须为正整数"
            logger.error(error_msg)
            raise ValueError(error_msg)
        
        self.segments_config = self.config['segments']
        if not isinstance(self.segments_config, list):
            error_msg = f"无效的 segments 配置值: {self.segments_config}，必须为列表"
            logger.error(error_msg)
            raise ValueError(error_msg)
        
        if not self.segments_config:
            error_msg = "segments 配置不能为空"
            logger.error(error_msg)
            raise ValueError(error_msg)
        
        # 按配置文件分段
        self.wled_state = self._create_wled_state()
        # 创建合并所有分段的状态对象，用于控制整条灯带
        self.wled_state_all = self.wled_state.merge_segments()
        logger.info(f"WLED控制器初始化完成，IP: {self.wled_ip}")
    
    def _load_config(self) -> Dict[str, Any]:
        """
        从YAML配置文件加载配置
        
        Returns:
            dict: 配置字典
            
        Raises:
            FileNotFoundError: 配置文件不存在
            yaml.YAMLError: YAML解析错误
            Exception: 其他加载错误
        """

        if not os.path.exists(self.config_path):
            error_msg = f"配置文件不存在: {self.config_path}"
            logger.error(error_msg)
            raise FileNotFoundError(error_msg)
        
        try:
            with open(self.config_path, 'r', encoding='utf-8') as file:
                config = yaml.safe_load(file)
                if config is None:
                    error_msg = f"配置文件为空或格式错误: {self.config_path}"
                    logger.error(error_msg)
                    raise ValueError(error_msg)
                logger.info(f"成功加载配置文件: {self.config_path}")
                return config
        except yaml.YAMLError as e:
            error_msg = f"YAML解析错误: {e}"
            logger.error(error_msg)
            raise
        except Exception as e:
            error_msg = f"加载配置文件失败: {e}"
            logger.error(error_msg)
            raise
    
    def _create_wled_state(self) -> WledState:
        """
        根据配置文件创建WledState对象
        
        Returns:
            WledState: 配置好的WLED状态对象
            
        Raises:
            KeyError: 分段配置缺少必需字段
            ValueError: 分段配置值无效
        """
        # 创建基础状态
        wled_state = WledState(
            on=True,
            brightness=255,
            transition=10
        )
        
        # 根据配置创建分段
        for i, segment_config in enumerate(self.segments_config):
            # 验证分段配置的必需字段
            required_fields = ['id', 'start', 'stop', 'name']
            for field in required_fields:
                if field not in segment_config:
                    error_msg = f"分段配置 {i} 缺少必需的字段 '{field}'"
                    logger.error(error_msg)
                    raise KeyError(error_msg)
            
            # 验证分段配置值
            segment_id = segment_config['id']
            if not isinstance(segment_id, int):
                error_msg = f"分段 {i} 的 id 必须是整数，当前值: {segment_id}"
                logger.error(error_msg)
                raise ValueError(error_msg)
            
            start = segment_config['start']
            if not isinstance(start, int) or start < 0:
                error_msg = f"分段 {i} 的 start 必须是非负整数，当前值: {start}"
                logger.error(error_msg)
                raise ValueError(error_msg)
            
            stop = segment_config['stop']
            if not isinstance(stop, int) or stop <= start:
                error_msg = f"分段 {i} 的 stop 必须是大于 start 的整数，当前值: stop={stop}, start={start}"
                logger.error(error_msg)
                raise ValueError(error_msg)
            
            name = segment_config['name']
            if not isinstance(name, str) or not name.strip():
                error_msg = f"分段 {i} 的 name 必须是非空字符串，当前值: {name}"
                logger.error(error_msg)
                raise ValueError(error_msg)
            
            segment = WledSegment(
                start=start,
                stop=stop,
                segment_id=segment_id,
                group=self.led_group_count,
                on=True,
                colors=[[0, 0, 255], [0, 0, 0], [0, 0, 0]],  # 默认蓝色
                effect=6,  # 默认彩虹特效
                speed=230,
                intensity=255,
                palette=0,
                reverse=False,
                selected=True
            )
            wled_state.add_segment(segment)
            logger.info(f"创建分段: {name} (ID: {segment.id}, LED: {segment.start}-{segment.stop-1})")
        
        logger.info(f"WledState创建完成，共{len(self.segments_config)}个分段")
        return wled_state
    
    def get_wled_state(self) -> WledState:
        """
        获取当前WLED状态对象
        
        Returns:
            WledState: 当前状态对象
        """
        return self.wled_state
    
    def get_segment_by_id(self, id: int) -> WledSegment:
        """
        根据ID获取分段
        
        Args:
            id: 分段ID
            
        Returns:
            WledSegment: 分段对象
            
        Raises:
            ValueError: 分段ID无效或分段不存在
        """
        if not isinstance(id, int) or id < 0:
            error_msg = f"分段ID必须是非负整数，当前值: {id}"
            logger.error(error_msg)
            raise ValueError(error_msg)
        
        return self.wled_state.get_segment(id)
    
    def set_segment_effect(self, segment_id: int, effect_id: int, 
                          speed: int = None, intensity: int = None):
        """
        设置指定分段的特效
        
        Args:
            segment_id: 分段ID
            effect_id: 特效ID
            speed: 特效速度 (0-255)
            intensity: 特效强度 (0-255)
            
        Raises:
            ValueError: 分段ID无效或分段不存在
        """
        if not isinstance(segment_id, int) or segment_id < 0:
            error_msg = f"分段ID必须是非负整数，当前值: {segment_id}"
            logger.error(error_msg)
            raise ValueError(error_msg)
        
        segment = self.wled_state.get_segment(segment_id)
        if not segment:
            error_msg = f"未找到分段ID: {segment_id}"
            logger.error(error_msg)
            raise ValueError(error_msg)
        
        segment.set_effect(effect_id, speed, intensity)
        logger.info(f"分段{segment_id}特效已设置为: {effect_id}")
    
    def set_global_brightness(self, brightness: int):
        """
        设置全局亮度
        
        Args:
            brightness: 亮度值 0-255
            
        Raises:
            ValueError: 亮度值无效
        """
        if not isinstance(brightness, int) or brightness < 0 or brightness > 255:
            error_msg = f"亮度值必须是0-255之间的整数，当前值: {brightness}"
            logger.error(error_msg)
            raise ValueError(error_msg)
        
        self.wled_state.set_global_brightness(brightness)
        logger.info(f"全局亮度已设置为: {brightness}")
    
    def toggle_power(self):
        """切换设备总开关状态"""
        self.wled_state.toggle_power()
        status = "开启" if self.wled_state.on else "关闭"
        logger.info(f"设备状态已切换为: {status}")
    
    def send_state(self, state: WledState = None, timeout: float = 5.0, 
                   retry_count: int = 3) -> bool:
        """
        发送状态到WLED设备
        
        Args:
            state: 要发送的状态对象，默认为当前状态
            timeout: 请求超时时间（秒）
            retry_count: 重试次数
            
        Returns:
            bool: 发送是否成功
        """
        if state is None:
            state = self.wled_state
        
        payload = state.to_dict()
        url = f"http://{self.wled_ip}/json/state"
        headers = {"Content-Type": "application/json"}
        
        for attempt in range(retry_count + 1):
            try:
                logger.info(f"发送状态到设备 {self.wled_ip} (尝试 {attempt + 1}/{retry_count + 1})")
                logger.info(f"发送状态: {payload}")
                response = requests.post(
                    url, 
                    headers=headers, 
                    json=payload, 
                    timeout=timeout,
                    verify=False
                )
                
                if response.status_code == 200:
                    logger.info(f"状态发送成功，状态码: {response.status_code}")
                    return True
                else:
                    logger.warning(f"状态发送失败，状态码: {response.status_code}")
                    
            except requests.exceptions.RequestException as e:
                logger.error(f"发送请求失败 (尝试 {attempt + 1}): {e}")
                if attempt < retry_count:
                    wait_time = 2 ** attempt  # 指数退避
                    logger.info(f"等待 {wait_time} 秒后重试...")
                    time.sleep(wait_time)
        
        logger.error(f"所有重试均失败，无法发送状态到设备 {self.wled_ip}")
        return False
    
    def __str__(self) -> str:
        """返回控制器的字符串表示"""
        return f"WledController(IP: {self.wled_ip}, 分段数: {len(self.segments_config)})"