Spaces:
Sleeping
Sleeping
| """Action Executor — translates ActionResult objects into game inputs. | |
| Two executors: | |
| DemoGameExecutor — queue-based, for Ursina game (preferred, cross-platform) | |
| ExternalGameExecutor — OS-level input injection for external games | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import logging | |
| import platform | |
| import time | |
| from typing import Optional | |
| from actions.action_schema import ActionResult, ActionType | |
| logger = logging.getLogger(__name__) | |
| # Key mapping for external game injection | |
| ACTION_KEY_MAP: dict[ActionType, object] = { | |
| ActionType.MOVE_FORWARD: "w", | |
| ActionType.MOVE_BACKWARD: "s", | |
| ActionType.MOVE_LEFT: "a", | |
| ActionType.MOVE_RIGHT: "d", | |
| ActionType.SPRINT: "shift", | |
| ActionType.JUMP: "space", | |
| ActionType.ROTATE_LEFT: ("mouse", "left", 200), | |
| ActionType.ROTATE_RIGHT: ("mouse", "right", 200), | |
| ActionType.LOOK_UP: ("mouse_y", "up", 150), | |
| ActionType.LOOK_DOWN: ("mouse_y", "down", 150), | |
| ActionType.EVADE_THREAT: "space", | |
| ActionType.INTERACT: ("mouse_click", "right"), | |
| ActionType.PICK_UP: ("mouse_click", "right"), | |
| ActionType.STOP: None, | |
| ActionType.IDLE: None, | |
| } | |
| class DemoGameExecutor: | |
| """Queue-based executor for the Ursina demo game. | |
| Uses a thread-safe queue.Queue so the asyncio orchestrator (background | |
| thread) and Ursina game loop (main thread) can share it safely on macOS. | |
| The game's update() loop drains this queue each frame. | |
| Never blocks the async loop. | |
| """ | |
| def __init__(self, command_queue: "queue.Queue") -> None: | |
| self._queue = command_queue | |
| self._executed = 0 | |
| async def execute(self, action: ActionResult) -> None: | |
| """Put action into the game command queue (drop-oldest if full).""" | |
| import queue as _queue | |
| cmd = { | |
| "action": action.action.value, | |
| "direction": action.direction, | |
| "magnitude": action.magnitude, | |
| "confidence": action.confidence, | |
| "say": action.say, | |
| "scene_description": action.scene_description, | |
| "target": action.target, | |
| "ts": time.perf_counter(), | |
| } | |
| try: | |
| if self._queue.full(): | |
| try: | |
| self._queue.get_nowait() | |
| except _queue.Empty: | |
| pass | |
| self._queue.put_nowait(cmd) | |
| except _queue.Full: | |
| pass | |
| self._executed += 1 | |
| logger.debug("Queued action: %s mag=%.2f", action.action.value, action.magnitude) | |
| def executed_count(self) -> int: | |
| return self._executed | |
| class ExternalGameExecutor: | |
| """OS-level input injection for controlling external game windows. | |
| Kill-switch: if mouse cursor within 50px of (0, 0), stop all injection. | |
| """ | |
| def __init__(self, config, hold_scale: float = 1.0) -> None: | |
| self._config = config | |
| self._is_windows = platform.system() == "Windows" | |
| self._executed = 0 | |
| self._hold_scale = hold_scale | |
| def _check_killswitch(self) -> bool: | |
| """Return True (kill active) if mouse is within 50px of top-left corner.""" | |
| try: | |
| if self._is_windows: | |
| try: | |
| import pydirectinput | |
| x, y = pydirectinput.position() | |
| except ImportError: | |
| import pyautogui | |
| x, y = pyautogui.position() | |
| else: | |
| import pyautogui | |
| x, y = pyautogui.position() | |
| return x <= 50 and y <= 50 | |
| except Exception: | |
| return False | |
| def _press_key(self, key: str, duration: float) -> None: | |
| """Press and hold key for duration seconds.""" | |
| try: | |
| if self._is_windows: | |
| try: | |
| import pydirectinput | |
| pydirectinput.keyDown(key) | |
| time.sleep(duration) | |
| pydirectinput.keyUp(key) | |
| return | |
| except ImportError: | |
| pass | |
| import pyautogui | |
| pyautogui.keyDown(key) | |
| time.sleep(duration) | |
| pyautogui.keyUp(key) | |
| except Exception as exc: | |
| logger.warning("Key press failed: %s", exc) | |
| async def execute(self, action: ActionResult) -> None: | |
| """Execute the action via OS-level input injection.""" | |
| if self._check_killswitch(): | |
| logger.warning("Kill-switch active — skipping OS injection") | |
| return | |
| key_or_spec = ACTION_KEY_MAP.get(action.action) | |
| if key_or_spec is None: | |
| return # STOP / IDLE — no input | |
| hold_duration = action.magnitude * 0.3 * self._hold_scale | |
| if isinstance(key_or_spec, tuple) and key_or_spec[0] == "mouse": | |
| # Mouse rotation | |
| _, direction, pixels = key_or_spec | |
| await asyncio.get_event_loop().run_in_executor( | |
| None, | |
| self._do_mouse_rotation, | |
| direction, | |
| int(pixels * action.magnitude), | |
| ) | |
| elif isinstance(key_or_spec, tuple) and key_or_spec[0] == "mouse_y": | |
| _, direction, pixels = key_or_spec | |
| dy = -int(pixels * action.magnitude) if direction == "up" else int(pixels * action.magnitude) | |
| await asyncio.get_event_loop().run_in_executor( | |
| None, self._do_mouse_rotation_y, dy, | |
| ) | |
| elif isinstance(key_or_spec, tuple) and key_or_spec[0] == "mouse_click": | |
| _, button = key_or_spec | |
| await asyncio.get_event_loop().run_in_executor( | |
| None, self._do_mouse_click, button, | |
| ) | |
| elif isinstance(key_or_spec, str): | |
| await asyncio.get_event_loop().run_in_executor( | |
| None, | |
| self._press_key, | |
| key_or_spec, | |
| max(0.05, hold_duration), | |
| ) | |
| self._executed += 1 | |
| def _do_mouse_rotation_y(self, dy: int) -> None: | |
| """Move mouse vertically to simulate camera pitch.""" | |
| try: | |
| if self._is_windows: | |
| try: | |
| import pydirectinput | |
| pydirectinput.move(0, dy, relative=True) | |
| return | |
| except ImportError: | |
| pass | |
| import pyautogui | |
| pyautogui.move(0, dy) | |
| except Exception as exc: | |
| logger.warning("Mouse Y rotation failed: %s", exc) | |
| def _do_mouse_click(self, button: str) -> None: | |
| """Single mouse click (left or right).""" | |
| try: | |
| if self._is_windows: | |
| try: | |
| import pydirectinput | |
| pydirectinput.click(button=button) | |
| return | |
| except ImportError: | |
| pass | |
| import pyautogui | |
| pyautogui.click(button=button) | |
| except Exception as exc: | |
| logger.warning("Mouse click failed: %s", exc) | |
| def _do_mouse_rotation(self, direction: str, pixels: int) -> None: | |
| """Move mouse horizontally to simulate camera rotation.""" | |
| try: | |
| if self._is_windows: | |
| try: | |
| import pydirectinput | |
| dx = pixels if direction == "right" else -pixels | |
| pydirectinput.move(dx, 0, relative=True) | |
| return | |
| except ImportError: | |
| pass | |
| import pyautogui | |
| dx = pixels if direction == "right" else -pixels | |
| pyautogui.move(dx, 0) | |
| except Exception as exc: | |
| logger.warning("Mouse rotation failed: %s", exc) | |