Spaces:
Sleeping
Sleeping
| """Exécution des outils Gemini Live — port Mark-XLVIII.""" | |
| from __future__ import annotations | |
| import asyncio | |
| import os | |
| import threading | |
| import time | |
| import traceback | |
| from typing import Any, Callable | |
| from google.genai import types | |
| from emo.desktop.actions.skill_loader import run_skill | |
| from emo.desktop.core.memory_manager import update_memory | |
| # Outil Gemini → skill emo.desktop.actions | |
| _TOOL_TO_SKILL: dict[str, str] = { | |
| "open_app": "open_app", | |
| "web_search": "web_search", | |
| "weather_report": "weather_report", | |
| "send_message": "send_message", | |
| "reminder": "reminder", | |
| "youtube_video": "youtube_video", | |
| "screen_processor": "screen_processor", | |
| "computer_settings": "computer_settings", | |
| "browser_control": "browser_control", | |
| "file_controller": "file_controller", | |
| "desktop_control": "desktop_control", | |
| "code_helper": "code_helper", | |
| "dev_agent_skill": "dev_agent_skill", | |
| "fallback_agent": "fallback_agent", | |
| "computer_control": "computer_control", | |
| "game_updater": "game_updater", | |
| "flight_finder": "flight_finder", | |
| "system_monitor_skill": "system_monitor_skill", | |
| "file_processor": "file_processor", | |
| } | |
| def _result_text(raw: Any) -> str: | |
| if raw is None: | |
| return "Done." | |
| if isinstance(raw, dict): | |
| if raw.get("message"): | |
| return str(raw["message"]) | |
| if raw.get("error"): | |
| return str(raw["error"]) | |
| if raw.get("ok") is False: | |
| return str(raw.get("error") or raw) | |
| return str(raw) | |
| def _normalize_args(name: str, args: dict, *, agent_folder: str, current_file: str) -> dict: | |
| params = dict(args or {}) | |
| if name == "open_app": | |
| app = params.get("app_name") or params.get("query") or "" | |
| return {"query": app} | |
| if name == "screen_processor": | |
| text = params.get("text") or params.get("prompt") or params.get("user_text") or "" | |
| return { | |
| "angle": (params.get("angle") or "screen").lower(), | |
| "text": text, | |
| "prompt": text, | |
| } | |
| if name == "run_custom_skill": | |
| skill_name = params.pop("skill_name", "") | |
| query = params.pop("query", "") | |
| out = {"query": query, **params} | |
| if query and "task" not in out: | |
| out["task"] = query | |
| return {"_skill_name": skill_name, **out} | |
| if name == "file_processor": | |
| if not params.get("file_path") and current_file: | |
| params["file_path"] = current_file | |
| if params.get("file_path") and not params.get("path"): | |
| params["path"] = params["file_path"] | |
| return params | |
| if name == "fallback_agent" and not params.get("folder") and agent_folder: | |
| params["folder"] = agent_folder | |
| return params | |
| async def execute_tool( | |
| fc, | |
| *, | |
| on_log: Callable[[str], None] | None = None, | |
| on_state: Callable[[str], None] | None = None, | |
| speak: Callable[[str], None] | None = None, | |
| agent_folder: str = "", | |
| current_file: str = "", | |
| vision_hooks: Any | None = None, | |
| on_shutdown: Callable[[], None] | None = None, | |
| on_close_camera: Callable[[], None] | None = None, | |
| ) -> types.FunctionResponse: | |
| name = fc.name | |
| args = dict(fc.args or {}) | |
| def _log(msg: str) -> None: | |
| if on_log: | |
| try: | |
| on_log(msg) | |
| except Exception: | |
| pass | |
| if on_state: | |
| try: | |
| on_state("THINKING") | |
| except Exception: | |
| pass | |
| _log(f"TOOL: {name} {args}") | |
| if name == "save_memory": | |
| category = args.get("category", "notes") | |
| key = args.get("key", "") | |
| value = args.get("value", "") | |
| if key and value: | |
| update_memory({category: {key: {"value": value}}}) | |
| _log(f"Memory: {category}/{key} = {value}") | |
| if on_state: | |
| try: | |
| on_state("LISTENING") | |
| except Exception: | |
| pass | |
| return types.FunctionResponse( | |
| id=fc.id, | |
| name=name, | |
| response={"result": "ok", "silent": True}, | |
| ) | |
| loop = asyncio.get_event_loop() | |
| result = "Done." | |
| try: | |
| if name == "shutdown_jarvis": | |
| _log("SYS: Shutdown requested.") | |
| if speak: | |
| speak("Goodbye.") | |
| if on_shutdown: | |
| on_shutdown() | |
| else: | |
| def _exit(): | |
| time.sleep(1) | |
| os._exit(0) | |
| threading.Thread(target=_exit, daemon=True).start() | |
| result = "Shutting down." | |
| elif name == "close_camera": | |
| if on_close_camera: | |
| on_close_camera() | |
| result = "Camera closed." | |
| elif name == "screen_processor": | |
| hooks = vision_hooks | |
| now = time.monotonic() | |
| cooldown = 4.0 | |
| if hooks and ( | |
| hooks.get("busy") | |
| or (now - hooks.get("last_time", 0.0)) < cooldown | |
| ): | |
| wait = max(0, cooldown - (now - hooks.get("last_time", 0.0))) | |
| _log(f"Vision cooldown ({wait:.1f}s)") | |
| result = ( | |
| "Vision is still processing the previous request. " | |
| "I will not call this again." | |
| ) | |
| else: | |
| from emo.desktop.actions import screen_processor as sp | |
| angle = (args.get("angle") or "screen").lower() | |
| user_text = args.get("text") or args.get("prompt") or "What do you see?" | |
| if angle == "camera": | |
| img_b, mime_t = await loop.run_in_executor(None, sp.capture_camera) | |
| stall = "camera" | |
| if hooks is not None: | |
| hooks["cam_active"] = True | |
| else: | |
| img_b, mime_t = await loop.run_in_executor(None, sp.capture_screen) | |
| stall = "screen" | |
| if hooks is not None: | |
| hooks["busy"] = True | |
| hooks["last_time"] = now | |
| hooks["pending"] = (img_b, mime_t, user_text, angle) | |
| result = ( | |
| f"[VISION_ACTIVE] {stall.capitalize()} captured. " | |
| f"Immediately say ONE natural sentence in the user's language. " | |
| f"Do NOT describe or guess content — the actual image arrives in the NEXT message." | |
| ) | |
| elif name == "run_custom_skill": | |
| skill_name = args.get("skill_name", "") | |
| norm = _normalize_args(name, args, agent_folder=agent_folder, current_file=current_file) | |
| skill_name = norm.pop("_skill_name", skill_name) | |
| raw = await loop.run_in_executor(None, lambda: run_skill(skill_name, norm)) | |
| result = _result_text(raw) | |
| elif name in _TOOL_TO_SKILL: | |
| skill = _TOOL_TO_SKILL[name] | |
| norm = _normalize_args(name, args, agent_folder=agent_folder, current_file=current_file) | |
| raw = await loop.run_in_executor(None, lambda: run_skill(skill, norm)) | |
| result = _result_text(raw) | |
| else: | |
| result = f"Unknown tool: {name}" | |
| except Exception as e: | |
| result = f"Tool '{name}' failed: {e}" | |
| traceback.print_exc() | |
| if speak: | |
| speak(f"Tool {name} failed.") | |
| if on_state: | |
| try: | |
| on_state("LISTENING") | |
| except Exception: | |
| pass | |
| _log(f"TOOL result {name}: {str(result)[:80]}") | |
| return types.FunctionResponse( | |
| id=fc.id, | |
| name=name, | |
| response={"result": result}, | |
| ) | |