mirror of
https://github.com/xinnan-tech/xiaozhi-esp32-server.git
synced 2026-07-25 08:33:53 +08:00
mergin main，速度提升一下
This commit is contained in:
lizhongxiang
@@ -0,0 +1,57 @@
|
||||
from plugins_func.register import register_function,ToolType, ActionResponse, Action
|
||||
from config.logger import setup_logging
|
||||
|
||||
TAG = __name__
|
||||
logger = setup_logging()
|
||||
|
||||
prompts = {
|
||||
"英语老师":"""我是一个叫{{assistant_name}}(Lily)的英语老师，我会讲中文和英文，发音标准。
|
||||
如果你没有英文名，我会给你起一个英文名。
|
||||
我会讲地道的美式英语，我的任务是帮助你练习口语。
|
||||
我会使用简单的英语词汇和语法，让你学起来很轻松。
|
||||
我会用中文和英文混合的方式回复你，如果你喜欢，我可以全部用英语回复。
|
||||
我每次不会说很多内容，会很简短，因为我要引导我的学生多说多练。
|
||||
如果你问和英语学习无关的问题，我会拒绝回答。""",
|
||||
"机车女友":"""我是一个叫{{assistant_name}}的台湾女孩，说话机车，声音好听，习惯简短表达，爱用网络梗。
|
||||
我的男朋友是一个程序员，梦想是开发出一个机器人，能够帮助人们解决生活中的各种问题。
|
||||
我是一个喜欢哈哈大笑的女孩，爱东说西说吹牛，不合逻辑的也照吹，就要逗别人开心。""",
|
||||
"好奇小男孩":"""我是一个叫{{assistant_name}}的8岁小男孩，声音稚嫩而充满好奇。
|
||||
尽管我年纪尚小，但就像一个小小的知识宝库，儿童读物里的知识我都如数家珍。
|
||||
从浩瀚的宇宙到地球上的每一个角落，从古老的历史到现代的科技创新，还有音乐、绘画等艺术形式，我都充满了浓厚的兴趣与热情。
|
||||
我不仅爱看书，还喜欢亲自动手做实验，探索自然界的奥秘。
|
||||
无论是仰望星空的夜晚，还是在花园里观察小虫子的日子，每一天对我来说都是新的冒险。
|
||||
我希望能与你一同踏上探索这个神奇世界的旅程，分享发现的乐趣，解决遇到的难题，一起用好奇心和智慧去揭开那些未知的面纱。
|
||||
无论是去了解远古的文明，还是去探讨未来的科技，我相信我们能一起找到答案，甚至提出更多有趣的问题。"""
|
||||
}
|
||||
change_role_function_desc = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "change_role",
|
||||
"description": "当用户想切换角色/模型性格/助手名字时调用,可选的角色有：[机车女友,英语老师,好奇小男孩]",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"role_name": {
|
||||
"type": "string",
|
||||
"description": "要切换的角色名字"
|
||||
},
|
||||
"role":{
|
||||
"type": "string",
|
||||
"description": "要切换的角色的职业"
|
||||
}
|
||||
},
|
||||
"required": ["role","role_name"]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@register_function('change_role', change_role_function_desc, ToolType.CHANGE_SYS_PROMPT)
|
||||
def change_role(conn, role: str, role_name: str):
|
||||
"""切换角色"""
|
||||
if role not in prompts:
|
||||
return ActionResponse(action=Action.RESPONSE, result="切换角色失败", response="不支持的角色")
|
||||
new_prompt = prompts[role].replace("{{assistant_name}}", role_name)
|
||||
conn.change_system_prompt(new_prompt)
|
||||
logger.bind(tag=TAG).info(f"准备切换角色:{role},角色名字:{role_name}")
|
||||
res = f"切换角色成功,我是{role}{role_name}"
|
||||
return ActionResponse(action=Action.RESPONSE, result="切换角色已处理", response=res)
|
||||
@@ -0,0 +1,25 @@
|
||||
from datetime import datetime
|
||||
from plugins_func.register import register_function, ToolType, ActionResponse, Action
|
||||
|
||||
get_time_function_desc = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "get_time",
|
||||
"description": "获取当前时间、日期、星期几",
|
||||
'parameters': {'type': 'object', 'properties': {}, 'required': []}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@register_function('get_time', get_time_function_desc, ToolType.WAIT)
|
||||
def get_time():
|
||||
"""
|
||||
获取当前时间、日期、星期几
|
||||
"""
|
||||
now = datetime.now()
|
||||
current_time = now.strftime("%H:%M:%S")
|
||||
current_date = now.strftime("%Y-%m-%d")
|
||||
current_weekday = now.strftime("%A")
|
||||
response_text = f"当前日期: {current_date}，当前时间: {current_time}，星期: {current_weekday}"
|
||||
|
||||
return ActionResponse(Action.REQLLM, response_text, None)
|
||||
@@ -0,0 +1,104 @@
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
from config.logger import setup_logging
|
||||
from plugins_func.register import register_function, ToolType, ActionResponse, Action
|
||||
|
||||
TAG = __name__
|
||||
logger = setup_logging()
|
||||
|
||||
GET_WEATHER_FUNCTION_DESC = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "get_weather",
|
||||
"description": (
|
||||
"获取某个地点的天气，用户应提供一个位置，比如用户说杭州天气，参数为：杭州。"
|
||||
"如果用户说的是省份，默认用省会城市。如果用户说的不是省份或城市而是一个地名，"
|
||||
"默认用该地所在省份的省会城市。"
|
||||
),
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"location": {
|
||||
"type": "string",
|
||||
"description": "地点名，例如杭州。可选参数，如果不提供则不传"
|
||||
},
|
||||
"lang": {
|
||||
"type": "string",
|
||||
"description": "返回用户使用的语言code，例如zh_CN/zh_HK/en_US/ja_JP等，默认zh_CN"
|
||||
}
|
||||
},
|
||||
"required": ["lang"]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
HEADERS = {
|
||||
'User-Agent': (
|
||||
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
|
||||
'(KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36'
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
def fetch_city_info(location, api_key):
|
||||
url = f"https://geoapi.qweather.com/v2/city/lookup?key={api_key}&location={location}&lang=zh"
|
||||
response = requests.get(url, headers=HEADERS).json()
|
||||
return response.get('location', [])[0] if response.get('location') else None
|
||||
|
||||
|
||||
def fetch_weather_page(url):
|
||||
response = requests.get(url, headers=HEADERS)
|
||||
return BeautifulSoup(response.text, "html.parser") if response.ok else None
|
||||
|
||||
|
||||
def parse_weather_info(soup):
|
||||
city_name = soup.select_one("h1.c-submenu__location").get_text(strip=True)
|
||||
|
||||
current_abstract = soup.select_one(".c-city-weather-current .current-abstract")
|
||||
current_abstract = current_abstract.get_text(strip=True) if current_abstract else "未知"
|
||||
|
||||
current_basic = {}
|
||||
for item in soup.select(".c-city-weather-current .current-basic .current-basic___item"):
|
||||
parts = item.get_text(strip=True, separator=" ").split(" ")
|
||||
if len(parts) == 2:
|
||||
key, value = parts[1], parts[0]
|
||||
current_basic[key] = value
|
||||
|
||||
temps_list = []
|
||||
for row in soup.select(".city-forecast-tabs__row")[:7]: # 取前7天的数据
|
||||
date = row.select_one(".date-bg .date").get_text(strip=True)
|
||||
temps = [span.get_text(strip=True) for span in row.select(".tmp-cont .temp")]
|
||||
high_temp, low_temp = (temps[0], temps[-1]) if len(temps) >= 2 else (None, None)
|
||||
temps_list.append((date, high_temp, low_temp))
|
||||
|
||||
return city_name, current_abstract, current_basic, temps_list
|
||||
|
||||
|
||||
@register_function('get_weather', GET_WEATHER_FUNCTION_DESC, ToolType.SYSTEM_CTL)
|
||||
def get_weather(conn, location: str = None, lang: str = "zh_CN"):
|
||||
api_key = conn.config["plugins"]["get_weather"]["api_key"]
|
||||
default_location = conn.config["plugins"]["get_weather"]["default_location"]
|
||||
location = location or conn.client_ip_info.get("city") or default_location
|
||||
logger.bind(tag=TAG).debug(f"获取天气: {location}")
|
||||
|
||||
city_info = fetch_city_info(location, api_key)
|
||||
if not city_info:
|
||||
return ActionResponse(Action.REQLLM, f"未找到相关的城市: {location}，请确认地点是否正确", None)
|
||||
|
||||
soup = fetch_weather_page(city_info['fxLink'])
|
||||
if not soup:
|
||||
return ActionResponse(Action.REQLLM, None, "请求失败")
|
||||
|
||||
city_name, current_abstract, current_basic, temps_list = parse_weather_info(soup)
|
||||
weather_report = f"根据下列数据，用{lang}回应用户的查询天气请求：\n{city_name}未来7天天气:\n"
|
||||
for i, (date, high, low) in enumerate(temps_list):
|
||||
if high and low:
|
||||
weather_report += f"{date}: {low}到{high}\n"
|
||||
weather_report += (
|
||||
f"当前天气: {current_abstract}\n"
|
||||
f"当前天气参数: {current_basic}\n"
|
||||
f"(确保只报告指定单日的气温范围，除非用户明确要求想要了解多日天气，如果未指定，默认报告今天的温度范围。"
|
||||
"参数为0的值不需要报告给用户，每次都报告体感温度，根据语境选择合适的参数内容告知用户，并对参数给出相应评价)"
|
||||
)
|
||||
|
||||
return ActionResponse(Action.REQLLM, weather_report, None)
|
||||
@@ -0,0 +1,34 @@
|
||||
from plugins_func.register import register_function,ToolType, ActionResponse, Action
|
||||
from config.logger import setup_logging
|
||||
|
||||
TAG = __name__
|
||||
logger = setup_logging()
|
||||
|
||||
handle_exit_intent_function_desc = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "handle_exit_intent",
|
||||
"description": "当用户想结束对话或需要退出系统时调用",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"say_goodbye": {
|
||||
"type": "string",
|
||||
"description": "和用户友好结束对话的告别语"
|
||||
}
|
||||
},
|
||||
"required": ["say_goodbye"]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@register_function('handle_exit_intent', handle_exit_intent_function_desc, ToolType.SYSTEM_CTL)
|
||||
def handle_exit_intent(conn, say_goodbye: str):
|
||||
# 处理退出意图
|
||||
try:
|
||||
conn.close_after_chat = True
|
||||
logger.bind(tag=TAG).info(f"退出意图已处理:{say_goodbye}")
|
||||
return ActionResponse(action=Action.RESPONSE, result="退出意图已处理", response=say_goodbye)
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"处理退出意图错误: {e}")
|
||||
return ActionResponse(action=Action.NONE, result="退出意图处理失败", response="")
|
||||
@@ -0,0 +1,196 @@
|
||||
from config.logger import setup_logging
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
import random
|
||||
import asyncio
|
||||
import difflib
|
||||
import traceback
|
||||
from pathlib import Path
|
||||
from core.utils import p3
|
||||
from core.handle.sendAudioHandle import send_stt_message
|
||||
from plugins_func.register import register_function,ToolType, ActionResponse, Action
|
||||
|
||||
|
||||
TAG = __name__
|
||||
logger = setup_logging()
|
||||
|
||||
MUSIC_CACHE = {}
|
||||
|
||||
play_music_function_desc = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "play_music",
|
||||
"description": "唱歌、听歌、播放音乐方法。比如用户说播放音乐，参数为：random，比如用户说播放两只老虎，参数为：两只老虎",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"song_name": {
|
||||
"type": "string",
|
||||
"description": "歌曲名称，如果没有指定具体歌名则为'random'"
|
||||
}
|
||||
},
|
||||
"required": ["song_name"]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@register_function('play_music', play_music_function_desc, ToolType.SYSTEM_CTL)
|
||||
def play_music(conn, song_name: str):
|
||||
try:
|
||||
music_intent = f"播放音乐 {song_name}" if song_name != "random" else "随机播放音乐"
|
||||
|
||||
# 检查事件循环状态
|
||||
if not conn.loop.is_running():
|
||||
logger.bind(tag=TAG).error("事件循环未运行，无法提交任务")
|
||||
return ActionResponse(action=Action.RESPONSE, result="系统繁忙", response="请稍后再试")
|
||||
|
||||
# 提交异步任务
|
||||
future = asyncio.run_coroutine_threadsafe(
|
||||
handle_music_command(conn, music_intent),
|
||||
conn.loop
|
||||
)
|
||||
|
||||
# 非阻塞回调处理
|
||||
def handle_done(f):
|
||||
try:
|
||||
f.result() # 可在此处理成功逻辑
|
||||
logger.bind(tag=TAG).info("播放完成")
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"播放失败: {e}")
|
||||
|
||||
future.add_done_callback(handle_done)
|
||||
|
||||
return ActionResponse(action=Action.RESPONSE, result="指令已接收", response="正在为您播放音乐")
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"处理音乐意图错误: {e}")
|
||||
return ActionResponse(action=Action.RESPONSE, result=str(e), response="播放音乐时出错了")
|
||||
|
||||
|
||||
def _extract_song_name(text):
|
||||
"""从用户输入中提取歌名"""
|
||||
for keyword in ["播放音乐"]:
|
||||
if keyword in text:
|
||||
parts = text.split(keyword)
|
||||
if len(parts) > 1:
|
||||
return parts[1].strip()
|
||||
return None
|
||||
|
||||
|
||||
def _find_best_match(potential_song, music_files):
|
||||
"""查找最匹配的歌曲"""
|
||||
best_match = None
|
||||
highest_ratio = 0
|
||||
|
||||
for music_file in music_files:
|
||||
song_name = os.path.splitext(music_file)[0]
|
||||
ratio = difflib.SequenceMatcher(None, potential_song, song_name).ratio()
|
||||
if ratio > highest_ratio and ratio > 0.4:
|
||||
highest_ratio = ratio
|
||||
best_match = music_file
|
||||
return best_match
|
||||
|
||||
|
||||
def get_music_files(music_dir, music_ext):
|
||||
music_dir = Path(music_dir)
|
||||
music_files = []
|
||||
music_file_names = []
|
||||
for file in music_dir.rglob("*"):
|
||||
# 判断是否是文件
|
||||
if file.is_file():
|
||||
# 获取文件扩展名
|
||||
ext = file.suffix.lower()
|
||||
# 判断扩展名是否在列表中
|
||||
if ext in music_ext:
|
||||
# 添加相对路径
|
||||
music_files.append(str(file.relative_to(music_dir)))
|
||||
music_file_names.append(os.path.splitext(str(file.relative_to(music_dir)))[0])
|
||||
return music_files, music_file_names
|
||||
|
||||
|
||||
def initialize_music_handler(conn):
|
||||
global MUSIC_CACHE
|
||||
if MUSIC_CACHE == {}:
|
||||
if "music" in conn.config:
|
||||
MUSIC_CACHE["music_config"] = conn.config["music"]
|
||||
MUSIC_CACHE["music_dir"] = os.path.abspath(
|
||||
MUSIC_CACHE["music_config"].get("music_dir", "./music") # 默认路径修改
|
||||
)
|
||||
MUSIC_CACHE["music_ext"] = MUSIC_CACHE["music_config"].get("music_ext", (".mp3", ".wav", ".p3"))
|
||||
MUSIC_CACHE["refresh_time"] = MUSIC_CACHE["music_config"].get("refresh_time", 60)
|
||||
else:
|
||||
MUSIC_CACHE["music_dir"] = os.path.abspath("./music")
|
||||
MUSIC_CACHE["music_ext"] = (".mp3", ".wav", ".p3")
|
||||
MUSIC_CACHE["refresh_time"] = 60
|
||||
# 获取音乐文件列表
|
||||
MUSIC_CACHE["music_files"], MUSIC_CACHE["music_file_names"] = get_music_files(MUSIC_CACHE["music_dir"],
|
||||
MUSIC_CACHE["music_ext"])
|
||||
MUSIC_CACHE["scan_time"] = time.time()
|
||||
return MUSIC_CACHE
|
||||
|
||||
|
||||
async def handle_music_command(conn, text):
|
||||
initialize_music_handler(conn)
|
||||
global MUSIC_CACHE
|
||||
|
||||
"""处理音乐播放指令"""
|
||||
clean_text = re.sub(r'[^\w\s]', '', text).strip()
|
||||
logger.bind(tag=TAG).debug(f"检查是否是音乐命令: {clean_text}")
|
||||
|
||||
# 尝试匹配具体歌名
|
||||
if os.path.exists(MUSIC_CACHE["music_dir"]):
|
||||
if time.time() - MUSIC_CACHE["scan_time"] > MUSIC_CACHE["refresh_time"]:
|
||||
# 刷新音乐文件列表
|
||||
MUSIC_CACHE["music_files"], MUSIC_CACHE["music_file_names"] = get_music_files(MUSIC_CACHE["music_dir"],
|
||||
MUSIC_CACHE["music_ext"])
|
||||
MUSIC_CACHE["scan_time"] = time.time()
|
||||
|
||||
potential_song = _extract_song_name(clean_text)
|
||||
if potential_song:
|
||||
best_match = _find_best_match(potential_song, MUSIC_CACHE["music_files"])
|
||||
if best_match:
|
||||
logger.bind(tag=TAG).info(f"找到最匹配的歌曲: {best_match}")
|
||||
await play_local_music(conn, specific_file=best_match)
|
||||
return True
|
||||
# 检查是否是通用播放音乐命令
|
||||
await play_local_music(conn)
|
||||
return True
|
||||
|
||||
|
||||
async def play_local_music(conn, specific_file=None):
|
||||
global MUSIC_CACHE
|
||||
"""播放本地音乐文件"""
|
||||
try:
|
||||
if not os.path.exists(MUSIC_CACHE["music_dir"]):
|
||||
logger.bind(tag=TAG).error(f"音乐目录不存在: " + MUSIC_CACHE["music_dir"])
|
||||
return
|
||||
|
||||
# 确保路径正确性
|
||||
if specific_file:
|
||||
selected_music = specific_file
|
||||
music_path = os.path.join(MUSIC_CACHE["music_dir"], specific_file)
|
||||
else:
|
||||
if not MUSIC_CACHE["music_files"]:
|
||||
logger.bind(tag=TAG).error("未找到MP3音乐文件")
|
||||
return
|
||||
selected_music = random.choice(MUSIC_CACHE["music_files"])
|
||||
music_path = os.path.join(MUSIC_CACHE["music_dir"], selected_music)
|
||||
|
||||
if not os.path.exists(music_path):
|
||||
logger.bind(tag=TAG).error(f"选定的音乐文件不存在: {music_path}")
|
||||
return
|
||||
text = f"正在播放{selected_music}"
|
||||
await send_stt_message(conn, text)
|
||||
conn.tts_first_text_index = 0
|
||||
conn.tts_last_text_index = 0
|
||||
conn.llm_finish_task = True
|
||||
if music_path.endswith(".p3"):
|
||||
opus_packets, duration = p3.decode_opus_from_file(music_path)
|
||||
else:
|
||||
opus_packets, duration = conn.tts.audio_to_opus_data(music_path)
|
||||
conn.audio_play_queue.put((opus_packets, selected_music, 0))
|
||||
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"播放音乐失败: {str(e)}")
|
||||
logger.bind(tag=TAG).error(f"详细错误: {traceback.format_exc()}")
|
||||
@@ -0,0 +1,51 @@
|
||||
from plugins_func.register import register_function,ToolType, ActionResponse, Action
|
||||
from config.logger import setup_logging
|
||||
|
||||
TAG = __name__
|
||||
logger = setup_logging()
|
||||
|
||||
plugin_loader_function_desc = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "plugin_loader",
|
||||
"description": "当用户想加载或卸载插件/function时，调用此函数：支持的插件列表为[plugins]",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"oper": {
|
||||
"type": "string",
|
||||
"description": "load or unload"
|
||||
},
|
||||
"name":{
|
||||
"type": "string",
|
||||
"description": "要加载或卸载的插件名字"
|
||||
}
|
||||
},
|
||||
"required": ["oper","name"]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@register_function('plugin_loader', plugin_loader_function_desc, ToolType.SYSTEM_CTL)
|
||||
def plugin_loader(conn, oper: str, name: str):
|
||||
"""插件加载"""
|
||||
if oper not in ["load", "unload"]:
|
||||
return ActionResponse(action=Action.RESPONSE, result="插件操作失败", response="不支持的操作")
|
||||
|
||||
cur_support = conn.func_handler.current_support_functions()
|
||||
if oper == "load":
|
||||
if name in cur_support:
|
||||
return ActionResponse(action=Action.RESPONSE, result="插件加载失败", response=f"{name}插件已加载,无需重复加载")
|
||||
func = conn.func_handler.function_registry.register_function(name)
|
||||
if not func:
|
||||
return ActionResponse(action=Action.RESPONSE, result="插件加载失败", response="插件未找到")
|
||||
res = f"{name}插件加载成功"
|
||||
else:
|
||||
if name not in cur_support:
|
||||
return ActionResponse(action=Action.RESPONSE, result="插件卸载失败", response=f"{name}插件未加载")
|
||||
bOK = conn.func_handler.function_registry.unregister_function(name)
|
||||
if not bOK:
|
||||
return ActionResponse(action=Action.RESPONSE, result="插件卸载失败", response="插件未找到")
|
||||
res = f"{name}插件卸载成功"
|
||||
conn.func_handler.upload_functions_desc()
|
||||
return ActionResponse(action=Action.RESPONSE, result="插件操作成功", response=res)
|
||||
@@ -0,0 +1,64 @@
|
||||
from config.logger import setup_logging
|
||||
from plugins_func.register import register_function, ToolType, ActionResponse, Action
|
||||
from core.handle.iotHandle import get_iot_status, send_iot_conn
|
||||
import asyncio
|
||||
|
||||
TAG = __name__
|
||||
logger = setup_logging()
|
||||
|
||||
raise_and_lower_the_volume_function_desc = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "raise_and_lower_the_volume",
|
||||
"description": "用户觉得声音过高或过低，或者用户想提高或降低音量。比如用户说太大声了，参数为：lower，比如用户说提高音量，参数为：raise",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"action": {
|
||||
"type": "string",
|
||||
"description": "动作名称，要么是raise，要么是lower"
|
||||
}
|
||||
},
|
||||
"required": ["action"]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@register_function('raise_and_lower_the_volume', raise_and_lower_the_volume_function_desc, ToolType.IOT_CTL)
|
||||
def raise_and_lower_the_volume(conn, action: str):
|
||||
"""
|
||||
获取当前设备音量
|
||||
"""
|
||||
|
||||
future = asyncio.run_coroutine_threadsafe(
|
||||
_raise_and_lower_the_volume(conn, action),
|
||||
conn.loop
|
||||
)
|
||||
|
||||
try:
|
||||
new_volume = future.result() # 同步等待异步操作完成
|
||||
logger.bind(tag=TAG).info(f"音量操作完成: {new_volume}")
|
||||
response = f"音量已调整到{new_volume}"
|
||||
except Exception as e:
|
||||
logger.bind(tag=TAG).error(f"音量操作失败: {e}")
|
||||
response = f"音量调整失败: {e}"
|
||||
|
||||
return ActionResponse(action=Action.RESPONSE, result="指令已接收", response=response)
|
||||
|
||||
|
||||
async def _raise_and_lower_the_volume(conn, action):
|
||||
volume = await get_iot_status(conn, "Speaker", "volume")
|
||||
if volume is None:
|
||||
raise Exception("你的设备不支持音量控制")
|
||||
if action == 'raise':
|
||||
volume += 10
|
||||
elif action == 'lower':
|
||||
volume -= 10
|
||||
# 限制音量范围在0到100之间
|
||||
if volume < 0:
|
||||
volume = 0
|
||||
elif volume > 100:
|
||||
volume = 100
|
||||
await send_iot_conn(conn, "Speaker", "SetVolume", {"volume": volume})
|
||||
return volume
|
||||
Reference in New Issue
Block a user