generated from coulomb/repo-seed
Add Economic Observatory web UI with ledger-backed API
Introduce ui/ dashboard (dark observatory layout), JSON API, and local dev server. All metrics load from expense and payment record ledgers. Links Claude design reference for visual alignment.
This commit is contained in:
72
projects/coulomb-pricing/observatory/server.py
Normal file
72
projects/coulomb-pricing/observatory/server.py
Normal file
@@ -0,0 +1,72 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
from .api import build_dashboard_payload
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
UI_DIR = ROOT / "ui"
|
||||
|
||||
|
||||
class ObservatoryHandler(BaseHTTPRequestHandler):
|
||||
data_dir: Path = ROOT / "data"
|
||||
|
||||
def _send(self, status: int, body: bytes, content_type: str) -> None:
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", content_type)
|
||||
self.send_header("Access-Control-Allow-Origin", "*")
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def do_GET(self) -> None:
|
||||
parsed = urlparse(self.path)
|
||||
if parsed.path == "/api/dashboard":
|
||||
query = parse_qs(parsed.query)
|
||||
period = query.get("period", [None])[0]
|
||||
payload = build_dashboard_payload(self.data_dir, period)
|
||||
self._send(200, json.dumps(payload).encode("utf-8"), "application/json")
|
||||
return
|
||||
|
||||
if parsed.path == "/":
|
||||
return self._serve_file(UI_DIR / "index.html", "text/html; charset=utf-8")
|
||||
|
||||
if parsed.path.startswith("/ui/"):
|
||||
relative = parsed.path.removeprefix("/ui/")
|
||||
target = UI_DIR / relative
|
||||
if target.exists() and target.is_file():
|
||||
content_type = "text/css" if target.suffix == ".css" else "application/javascript"
|
||||
return self._serve_file(target, f"{content_type}; charset=utf-8")
|
||||
|
||||
self._send(404, b"Not found", "text/plain")
|
||||
|
||||
def _serve_file(self, path: Path, content_type: str) -> None:
|
||||
self._send(200, path.read_bytes(), content_type)
|
||||
|
||||
def log_message(self, format: str, *args) -> None:
|
||||
return
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description="Coulomb Economic Observatory UI server")
|
||||
parser.add_argument("--host", default="127.0.0.1")
|
||||
parser.add_argument("--port", type=int, default=8765)
|
||||
parser.add_argument("--data-dir", type=Path, default=ROOT / "data")
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
ObservatoryHandler.data_dir = args.data_dir
|
||||
server = ThreadingHTTPServer((args.host, args.port), ObservatoryHandler)
|
||||
print(f"Economic Observatory UI: http://{args.host}:{args.port}/")
|
||||
print(f"API: http://{args.host}:{args.port}/api/dashboard")
|
||||
try:
|
||||
server.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
print("\nStopped.")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user