219 lines
7.9 KiB
Python
219 lines
7.9 KiB
Python
"""
|
|
Mission Control API — Data layer for Hermes WebUI Mission Control extension.
|
|
Provides priorities, tasks, feed, and dashboard status management.
|
|
"""
|
|
|
|
import json
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from api.helpers import j
|
|
|
|
# ── State file ────────────────────────────────────────────────────────────────
|
|
_MC_DATA_FILE = Path.home() / ".hermes" / "data" / "mc-data.json"
|
|
_MC_LOCK = threading.RLock()
|
|
|
|
# ── Default structure ─────────────────────────────────────────────────────────
|
|
DEFAULT_MC_DATA = {
|
|
"priorities": [],
|
|
"tasks": [],
|
|
"feed": [],
|
|
}
|
|
|
|
|
|
def _load_mc_data() -> dict:
|
|
"""Load Mission Control data from disk."""
|
|
with _MC_LOCK:
|
|
if not _MC_DATA_FILE.exists():
|
|
return DEFAULT_MC_DATA.copy()
|
|
try:
|
|
with open(_MC_DATA_FILE, "r") as f:
|
|
return json.load(f)
|
|
except (json.JSONDecodeError, IOError):
|
|
return DEFAULT_MC_DATA.copy()
|
|
|
|
|
|
def _save_mc_data(data: dict) -> None:
|
|
"""Save Mission Control data to disk."""
|
|
with _MC_LOCK:
|
|
_MC_DATA_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(_MC_DATA_FILE, "w") as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
|
|
# ── Priority helpers ──────────────────────────────────────────────────────────
|
|
|
|
def get_priorities() -> list[dict]:
|
|
"""Return all priorities sorted by id."""
|
|
data = _load_mc_data()
|
|
return sorted(data.get("priorities", []), key=lambda p: p.get("id", 0))
|
|
|
|
|
|
def create_priority(name: str, color: str = "#808080") -> dict:
|
|
"""Add a new priority. Returns the created priority."""
|
|
data = _load_mc_data()
|
|
priorities = data.get("priorities", [])
|
|
new_id = max([p.get("id", 0) for p in priorities], default=0) + 1
|
|
priority = {"id": new_id, "name": name, "color": color}
|
|
priorities.append(priority)
|
|
data["priorities"] = priorities
|
|
_save_mc_data(data)
|
|
_add_feed_event(f"Priority created: {name}")
|
|
return priority
|
|
|
|
|
|
def update_priority(priority_id: int, name: str = None, color: str = None, done: bool = None) -> dict | None:
|
|
"""Update an existing priority. Returns updated priority or None if not found."""
|
|
data = _load_mc_data()
|
|
priorities = data.get("priorities", [])
|
|
for p in priorities:
|
|
if p.get("id") == priority_id:
|
|
if name is not None:
|
|
p["name"] = name
|
|
if color is not None:
|
|
p["color"] = color
|
|
if done is not None:
|
|
p["done"] = done
|
|
if done:
|
|
_add_feed_event(f"Priority completed: {p['name']}")
|
|
data["priorities"] = priorities
|
|
_save_mc_data(data)
|
|
return p
|
|
return None
|
|
|
|
|
|
def delete_priority(priority_id: int) -> bool:
|
|
"""Delete a priority. Returns True if found and deleted."""
|
|
data = _load_mc_data()
|
|
priorities = data.get("priorities", [])
|
|
original_len = len(priorities)
|
|
priorities = [p for p in priorities if p.get("id") != priority_id]
|
|
if len(priorities) < original_len:
|
|
data["priorities"] = priorities
|
|
_save_mc_data(data)
|
|
return True
|
|
return False
|
|
|
|
|
|
# ── Task helpers ──────────────────────────────────────────────────────────────
|
|
|
|
def get_tasks() -> list[dict]:
|
|
"""Return all tasks sorted by priority then id."""
|
|
data = _load_mc_data()
|
|
return sorted(data.get("tasks", []), key=lambda t: (t.get("priority", 999), t.get("id", 0)))
|
|
|
|
|
|
def create_task(title: str, priority: int = 1, status: str = "backlog") -> dict:
|
|
"""Create a new task. Returns the created task."""
|
|
data = _load_mc_data()
|
|
tasks = data.get("tasks", [])
|
|
new_id = max([t.get("id", 0) for t in tasks], default=0) + 1
|
|
task = {"id": new_id, "title": title, "priority": priority, "status": status}
|
|
tasks.append(task)
|
|
data["tasks"] = tasks
|
|
_save_mc_data(data)
|
|
_add_feed_event(f"Task created: {title}")
|
|
return task
|
|
|
|
|
|
def update_task(task_id: int, **kwargs) -> dict | None:
|
|
"""Update a task by id. kwargs: title, priority, status. Returns updated task or None."""
|
|
data = _load_mc_data()
|
|
tasks = data.get("tasks", [])
|
|
for t in tasks:
|
|
if t.get("id") == task_id:
|
|
old_status = t.get("status")
|
|
for key in ("title", "priority", "status"):
|
|
if key in kwargs:
|
|
t[key] = kwargs[key]
|
|
new_status = t.get("status")
|
|
# Feed events for status transitions
|
|
if old_status != new_status:
|
|
if new_status == "done":
|
|
_add_feed_event(f"Task completed: {t['title']}")
|
|
elif new_status == "progress":
|
|
_add_feed_event(f"Task started: {t['title']}")
|
|
data["tasks"] = tasks
|
|
_save_mc_data(data)
|
|
return t
|
|
return None
|
|
|
|
|
|
def delete_task(task_id: int) -> bool:
|
|
"""Delete a task. Returns True if found and deleted."""
|
|
data = _load_mc_data()
|
|
tasks = data.get("tasks", [])
|
|
original_len = len(tasks)
|
|
tasks = [t for t in tasks if t.get("id") != task_id]
|
|
if len(tasks) < original_len:
|
|
data["tasks"] = tasks
|
|
_save_mc_data(data)
|
|
return True
|
|
return False
|
|
|
|
|
|
# ── Feed helpers ──────────────────────────────────────────────────────────────
|
|
|
|
def get_feed(limit: int = 50) -> list[dict]:
|
|
"""Return recent feed events, newest first."""
|
|
data = _load_mc_data()
|
|
feed = data.get("feed", [])
|
|
return sorted(feed, key=lambda f: f.get("timestamp", ""), reverse=True)[:limit]
|
|
|
|
|
|
def _add_feed_event(event: str) -> None:
|
|
"""Add a timestamped feed event."""
|
|
data = _load_mc_data()
|
|
feed = data.get("feed", [])
|
|
feed.append({
|
|
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
"event": event,
|
|
})
|
|
# Keep only last 200 events
|
|
data["feed"] = feed[-200:]
|
|
_save_mc_data(data)
|
|
|
|
|
|
# ── Dashboard status ──────────────────────────────────────────────────────────
|
|
|
|
def get_dashboard_status() -> dict:
|
|
"""Return aggregated dashboard status for Mission Control."""
|
|
data = _load_mc_data()
|
|
priorities = data.get("priorities", [])
|
|
tasks = data.get("tasks", [])
|
|
|
|
priorities_total = len(priorities)
|
|
priorities_done = sum(1 for p in priorities if p.get("done"))
|
|
|
|
tasks_backlog = sum(1 for t in tasks if t.get("status") == "backlog")
|
|
tasks_progress = sum(1 for t in tasks if t.get("status") == "progress")
|
|
tasks_done = sum(1 for t in tasks if t.get("status") == "done")
|
|
|
|
feed = get_feed(limit=5)
|
|
latest_event = feed[0]["event"] if feed else "No recent activity"
|
|
|
|
# Health assessment
|
|
if tasks_done == 0 and tasks_backlog == 0 and tasks_progress == 0:
|
|
health = "empty"
|
|
elif tasks_progress > 0 and tasks_done > 0:
|
|
health = "healthy"
|
|
elif tasks_progress > 0:
|
|
health = "active"
|
|
elif tasks_backlog > 0:
|
|
health = "warning"
|
|
else:
|
|
health = "ok"
|
|
|
|
return {
|
|
"priorities_total": priorities_total,
|
|
"priorities_done": priorities_done,
|
|
"tasks_backlog": tasks_backlog,
|
|
"tasks_progress": tasks_progress,
|
|
"tasks_done": tasks_done,
|
|
"latest_feed_event": latest_event,
|
|
"dashboard_health": health,
|
|
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
}
|