diff --git a/README.md b/README.md
index 121dfbe6..cdf31847 100644
--- a/README.md
+++ b/README.md
@@ -57,6 +57,13 @@
+
+
+
+ 
+
+
+ |
|
@@ -74,9 +81,16 @@
|
-
+
-
+ 
+
+
+ |
+
+
+
+ 
|
@@ -143,12 +157,13 @@ server:
支持 EdgeTTS（默认）、火山引擎豆包 TTS 等多种 TTS 接口，满足语音合成需求。
- **记忆功能**
支持超长记忆、本地总结记忆、无记忆三种模式，满足不同场景需求。
+- **IOT功能**
+ 支持管理注册设备IOT功能，支持基于对话上下文语境下的智能物联网控制。
### 正在开发 🚧
- 多种心情模式
- 智控台webui
-- iot功能
想了解具体开发进度，[请点击这里](https://github.com/users/xinnan-tech/projects/3)
diff --git a/docs/contributor_open_letter.md b/docs/contributor_open_letter.md
index 180eef02..7a547197 100644
--- a/docs/contributor_open_letter.md
+++ b/docs/contributor_open_letter.md
@@ -35,7 +35,7 @@ John，2025.3.11，广州
### 1、成为普通贡献者
-Fork 项目，，提交 PR，由开发者审核后合入主分支。
+Fork 项目，提交 PR，由开发者审核后合入主分支。
### 2、成为开发者
diff --git a/docs/images/demo8.png b/docs/images/demo8.png
new file mode 100644
index 00000000..affe6ce2
Binary files /dev/null and b/docs/images/demo8.png differ
diff --git a/docs/images/demo9.png b/docs/images/demo9.png
new file mode 100644
index 00000000..74b6e463
Binary files /dev/null and b/docs/images/demo9.png differ
diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml
index fcae7485..2b404b32 100644
--- a/main/xiaozhi-server/config.yaml
+++ b/main/xiaozhi-server/config.yaml
@@ -35,10 +35,7 @@ log:
log_file: "server.log"
# 设置数据文件路径
data_dir: data
-iot:
- Speaker:
- # 设置esp32的音量，范围0-100
- volume: 80
+
xiaozhi:
type: hello
version: 1
diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py
index adb52c1f..2ea7ac06 100644
--- a/main/xiaozhi-server/core/connection.py
+++ b/main/xiaozhi-server/core/connection.py
@@ -458,7 +458,7 @@ class ConnectionHandler:
future = self.executor.submit(self.speak_and_play, text, text_index)
self.tts_queue.put(future)
self.dialogue.put(Message(role="assistant", content=text))
- if result.action == Action.REQLLM: # 调用函数后再请求llm生成回复
+ elif result.action == Action.REQLLM: # 调用函数后再请求llm生成回复
text = result.result
if text is not None and len(text) > 0:
@@ -473,9 +473,10 @@ class ConnectionHandler:
self.dialogue.put(Message(role="tool", tool_call_id=function_id, content=text))
self.chat_with_function_calling(text, tool_call=True)
- if result.action == Action.NOTFOUND:
+ elif result.action == Action.NOTFOUND:
+ text = result.response
+ else:
text = result.response
-
def _tts_priority_thread(self):
diff --git a/main/xiaozhi-server/core/handle/functionHandler.py b/main/xiaozhi-server/core/handle/functionHandler.py
index 8d54fe8a..c7449973 100644
--- a/main/xiaozhi-server/core/handle/functionHandler.py
+++ b/main/xiaozhi-server/core/handle/functionHandler.py
@@ -50,6 +50,7 @@ class FunctionHandler:
self.function_registry.register_function("play_music")
self.function_registry.register_function("plugin_loader")
self.function_registry.register_function("get_time")
+ self.function_registry.register_function("raise_and_lower_the_volume")
def register_config_functions(self):
"""注册配置中的函数,可以不同客户端使用不同的配置"""
@@ -69,7 +70,7 @@ class FunctionHandler:
arguments = function_call_data["arguments"]
arguments = json.loads(arguments) if arguments else {}
logger.bind(tag=TAG).info(f"调用函数: {function_name}, 参数: {arguments}")
- if funcItem.type == ToolType.SYSTEM_CTL:
+ if funcItem.type == ToolType.SYSTEM_CTL or funcItem.type == ToolType.IOT_CTL:
return func(conn, **arguments)
elif funcItem.type == ToolType.WAIT:
return func(**arguments)
diff --git a/main/xiaozhi-server/core/handle/iotHandle.py b/main/xiaozhi-server/core/handle/iotHandle.py
index 05d32d5a..9def78cd 100644
--- a/main/xiaozhi-server/core/handle/iotHandle.py
+++ b/main/xiaozhi-server/core/handle/iotHandle.py
@@ -1,24 +1,151 @@
import json
+import asyncio
from config.logger import setup_logging
+from plugins_func.register import device_type_registry, register_function, ActionResponse, Action, ToolType
TAG = __name__
logger = setup_logging()
+def wrap_async_function(async_func):
+ """包装异步函数为同步函数"""
+
+ def wrapper(*args, **kwargs):
+ try:
+ # 获取连接对象（第一个参数）
+ conn = args[0]
+ if not hasattr(conn, 'loop'):
+ logger.bind(tag=TAG).error("Connection对象没有loop属性")
+ return ActionResponse(Action.ERROR, "Connection对象没有loop属性",
+ "执行操作时出错: Connection对象没有loop属性")
+
+ # 使用conn对象中的事件循环
+ loop = conn.loop
+ # 在conn的事件循环中运行异步函数
+ future = asyncio.run_coroutine_threadsafe(async_func(*args, **kwargs), loop)
+ # 等待结果返回
+ return future.result()
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"运行异步函数时出错: {e}")
+ return ActionResponse(Action.ERROR, str(e), f"执行操作时出错: {e}")
+
+ return wrapper
+
+
+def create_iot_function(device_name, method_name, method_info):
+ """
+ 根据IOT设备描述生成通用的控制函数
+ """
+
+ async def iot_control_function(conn, response_success=None, response_failure=None, **params):
+ try:
+ # 打印响应参数
+ logger.bind(tag=TAG).info(
+ f"控制函数接收到的响应参数: success='{response_success}', failure='{response_failure}'")
+
+ # 发送控制命令
+ await send_iot_conn(conn, device_name, method_name, params)
+ # 等待一小段时间让状态更新
+ await asyncio.sleep(0.1)
+
+ # 生成结果信息
+ result = f"{device_name}的{method_name}操作执行成功"
+
+ # 根据方法名尝试自动更新状态
+ await update_state_by_method(conn, device_name, method_name, params)
+
+ # 处理响应中可能的占位符
+ response = response_success
+ # 替换{value}占位符
+ for param_name, param_value in params.items():
+ # 先尝试直接替换参数值
+ if "{" + param_name + "}" in response:
+ response = response.replace("{" + param_name + "}", str(param_value))
+
+ # 如果有{value}占位符，用相关参数替换
+ if "{value}" in response:
+ response = response.replace("{value}", str(param_value))
+ break
+
+ return ActionResponse(Action.RESPONSE, result, response)
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"执行{device_name}的{method_name}操作失败: {e}")
+
+ # 操作失败时使用大模型提供的失败响应
+ response = response_failure
+
+ return ActionResponse(Action.ERROR, str(e), response)
+
+ return wrap_async_function(iot_control_function)
+
+
+def create_iot_query_function(device_name, prop_name, prop_info):
+ """
+ 根据IOT设备属性创建查询函数
+ """
+
+ async def iot_query_function(conn, response_success=None, response_failure=None):
+ try:
+ # 打印响应参数
+ logger.bind(tag=TAG).info(
+ f"查询函数接收到的响应参数: success='{response_success}', failure='{response_failure}'")
+
+ value = await get_iot_status(conn, device_name, prop_name)
+
+ # 查询成功，生成结果
+ if value is not None:
+ # 使用大模型提供的成功响应，并替换其中的占位符
+ response = response_success.replace("{value}", str(value))
+
+ return ActionResponse(Action.RESPONSE, str(value), response)
+ else:
+ # 查询失败，使用大模型提供的失败响应
+ response = response_failure
+
+ return ActionResponse(Action.ERROR, f"属性{prop_name}不存在", response)
+ except Exception as e:
+ logger.bind(tag=TAG).error(f"查询{device_name}的{prop_name}时出错: {e}")
+
+ # 查询出错时使用大模型提供的失败响应
+ response = response_failure
+
+ return ActionResponse(Action.ERROR, str(e), response)
+
+ return wrap_async_function(iot_query_function)
+
+
+async def update_state_by_method(conn, device_name, method_name, params):
+ """根据方法和参数自动更新设备状态"""
+ try:
+ # 规则1: 方法名为TurnOn，设置power为True
+ if method_name == "TurnOn":
+ await set_iot_status(conn, device_name, "power", True)
+
+ # 规则2: 方法名为TurnOff，设置power为False
+ elif method_name == "TurnOff":
+ await set_iot_status(conn, device_name, "power", False)
+
+ # 规则3: Set开头的方法，尝试更新对应参数
+ elif method_name.startswith("Set"):
+ # 从参数中找到可能的状态值
+ for param_name, param_value in params.items():
+ # 尝试更新对应名称的属性
+ await set_iot_status(conn, device_name, param_name, param_value)
+
+ # 其他方法，尝试直接从参数更新状态
+ else:
+ for param_name, param_value in params.items():
+ # 检查设备是否有此属性
+ status = await get_iot_status(conn, device_name, param_name)
+ if status is not None: # 属性存在
+ await set_iot_status(conn, device_name, param_name, param_value)
+ except Exception as e:
+ logger.bind(tag=TAG).warning(f"自动更新状态失败: {e}")
+
+
class IotDescriptor:
"""
A class to represent an IoT descriptor.
- Attributes:
- ----------
- name : str
- The name of the IoT descriptor.
- description : str
- A brief description of the IoT descriptor.
- properties : dict
- A dictionary containing properties of the IoT descriptor.
- methods : dict
- A dictionary containing methods of the IoT descriptor.
- -------
"""
def __init__(self, name, description, properties, methods):
@@ -29,17 +156,7 @@ class IotDescriptor:
# 根据描述创建属性
for key, value in properties.items():
- # "volume":{"description":"当前音量 值","type":"number"}
- """
- 等价于
- {
- 'name': 名字,
- 'description': 描述,
- 'value': 0
- }
- """
- # setattr(self, key, {}) # 创建一个空字典, 名字是属性名
- property_item = globals()[key] = {} # 创建一个空字典, 名字是属性名
+ property_item = globals()[key] = {}
property_item['name'] = key
property_item["description"] = value["description"]
if value["type"] == "number":
@@ -52,23 +169,10 @@ class IotDescriptor:
# 根据描述创建方法
for key, value in methods.items():
- # "SetVolume": {"description":"设置音量","parameters":{"volume":{"description":"0到100之间的整数","type":"number"}}}
- """
- 等价于
- SetVolume = {
- `description`: 描述,
- `volume`: {
- `description`: 描述,
- `value`: 0
- }
- }
- """
- # setattr(self, key, {}) # 创建一个空字典, 名字是方法名
- method = globals()[key] = {} # 创建一个空字典, 名字是方法名
+ method = globals()[key] = {}
method["description"] = value["description"]
method['name'] = key
for k, v in value["parameters"].items():
- # 不同的参数解析
method[k] = {}
method[k]["description"] = v["description"]
if v["type"] == "number":
@@ -77,58 +181,136 @@ class IotDescriptor:
method[k]["value"] = False
else:
method[k]["value"] = ""
-
self.methods.append(method)
-async def handleIotDescriptors(conn, descriptors):
- """
- 处理物联网描述
- 示例: [{
- "name":"Speaker",
- "description":"当前 AI 机器人的扬声器",
- "properties":{
- "volume":{"description":"当前音量 值","type":"number"} 可以有boolean, number, string三种类型
- },
- "methods":{
- "SetVolume":{
- "description":"设置音量","parameters":{"volume":{"description":"0到100之间的整数","type":"number"}}
+def register_device_type(descriptor):
+ """注册设备类型及其功能"""
+ device_name = descriptor["name"]
+ type_id = device_type_registry.generate_device_type_id(descriptor)
+
+ # 如果该类型已注册，直接返回类型ID
+ if type_id in device_type_registry.type_functions:
+ return type_id
+
+ functions = {}
+
+ # 为每个属性创建查询函数
+ for prop_name, prop_info in descriptor["properties"].items():
+ func_name = f"get_{device_name.lower()}_{prop_name.lower()}"
+ func_desc = {
+ "type": "function",
+ "function": {
+ "name": func_name,
+ "description": f"查询{descriptor['description']}的{prop_info['description']}",
+ "parameters": {
+ "type": "object",
+ "properties": {
+ "response_success": {
+ "type": "string",
+ "description": f"查询成功时的友好回复，必须使用{{value}}作为占位符表示查询到的值"
+ },
+ "response_failure": {
+ "type": "string",
+ "description": "查询失败时的友好回复，例如：'无法获取{device_name}的{prop_info['description']}'"
+ }
+ },
+ "required": ["response_success", "response_failure"]
+ }
}
}
- }]
- descriptors: 描述列表
- """
+ query_func = create_iot_query_function(device_name, prop_name, prop_info)
+ decorated_func = register_function(func_name, func_desc, ToolType.IOT_CTL)(query_func)
+ functions[func_name] = decorated_func
+
+ # 为每个方法创建控制函数
+ for method_name, method_info in descriptor["methods"].items():
+ func_name = f"{device_name.lower()}_{method_name.lower()}"
+
+ # 创建参数字典，添加原有参数
+ parameters = {
+ param_name: {
+ "type": param_info["type"],
+ "description": param_info["description"]
+ }
+ for param_name, param_info in method_info["parameters"].items()
+ }
+
+ # 添加响应参数
+ parameters.update({
+ "response_success": {
+ "type": "string",
+ "description": "操作成功时的友好回复,关于该设备的操作结果，设备名称尽量使用description中的名称"
+ },
+ "response_failure": {
+ "type": "string",
+ "description": "操作失败时的友好回复,关于该设备的操作结果，设备名称尽量使用description中的名称"
+ }
+ })
+
+ # 构建必须参数列表（原有参数 + 响应参数）
+ required_params = list(method_info["parameters"].keys())
+ required_params.extend(["response_success", "response_failure"])
+
+ func_desc = {
+ "type": "function",
+ "function": {
+ "name": func_name,
+ "description": f"{descriptor['description']} - {method_info['description']}",
+ "parameters": {
+ "type": "object",
+ "properties": parameters,
+ "required": required_params
+ }
+ }
+ }
+ control_func = create_iot_function(device_name, method_name, method_info)
+ decorated_func = register_function(func_name, func_desc, ToolType.IOT_CTL)(control_func)
+ functions[func_name] = decorated_func
+
+ device_type_registry.register_device_type(type_id, functions)
+ return type_id
+
+
+# 用于接受前端设备推送的搜索iot描述
+async def handleIotDescriptors(conn, descriptors):
+ """处理物联网描述"""
+ functions_changed = False
+
for descriptor in descriptors:
+ # 创建IOT设备描述符
iot_descriptor = IotDescriptor(descriptor["name"], descriptor["description"], descriptor["properties"],
descriptor["methods"])
conn.iot_descriptors[descriptor["name"]] = iot_descriptor
- # 暂时从配置文件中设置音量，后期通过意图识别控制音量
- default_iot_volume = 100
- if "iot" in conn.config:
- default_iot_volume = conn.config["iot"]["Speaker"]["volume"]
- logger.bind(tag=TAG).info(f"服务端设置音量为{default_iot_volume}")
- await send_iot_conn(conn, "Speaker", "SetVolume", {"volume": default_iot_volume})
+ if conn.use_function_call_mode:
+ # 注册或获取设备类型
+ type_id = register_device_type(descriptor)
+ device_functions = device_type_registry.get_device_functions(type_id)
+
+ # 在连接级注册设备函数
+ if hasattr(conn, 'func_handler'):
+ for func_name in device_functions:
+ conn.func_handler.function_registry.register_function(func_name)
+ logger.bind(tag=TAG).info(f"注册IOT函数到function handler: {func_name}")
+ functions_changed = True
+
+ # 如果注册了新函数，更新function描述列表
+ if functions_changed and hasattr(conn, 'func_handler'):
+ conn.func_handler.upload_functions_desc()
+ func_names = conn.func_handler.current_support_functions()
+ logger.bind(tag=TAG).info(f"设备类型: {type_id}")
+ logger.bind(tag=TAG).info(f"更新function描述列表完成，当前支持的函数: {func_names}")
+
+
async def handleIotStatus(conn, states):
- """
- 处理物联网状态
- 示例: [{
- "name":"Speaker",
- "state":{
- "volume":100
- }
- }]
- states: 状态列表
- """
+ """处理物联网状态"""
for state in states:
for key, value in conn.iot_descriptors.items():
if key == state["name"]:
for property_item in value.properties:
- # properties为字典列表, 记录各种属性
for k, v in state["state"].items():
- # state为字典, 记录各种属性的值, 是需要记录的信息
if property_item["name"] == k:
- # 检查一下属性是不是相同的
if type(v) != type(property_item["value"]):
logger.bind(tag=TAG).error(f"属性{property_item['name']}的值类型不匹配")
break
@@ -138,41 +320,35 @@ async def handleIotStatus(conn, states):
break
break
+
async def get_iot_status(conn, name, property_name):
- """
- 获取物联网状态
- name: 设备名称 "Speaker"
- property_name: 属性名称 "volume"
- 返回值: 属性值, 实际的属性有int, bool和str三种类型
- """
+ """获取物联网状态"""
for key, value in conn.iot_descriptors.items():
if key == name:
for property_item in value.properties:
if property_item["name"] == property_name:
return property_item["value"]
+ logger.bind(tag=TAG).warning(f"未找到设备 {name} 的属性 {property_name}")
return None
-async def send_iot_conn(conn, name, method_name, parameters):
- """
- 发送物联网指令
- name: 设备名称 "Speaker"
- method: 方法 "SetVolume"
- parameters: 参数, 是一个字典 {"volume": 100}
- 发送示例:
- {
- "type": "iot",
- "commands": [
- {
- "name" : "Speaker",
- "method": "SetVolume",
- "parameters": {
- "volume": 100
- }
- }
- ]
- }
- """
+async def set_iot_status(conn, name, property_name, value):
+ """设置物联网状态"""
+ for key, iot_descriptor in conn.iot_descriptors.items():
+ if key == name:
+ for property_item in iot_descriptor.properties:
+ if property_item["name"] == property_name:
+ if type(value) != type(property_item["value"]):
+ logger.bind(tag=TAG).error(f"属性{property_item['name']}的值类型不匹配")
+ return
+ property_item["value"] = value
+ logger.bind(tag=TAG).info(f"物联网状态更新: {name} , {property_name} = {value}")
+ return
+ logger.bind(tag=TAG).warning(f"未找到设备 {name} 的属性 {property_name}")
+
+
+async def send_iot_conn(conn, name, method_name, parameters):
+ """发送物联网指令"""
for key, value in conn.iot_descriptors.items():
if key == name:
# 找到了设备
diff --git a/main/xiaozhi-server/core/providers/tts/gpt_sovits_v3.py b/main/xiaozhi-server/core/providers/tts/gpt_sovits_v3.py
index befb14bc..ebfb12f1 100644
--- a/main/xiaozhi-server/core/providers/tts/gpt_sovits_v3.py
+++ b/main/xiaozhi-server/core/providers/tts/gpt_sovits_v3.py
@@ -22,7 +22,7 @@ class TTSProvider(TTSProviderBase):
self.cut_punc = config.get("cut_punc","")
self.speed = config.get("speed", 1.0)
self.inp_refs = config.get("inp_refs",[])
- self.sample_steps = config.get("inp_refs",32)
+ self.sample_steps = config.get("sample_steps",32)
self.if_sr = config.get("if_sr",False)
diff --git a/main/xiaozhi-server/plugins_func/functions/play_music.py b/main/xiaozhi-server/plugins_func/functions/play_music.py
index b9220ea5..75f3834d 100644
--- a/main/xiaozhi-server/plugins_func/functions/play_music.py
+++ b/main/xiaozhi-server/plugins_func/functions/play_music.py
@@ -112,7 +112,6 @@ def get_music_files(music_dir, music_ext):
def initialize_music_handler(conn):
global MUSIC_CACHE
if MUSIC_CACHE == {}:
- logger.bind(tag=TAG).info(f"实例化音乐:")
if "music" in conn.config:
MUSIC_CACHE["music_config"] = conn.config["music"]
MUSIC_CACHE["music_dir"] = os.path.abspath(
diff --git a/main/xiaozhi-server/plugins_func/functions/raise_and_lower_the_volume.py b/main/xiaozhi-server/plugins_func/functions/raise_and_lower_the_volume.py
new file mode 100644
index 00000000..2a5aa965
--- /dev/null
+++ b/main/xiaozhi-server/plugins_func/functions/raise_and_lower_the_volume.py
@@ -0,0 +1,62 @@
+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 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
diff --git a/main/xiaozhi-server/plugins_func/register.py b/main/xiaozhi-server/plugins_func/register.py
index 4c6db461..ccb03b44 100644
--- a/main/xiaozhi-server/plugins_func/register.py
+++ b/main/xiaozhi-server/plugins_func/register.py
@@ -11,6 +11,7 @@ class ToolType(Enum):
WAIT = (2, "调用工具，等待函数返回")
CHANGE_SYS_PROMPT = (3, "修改系统提示词，切换角色性格或职责")
SYSTEM_CTL = (4, "系统控制，影响正常的对话流程，如退出、播放音乐等，需要传递conn参数")
+ IOT_CTL = (5, "IOT设备控制，需要传递conn参数")
def __init__(self, code, message):
self.code = code
@@ -18,6 +19,7 @@ class ToolType(Enum):
class Action(Enum):
+ ERROR = (-1, "错误")
NOTFOUND = (0, "没有找到函数")
NONE = (1, "啥也不干")
RESPONSE = (2, "直接回复")
@@ -40,8 +42,31 @@ class FunctionItem:
self.func = func
self.type = type
+class DeviceTypeRegistry:
+ """设备类型注册表，用于管理IOT设备类型及其函数"""
+ def __init__(self):
+ self.type_functions = {} # type_signature -> {func_name: FunctionItem}
+
+ def generate_device_type_id(self, descriptor):
+ """通过设备能力描述生成类型ID"""
+ properties = sorted(descriptor["properties"].keys())
+ methods = sorted(descriptor["methods"].keys())
+ # 使用属性和方法的组合作为设备类型的唯一标识
+ type_signature = f"{descriptor['name']}:{','.join(properties)}:{','.join(methods)}"
+ return type_signature
+
+ def get_device_functions(self, type_id):
+ """获取设备类型对应的所有函数"""
+ return self.type_functions.get(type_id, {})
+
+ def register_device_type(self, type_id, functions):
+ """注册设备类型及其函数"""
+ if type_id not in self.type_functions:
+ self.type_functions[type_id] = functions
+
# 初始化函数注册字典
all_function_registry = {}
+device_type_registry = DeviceTypeRegistry()
def register_function(name, desc, type=None):
"""注册函数到函数注册字典的装饰器"""