tv_show_v1/hardware/ha_devices_module.py

import yaml
import requests
import os
import json
from utils.logger_config import logger

class HADeviceController:
    def __init__(self, config_path='ha_config_prod.yaml'):
        self.config_path = config_path
        self.config = self._load_config()
        self.ha_config = self.config.get('home_assistant', {})
        self.base_url = self.ha_config.get('url', '').rstrip('/')
        self.token = self.ha_config.get('token', '')
        self.headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json",
        }
        self.devices = self.config.get('ha_devices', {})

    def _load_config(self):
        if not os.path.exists(self.config_path):
            logger.error(f"配置文件不存在: {self.config_path}")
            return {}
        try:
            with open(self.config_path, 'r', encoding='utf-8') as f:
                return yaml.safe_load(f) or {}
        except Exception as e:
            logger.error(f"加载配置文件失败: {e}")
            return {}

    def _get_device_state(self, entity_id):
        """获取设备当前状态"""
        if not self.base_url or not self.token:
            return None
        
        url = f"{self.base_url}/api/states/{entity_id}"
        try:
            logger.info(f"正在检查设备状态: {entity_id}")
            response = requests.get(url, headers=self.headers, timeout=5)
            if response.status_code == 200:
                data = response.json()
                state = data.get('state')
                logger.info(f"设备 {entity_id} 当前状态: {state}")
                return state
            else:
                logger.error(f"获取设备状态失败: {entity_id}, code={response.status_code}")
                return None
        except Exception as e:
            logger.error(f"获取设备状态异常: {entity_id}, {e}")
            return None

    def _call_ha_service(self, entity_id, service="turn_on", domain="switch"):
        if not self.base_url or not self.token:
            logger.error("Home Assistant 配置缺失 (url or token)")
            return False
        
        if not entity_id:
            logger.warning("未配置 entity_id")
            return False

        url = f"{self.base_url}/api/services/{domain}/{service}"
        payload = {"entity_id": entity_id}
        
        logger.info(f"准备调用HA接口: URL={url}, Entity={entity_id}, Service={domain}.{service}")
        
        try:
            response = requests.post(url, headers=self.headers, json=payload, timeout=5)
            if response.status_code in [200, 201]:
                logger.info(f"HA调用成功: {entity_id} -> {domain}.{service}, 响应: {response.text}")
                return True
            else:
                logger.error(f"HA调用失败: {response.status_code} - {response.text}")
                return False
        except Exception as e:
            logger.error(f"HA请求异常: {e}")
            return False

    def control_device(self, device_key, action="turn_on"):
        """
        通用设备控制方法
        :param device_key: 配置文件 ha_devices 中的 key (如 'entrance_light_1')
        :param action: 动作 'turn_on' 或 'turn_off'
        """
        device_info = self.devices.get(device_key)
        if not device_info:
            logger.error(f"未找到设备配置: {device_key}")
            return False
        
        entity_id = device_info.get('entity_id')
        if not entity_id:
             logger.error(f"设备 {device_key} 未配置 entity_id")
             return False

        # 检查设备当前状态
        current_state = self._get_device_state(entity_id)
        if current_state:
            if action == "turn_on" and current_state == "on":
                logger.info(f"设备 {device_key} ({entity_id}) 已经是开启状态，跳过开启操作")
                return True
            if action == "turn_off" and current_state == "off":
                logger.info(f"设备 {device_key} ({entity_id}) 已经是关闭状态，跳过关闭操作")
                return True
             
        # 自动判断 domain
        domain = entity_id.split('.')[0]
        
        return self._call_ha_service(entity_id, service=action, domain=domain)

    # 具体设备控制方法
    def turn_on_entrance_lights(self):
        """打开一楼大门玄关顶灯 (全部)"""
        r1 = self.control_device('entrance_light_1', 'turn_on')
        r2 = self.control_device('entrance_light_2', 'turn_on')
        return r1 and r2

    def turn_off_entrance_lights(self):
        """关闭一楼大门玄关顶灯 (全部)"""
        r1 = self.control_device('entrance_light_1', 'turn_off')
        r2 = self.control_device('entrance_light_2', 'turn_off')
        return r1 and r2
        
    def turn_on_exhibition_spotlight(self):
        """打开一楼大门玄关射灯"""
        return self.control_device('exhibition_spotlight', 'turn_on')
        
    def turn_off_exhibition_spotlight(self):
        """关闭一楼大门玄关射灯"""
        return self.control_device('exhibition_spotlight', 'turn_off')
        
    def turn_on_exhibition_ceiling_lights(self):
        """打开一楼展厅顶灯 (全部)"""
        r1 = self.control_device('exhibition_ceiling_light_1', 'turn_on')
        r2 = self.control_device('exhibition_ceiling_light_2', 'turn_on')
        r3 = self.control_device('exhibition_ceiling_light_3', 'turn_on')
        return r1 and r2 and r3

    def turn_off_exhibition_ceiling_lights(self):
        """关闭一楼展厅顶灯 (全部)"""
        r1 = self.control_device('exhibition_ceiling_light_1', 'turn_off')
        r2 = self.control_device('exhibition_ceiling_light_2', 'turn_off')
        r3 = self.control_device('exhibition_ceiling_light_3', 'turn_off')
        return r1 and r2 and r3

    def set_exhibition_ceiling_light_level(self, level):
        """
        设置一楼展厅顶灯等级 (0-3)
        0: 全关
        1: 开启灯1 (关2,3)
        2: 开启灯1,2 (关3)
        3: 全开 (1,2,3)
        """
        logger.info(f"设置展厅顶灯等级: {level}")
        
        # 确保 level 在 0-3 之间
        try:
            level = int(level)
            if level < 0: level = 0
            if level > 3: level = 3
        except (ValueError, TypeError):
            logger.error(f"无效的灯光等级: {level}")
            return False

        r1, r2, r3 = True, True, True

        if level == 0:
            # 0: 全关
            r1 = self.control_device('exhibition_ceiling_light_1', 'turn_off')
            r2 = self.control_device('exhibition_ceiling_light_2', 'turn_off')
            r3 = self.control_device('exhibition_ceiling_light_3', 'turn_off')
        
        elif level == 1:
            # 1: 开启灯1 (关2,3)
            r1 = self.control_device('exhibition_ceiling_light_1', 'turn_on')
            r2 = self.control_device('exhibition_ceiling_light_2', 'turn_off')
            r3 = self.control_device('exhibition_ceiling_light_3', 'turn_off')
            
        elif level == 2:
            # 2: 开启灯1,2 (关3)
            r1 = self.control_device('exhibition_ceiling_light_1', 'turn_on')
            r2 = self.control_device('exhibition_ceiling_light_2', 'turn_on')
            r3 = self.control_device('exhibition_ceiling_light_3', 'turn_off')
            
        elif level == 3:
            # 3: 全开
            r1 = self.control_device('exhibition_ceiling_light_1', 'turn_on')
            r2 = self.control_device('exhibition_ceiling_light_2', 'turn_on')
            r3 = self.control_device('exhibition_ceiling_light_3', 'turn_on')

        return r1 and r2 and r3

    def turn_on_exhibition_desktop_switch(self):
        """打开展厅桌面的灯座总开关"""
        return self.control_device('exhibition_desktop_switch', 'turn_on')

    def turn_off_exhibition_desktop_switch(self):
        """关闭展厅桌面的灯座总开关"""
        return self.control_device('exhibition_desktop_switch', 'turn_off')

    def turn_on_exhibition_3d_fan(self):
        """打开展厅桌面3D风扇投影"""
        return self.control_device('exhibition_3d_fan', 'turn_on')

    def turn_off_exhibition_3d_fan(self):
        """关闭展厅桌面3D风扇投影"""
        return self.control_device('exhibition_3d_fan', 'turn_off')

    def turn_on_exhibition_stand_light_strip(self):
        """打开展台桌子灯带"""
        return self.control_device('exhibition_stand_light_strip', 'turn_on')

    def turn_off_exhibition_stand_light_strip(self):
        """关闭展台桌子灯带"""
        return self.control_device('exhibition_stand_light_strip', 'turn_off')
        
    def turn_on_all(self):
        """打开所有配置的HA设备"""
        results = []
        for key in self.devices:
            results.append(self.control_device(key, 'turn_on'))
        return all(results) if results else False
        
    def turn_off_all(self):
        """关闭所有配置的HA设备"""
        results = []
        for key in self.devices:
            results.append(self.control_device(key, 'turn_off'))
        return all(results) if results else False

# 单例实例
ha_device_controller = HADeviceController()

# 模块级便捷函数
def control_device(device_key, action="turn_on"):
    return ha_device_controller.control_device(device_key, action)

def turn_on_entrance_lights():
    return ha_device_controller.turn_on_entrance_lights()

def turn_off_entrance_lights():
    return ha_device_controller.turn_off_entrance_lights()

def turn_on_exhibition_spotlight():
    return ha_device_controller.turn_on_exhibition_spotlight()

def turn_off_exhibition_spotlight():
    return ha_device_controller.turn_off_exhibition_spotlight()

def turn_on_exhibition_ceiling_lights():
    return ha_device_controller.turn_on_exhibition_ceiling_lights()

def turn_off_exhibition_ceiling_lights():
    return ha_device_controller.turn_off_exhibition_ceiling_lights()

def set_exhibition_ceiling_light_level(level):
    return ha_device_controller.set_exhibition_ceiling_light_level(level)

def turn_on_exhibition_desktop_switch():
    return ha_device_controller.turn_on_exhibition_desktop_switch()

def turn_off_exhibition_desktop_switch():
    return ha_device_controller.turn_off_exhibition_desktop_switch()

def turn_on_exhibition_3d_fan():
    return ha_device_controller.turn_on_exhibition_3d_fan()

def turn_off_exhibition_3d_fan():
    return ha_device_controller.turn_off_exhibition_3d_fan()

def turn_on_exhibition_stand_light_strip():
    return ha_device_controller.turn_on_exhibition_stand_light_strip()

def turn_off_exhibition_stand_light_strip():
    return ha_device_controller.turn_off_exhibition_stand_light_strip()

def turn_on_all():
    return ha_device_controller.turn_on_all()

def turn_off_all():
    return ha_device_controller.turn_off_all()