status.py

#!/usr/bin/env python3.12
"""ULTRON Status Dashboard - see everything at a glance."""

import json
import subprocess
from datetime import datetime, timezone
from pathlib import Path

import requests
import yaml

BASE_DIR = Path(__file__).resolve().parent

TIMERS = [
    "curiosity.timer",
    "session-watcher.timer",
    "curiosity-surface.timer",
    "daily-briefing.timer",
    "goal-engine.timer",
    "intel.timer",
    "ultron-reflect.timer",
    "ultron-heal.timer",
    "ultron-goalgen.timer",
    "ultron-bridge.timer",
    "ultron-consolidate.timer",
]

SERVICES = [
    "ultron-trigger.service",
    "ultron-inbox.service",
]


def header(text):
    print(f"\n\033[1;36m{text}\033[0m")
    print("\033[36m" + "─" * 50 + "\033[0m")


def ok(text):
    print(f"  \033[32m●\033[0m {text}")


def warn(text):
    print(f"  \033[33m●\033[0m {text}")


def fail(text):
    print(f"  \033[31m●\033[0m {text}")


def check_service(name):
    try:
        r = subprocess.run(
            ["systemctl", "--user", "is-active", name],
            capture_output=True, text=True, timeout=5,
        )
        return r.stdout.strip()
    except Exception:
        return "unknown"


def get_next_run(timer):
    try:
        r = subprocess.run(
            ["systemctl", "--user", "show", timer, "--property=NextElapseUSecRealtime"],
            capture_output=True, text=True, timeout=5,
        )
        val = r.stdout.strip().split("=", 1)[-1]
        return val if val else "unknown"
    except Exception:
        return "unknown"


def main():
    print("\n\033[1;35m╔══════════════════════════════════╗\033[0m")
    print("\033[1;35m║      ULTRON STATUS DASHBOARD     ║\033[0m")
    print("\033[1;35m╚══════════════════════════════════╝\033[0m")
    print(f"  {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}")

    # --- Timers ---
    header("TIMERS")
    for timer in TIMERS:
        status = check_service(timer)
        if status == "active":
            ok(f"{timer:<30s} active")
        elif status == "inactive":
            warn(f"{timer:<30s} inactive")
        else:
            fail(f"{timer:<30s} {status}")

    # --- Services ---
    header("SERVICES")
    for svc in SERVICES:
        status = check_service(svc)
        if status == "active":
            ok(f"{svc:<30s} running")
        else:
            fail(f"{svc:<30s} {status}")

    # --- AOMS ---
    header("AOMS")
    try:
        resp = requests.get("http://localhost:9100/stats", timeout=5)
        resp.raise_for_status()
        stats = resp.json()
        ok(f"Total memories: {stats.get('total_entries', '?')}")
        by_tier = stats.get("by_tier", {})
        if by_tier:
            ok(f"  Episodic: {by_tier.get('episodic', 0)}  Semantic: {by_tier.get('semantic', 0)}  Procedural: {by_tier.get('procedural', 0)}")
        ok(f"Last 24h: {stats.get('recent_24h', '?')} new")
    except Exception:
        fail("AOMS unreachable")

    # --- Ollama ---
    header("OLLAMA")
    try:
        resp = requests.get("http://localhost:11434/api/tags", timeout=5)
        resp.raise_for_status()
        models = [m["name"] for m in resp.json().get("models", [])]
        ok(f"Online — {len(models)} models: {', '.join(models[:5])}")
    except Exception:
        fail("Ollama unreachable")

    # --- Goals ---
    header("GOALS")
    try:
        goals_data = yaml.safe_load((BASE_DIR / "goals.yaml").read_text()) or {}
        for goal in goals_data.get("goals", []):
            tasks = goal.get("tasks", [])
            done = sum(1 for t in tasks if t["status"] == "completed")
            failed = sum(1 for t in tasks if t["status"] == "failed")
            total = len(tasks)
            pct = int((done / total) * 100) if total else 0
            pri = goal.get("priority", "?")[0].upper()
            status = goal.get("status", "active")

            bar_width = 16
            filled = int(bar_width * pct / 100)
            bar = "\033[32m" + "█" * filled + "\033[90m" + "░" * (bar_width - filled) + "\033[0m"

            line = f"[{pri}] {bar} {pct:>3d}% {goal['name'][:40]}"
            if failed:
                line += f" ({failed} failed)"
            if status != "active":
                line += f" [{status}]"

            fn = ok if pct == 100 else (warn if failed else ok)
            fn(line)
    except Exception:
        fail("Could not read goals.yaml")

    # --- Curiosity Queue ---
    header("CURIOSITY QUEUE")
    try:
        queue = json.loads((BASE_DIR / "queue.json").read_text())
        questions = queue.get("questions", [])
        ok(f"{len(questions)} questions queued")
        for q in questions[:3]:
            print(f"    → {q[:60]}")
    except Exception:
        warn("No queue")

    # --- Learnings ---
    header("LEARNINGS")
    learnings_file = BASE_DIR / "learnings.json"
    if learnings_file.exists():
        try:
            data = json.loads(learnings_file.read_text())
            ok(f"Successes: {len(data.get('successes', []))}  "
               f"Failures: {len(data.get('failures', []))}  "
               f"Anti-patterns: {len(data.get('anti_patterns', []))}")
        except Exception:
            warn("Could not read learnings")
    else:
        ok("No learnings yet (clean slate)")

    # --- OpenClaw Cron ---
    header("OPENCLAW CRON JOBS")
    try:
        cron_path = Path.home() / ".openclaw" / "cron" / "jobs.json"
        if cron_path.exists():
            cron_data = json.loads(cron_path.read_text())
            for job in cron_data.get("jobs", []):
                name = job.get("name", "?")
                enabled = job.get("enabled", False)
                state = job.get("state", {})
                last_status = state.get("lastRunStatus", "never")
                if enabled:
                    fn = ok if last_status == "ok" else warn
                    fn(f"{name:<35s} enabled  last={last_status}")
                else:
                    warn(f"{name:<35s} disabled")
        else:
            warn("No cron jobs file found")
    except Exception:
        fail("Could not read OpenClaw cron jobs")

    # --- Recent logs ---
    header("RECENT ACTIVITY")
    log_dir = BASE_DIR / "logs"
    if log_dir.exists():
        today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
        for log_name in ["goal_engine.log", "actions.log", "session_watcher.log"]:
            log_file = log_dir / log_name
            if log_file.exists():
                lines = log_file.read_text().strip().split("\n")
                today_lines = [l for l in lines[-20:] if today in l]
                if today_lines:
                    ok(f"{log_name}: {len(today_lines)} entries today")
    else:
        warn("No logs yet")

    print()


if __name__ == "__main__":
    main()