#!/usr/bin/env python3
"""Orchestration Dashboard Server - Serves projects and provides API"""
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from pathlib import Path
import json
from datetime import datetime
from typing import List, Dict, Optional
import os
app = FastAPI(title="Orchestration Dashboard")
# Base paths
BASE_DIR = Path("/home/jooyoung/.openclaw/workspace-260312/orchestration")
PROJECTS_DIR = BASE_DIR / "projects"
REGISTRY_FILE = BASE_DIR / "registry.json"
# Ensure directories exist
PROJECTS_DIR.mkdir(parents=True, exist_ok=True)
def load_registry() -> List[Dict]:
"""Load project registry from file"""
if not REGISTRY_FILE.exists():
return []
try:
with open(REGISTRY_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception:
return []
def save_registry(registry: List[Dict]):
"""Save project registry to file"""
with open(REGISTRY_FILE, 'w', encoding='utf-8') as f:
json.dump(registry, f, indent=2, ensure_ascii=False)
def scan_projects() -> List[Dict]:
"""Auto-detect projects in projects/ directory"""
projects = []
if not PROJECTS_DIR.exists():
return projects
for project_dir in PROJECTS_DIR.iterdir():
if project_dir.is_dir():
# Check if it has an index.html
index_file = project_dir / "index.html"
has_game = index_file.exists()
# Get metadata if exists
meta_file = project_dir / "metadata.json"
metadata = {}
if meta_file.exists():
try:
with open(meta_file, 'r', encoding='utf-8') as f:
metadata = json.load(f)
except Exception:
pass
projects.append({
"name": project_dir.name,
"path": str(project_dir),
"has_game": has_game,
"created": metadata.get("created", "unknown"),
"description": metadata.get("description", ""),
"status": "ready" if has_game else "empty"
})
return projects
@app.get("/", response_class=HTMLResponse)
async def dashboard():
"""Main orchestration dashboard"""
projects = scan_projects()
# Generate project cards HTML
cards_html = ""
for project in projects:
status_badge = "✅" if project["has_game"] else "⚠️"
play_link = f'Play' if project["has_game"] else 'No Game'
cards_html += f'''
{status_badge} {project["name"]}
{project["description"] or "No description"}
Status: {project["status"]}
Created: {project["created"]}
'''
if not cards_html:
cards_html = 'No projects yet. Create one using the orchestration system!
'
html = f'''
Orchestration Dashboard
'''
return HTMLResponse(content=html)
@app.get("/api/health")
async def health():
"""Health check endpoint"""
return {
"status": "ok",
"timestamp": datetime.now().isoformat(),
"projects_count": len(scan_projects())
}
@app.get("/api/projects")
async def get_projects():
"""Get all projects"""
return {
"projects": scan_projects(),
"timestamp": datetime.now().isoformat()
}
@app.get("/api/projects/{project_name}/status")
async def get_project_status(project_name: str):
"""Get individual project status"""
projects = scan_projects()
project = next((p for p in projects if p["name"] == project_name), None)
if not project:
raise HTTPException(status_code=404, detail=f"Project '{project_name}' not found")
return project
# Dynamic project serving - scan both projects/ and serving/ directories
SERVING_DIR = BASE_DIR / "serving"
SERVING_DIR.mkdir(parents=True, exist_ok=True)
def _mount_project_dirs():
"""Mount all project directories for static serving"""
for scan_dir in [PROJECTS_DIR, SERVING_DIR]:
if not scan_dir.exists():
continue
for project_dir in scan_dir.iterdir():
# Resolve symlinks
real_dir = project_dir.resolve() if project_dir.is_symlink() else project_dir
if real_dir.is_dir() and (real_dir / "index.html").exists():
route = f"/p/{project_dir.name}"
try:
app.mount(route, StaticFiles(directory=str(real_dir), html=True), name=f"serve_{project_dir.name}")
except Exception:
pass
_mount_project_dirs()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8600)