update:优化ha工具获取密钥的方式

This commit is contained in:
hrz
2025-06-12 21:43:30 +08:00
parent e943b07344
commit 9c3e1a6ff6
4 changed files with 165 additions and 151 deletions
@@ -20,40 +20,39 @@ hass_set_state_function_desc = {
"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"
"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%"
"description": "只有在设置音量,设置亮度时候才需要,有效值为1-100,对应音量和亮度的1%-100%",
},
"is_muted": {
"type": "string",
"description": "只有在设置静音操作时才需要,设置静音的时候该值为true,取消静音时该值为false"
"description": "只有在设置静音操作时才需要,设置静音的时候该值为true,取消静音时该值为false",
},
"rgb_color": {
"type": "list",
"description": "只有在设置颜色时需要,这里填目标颜色的rgb值",
},
"rgb_color":{
"type":"list",
"description": "只有在设置颜色时需要,这里填目标颜色的rgb值"
}
},
"required": ["type"]
"required": ["type"],
},
"entity_id": {
"type": "string",
"description": "需要操作的设备id,homeassistant里的entity_id"
}
"description": "需要操作的设备id,homeassistant里的entity_id",
},
},
"required": ["state", "entity_id"]
}
}
"required": ["state", "entity_id"],
},
},
}
@register_function('hass_set_state', hass_set_state_function_desc, ToolType.SYSTEM_CTL)
def hass_set_state(conn, entity_id='', state={}):
@register_function("hass_set_state", hass_set_state_function_desc, ToolType.SYSTEM_CTL)
def hass_set_state(conn, entity_id="", state={}):
try:
future = asyncio.run_coroutine_threadsafe(
handle_hass_set_state(conn, entity_id, state),
conn.loop
handle_hass_set_state(conn, entity_id, state), conn.loop
)
ha_response = future.result()
return ActionResponse(Action.REQLLM, ha_response, None)
@@ -62,21 +61,21 @@ def hass_set_state(conn, entity_id='', state={}):
async def handle_hass_set_state(conn, entity_id, state):
HASS_CACHE = initialize_hass_handler(conn)
api_key = HASS_CACHE['api_key']
base_url = HASS_CACHE['base_url']
'''
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':
action = ""
arg = ""
value = ""
if state["type"] == "turn_on":
description = "设备已打开"
if domain == "cover":
action = "open_cover"
@@ -84,91 +83,87 @@ async def handle_hass_set_state(conn, entity_id, state):
action = "start"
else:
action = "turn_on"
elif state['type'] == 'turn_off':
elif state["type"] == "turn_off":
description = "设备已关闭"
if domain == 'cover':
if domain == "cover":
action = "close_cover"
elif domain == 'vacuum':
elif domain == "vacuum":
action = "stop"
else:
action = "turn_off"
elif state['type'] == 'brightness_up':
elif state["type"] == "brightness_up":
description = "灯光已调亮"
action = 'turn_on'
arg = 'brightness_step_pct'
action = "turn_on"
arg = "brightness_step_pct"
value = 10
elif state['type'] == 'brightness_down':
elif state["type"] == "brightness_down":
description = "灯光已调暗"
action = 'turn_on'
arg = 'brightness_step_pct'
action = "turn_on"
arg = "brightness_step_pct"
value = -10
elif state['type'] == 'brightness_value':
elif state["type"] == "brightness_value":
description = f"亮度已调整到{state['input']}"
action = 'turn_on'
arg = 'brightness_pct'
value = state['input']
elif state['type'] == 'set_color':
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':
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':
action = "turn_on"
arg = "kelvin"
value = state["input"]
elif state["type"] == "volume_up":
description = "音量已调大"
action = state['type']
elif state['type'] == 'volume_down':
action = state["type"]
elif state["type"] == "volume_down":
description = "音量已调小"
action = state['type']
elif state['type'] == 'volume_set':
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':
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':
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':
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'
if domain == "media_player":
action = "media_play"
if domain == "vacuum":
action = "start"
else:
return f"{domain} {state.type}功能尚未支持"
if arg == '':
if arg == "":
data = {
"entity_id": entity_id,
}
else:
data = {
"entity_id": entity_id,
arg: value
}
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"
}
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
response = requests.post(url, headers=headers, json=data)
logger.bind(tag=TAG).info(f"设置状态:{description},url:{url},return_code:{response.status_code}")
logger.bind(tag=TAG).info(
f"设置状态:{description},url:{url},return_code:{response.status_code}"
)
if response.status_code == 200:
return description
else: