"""GenRecon web service — FastAPI. POST /api/jobs multipart `video` -> {id} GET /api/jobs/{id} status json GET /api/jobs/{id}/scene.glb the reconstructed scene GET /api/jobs/{id}/log raw pipeline log (debug) GET /api/health / the PWA (static) One serial worker thread (single GPU). Job state persists to //status.json so it survives a restart. """ from __future__ import annotations import json import os import subprocess import sys import time import uuid from pathlib import Path from fastapi import FastAPI, UploadFile, File, HTTPException from fastapi.responses import FileResponse, JSONResponse, PlainTextResponse from fastapi.staticfiles import StaticFiles import config import pipeline app = FastAPI(title="GenRecon", version="1.0") RUN_JOB = str(Path(__file__).resolve().parent / "run_job.py") def _job_dir(job_id: str) -> Path: return config.JOBS_DIR / job_id def _status_path(job_id: str) -> Path: return _job_dir(job_id) / "status.json" def _read(job_id: str) -> dict | None: p = _status_path(job_id) if not p.exists(): return None try: return json.loads(p.read_text()) except Exception: return None def _write(job_id: str, **patch) -> dict: st = _read(job_id) or {"id": job_id} st.update(patch) st["updated"] = time.time() _status_path(job_id).write_text(json.dumps(st)) return st def _log_line(job_id: str, line: str) -> None: with open(_job_dir(job_id) / "pipeline.log", "a") as f: f.write(line + "\n") def _pid_alive(pid: int) -> bool: try: os.kill(pid, 0) return True except (OSError, ValueError): return False def _runner_alive(job_id: str) -> bool: try: return _pid_alive(int((_job_dir(job_id) / "worker.pid").read_text().strip())) except Exception: return False def _launch_runner(job_id: str) -> None: """Run the reconstruction in a DETACHED process that outlives this web service (KillMode=process on the unit), self-serialized on the GPU by an flock. A redeploy / restart of genrecon-app therefore never kills an in-flight scan.""" subprocess.Popen( [sys.executable, RUN_JOB, job_id], cwd=str(Path(RUN_JOB).parent), stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, close_fds=True, ) @app.on_event("startup") def _startup() -> None: # Re-attach: any job that isn't finished gets a live runner. If its detached # runner survived our restart (KillMode=process) we leave it alone; otherwise # we (re)launch — so a service restart never strands a reconstruction. for d in sorted(config.JOBS_DIR.glob("*/")): st = _read(d.name) if not st or st.get("status") in ("done", "error"): continue if not _runner_alive(d.name): _launch_runner(d.name) @app.get("/api/health") def health() -> dict: active = sum(1 for d in config.JOBS_DIR.glob("*/") if (_read(d.name) or {}).get("status") in ("queued", "running")) return {"ok": True, "mock": config.MOCK, "active": active} @app.post("/api/jobs") async def create_job(video: UploadFile = File(...)) -> dict: job_id = uuid.uuid4().hex[:12] d = _job_dir(job_id) d.mkdir(parents=True, exist_ok=True) ext = (Path(video.filename or "").suffix or ".mp4").lower() dest = d / f"input{ext}" size = 0 cap = config.MAX_UPLOAD_MB * 1024 * 1024 with open(dest, "wb") as f: while chunk := await video.read(1 << 20): size += len(chunk) if size > cap: f.close() import shutil shutil.rmtree(d, ignore_errors=True) raise HTTPException(413, f"upload exceeds {config.MAX_UPLOAD_MB} MB") f.write(chunk) _write(job_id, status="queued", stage=None, progress=0, message="queued", created=time.time(), filename=video.filename, error=None) _launch_runner(job_id) return {"id": job_id, "status": "queued"} @app.get("/api/jobs/{job_id}") def get_job(job_id: str) -> dict: st = _read(job_id) if not st: raise HTTPException(404, "no such job") return st @app.get("/api/jobs/{job_id}/scene.glb") def get_scene(job_id: str): p = _job_dir(job_id) / "out" / "scene.glb" if not p.exists(): raise HTTPException(404, "scene not ready") return FileResponse(p, media_type="model/gltf-binary", filename="scene.glb") @app.get("/api/jobs/{job_id}/log") def get_log(job_id: str): p = _job_dir(job_id) / "pipeline.log" return PlainTextResponse(p.read_text() if p.exists() else "") # The PWA. Mounted last so /api/* wins. html=True serves index.html at /. if config.FRONTEND_DIR.exists(): app.mount("/", StaticFiles(directory=str(config.FRONTEND_DIR), html=True), name="frontend") else: # pragma: no cover @app.get("/") def _no_frontend() -> JSONResponse: return JSONResponse({"error": "frontend not found", "expected": str(config.FRONTEND_DIR)}, 500)