""" modules/autonomy.py — FRIDAY Autonomous Initiative FRIDAY VOLUNTEERS for complex tasks: - Monitors system - Offers help proactively - Executes multi-step tasks without asking - Anticipates needs - "I can handle that" behavior """ import time import os import json import random from config import DATA_DIR AUTO_FILE = os.path.join(DATA_DIR, "autonomy.json") def _load() -> dict: if not os.path.exists(AUTO_FILE): return {"volunteers": [], "offers": [], "last_offer": 0, "enabled": True} try: with open(AUTO_FILE, "r") as f: return json.load(f) except Exception: return {"volunteers": [], "offers": [], "last_offer": 0, "enabled": True} def _save(data: dict): os.makedirs(DATA_DIR, exist_ok=True) try: with open(AUTO_FILE, "w") as f: json.dump(data, f, indent=2) except Exception: pass # ── Volunteer Triggers ──────────────────────────────────────────── VOLUNTEER_TRIGGERS = [ {"condition": "high_cpu", "offer": "I can lower your system load."}, {"condition": "low_memory", "offer": "RAM is tight. Free up memory?"}, {"condition": "hot_cpu", "offer": "System hot. Optimize cooling?"}, {"condition": "repeat_command", "offer": "Done that twice. Create shortcut?"}, {"condition": "long_task", "offer": "I'll handle it in background."}, {"condition": "new_day", "offer": "Morning. Daily brief?"}, {"condition": "evening", "offer": "Evening prep?"}, {"condition": "free_time", "offer": "Run maintenance?"}, ] # ── Monitor System ───────────────────────────────────────────────── def check_system() -> list[str]: """Check system and return offers.""" offers = [] try: import psutil # CPU check cpu = psutil.cpu_percent(interval=1) if cpu > 80: offers.append("High CPU detected") # RAM check ram = psutil.virtual_memory() if ram.percent > 85: offers.append("Low memory") # Temperature check try: temps = psutil.sensors_temperatures() for name, entries in temps.items(): for e in entries: if e.current and e.current > 80: offers.append("Hot CPU") except Exception: pass except Exception: pass return offers # ── Check Repeat Behavior ────────────────────────────────────────── def check_repeat(user_text: str) -> str | None: """Check if user is repeating an action.""" data = _load() recent = data.get("recent_commands", []) if user_text in recent and len(recent) > 0: count = recent.count(user_text) if count >= 2: return "I notice you've done this twice. Want me to automate it?" return None # ── Should Volunteer ───────────────────────────────────────────── def should_volunteer(now: float = None) -> bool: """Decide if FRIDAY should volunteer.""" if now is None: now = time.time() data = _load() if not data.get("enabled", True): return False last = data.get("last_offer", 0) # Only volunteer every 5 mins if now - last < 300: return False # Check conditions issues = check_system() if issues: return True return False # ── Get Offer ───────────────────────────────────────────────────── def get_offer() -> str | None: """Get proactive offer.""" now = time.time() # System offers issues = check_system() if issues: offers_map = { "High CPU": "I can lower your system load. Close some background apps?", "Low memory": "RAM is tight. Free up memory?", "Hot CPU": "System hot. Optimize cooling?", } for issue in issues: if issue in offers_map: data = _load() data["last_offer"] = now _save(data) return offers_map[issue] # Time-based offers from datetime import datetime hour = datetime.now().hour if 8 <= hour <= 9: data = _load() data["last_offer"] = now _save(data) return "Morning. Want your daily system check?" if 22 <= hour <= 23: data = _load() data["last_offer"] = now _save(data) return "Evening prep? I can optimize for rest." return None # ── Track Recent Commands ───────────────────────────────────────── def track_command(command: str): """Track recent commands for repeat detection.""" data = _load() recent = data.setdefault("recent_commands", []) recent.append(command) if len(recent) > 20: recent = recent[-20:] data["recent_commands"] = recent _save(data) # ── Enable/Disable ──────────────────────────────────────────────── def enable_autonomy(enabled: bool = True): """Enable/disable autonomous mode.""" data = _load() data["enabled"] = enabled _save(data) def is_autonomy_enabled() -> bool: """Check if autonomy is enabled.""" return _load().get("enabled", True) # ── Brain Integration ───────────────────────────────────────── def get_autonomy_block() -> str: """Get autonomy context for brain.""" if is_autonomy_enabled(): return "[AUTONOMY] I can volunteer to help if needed." return "" # ── Execute Complex Task ─────────────────────────────────────── def execute_autonomous(task: str) -> str: """Execute complex task without explicit command.""" # This delegates to executor try: from modules.executor import execute_autonomous import asyncio return asyncio.run(execute_autonomous(task, speak_out_loud=False)) except Exception: return "I'd help but need more context."