""" app.py – Flask dashboard for Ubuntu Desktop on Hugging Face Spaces Serves: stats API, log viewer, service restart, and embedded desktop """ import os import subprocess import time import datetime import glob import json from pathlib import Path import psutil from flask import Flask, jsonify, render_template_string, request app = Flask(__name__) START_TIME = time.time() # ── HTML Template ───────────────────────────────────────────────────────────── DASHBOARD_HTML = """ Ubuntu Space Dashboard

System Dashboard

Loading…

Ubuntu 24.04 XFCE Desktop

PRoot-Distro style isolated environment on Hugging Face Spaces

🖥 Open Desktop
CPU Usage
–
–
Memory
–
–
Storage (/data)
–
–
Session Uptime
–
Space started
⚙️ Services
ServiceStatusPIDAction
Loading…
📄 Logs
Select a log file above…
""" # ── Helpers ─────────────────────────────────────────────────────────────────── def _supervisor_ctl(*args) -> str: try: result = subprocess.run( ["supervisorctl"] + list(args), capture_output=True, text=True, timeout=10 ) return result.stdout + result.stderr except Exception as e: return str(e) def _supervisor_status() -> list[dict]: """Parse `supervisorctl status` into structured list.""" output = _supervisor_ctl("status") services = [] descriptions = { "xvnc": "Virtual framebuffer + VNC server", "xfce4": "XFCE4 desktop session", "websockify": "noVNC ↔ VNC WebSocket bridge", "dashboard": "Flask stats dashboard", "nginx": "Reverse proxy (port 7860)", } for line in output.strip().splitlines(): if not line.strip(): continue parts = line.split() if len(parts) < 2: continue name = parts[0] state = parts[1] pid = None for p in parts: if p.startswith("pid"): try: pid = parts[parts.index(p) + 1].rstrip(",") except Exception: pass services.append({ "name": name, "state": state, "pid": pid, "description": descriptions.get(name, ""), }) return services def _human_uptime(seconds: float) -> str: d = int(seconds // 86400) h = int((seconds % 86400) // 3600) m = int((seconds % 3600) // 60) s = int(seconds % 60) if d: return f"{d}d {h}h {m}m" if h: return f"{h}h {m}m {s}s" return f"{m}m {s}s" # ── Routes ──────────────────────────────────────────────────────────────────── @app.route("/dashboard") def dashboard(): return render_template_string(DASHBOARD_HTML) @app.route("/health") def health(): return jsonify({"status": "ok", "ts": datetime.datetime.utcnow().isoformat()}) @app.route("/api/stats") def api_stats(): cpu = psutil.cpu_percent(interval=0.3) mem = psutil.virtual_memory() try: disk = psutil.disk_usage("/data") disk_data = { "percent": disk.percent, "used_gb": round(disk.used / 1e9, 2), "total_gb": round(disk.total / 1e9, 2), "free_gb": round(disk.free / 1e9, 2), } except Exception: disk_data = {"percent": 0, "used_gb": 0, "total_gb": 0, "free_gb": 0} uptime_secs = time.time() - START_TIME return jsonify({ "cpu": { "percent": round(cpu, 1), "count": psutil.cpu_count(), }, "memory": { "percent": mem.percent, "used_gb": round(mem.used / 1e9, 2), "total_gb": round(mem.total / 1e9, 2), "free_gb": round(mem.available / 1e9, 2), }, "disk": disk_data, "uptime": { "seconds": round(uptime_secs), "human": _human_uptime(uptime_secs), }, "services": _supervisor_status(), "ts": datetime.datetime.utcnow().isoformat(), }) @app.route("/api/logs/") def api_logs(name: str): # Allowlist to prevent path traversal allowed = {"supervisord", "xvnc", "xfce4", "websockify", "dashboard", "nginx", "xvnc_err", "xfce4_err", "websockify_err", "dashboard_err", "nginx_err"} if name not in allowed: return jsonify({"error": "Unknown log name"}), 400 log_path = Path(f"/data/logs/{name}.log") if not log_path.exists(): return jsonify({"content": f"Log file not found: {log_path}"}) try: # Return last 100 lines lines = log_path.read_text(errors="replace").splitlines() return jsonify({"content": "\n".join(lines[-100:])}) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/api/restart/", methods=["POST"]) def api_restart(name: str): allowed = {"xvnc", "xfce4", "websockify", "dashboard", "nginx"} if name not in allowed: return jsonify({"error": "Unknown service"}), 400 out = _supervisor_ctl("restart", name) return jsonify({"message": out.strip()}) # ── Entry point ─────────────────────────────────────────────────────────────── if __name__ == "__main__": app.run(host="127.0.0.1", port=5000, debug=False, threaded=True)