Spaces:
Running
Running
| """Public UI and credential-free health report; operator files are not served.""" | |
| from __future__ import annotations | |
| import json | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| import sys | |
| import time | |
| ROOT = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(ROOT)) | |
| def health_report(root: Path = ROOT): | |
| def read(path): | |
| try: | |
| return json.loads((root / path).read_text()) | |
| except (OSError, ValueError): | |
| return {} | |
| supervisor = read(".cloud-state/supervisor.json") | |
| checkpoint = read(".cloud-state/checkpoint.json") | |
| status = read("space/results/online_status.json") | |
| publication = read("space/results/publication_status.json") or status | |
| if status.get("push_status") == "failed": | |
| publication = status | |
| stamp = checkpoint.get("checkpointed_at") | |
| def age_seconds(stamp): | |
| try: | |
| return (datetime.now(timezone.utc) - datetime.fromisoformat(stamp)).total_seconds() | |
| except (ValueError, TypeError): | |
| return None | |
| age = age_seconds(stamp) | |
| push_age = age_seconds(publication.get("pushed_at")) | |
| failures = status.get("failed_forecasts", []) | |
| model_failures = sum(item.get("model") != "*" for item in failures) | |
| healthy = bool(supervisor.get("worker_running") and supervisor.get("evaluator_enabled") | |
| and 0 <= time.time() - supervisor.get("updated_at", 0) < 90 | |
| and age is not None and 0 <= age < 3600 | |
| and status.get("aggregate_status") == "ok" and publication.get("push_status") == "ok" | |
| and push_age is not None and 0 <= push_age < 3600 and model_failures == 0) | |
| return {"ok": healthy, "source_revision": supervisor.get("source_revision"), | |
| "evaluator_enabled": supervisor.get("evaluator_enabled", False), | |
| "worker_running": supervisor.get("worker_running", False), | |
| "checkpointed_at": stamp, "checkpoint_age_seconds": age, | |
| "last_evaluation_at": status.get("finished_at"), | |
| "evaluation_status": status.get("status"), | |
| "model_failure_count": model_failures, | |
| "unavailable_source_count": len(failures) - model_failures, | |
| "pending_tasks": status.get("pending_tasks"), | |
| "push_status": publication.get("push_status"), "pushed_at": publication.get("pushed_at")} | |
| def main(): | |
| import gradio as gr | |
| from fastapi import FastAPI | |
| from fastapi.responses import JSONResponse | |
| import uvicorn | |
| from space.app import demo | |
| app = FastAPI() | |
| def health(): | |
| report = health_report() | |
| return JSONResponse(report, status_code=200 if report["ok"] else 503) | |
| blocked = [str(ROOT / p) for p in ("space/results", "space/vendor/ts_bench/data", | |
| ".cloud-state", ".cache", ".env")] | |
| app = gr.mount_gradio_app(app, demo.queue(default_concurrency_limit=20), path="/", | |
| blocked_paths=[*blocked, "/proc"]) | |
| uvicorn.run(app, host="0.0.0.0", port=7860) | |
| if __name__ == "__main__": | |
| main() | |