inital commit at migration

.dockerignore

@@ -0,0 +1,5 @@
+.env
+.env.*
+__pycache__/
+.venv/
+test_app.py

.env.example

@@ -0,0 +1,5 @@
+CORE_REPO_ID=your-account/sporalize-core
+STORAGE_REPO_ID=your-account/sporalize-application-storage
+HF_TOKEN=
+ANALYSIS_PROXY_SECRET=
+BACKEND_URL=http://localhost:4000

.gitignore

@@ -0,0 +1,3 @@
+.env
+__pycache__/
+.venv/

Dockerfile

@@ -0,0 +1,11 @@
+FROM python:3.10-slim
+ENV PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1 HF_HOME=/home/appuser/.cache/huggingface
+RUN apt-get update && apt-get install -y --no-install-recommends libglib2.0-0 libgl1 && rm -rf /var/lib/apt/lists/*
+RUN useradd --create-home --uid 1000 appuser
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+COPY --chown=appuser:appuser app.py .
+USER appuser
+EXPOSE 7860
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860", "--workers", "1"]

README.md

@@ -0,0 +1,21 @@
+---
+title: Sporalize AI API
+sdk: docker
+app_port: 7860
+---
+
+# Sporalize AI API
+
+Deploy `AI/API` as a public Docker Space and `AI/Core` as a private dataset. The API loads `pipeline.py` and weights from Core at startup. The public service is accessible only through the Node backend using `ANALYSIS_PROXY_SECRET`; the health endpoint is public. Supabase credentials stay in Node.
+
+Copy `.env.example` to `.env` for local use. Set `HF_TOKEN` with read access to Core and write access to a new private storage dataset. Set `BACKEND_URL` and the same `ANALYSIS_PROXY_SECRET` as Node. `CORE_REVISION` can pin the private Core dataset revision.
+
+Model settings are constants near the top of `app.py`: `POSE_MODEL_FILE`, `POSE_MODEL_REPO_ID`, `YOLO_MODEL_FILE`, `YOLO_MODEL_REPO_ID`, and `YOLO_IMGSZ`. Change those values to select weights or detector input size. Video uploads do not have an application-level size limit.
+
+The storage dataset contains only the pipeline's `results.json` and source videos under `sessions/<session UUID>/`. Calibration and action input files are temporary. `results.json` is the direct pipeline return after NumPy values are serialized and non-finite values become `null`.
+
+`POST /api/analyze` accepts `playerId`, `clientId`, `videoOrders`, either uploaded `videos` or `videoDriveLinks`, `actionsJson`, `calibration`, and optional `sessionName`. Drive videos must be publicly downloadable. The service asks Node to authorize/create the session, saves results and videos to the HF dataset, then sends the path and action frame/FPS details to Node. Node owns the database and stores frame indices and FPS.
+
+The authorized session endpoint adapts stored reports to the Results screen shape and provides private video URLs. Other endpoints include `/api/progress/{clientId}`, `/api/active-analyses`, `/api/cancel/{clientId}`, and `/api/retry-save/{clientId}`. All except `/healthz` require trusted proxy headers. Run one process because jobs and retry payloads are held in memory and do not survive a restart.
+
+Apply `../../Database/migrations/20261016_node_analysis.sql` in Supabase. The older `20261016_analysis_persistence.sql` migration is not used.

app.py

@@ -0,0 +1,409 @@
+import hmac
+import hashlib
+import importlib.util
+import json
+import logging
+import math
+import os
+import re
+import shutil
+import tempfile
+import time
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import asynccontextmanager
+from datetime import datetime, timezone
+from functools import lru_cache
+from pathlib import Path
+from urllib.request import urlretrieve
+from uuid import UUID, uuid5
+from zipfile import ZipFile
+
+import httpx
+import numpy as np
+from dotenv import load_dotenv
+from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile
+from fastapi.responses import FileResponse, JSONResponse
+from huggingface_hub import HfApi, hf_hub_download, snapshot_download
+
+load_dotenv(Path(__file__).with_name(".env"))
+jobs = {}
+executor = ThreadPoolExecutor(max_workers=1)
+logger = logging.getLogger("sporalize")
+POSE_MODEL_FILE = "Weights/rtmpose-x-halpe26.onnx"
+POSE_MODEL_REPO_ID = ""
+POSE_MODEL_HUB_FILE = "rtmpose-x-halpe26.onnx"
+POSE_MODEL_REPO_TYPE = "model"
+YOLO_MODEL_FILE = "Weights/yolo11x.pt"
+YOLO_MODEL_REPO_ID = "Ultralytics/YOLO11"
+YOLO_MODEL_HUB_FILE = "yolo11x.pt"
+YOLO_MODEL_REPO_TYPE = "model"
+YOLO_IMGSZ = 896
+
+
+def hub():
+    return HfApi(token=os.environ["HF_TOKEN"])
+
+
+def backend(path, user_id, method="GET", body=None):
+    for attempt in range(3):
+        try:
+            response = httpx.request(method, f'{os.environ["BACKEND_URL"].rstrip("/")}/internal/analysis{path}',
+                json=body, timeout=60, headers={"x-sporalize-proxy-secret": os.environ["ANALYSIS_PROXY_SECRET"], "x-sporalize-user-id": user_id})
+            response.raise_for_status()
+            return response.json() if response.content else None
+        except (httpx.TransportError, httpx.HTTPStatusError) as error:
+            if attempt == 2 or isinstance(error, httpx.HTTPStatusError) and error.response.status_code < 500:
+                raise
+            time.sleep(2 ** attempt)
+
+
+@lru_cache(maxsize=1)
+def runtime():
+    root = Path(snapshot_download(repo_id=os.environ["CORE_REPO_ID"], repo_type="dataset",
+        revision=os.environ.get("CORE_REVISION"), token=os.environ["HF_TOKEN"],
+        allow_patterns=["pipeline.py", "Weights/*"]))
+    spec = importlib.util.spec_from_file_location("sporalize_core", root / "pipeline.py")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    pose_file = POSE_MODEL_FILE
+    yolo_file = YOLO_MODEL_FILE
+    pose = (root / pose_file).resolve()
+    yolo = (root / yolo_file).resolve()
+    if root.resolve() not in pose.parents or root.resolve() not in yolo.parents:
+        raise RuntimeError("Model files must be inside the Core repository")
+    if not pose.is_file():
+        cache = Path(os.environ.get("HF_HOME", str(Path.home() / ".cache/huggingface"))) / "sporalize"
+        cache.mkdir(parents=True, exist_ok=True)
+        if not (cache / Path(pose_file).name).is_file():
+            if POSE_MODEL_REPO_ID:
+                downloaded = hf_hub_download(POSE_MODEL_REPO_ID, POSE_MODEL_HUB_FILE,
+                    repo_type=POSE_MODEL_REPO_TYPE, token=os.environ["HF_TOKEN"])
+                shutil.copyfile(downloaded, cache / Path(pose_file).name)
+            elif pose_file == "Weights/rtmpose-x-halpe26.onnx":
+                with tempfile.TemporaryDirectory() as temporary:
+                    archive = Path(temporary) / "pose.zip"
+                    urlretrieve("https://download.openmmlab.com/mmpose/v1/projects/rtmposev1/onnx_sdk/rtmpose-x_simcc-body7_pt-body7-halpe26_700e-384x288-7fb6e239_20230606.zip", archive)
+                    with ZipFile(archive) as bundle:
+                        member = next(name for name in bundle.namelist() if name.endswith("end2end.onnx"))
+                        with bundle.open(member) as source, (cache / Path(pose_file).name).open("wb") as target:
+                            shutil.copyfileobj(source, target)
+            else:
+                raise RuntimeError("Configured pose model is missing from Core and no pose model repository was configured")
+        pose = cache / Path(pose_file).name
+    if not yolo.is_file():
+        yolo = Path(hf_hub_download(YOLO_MODEL_REPO_ID, YOLO_MODEL_HUB_FILE,
+            repo_type=YOLO_MODEL_REPO_TYPE, token=os.environ["HF_TOKEN"]))
+    return module.run_pipeline, {"POSE_PATH": str(pose), "YOLO_PATH": str(yolo)}
+
+
+@asynccontextmanager
+async def lifespan(app):
+    for name in ("CORE_REPO_ID", "STORAGE_REPO_ID", "HF_TOKEN", "BACKEND_URL"):
+        if not os.environ.get(name):
+            raise RuntimeError(f"{name} is required")
+    if len(os.environ.get("ANALYSIS_PROXY_SECRET", "")) < 32:
+        raise RuntimeError("ANALYSIS_PROXY_SECRET must contain at least 32 characters")
+    if os.environ["STORAGE_REPO_ID"].lower() == "shoraky/sporalizelabs-storage":
+        raise RuntimeError("Use a separate storage dataset for this application")
+    if not 32 <= len(os.environ["ANALYSIS_PROXY_SECRET"]) <= 4096:
+        raise RuntimeError("ANALYSIS_PROXY_SECRET length is invalid")
+    runtime()
+    yield
+
+
+app = FastAPI(title="Sporalize AI", lifespan=lifespan)
+
+
+@app.exception_handler(httpx.HTTPStatusError)
+async def backend_error(request: Request, error: httpx.HTTPStatusError):
+    status = error.response.status_code
+    return JSONResponse({"detail": "Backend request failed"}, status_code=status if status < 500 else 502)
+
+
+@app.middleware("http")
+async def authenticate(request: Request, call_next):
+    if request.url.path != "/healthz":
+        secret = os.environ.get("ANALYSIS_PROXY_SECRET", "")
+        if len(secret) < 32 or not hmac.compare_digest(secret, request.headers.get("x-sporalize-proxy-secret", "")):
+            return JSONResponse({"detail": "Unauthorized"}, status_code=403)
+        try:
+            request.state.user_id = str(UUID(request.headers.get("x-sporalize-user-id", "")))
+        except ValueError:
+            return JSONResponse({"detail": "Invalid user identity"}, status_code=403)
+    return await call_next(request)
+
+
+def json_value(value):
+    if isinstance(value, np.ndarray):
+        return json_value(value.tolist())
+    if isinstance(value, np.generic):
+        return json_value(value.item())
+    if isinstance(value, dict):
+        return {str(key): json_value(item) for key, item in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [json_value(item) for item in value]
+    if isinstance(value, float) and not math.isfinite(value):
+        return None
+    return value
+
+
+def drive_file_id(link):
+    value = link.strip()
+    if re.fullmatch(r"[A-Za-z0-9_-]{10,}", value):
+        return value
+    if not any(host in value for host in ("drive.google.com", "drive.usercontent.google.com")):
+        raise ValueError("Not a Google Drive file link")
+    match = re.search(r"/file/d/([A-Za-z0-9_-]{10,})|/d/([A-Za-z0-9_-]{10,})|[?&]id=([A-Za-z0-9_-]{10,})", value)
+    if not match:
+        raise ValueError("Could not find a Google Drive file ID")
+    return next(group for group in match.groups() if group)
+
+
+async def download_drive_file(link, destination):
+    file_id = drive_file_id(link)
+    url = f"https://drive.google.com/uc?export=download&confirm=t&id={file_id}"
+    total = 0
+    async with httpx.AsyncClient(follow_redirects=True, timeout=300) as client:
+        async with client.stream("GET", url) as response:
+            response.raise_for_status()
+            content_type = response.headers.get("content-type", "").lower()
+            if "text/html" in content_type:
+                raise ValueError("Google Drive did not return a public video; check link access")
+            with destination.open("wb") as output:
+                async for chunk in response.aiter_bytes(1024 * 1024):
+                    total += len(chunk)
+                    output.write(chunk)
+    if total == 0:
+        raise ValueError("Google Drive returned an empty file")
+    return file_id, total
+
+
+def run_job(client_id, work, cameras):
+    job = jobs[client_id]
+    session_id = job["sessionId"]
+    saved = False
+    try:
+        if job.get("cancel"):
+            raise InterruptedError()
+        job.update(status="running", progress=5, phase="Analyzing")
+        pipeline, weights = runtime()
+
+        def progress(action, total, frame, frames):
+            job.update(progress=5 + 85 * (action + (frame + 1) / max(frames, 1)) / max(total, 1),
+                       updatedAt=int(time.time() * 1000))
+            return not job.get("cancel")
+
+        reports = pipeline(cameras, {**weights, "CALIBRATION_PATH": str(work / "calibration.npz"), "ACTIONS_PATH": str(work / "actions.json")}, {"YOLO_IMGSZ": YOLO_IMGSZ}, progress)
+        if job.get("cancel"):
+            raise InterruptedError()
+        (work / "results.json").write_text(json.dumps(json_value(reports), allow_nan=False), encoding="utf-8")
+        job.update(progress=95, phase="Saving")
+        hub().upload_folder(repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset",
+            folder_path=str(work), path_in_repo=f"sessions/{session_id}",
+            allow_patterns=["results.json", "videos/*"])
+        saved = True
+        completion = {"status": "completed", "path": f"sessions/{session_id}", "actions": [
+            {"index": index, **{key: json_value(report[key]) for key in ("action", "start_frame", "end_frame", "fps", "error") if key in report}}
+            for index, report in enumerate(reports)]}
+        job["completion"] = completion
+        backend(f"/{session_id}", job["userId"], "PUT", completion)
+        job.pop("completion", None)
+        job.update(status="completed", phase="Completed", progress=100, resultUrl=f"/api/sessions/{session_id}")
+    except Exception as error:
+        status = "cancelled" if isinstance(error, InterruptedError) else "failed"
+        job.update(status=status, phase="Sync failed" if saved else status.capitalize(), error="Files saved; retry backend sync" if saved else "Analysis cancelled" if status == "cancelled" else "Analysis or saving failed")
+        if saved:
+            job["status"] = "failed"
+        logger.exception("Session %s %s", session_id, status)
+        try:
+            if not saved:
+                backend(f"/{session_id}", job["userId"], "PUT", {"status": status})
+        except Exception:
+            logger.exception("Could not update session %s", session_id)
+    finally:
+        job["updatedAt"] = int(time.time() * 1000)
+        shutil.rmtree(work, ignore_errors=True)
+
+
+@app.get("/healthz")
+def health():
+    return {"ok": True}
+
+
+@app.post("/api/drive/probe")
+def probe_drive(payload: dict):
+    links = payload.get("links") or ([payload["link"]] if payload.get("link") else [])
+    if not isinstance(links, list) or not 1 <= len(links) <= 16:
+        raise HTTPException(400, "Provide between 1 and 16 Drive links")
+    results = []
+    for link in links:
+        result = {"link": str(link)}
+        try:
+            file_id = drive_file_id(str(link))
+            with httpx.stream("GET", f"https://drive.google.com/uc?export=download&confirm=t&id={file_id}", follow_redirects=True, timeout=30) as response:
+                content_type = response.headers.get("content-type", "")
+                size = int(response.headers.get("content-length", "0") or 0)
+                if not response.is_success or "text/html" in content_type.lower():
+                    raise ValueError("File is not publicly downloadable")
+            result.update(ok=True, fileId=file_id, fileName=f"{file_id}.mp4", sizeBytes=size or None, contentType=content_type)
+        except Exception as error:
+            result.update(ok=False, code="unavailable", error=str(error))
+        results.append(result)
+    return {"results": results}
+
+
+@app.post("/api/analyze", status_code=202)
+async def analyze(request: Request, playerId: str = Form(...), clientId: UUID = Form(...),
+                  videoOrders: list[int] = Form(...), videos: list[UploadFile] = File(None),
+                  videoDriveLinks: list[str] = Form(None),
+                  actionsJson: UploadFile = File(...), calibration: UploadFile = File(...),
+                  sessionName: str = Form("")):
+    videos = videos or []
+    videoDriveLinks = videoDriveLinks or []
+    if videos and videoDriveLinks or len(videos or videoDriveLinks) < 2 or len(videos or videoDriveLinks) != len(videoOrders) or len(set(videoOrders)) != len(videoOrders):
+        raise HTTPException(422, "Provide at least two videos with unique camera orders")
+    client_id, user_id = str(clientId), request.state.user_id
+    if client_id in jobs:
+        raise HTTPException(409, "Analysis already exists")
+    reference = playerId.strip()
+    if not reference or len(reference) > 100 or len(sessionName) > 150:
+        raise HTTPException(422, "Invalid player reference or session name")
+    session = backend("", user_id, "POST", {"clientId": client_id, "playerId": reference, "name": sessionName})
+    session_id, player_id = session["session_id"], session["player_id"]
+    now = int(time.time() * 1000)
+    jobs[client_id] = {"clientId": client_id, "sessionId": session_id, "playerId": player_id,
+        "userId": user_id, "status": "uploading", "phase": "Uploading", "progress": 0, "createdAt": now, "updatedAt": now}
+    work = Path(tempfile.mkdtemp(prefix="sporalize-"))
+    try:
+        (work / "videos").mkdir()
+        cameras = {}
+        uploads = [(actionsJson, work / "actions.json"), (calibration, work / "calibration.npz")]
+        video_names = []
+        sources = list(zip(videoOrders, videos)) if videos else list(zip(videoOrders, videoDriveLinks))
+        for order, source in sources:
+            if videos:
+                video = source
+                suffix = Path(video.filename or "video.mp4").suffix.lower()
+            else:
+                video = None
+                suffix = ".mp4"
+            if suffix not in {".mp4", ".mov", ".avi", ".mkv", ".webm", ".m4v"}:
+                raise HTTPException(422, "Unsupported video format")
+            target = work / "videos" / f"camera_{order}{suffix}"
+            video_names.append(target.name)
+            cameras[order] = str(target)
+            if video is not None:
+                uploads.append((video, target))
+            else:
+                await download_drive_file(source, target)
+        for upload, target in uploads:
+            with target.open("wb") as output:
+                while chunk := await upload.read(1024 * 1024):
+                    output.write(chunk)
+        jobs[client_id].update(status="queued", phase="Queued")
+        jobs[client_id]["videoNames"] = video_names
+        executor.submit(run_job, client_id, work, cameras)
+        return {"accepted": True, "status": "queued", "clientId": client_id, "sessionId": session_id}
+    except Exception:
+        jobs[client_id].update(status="failed", phase="Failed")
+        shutil.rmtree(work)
+        try:
+            backend(f"/{session_id}", user_id, "PUT", {"status": "failed"})
+        except Exception:
+            logger.exception("Could not report failed upload %s", session_id)
+        raise
+
+
+def owned_job(client_id, user_id):
+    job = jobs.get(client_id)
+    if not job or job["userId"] != user_id:
+        raise HTTPException(404, "Analysis not found")
+    return job
+
+
+@app.get("/api/progress/{client_id}")
+def progress(client_id: str, request: Request):
+    return {key: value for key, value in owned_job(client_id, request.state.user_id).copy().items() if key not in {"userId", "cancel", "completion"}}
+
+
+@app.post("/api/retry-save/{client_id}")
+def retry_save(client_id: str, request: Request):
+    job = owned_job(client_id, request.state.user_id)
+    if job.get("completion"):
+        backend(f'/{job["sessionId"]}', job["userId"], "PUT", job["completion"])
+        job.pop("completion", None)
+        job.update(status="completed", phase="Completed", progress=100, resultUrl=f'/api/sessions/{job["sessionId"]}')
+    return progress(client_id, request)
+
+
+@app.get("/api/active-analyses")
+def active(request: Request):
+    return {"jobs": [progress(key, request) for key, job in list(jobs.items()) if job["userId"] == request.state.user_id and job["status"] in {"uploading", "queued", "running"}]}
+
+
+@app.post("/api/cancel/{client_id}")
+def cancel(client_id: str, request: Request):
+    job = owned_job(client_id, request.state.user_id)
+    if job["status"] not in {"completed", "failed", "cancelled", "sync_failed"}:
+        job["cancel"] = True
+    return {"status": job["status"], "clientId": client_id}
+
+
+def session_row(session_id, user_id):
+    row = backend(f"/{session_id}", user_id)
+    if row["path"] != f"sessions/{session_id}":
+        raise HTTPException(409, "Session uses a different storage layout")
+    return row
+
+
+@app.get("/api/sessions/{session_id}")
+def results(session_id: UUID, request: Request):
+    row = session_row(session_id, request.state.user_id)
+    path = hf_hub_download(repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset",
+        filename=f'{row["path"]}/results.json', token=os.environ["HF_TOKEN"])
+    reports = json.loads(Path(path).read_text(encoding="utf-8"))
+    prefix = f'{row["path"]}/videos/camera_'
+    camera_files = sorted({name[len(prefix):] for name in hub().list_repo_files(
+        repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset") if name.startswith(prefix)})
+    fps_values = [float(report.get("fps") or 30) for report in reports if report.get("fps")]
+    actions, failed = [], []
+    for index, report in enumerate(reports):
+        start = int(report.get("start_frame") or 0)
+        end = int(report.get("end_frame") or start)
+        action_fps = float(report.get("fps") or (fps_values[0] if fps_values else 30))
+        label = report.get("action", "Unknown")
+        if report.get("error"):
+            failed.append({"id": str(uuid5(session_id, str(index))), "label": label, "start": str(start), "end": str(end), "error": report["error"]})
+            continue
+        analytics = report.get("analytics") or {}
+        metrics = [{"name": key.replace("_", " ").title(), "value": value, "unit": ""} for key, value in analytics.items() if isinstance(value, (int, float, str, bool)) or value is None]
+        clips = [{"cameraId": int(name.split("_", 1)[1].split(".", 1)[0]), "url": f"/storage/{session_id}/videos/{name}"} for name in camera_files]
+        digest = hashlib.sha256(f"{session_id}:{index}".encode()).hexdigest()
+        action_id = f"{digest[:8]}-{digest[8:12]}-5{digest[13:16]}-8{digest[17:20]}-{digest[20:32]}"
+        actions.append({"id": action_id, "label": label,
+            "start": str(start), "end": str(end), "fps": action_fps,
+            "startFrame": start, "endFrame": end, "startSeconds": start / action_fps,
+            "endSeconds": (end + 1) / action_fps, "totalFrames": max(0, end - start + 1),
+            "preFrames": 0, "inFrame": max(0, (end - start) // 2), "postFrames": 0,
+            "cameraClips": clips, "sourceCameraClips": [], "preMetrics": [],
+            "preActionMetrics": [], "inActionMetrics": metrics, "postMetrics": [],
+            "postActionMetrics": [], "fullIntervalMetrics": [],
+            "skeleton": report.get("skeleton_history", []), "rawAnalytics": analytics})
+    created = row.get("created_at")
+    created_at = datetime.fromisoformat(created.replace("Z", "+00:00")) if created else datetime.now(timezone.utc)
+    if created_at.tzinfo is None:
+        created_at = created_at.replace(tzinfo=timezone.utc)
+    return {"id": str(session_id), "playerId": row["player_id"], "playerName": "",
+        "name": row.get("name", "Analysis session"),
+        "createdAt": int(created_at.timestamp() * 1000),
+        "targetSize": [0, 0], "cameraCount": len(camera_files), "actions": actions, "failedActions": failed}
+
+
+@app.api_route("/storage/{session_id}/videos/{filename}", methods=["GET", "HEAD"])
+def video(session_id: UUID, filename: str, request: Request):
+    row = session_row(session_id, request.state.user_id)
+    if Path(filename).name != filename or "\\" in filename:
+        raise HTTPException(404, "Video not found")
+    path = hf_hub_download(repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset",
+        filename=f'{row["path"]}/videos/{filename}', token=os.environ["HF_TOKEN"])
+    return FileResponse(path)

requirements.txt

@@ -0,0 +1,14 @@
+fastapi>=0.115.0
+uvicorn>=0.30.0
+python-multipart>=0.0.9
+python-dotenv>=1.0.0
+huggingface_hub>=0.34.0
+httpx>=0.27.0
+numpy
+scipy
+opencv-python-headless
+onnxruntime
+rtmlib>=0.0.14
+torch
+ultralytics
+lap>=0.5.12