230 lines
10 KiB
Python
230 lines
10 KiB
Python
from langchain_core.tools import tool
|
||
from miio import DeviceFactory
|
||
from miio.miot_device import MiotDevice
|
||
import json
|
||
from pydantic import BaseModel, Field
|
||
import time
|
||
import logging
|
||
import threading
|
||
|
||
# 配置日志
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 设备配置
|
||
LAMP_IP = "192.168.110.122"
|
||
LAMP_TOKEN = "4a90f98aaa1273ca34685d66d6e13958"
|
||
LAMP_MODEL = "yeelink.light.bslamp2"
|
||
|
||
try:
|
||
device = DeviceFactory.create(LAMP_IP, LAMP_TOKEN)
|
||
logger.info(f"使用 DeviceFactory 创建设备成功: {LAMP_MODEL}")
|
||
except Exception as e:
|
||
logger.warning(f"DeviceFactory 创建失败，使用 MiotDevice: {e}")
|
||
|
||
# 添加线程锁，确保同一时间只有一个操作
|
||
device_lock = threading.Lock()
|
||
|
||
|
||
@tool("get_lamp_status", description="获取床头灯当前状态，包括电源、亮度、色温、颜色等信息")
|
||
def get_lamp_status():
|
||
"""获取床头灯设备状态并以 JSON 格式返回"""
|
||
try:
|
||
with device_lock: # 使用锁确保串行执行
|
||
# 使用 status() 方法获取 DeviceStatus 对象
|
||
device_status = device.status()
|
||
|
||
# 构建状态字典
|
||
status = {
|
||
"power": device_status.power if hasattr(device_status, 'power') else None,
|
||
"is_on": device_status.is_on if hasattr(device_status, 'is_on') else None,
|
||
"brightness": device_status.brightness if hasattr(device_status, 'brightness') else None,
|
||
"color_temp": device_status.color_temp if hasattr(device_status, 'color_temp') else None,
|
||
"color_mode": device_status.color_mode if hasattr(device_status, 'color_mode') else None,
|
||
"rgb": device_status.rgb if hasattr(device_status, 'rgb') else None,
|
||
"online": True,
|
||
"model": LAMP_MODEL
|
||
}
|
||
|
||
return json.dumps(status, indent=2, ensure_ascii=False)
|
||
except Exception as e:
|
||
logger.error(f"获取床头灯状态失败: {e}")
|
||
error_status = {
|
||
"error": f"获取设备状态失败: {str(e)}",
|
||
"message": "请检查：\n1. 设备是否已开启并连接到网络\n2. 设备IP地址是否配置正确（当前配置：{ip}）\n3. 设备Token是否正确".format(ip=LAMP_IP),
|
||
"online": False,
|
||
"model": LAMP_MODEL
|
||
}
|
||
return json.dumps(error_status, indent=2, ensure_ascii=False)
|
||
|
||
|
||
class PowerArgs(BaseModel):
|
||
power: bool = Field(..., description="床头灯电源状态，true 开启，false 关闭")
|
||
|
||
|
||
@tool("set_lamp_power", args_schema=PowerArgs, description="开启或关闭床头灯。power=true 开启，power=false 关闭")
|
||
def set_lamp_power(power: bool):
|
||
"""开启或关闭床头灯"""
|
||
try:
|
||
with device_lock: # 使用锁确保串行执行
|
||
if power:
|
||
result = device.on()
|
||
else:
|
||
result = device.off()
|
||
action = "开启" if power else "关闭"
|
||
logger.info(f"床头灯已{action}")
|
||
return json.dumps({
|
||
"message": f"床头灯已{action}",
|
||
"power": power,
|
||
"result": str(result)
|
||
}, indent=2, ensure_ascii=False)
|
||
except Exception as e:
|
||
logger.error(f"设置床头灯电源失败: {e}")
|
||
return json.dumps({
|
||
"error": f"设置电源状态失败: {str(e)}",
|
||
"message": "请检查：\n1. 设备是否已开启并连接到网络\n2. 设备IP地址是否配置正确（当前配置：{ip}）\n3. 设备Token是否正确".format(ip=LAMP_IP),
|
||
"online": False,
|
||
"model": LAMP_MODEL
|
||
}, indent=2, ensure_ascii=False)
|
||
|
||
|
||
class BrightnessArgs(BaseModel):
|
||
brightness: int = Field(..., ge=1, le=100, description="亮度值，范围 1-100")
|
||
|
||
|
||
@tool("set_lamp_brightness", args_schema=BrightnessArgs, description="设置床头灯亮度（1-100）")
|
||
def set_lamp_brightness(brightness: int):
|
||
"""设置床头灯亮度"""
|
||
try:
|
||
with device_lock:
|
||
# 参考 Iot.py - 设置亮度 (siid=2, piid=2)
|
||
result = device.send("set_bright", [brightness])
|
||
logger.info(f"亮度已设置为{brightness}%")
|
||
return json.dumps({
|
||
"message": f"亮度已设置为{brightness}%",
|
||
"brightness": brightness,
|
||
"result": str(result)
|
||
}, indent=2, ensure_ascii=False)
|
||
except Exception as e:
|
||
logger.error(f"设置床头灯亮度失败: {e}")
|
||
return json.dumps({
|
||
"error": f"设置亮度失败: {str(e)}",
|
||
"message": "请检查：\n1. 设备是否已开启并连接到网络\n2. 设备IP地址是否配置正确（当前配置：{ip}）\n3. 设备Token是否正确".format(ip=LAMP_IP),
|
||
"online": False,
|
||
"model": LAMP_MODEL
|
||
}, indent=2, ensure_ascii=False)
|
||
|
||
|
||
class ColorTempArgs(BaseModel):
|
||
color_temp: int = Field(..., ge=1700, le=6500, description="色温值，范围 1700-6500K")
|
||
|
||
|
||
@tool("set_lamp_color_temp", args_schema=ColorTempArgs, description="设置床头灯色温（1700-6500K，暖光到冷光）")
|
||
def set_lamp_color_temp(color_temp: int):
|
||
"""设置床头灯色温"""
|
||
try:
|
||
with device_lock:
|
||
result = device.send("set_ct_abx", [color_temp, "smooth", 500])
|
||
temp_desc = "暖光" if color_temp < 3000 else "中性光" if color_temp < 5000 else "冷光"
|
||
logger.info(f"色温已设置为{color_temp}K ({temp_desc})")
|
||
return json.dumps({
|
||
"message": f"色温已设置为{color_temp}K ({temp_desc})",
|
||
"color_temp": color_temp,
|
||
"description": temp_desc,
|
||
"result": str(result)
|
||
}, indent=2, ensure_ascii=False)
|
||
except Exception as e:
|
||
logger.error(f"设置床头灯色温失败: {e}")
|
||
return json.dumps({
|
||
"error": f"设置色温失败: {str(e)}",
|
||
"message": "请检查：\n1. 设备是否已开启并连接到网络\n2. 设备IP地址是否配置正确（当前配置：{ip}）\n3. 设备Token是否正确".format(ip=LAMP_IP),
|
||
"online": False,
|
||
"model": LAMP_MODEL
|
||
}, indent=2, ensure_ascii=False)
|
||
|
||
|
||
class ColorArgs(BaseModel):
|
||
red: int = Field(..., ge=0, le=255, description="红色值，范围 0-255")
|
||
green: int = Field(..., ge=0, le=255, description="绿色值，范围 0-255")
|
||
blue: int = Field(..., ge=0, le=255, description="蓝色值，范围 0-255")
|
||
|
||
|
||
@tool("set_lamp_color", args_schema=ColorArgs, description="设置床头灯RGB颜色（红、绿、蓝各0-255）")
|
||
def set_lamp_color(red: int, green: int, blue: int):
|
||
"""设置床头灯RGB颜色"""
|
||
try:
|
||
with device_lock:
|
||
color_value = (red << 16) | (green << 8) | blue
|
||
result = device.send("set_rgb", [color_value])
|
||
logger.info(f"颜色已设置为 RGB({red}, {green}, {blue})")
|
||
return json.dumps({
|
||
"message": f"颜色已设置为 RGB({red}, {green}, {blue})",
|
||
"red": red,
|
||
"green": green,
|
||
"blue": blue,
|
||
"color_hex": f"#{red:02x}{green:02x}{blue:02x}",
|
||
"result": str(result)
|
||
}, indent=2, ensure_ascii=False)
|
||
except Exception as e:
|
||
logger.error(f"设置床头灯颜色失败: {e}")
|
||
return json.dumps({
|
||
"error": f"设置颜色失败: {str(e)}",
|
||
"message": "请检查：\n1. 设备是否已开启并连接到网络\n2. 设备IP地址是否配置正确（当前配置：{ip}）\n3. 设备Token是否正确".format(ip=LAMP_IP),
|
||
"online": False,
|
||
"model": LAMP_MODEL
|
||
}, indent=2, ensure_ascii=False)
|
||
|
||
|
||
class SceneArgs(BaseModel):
|
||
scene: str = Field(..., description="场景名称: 'reading' (阅读), 'sleep' (睡眠), 'romantic' (浪漫), 'night' (夜灯)")
|
||
|
||
|
||
@tool("set_lamp_scene", args_schema=SceneArgs, description="设置床头灯预设场景（阅读/睡眠/浪漫/夜灯）")
|
||
def set_lamp_scene(scene: str):
|
||
"""设置床头灯预设场景"""
|
||
# 定义预设场景
|
||
scenes = {
|
||
"reading": {"brightness": 100, "color_temp": 4000, "desc": "阅读模式：100%亮度，4000K中性光"},
|
||
"sleep": {"brightness": 10, "color_temp": 2000, "desc": "睡眠模式：10%亮度，2000K暖光"},
|
||
"romantic": {"brightness": 30, "color": (255, 100, 100), "desc": "浪漫模式：30%亮度，粉红色"},
|
||
"night": {"brightness": 5, "color_temp": 1700, "desc": "夜灯模式：5%亮度，1700K极暖光"}
|
||
}
|
||
|
||
if scene not in scenes:
|
||
return json.dumps({
|
||
"error": f"未知场景: {scene}",
|
||
"available_scenes": list(scenes.keys())
|
||
}, indent=2, ensure_ascii=False)
|
||
|
||
try:
|
||
with device_lock:
|
||
scene_config = scenes[scene]
|
||
|
||
# 设置亮度
|
||
device.set_property_by(2, 2, scene_config["brightness"])
|
||
time.sleep(0.3) # 给设备一点响应时间
|
||
|
||
# 设置色温或颜色
|
||
if "color_temp" in scene_config:
|
||
device.set_property_by(2, 3, scene_config["color_temp"])
|
||
elif "color" in scene_config:
|
||
r, g, b = scene_config["color"]
|
||
color_value = (r << 16) | (g << 8) | b
|
||
device.set_property_by(2, 5, color_value)
|
||
|
||
logger.info(f"场景已设置为: {scene}")
|
||
return json.dumps({
|
||
"message": f"场景已设置为: {scene}",
|
||
"scene": scene,
|
||
"description": scene_config["desc"],
|
||
"config": scene_config
|
||
}, indent=2, ensure_ascii=False)
|
||
except Exception as e:
|
||
logger.error(f"设置床头灯场景失败: {e}")
|
||
return json.dumps({
|
||
"error": f"设置场景失败: {str(e)}",
|
||
"message": "请检查：\n1. 设备是否已开启并连接到网络\n2. 设备IP地址是否配置正确（当前配置：{ip}）\n3. 设备Token是否正确".format(ip=LAMP_IP),
|
||
"online": False,
|
||
"model": LAMP_MODEL
|
||
}, indent=2, ensure_ascii=False)
|
||
|