Pine/DPM
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat: 增加 screen_id 参数以支持针对特定屏幕的工具执行，优化大屏控制逻辑

Browse Source
This commit is contained in:
Pine
2026-08-19 14:28:10 +08:00
parent e72442dae0
commit 3688d5d0b1
7 changed files with 52 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/ai_tools.py
+12 -12
View File
@@ -14,34 +14,34 @@ log = logging.getLogger("dpm.ai_tools")
# ---------------- 工具执行 ----------------
def execute_tool(tool):
"""执行一个工具调用：发布 MQTT 控制指令，返回是否发布成功"""
def execute_tool(tool, screen_id=""):
"""执行一个工具调用：发布 MQTT 控制指令（带发起屏幕 screen_id，只发到该屏），返回是否发布成功"""
t = tool.get("type")
params = tool.get("params") or {}
if t == "navigate":
return hub.publish_command("navigate", {"page": params.get("page", "home")}).get("published", False)
return hub.publish_command("navigate", {"page": params.get("page", "home")}, screen_id=screen_id).get("published", False)
if t == "control":
action = params.get("action")
if action in ("play", "pause", "next", "prev"):
return hub.publish_command(action).get("published", False)
return hub.publish_command(action, screen_id=screen_id).get("published", False)
if action == "set_mode":
return hub.publish_command("set_mode", {"mode": params.get("mode", "auto")}).get("published", False)
return hub.publish_command("set_mode", {"mode": params.get("mode", "auto")}, screen_id=screen_id).get("published", False)
return False
if t == "alert":
return hub.publish_command("alert", {"title": params.get("title", "提示"), "content": params.get("content", "")}).get("published", False)
return hub.publish_command("alert", {"title": params.get("title", "提示"), "content": params.get("content", "")}, screen_id=screen_id).get("published", False)
if t == "show_card":
return hub.publish_command("show_card", params).get("published", False)
return hub.publish_command("show_card", params, screen_id=screen_id).get("published", False)
return False
def run_tools(tools):
"""执行一组工具，返回 [{type, ok}] 并汇总是否全部发布成功"""
def run_tools(tools, screen_id=""):
"""执行一组工具（带发起屏幕 screen_id），返回 [{type, ok}] 并汇总是否全部发布成功"""
results = []
all_ok = True
for tool in tools or []:
ok = bool(execute_tool(tool))
log.info("ai_tools: 执行工具 %s params=%s -> %s", tool.get("type"),
json.dumps(tool.get("params") or {}, ensure_ascii=False), ok)
ok = bool(execute_tool(tool, screen_id))
log.info("ai_tools: 执行工具 %s params=%s screen=%s -> %s", tool.get("type"),
json.dumps(tool.get("params") or {}, ensure_ascii=False), screen_id or "all", ok)
results.append({"type": tool.get("type"), "ok": ok})
all_ok = all_ok and ok
return results, all_ok