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)})"