modules/autonomy.py

"""

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."