""" actions.py — Local OS Command Execution Responsibilities: 1. Provide a registry of local actions the assistant can perform. 2. Map action names from the LLM's JSON commands to Python functions. 3. Execute commands and return a spoken summary for TTS feedback. Supported actions: open_app — Launch a desktop application set_timer — Start a countdown timer with audible alarm get_time — Return the current time get_date — Return today's date get_weather — (stub) Return weather info create_reminder — (stub) Save a reminder note control_volume — Adjust system volume (Linux/macOS) search_web — Open a web search in the default browser calculate — Evaluate a math expression safely shutdown — System shutdown (with confirmation) """ import logging import os import platform import subprocess import threading import webbrowser from datetime import datetime logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Action Registry # --------------------------------------------------------------------------- _REGISTRY: dict[str, callable] = {} def register(name: str): """Decorator to register an action function by name.""" def decorator(func): _REGISTRY[name] = func return func return decorator def execute(action: str, params: dict | None = None) -> str: """ Execute a registered action and return a spoken summary. Args: action: The action name (e.g., "open_app"). params: Optional dict of parameters. Returns: A short text description of what was done (for TTS feedback). """ params = params or {} func = _REGISTRY.get(action) if not func: logger.warning("Unknown action: %s", action) return f"Sorry, I don't know how to {action.replace('_', ' ')}." try: result = func(**params) logger.info("Action '%s' executed: %s", action, result) return result except Exception as exc: logger.exception("Action '%s' failed", action) return f"Something went wrong: {exc}" # --------------------------------------------------------------------------- # Action Implementations # --------------------------------------------------------------------------- @register("get_time") def get_time(**_) -> str: now = datetime.now().strftime("%-I:%M %p") return f"It's currently {now}." @register("get_date") def get_date(**_) -> str: today = datetime.now().strftime("%A, %B %d, %Y") return f"Today is {today}." @register("open_app") def open_app(app_name: str = "", **_) -> str: if not app_name: return "What app would you like me to open?" system = platform.system() app_lower = app_name.lower().strip() try: if system == "Darwin": # macOS subprocess.Popen(["open", "-a", app_name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) elif system == "Windows": subprocess.Popen(f"start {app_name}", shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) else: # Linux # Try common app launchers app_map = { "chrome": "google-chrome", "firefox": "firefox", "terminal": "gnome-terminal", "files": "nautilus", "calculator": "gnome-calculator", "settings": "gnome-control-center", "browser": "xdg-open", "vs code": "code", "vscode": "code", } cmd = app_map.get(app_lower, app_lower) subprocess.Popen( [cmd], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ) return f"Opening {app_name}." except FileNotFoundError: return f"Sorry, I couldn't find {app_name} on this system." @register("set_timer") def set_timer(seconds: int = 60, label: str = "Timer", **_) -> str: """ Start a background timer that rings after the given number of seconds. Uses the terminal bell when the timer completes. """ try: duration = int(seconds) except (ValueError, TypeError): return "I need a number of seconds for the timer." def _timer_thread(): import time logger.info("Timer '%s' started for %d seconds", label, duration) time.sleep(duration) # Terminal bell print(f"\a") logger.info("Timer '%s' finished!", label) # Try to use TTS to announce (if available — soft dependency) try: import pygame pygame.mixer.init() # Generate a simple beep import numpy as np sample_rate = 22050 t = np.linspace(0, 0.5, int(sample_rate * 0.5), dtype=np.float32) tone = np.sin(2 * np.pi * 880 * t) * 0.5 tone = (tone * 32767).astype(np.int16) # Save and play import soundfile as sf beep_path = f"/tmp/echo_timer_{os.urandom(4).hex()}.wav" sf.write(beep_path, tone, sample_rate) pygame.mixer.music.load(beep_path) pygame.mixer.music.play() while pygame.mixer.music.get_busy(): pygame.time.wait(50) pygame.mixer.quit() os.unlink(beep_path) except Exception: pass # Fall back to terminal bell only threading.Thread(target=_timer_thread, daemon=True, name=f"timer-{label}").start() minutes, secs = divmod(duration, 60) if minutes: return f"{label} set for {minutes} minute{'s' if minutes != 1 else ''} and {secs} seconds." return f"{label} set for {secs} seconds." @register("get_weather") def get_weather(location: str = "", **_) -> str: """Stub — can be expanded with a weather API integration.""" return ( "I don't have a weather service connected yet. " "You can ask me again once the weather API is configured." ) @register("create_reminder") def create_reminder(text: str = "", **_) -> str: """Save a reminder to a local file.""" if not text: return "What would you like me to remind you about?" reminders_dir = Path.home() / ".echo" / "reminders" reminders_dir.mkdir(parents=True, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") reminder_file = reminders_dir / f"{timestamp}.txt" reminder_file.write_text(f"[{datetime.now().isoformat()}] {text}\n") return f"Reminder saved: {text}" @register("control_volume") def control_volume(level: int = 50, **_) -> str: """Adjust system volume (Linux/macOS only).""" try: vol = int(level) vol = max(0, min(100, vol)) except (ValueError, TypeError): return "Please specify a volume level between 0 and 100." system = platform.system() try: if system == "Darwin": subprocess.run(["osascript", "-e", f"set volume output volume {vol}"], check=True, capture_output=True) elif system == "Linux": subprocess.run( ["pactl", "set-sink-volume", "@DEFAULT_SINK@", f"{vol}%"], check=True, capture_output=True, ) else: return "Volume control isn't supported on Windows yet." return f"Volume set to {vol}%." except subprocess.CalledProcessError: return "I couldn't adjust the volume." @register("search_web") def search_web(query: str = "", **_) -> str: if not query: return "What would you like to search for?" url = f"https://www.google.com/search?q={query.replace(' ', '+')}" webbrowser.open(url) return f"Searching the web for: {query}" @register("calculate") def calculate(expression: str = "", **_) -> str: if not expression: return "What would you like me to calculate?" # Whitelist only safe math operations import ast allowed = { ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Pow, ast.USub, ast.UAdd, ast.Constant, ast.Num, } try: tree = ast.parse(expression.strip(), mode="eval") for node in ast.walk(tree): if type(node) not in allowed: raise ValueError("Unsafe expression") result = eval(compile(tree, "", "eval")) # noqa: S307 return f"The result is {result}" except ZeroDivisionError: return "You can't divide by zero." except Exception: return f"I couldn't calculate that expression." @register("shutdown") def shutdown(confirm: bool = False, **_) -> str: if not confirm: return "Are you sure? Please confirm the shutdown command." system = platform.system() try: if system == "Darwin": subprocess.run(["sudo", "shutdown", "-h", "now"], check=True) elif system == "Windows": subprocess.run(["shutdown", "/s", "/t", "5"], check=True) else: subprocess.run(["sudo", "shutdown", "-h", "now"], check=True) return "Shutting down now. Goodbye!" except Exception: return "I don't have permission to shut down the system."