genrecon/backend/app.py

"""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
<JOBS_DIR>/<id>/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)