"""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) @property 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)