import httpx
|
from config.logger import setup_logging
|
from plugins_func.functions.hass_init import initialize_hass_handler
|
from plugins_func.register import register_function, ToolType, ActionResponse, Action
|
from typing import TYPE_CHECKING
|
|
if TYPE_CHECKING:
|
from core.connection import ConnectionHandler
|
|
TAG = __name__
|
logger = setup_logging()
|
|
hass_get_state_function_desc = {
|
"type": "function",
|
"function": {
|
"name": "hass_get_state",
|
"description": "获取homeassistant里设备的状态,包括查询灯光亮度、颜色、色温,媒体播放器的音量,设备的暂停、继续操作",
|
"parameters": {
|
"type": "object",
|
"properties": {
|
"entity_id": {
|
"type": "string",
|
"description": "需要操作的设备id,homeassistant里的entity_id",
|
}
|
},
|
"required": ["entity_id"],
|
},
|
},
|
}
|
|
hass_set_state_function_desc = {
|
"type": "function",
|
"function": {
|
"name": "hass_set_state",
|
"description": "设置homeassistant里设备的状态,包括开、关,调整灯光亮度、颜色、色温,调整播放器的音量,设备的暂停、继续、静音操作",
|
"parameters": {
|
"type": "object",
|
"properties": {
|
"state": {
|
"type": "object",
|
"properties": {
|
"type": {
|
"type": "string",
|
"description": "需要操作的动作,打开设备:turn_on,关闭设备:turn_off,增加亮度:brightness_up,降低亮度:brightness_down,设置亮度:brightness_value,增加音量:volume_up,降低音量:volume_down,设置音量:volume_set,设置色温:set_kelvin,设置颜色:set_color,设备暂停:pause,设备继续:continue,静音/取消静音:volume_mute",
|
},
|
"input": {
|
"type": "integer",
|
"description": "只有在设置音量,设置亮度时候才需要,有效值为1-100,对应音量和亮度的1%-100%",
|
},
|
"is_muted": {
|
"type": "string",
|
"description": "只有在设置静音操作时才需要,设置静音的时候该值为true,取消静音时该值为false",
|
},
|
"rgb_color": {
|
"type": "array",
|
"items": {"type": "integer"},
|
"description": "只有在设置颜色时需要,这里填目标颜色的rgb值",
|
},
|
},
|
"required": ["type"],
|
},
|
"entity_id": {
|
"type": "string",
|
"description": "需要操作的设备id,homeassistant里的entity_id",
|
},
|
},
|
"required": ["state", "entity_id"],
|
},
|
},
|
}
|
|
|
@register_function("hass_get_state", hass_get_state_function_desc, ToolType.SYSTEM_CTL)
|
async def hass_get_state(conn: "ConnectionHandler", entity_id=""):
|
try:
|
ha_response = await handle_hass_get_state(conn, entity_id)
|
return ActionResponse(Action.REQLLM, ha_response, None)
|
except httpx.TimeoutException:
|
logger.bind(tag=TAG).error("获取Home Assistant状态超时")
|
return ActionResponse(Action.ERROR, "请求超时", None)
|
except Exception as e:
|
error_msg = f"执行Home Assistant操作失败"
|
logger.bind(tag=TAG).error(error_msg)
|
return ActionResponse(Action.ERROR, error_msg, None)
|
|
|
@register_function("hass_set_state", hass_set_state_function_desc, ToolType.SYSTEM_CTL)
|
async def hass_set_state(conn: "ConnectionHandler", entity_id="", state=None):
|
if state is None:
|
state = {}
|
try:
|
ha_response = await handle_hass_set_state(conn, entity_id, state)
|
return ActionResponse(Action.REQLLM, ha_response, None)
|
except httpx.TimeoutException:
|
logger.bind(tag=TAG).error("设置Home Assistant状态超时")
|
return ActionResponse(Action.ERROR, "请求超时", None)
|
except Exception as e:
|
error_msg = f"执行Home Assistant操作失败"
|
logger.bind(tag=TAG).error(error_msg)
|
return ActionResponse(Action.ERROR, error_msg, None)
|
|
|
async def handle_hass_get_state(conn: "ConnectionHandler", entity_id):
|
ha_config = initialize_hass_handler(conn)
|
api_key = ha_config.get("api_key")
|
base_url = ha_config.get("base_url")
|
url = f"{base_url}/api/states/{entity_id}"
|
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
|
|
async with httpx.AsyncClient(timeout=httpx.Timeout(5.0, connect=3.0)) as client:
|
response = await client.get(url, headers=headers)
|
|
if response.status_code == 200:
|
responsetext = "设备状态:" + response.json()["state"] + " "
|
logger.bind(tag=TAG).info(f"api返回内容: {response.json()}")
|
|
if "media_title" in response.json()["attributes"]:
|
responsetext = (
|
responsetext
|
+ "正在播放的是:"
|
+ str(response.json()["attributes"]["media_title"])
|
+ " "
|
)
|
if "volume_level" in response.json()["attributes"]:
|
responsetext = (
|
responsetext
|
+ "音量是:"
|
+ str(response.json()["attributes"]["volume_level"])
|
+ " "
|
)
|
if "color_temp_kelvin" in response.json()["attributes"]:
|
responsetext = (
|
responsetext
|
+ "色温是:"
|
+ str(response.json()["attributes"]["color_temp_kelvin"])
|
+ " "
|
)
|
if "rgb_color" in response.json()["attributes"]:
|
responsetext = (
|
responsetext
|
+ "rgb颜色是:"
|
+ str(response.json()["attributes"]["rgb_color"])
|
+ " "
|
)
|
if "brightness" in response.json()["attributes"]:
|
responsetext = (
|
responsetext
|
+ "亮度是:"
|
+ str(response.json()["attributes"]["brightness"])
|
+ " "
|
)
|
logger.bind(tag=TAG).info(f"查询返回内容: {responsetext}")
|
return responsetext
|
else:
|
return f"切换失败，错误码: {response.status_code}"
|
|
|
async def handle_hass_set_state(conn: "ConnectionHandler", entity_id, state):
|
ha_config = initialize_hass_handler(conn)
|
api_key = ha_config.get("api_key")
|
base_url = ha_config.get("base_url")
|
"""
|
state = { "type":"brightness_up","input":"80","is_muted":"true"}
|
"""
|
domains = entity_id.split(".")
|
if len(domains) > 1:
|
domain = domains[0]
|
else:
|
return "执行失败，错误的设备id"
|
action = ""
|
arg = ""
|
value = ""
|
if state["type"] == "turn_on":
|
description = "设备已打开"
|
if domain == "cover":
|
action = "open_cover"
|
elif domain == "vacuum":
|
action = "start"
|
else:
|
action = "turn_on"
|
elif state["type"] == "turn_off":
|
description = "设备已关闭"
|
if domain == "cover":
|
action = "close_cover"
|
elif domain == "vacuum":
|
action = "stop"
|
else:
|
action = "turn_off"
|
elif state["type"] == "brightness_up":
|
description = "灯光已调亮"
|
action = "turn_on"
|
arg = "brightness_step_pct"
|
value = 10
|
elif state["type"] == "brightness_down":
|
description = "灯光已调暗"
|
action = "turn_on"
|
arg = "brightness_step_pct"
|
value = -10
|
elif state["type"] == "brightness_value":
|
description = f"亮度已调整到{state['input']}"
|
action = "turn_on"
|
arg = "brightness_pct"
|
value = state["input"]
|
elif state["type"] == "set_color":
|
description = f"颜色已调整到{state['rgb_color']}"
|
action = "turn_on"
|
arg = "rgb_color"
|
value = state["rgb_color"]
|
elif state["type"] == "set_kelvin":
|
description = f"色温已调整到{state['input']}K"
|
action = "turn_on"
|
arg = "kelvin"
|
value = state["input"]
|
elif state["type"] == "volume_up":
|
description = "音量已调大"
|
action = state["type"]
|
elif state["type"] == "volume_down":
|
description = "音量已调小"
|
action = state["type"]
|
elif state["type"] == "volume_set":
|
description = f"音量已调整到{state['input']}"
|
action = state["type"]
|
arg = "volume_level"
|
value = state["input"]
|
if state["input"] >= 1:
|
value = state["input"] / 100
|
elif state["type"] == "volume_mute":
|
description = f"设备已静音"
|
action = state["type"]
|
arg = "is_volume_muted"
|
value = state["is_muted"]
|
elif state["type"] == "pause":
|
description = f"设备已暂停"
|
action = state["type"]
|
if domain == "media_player":
|
action = "media_pause"
|
if domain == "cover":
|
action = "stop_cover"
|
if domain == "vacuum":
|
action = "pause"
|
elif state["type"] == "continue":
|
description = f"设备已继续"
|
if domain == "media_player":
|
action = "media_play"
|
if domain == "vacuum":
|
action = "start"
|
else:
|
return f"{domain} {state['type']}功能尚未支持"
|
|
if arg == "":
|
data = {
|
"entity_id": entity_id,
|
}
|
else:
|
data = {"entity_id": entity_id, arg: value}
|
url = f"{base_url}/api/services/{domain}/{action}"
|
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
|
|
async with httpx.AsyncClient(timeout=httpx.Timeout(5.0, connect=3.0)) as client:
|
response = await client.post(url, headers=headers, json=data)
|
|
logger.bind(tag=TAG).info(
|
f"设置状态:{description},url:{url},return_code:{response.status_code}"
|
)
|
if response.status_code == 200:
|
return description
|
else:
|
return f"设置失败，错误码: {response.status_code}"
|
