a16

.github/workflows/push_to_hf_space.yml

@@ -61,7 +61,7 @@ jobs:
       # -------------------------
       - name: Run unit tests
         run: |
-          pytest A4/ -v --tb=short
+          pytest A4/ A16/ -v --tb=short
 
       # -------------------------
       # 7. Push to HuggingFace

A12/service/ui.py

@@ -37,4 +37,4 @@ Outputs:
         )
 
     except Exception as e:
-        return None, None, None, {"error": str(e)}, f"### Error\n{e}"
+        return None, None, {"error": str(e)}, f"### Error\n{e}"

A15/inference.py

@@ -0,0 +1,124 @@
+"""A15 scoring — reusable inference helpers.
+
+Extracted from ``app.py`` so that other endpoints (notably A16) can reuse the
+deployed 0–4 regression scorer without importing the Gradio app module.
+
+The deployed champion architecture (Dense_medium) and scaling are described in
+``A15_results/training_summary.json``. Models live in ``<repo_root>/models/``:
+
+    - ``scoring_model.keras``  — Dense regressor (input: 390 = 10×13×3)
+    - ``scoring_scaler.pkl``   — StandardScaler fitted on flattened frames
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Tuple
+
+import numpy as np
+
+# 13-joint Kinect ordering used during A15 training.
+A15_JOINTS = [
+    'head', 'left_shoulder', 'left_elbow', 'right_shoulder', 'right_elbow',
+    'left_hand', 'right_hand', 'left_hip', 'right_hip',
+    'left_knee', 'right_knee', 'left_foot', 'right_foot',
+]
+A15_C = 10  # frames per clip the scorer was trained on
+
+_REPO_ROOT = Path(__file__).resolve().parent.parent
+_MODEL_PATH = _REPO_ROOT / 'models' / 'scoring_model.keras'
+_SCALER_PATH = _REPO_ROOT / 'models' / 'scoring_scaler.pkl'
+
+_MODEL = None
+_SCALER = None
+
+
+def load_a15_scorer():
+    """Lazy-load the deployed A15 regression scorer.
+
+    Returns
+    -------
+    (model, scaler) : tuple
+        Keras model and fitted StandardScaler. Cached after first call.
+    """
+    global _MODEL, _SCALER
+    if _MODEL is not None and _SCALER is not None:
+        return _MODEL, _SCALER
+
+    import joblib  # local import keeps app startup light
+    from tensorflow import keras
+    from tensorflow.keras import layers
+
+    try:
+        _MODEL = keras.models.load_model(str(_MODEL_PATH))
+    except (TypeError, ValueError):
+        # Saved with a newer Keras (extra ``quantization_config`` kwarg);
+        # rebuild Dense_medium and load weights only. Architecture matches
+        # ``A15_results/training_summary.json``'s deployed champion.
+        inp = keras.Input(shape=(390,))
+        x = layers.Dense(64, activation='relu')(inp)
+        x = layers.Dropout(0.2)(x)
+        out = layers.Dense(1, activation='linear')(x)
+        _MODEL = keras.Model(inp, out, name='Dense')
+        _MODEL.load_weights(str(_MODEL_PATH))
+
+    _SCALER = joblib.load(str(_SCALER_PATH))
+    return _MODEL, _SCALER
+
+
+def sample_frames(df) -> np.ndarray:
+    """Sample ``A15_C`` equally-spaced frames from a cut 3D dataframe.
+
+    Returns array of shape ``(A15_C, len(A15_JOINTS), 3)``.
+    """
+    df = df.copy()
+    df.columns = df.columns.str.strip()
+    idx = np.linspace(0, len(df) - 1, A15_C).astype(int)
+    sub = df.iloc[idx]
+    frames = []
+    for _, row in sub.iterrows():
+        frames.append([
+            [row[f'{j}_x'], row[f'{j}_y'], row[f'{j}_z']]
+            for j in A15_JOINTS
+        ])
+    return np.array(frames, dtype=np.float32)
+
+
+def score_band(score: float) -> str:
+    """Map a 0–4 score onto a traffic-light band."""
+    if score < 1.0:
+        return "GREEN — acceptable form (0-1)"
+    if score < 2.0:
+        return "AMBER — borderline (1-2)"
+    return "RED — poor form (2-4)"
+
+
+def predict_score(df) -> Tuple[float, str, float]:
+    """End-to-end scoring: cut 3D dataframe → (clipped score, band, NN ms).
+
+    Parameters
+    ----------
+    df : pandas.DataFrame
+        Cut 3D dataframe with columns ``<joint>_x/y/z`` for the 13 A15 joints.
+
+    Returns
+    -------
+    (score, band, nn_ms) : tuple
+        ``score`` is clipped to ``[0, 4]``. ``nn_ms`` is the wall-clock time
+        spent inside ``model.predict`` only (excludes sampling / scaling).
+    """
+    import time
+
+    model, scaler = load_a15_scorer()
+    frames = sample_frames(df)
+    flat = frames.reshape(1, -1)
+    scaled = scaler.transform(flat).astype(np.float32)
+    if len(model.input_shape) == 3:
+        scaled = scaled.reshape(1, A15_C, len(A15_JOINTS) * 3)
+
+    t0 = time.perf_counter()
+    raw = float(model.predict(scaled, verbose=0).flatten()[0])
+    nn_ms = (time.perf_counter() - t0) * 1000.0
+
+    score = float(np.clip(raw, 0.0, 4.0))
+    return score, score_band(score), nn_ms

A16/A16_Report.ipynb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:4c1c557bbb157bbc918fea0c355c38d55cf8ae6951c7317608e8b1a57131dcb3
+size 12468

A16/MANUAL_TEST_CHECKLIST.md

@@ -0,0 +1,147 @@
+# A16 Final Endpoint — Manual Test Checklist
+
+Use this list to verify the A16 tab end-to-end before the presentation.
+Tick each box. If a step fails, capture what you saw and let the dev know
+(do not hallucinate fixes).
+
+**Scope:** UI behaviour, response shape, and integration. Model accuracy
+is out of scope for this checklist (covered by A10/A11/A13/A15 reports).
+
+---
+
+## 0. Pre-flight
+
+- [ ] `python -m pytest A4/ A16/ -v` passes locally. Expect **11 A16 tests + the A4 tests** all green.
+- [ ] Model artefacts present in `models/`:
+  - [ ] `week16_result.h5`
+  - [ ] `week15_2d_to_3d.h5`
+  - [ ] `week17_start_and_stop.h5`
+  - [ ] `week17_start_and_stop.pkl`
+  - [ ] `A_CNN.keras`
+  - [ ] `scoring_model.keras`
+  - [ ] `scoring_scaler.pkl`
+  - [ ] `week16_scaler_X.pkl`, `week16_scaler_y.pkl`
+- [ ] MediaPipe model present: `A14/pose_landmarker_lite.task`.
+- [ ] `python app.py` launches without traceback. Console shows the existing
+      `Initialising Exercise Pipeline` banner only **when** a tab triggers it
+      (lazy load), **not** at startup.
+
+---
+
+## 1. Tab presence & layout
+
+Open `http://127.0.0.1:7860` (local) or the HF Space URL (deployed).
+
+- [ ] A new tab labelled **"A16 Final Endpoint"** is visible to the right
+      of `Exercise Scoring (A15)`.
+- [ ] The tab contains, top to bottom on each column:
+  - **Left column:** intro markdown, a `Video` upload, a `Recording quality
+    threshold` slider (default 0.6, range 0.1–0.9), a primary `Run A16
+    endpoint` button.
+  - **Right column:** `Status` textbox (read-only), a Markdown panel, a
+    `3D skeleton animation` video output, a `Full response (A16 schema)`
+    JSON viewer.
+
+---
+
+## 2. Happy path (good recording)
+
+Use any good-quality exercise clip you have used for A14 / A15 demos.
+
+- [ ] Upload the video, leave threshold at `0.6`, click **Run A16 endpoint**.
+- [ ] During processing the console prints the existing
+      `[1] Loading MediaPipe …` banner once.
+- [ ] When it finishes:
+  - [ ] **Status** textbox starts with `OK — score <number> (<BAND>...)`.
+  - [ ] **Summary** Markdown shows non-`None` values for Recording, Segment,
+        Classification, Score and Timing sections.
+  - [ ] **3D skeleton animation** plays (it is the same `<stem>_skeleton.mp4`
+        you'd get from A14).
+  - [ ] **Full response JSON** has:
+    - `endpoint == "A16"`, `variant == "3D"`, `schema_version == "1.0.0"`
+    - `status == "OK"`
+    - `score.value` is a number in `[0, 4]`
+    - `score.band` matches the threshold: `<1` GREEN, `1–2` AMBER, `≥2` RED
+    - `timing_ms.upstream_ms` > 0 and `timing_ms.total_ms` ≥ `upstream_ms`
+    - `artefacts.cut_3d_csv` and `artefacts.full_3d_csv` are non-null paths
+    - `warnings` is an empty list
+
+> If the score is `null` but everything else looks fine, check the
+> `status` field — `ERROR_SCORER` or `ERROR_TOO_SHORT_AFTER_CUT` is the
+> expected failure mode and should already show a clear message.
+
+---
+
+## 3. Ugly recording path
+
+Use a deliberately bad clip (occluded body, very dark, partial frame).
+If you don't have one, raise the threshold to `0.9` on any normal clip.
+
+- [ ] Click **Run A16 endpoint**.
+- [ ] **Status** textbox starts with `REJECTED — ugly recording (conf <n>)`.
+- [ ] **Full response JSON**:
+  - `status == "REJECTED_UGLY_RECORDING"`
+  - `recording.quality_label == "UGLY"`
+  - `recording.quality_confidence` < `recording.threshold`
+  - `score.value` is `null`
+  - `classification.label` is `null`
+  - `segment.start_frame` is `null`
+  - `timing_ms.scorer_nn_ms == 0.0`
+
+---
+
+## 4. No-video path
+
+- [ ] Click **Run A16 endpoint** without uploading anything.
+- [ ] **Status** shows `ERROR_NO_VIDEO — No video provided.`
+- [ ] **Full response JSON** `status == "ERROR_NO_VIDEO"`, all sections present but null.
+- [ ] **No traceback** appears in the console.
+
+---
+
+## 5. Existing tabs still work (regression check)
+
+Quickly verify nothing regressed in the other tabs:
+
+- [ ] `📸 Image Processing` — still renders an annotated image.
+- [ ] `🎥 Video Processing` — still renders an annotated video.
+- [ ] `🧪 Video Pipeline` (A12) — still runs end-to-end.
+- [ ] `Exercise Analysis (A14)` — still runs end-to-end.
+- [ ] `Exercise Scoring (A15)` — still returns a score.
+
+---
+
+## 6. Deployment
+
+After pushing to `main`:
+
+- [ ] GitHub Actions run `Sync to Hugging Face hub` is green:
+  - [ ] `Lint .py files` passes
+  - [ ] `Lint notebooks` passes
+  - [ ] `Run unit tests` passes (now runs `pytest A4/ A16/`)
+  - [ ] `Push to hub` succeeds
+- [ ] The HF Space rebuilds and the **A16 Final Endpoint** tab appears live.
+
+---
+
+## 7. What to capture for the presentation
+
+- [ ] Screenshot of the A16 tab on a happy-path run (status + summary + JSON).
+- [ ] Screenshot of the ugly rejection path (status + JSON).
+- [ ] Mermaid architecture diagram from `A16_Report.ipynb` §1.
+- [ ] One-line timing numbers from the happy-path JSON (`upstream_ms`,
+      `scorer_nn_ms`, `total_ms`).
+
+---
+
+## 8. If something looks wrong
+
+Do **not** guess. Note exactly:
+
+1. Which step failed (section + checkbox text).
+2. What the `status` and `message` fields said.
+3. The first 5 lines of the console output around the failure.
+4. Whether the same clip works in the A14 or A15 tab.
+
+Then ping the dev with that info — most failures map directly to a
+known model-loading issue documented in `A16_Report.ipynb` §8.

A16/service/__init__.py

@@ -0,0 +1 @@
+"""A16 final-week capstone endpoint package."""

A16/service/endpoint.py

@@ -0,0 +1,274 @@
+"""A16 — Final unified server endpoint (capstone, week 22 deliverable).
+
+Single entry point that runs the full chain in one call:
+
+    video → MediaPipe pose → PoseNet→Kinect 2D → 2D→3D
+        → start/stop cut → recording-quality (ugly) gate
+        → good/bad classifier → 0–4 score (A15)
+
+Two alternatives are scaffolded; only the **3D** variant is wired today. The
+2D-only fast path is reserved as the named stretch alternative (see
+``run_pipeline_2d``) so the public surface is stable when it lands.
+
+Design notes
+------------
+- Heavy lifting is **delegated** to the existing :class:`ExercisePipeline`
+  (``exercise_pipeline.py``). A16 only adds: (1) A15 scoring on top of the cut
+  3D CSV, (2) a unified response schema, (3) per-stage timing instrumentation,
+  (4) a stable JSON contract that the UI / future REST clients can rely on.
+- Models that fail to load are reported in the response (``warnings``) instead
+  of crashing the endpoint — the wider service must stay demo-able even when
+  individual deep-learning artefacts are out of sync with the installed Keras.
+"""
+
+from __future__ import annotations
+
+import time
+import traceback
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+REPO_ROOT = Path(__file__).resolve().parent.parent.parent
+OUTPUTS_DIR = REPO_ROOT / "outputs"
+
+# Stable response schema — bump on breaking changes.
+A16_RESPONSE_VERSION = "1.0.0"
+
+# Status enum kept narrow on purpose; UI / tests pattern-match on these.
+STATUS_OK = "OK"
+STATUS_REJECTED_UGLY = "REJECTED_UGLY_RECORDING"
+STATUS_ERROR_NO_VIDEO = "ERROR_NO_VIDEO"
+STATUS_ERROR_PIPELINE = "ERROR_PIPELINE"
+STATUS_ERROR_SCORER = "ERROR_SCORER"
+STATUS_ERROR_TOO_SHORT = "ERROR_TOO_SHORT_AFTER_CUT"
+
+
+def _empty_timing() -> Dict[str, float]:
+    return {
+        "upstream_ms": 0.0,
+        "scorer_nn_ms": 0.0,
+        "scorer_total_ms": 0.0,
+        "total_ms": 0.0,
+    }
+
+
+def _base_response(
+    status: str,
+    *,
+    variant: str = "3D",
+    message: str = "",
+) -> Dict[str, Any]:
+    """Skeleton response — every endpoint return goes through this."""
+    return {
+        "schema_version": A16_RESPONSE_VERSION,
+        "endpoint": "A16",
+        "variant": variant,
+        "status": status,
+        "message": message,
+        "recording": {
+            "quality_label": None,
+            "quality_confidence": None,
+            "threshold": None,
+        },
+        "segment": {
+            "start_frame": None,
+            "stop_frame": None,
+            "duration_frames": None,
+            "duration_sec": None,
+            "total_frames": None,
+        },
+        "classification": {
+            "label": None,
+            "confidence": None,
+        },
+        "score": {
+            "value": None,
+            "band": None,
+            "scale": "0=best, 4=worst",
+        },
+        "artefacts": {
+            "full_3d_csv": None,
+            "cut_3d_csv": None,
+            "skeleton_mp4": None,
+        },
+        "timing_ms": _empty_timing(),
+        "warnings": [],
+    }
+
+
+def _attach_upstream_fields(resp: Dict[str, Any], upstream: Dict[str, Any]) -> None:
+    """Copy the relevant slice of the ``ExercisePipeline`` result into resp."""
+    resp["recording"]["quality_label"] = upstream.get("recording_quality")
+    resp["recording"]["quality_confidence"] = upstream.get("recording_confidence")
+    resp["recording"]["threshold"] = upstream.get("recording_threshold")
+
+    resp["segment"]["start_frame"] = upstream.get("start_frame")
+    resp["segment"]["stop_frame"] = upstream.get("stop_frame")
+    resp["segment"]["duration_frames"] = upstream.get("exercise_frames")
+    resp["segment"]["duration_sec"] = upstream.get("exercise_duration_sec")
+    resp["segment"]["total_frames"] = upstream.get("total_frames")
+
+    resp["classification"]["label"] = upstream.get("quality_label")
+    resp["classification"]["confidence"] = upstream.get("quality_confidence")
+
+
+def _resolve_artefacts(resp: Dict[str, Any], video_path: str) -> Optional[Path]:
+    """Populate artefact paths from the upstream stage. Returns cut CSV path."""
+    stem = Path(video_path).stem
+    full_csv = OUTPUTS_DIR / f"{stem}_3d_points.csv"
+    cut_csv = OUTPUTS_DIR / f"{stem}_cut_3d_points.csv"
+    skel_mp4 = OUTPUTS_DIR / f"{stem}_skeleton.mp4"
+
+    resp["artefacts"]["full_3d_csv"] = str(full_csv) if full_csv.exists() else None
+    resp["artefacts"]["cut_3d_csv"] = str(cut_csv) if cut_csv.exists() else None
+    resp["artefacts"]["skeleton_mp4"] = str(skel_mp4) if skel_mp4.exists() else None
+    return cut_csv if cut_csv.exists() else None
+
+
+def run_pipeline_3d(
+    video_path: Optional[str],
+    quality_threshold: float = 0.6,
+) -> Dict[str, Any]:
+    """Run the full 3D A16 pipeline on one video.
+
+    Parameters
+    ----------
+    video_path : str or None
+        Path to the input video. ``None`` returns a structured error.
+    quality_threshold : float
+        Recording-quality threshold forwarded to :class:`ExercisePipeline`.
+
+    Returns
+    -------
+    dict
+        A response dictionary matching the schema in :data:`A16_RESPONSE_VERSION`.
+        Errors are reported via ``status`` + ``message`` rather than raised.
+    """
+    t_total = time.perf_counter()
+    resp = _base_response(STATUS_OK, variant="3D")
+
+    if not video_path:
+        resp["status"] = STATUS_ERROR_NO_VIDEO
+        resp["message"] = "No video provided."
+        resp["timing_ms"]["total_ms"] = (time.perf_counter() - t_total) * 1000.0
+        return resp
+
+    # ---- Stage 1-5: upstream pipeline (pose → 3D → cut → ugly/good-bad) ----
+    t_up = time.perf_counter()
+    try:
+        # Local import keeps test-time mocking easy and avoids importing
+        # TensorFlow when this module is merely inspected.
+        from exercise_pipeline import ExercisePipeline
+
+        pipeline = ExercisePipeline(quality_threshold=quality_threshold)
+        try:
+            upstream = pipeline.process_video(video_path)
+        finally:
+            pipeline.close()
+    except Exception as exc:  # pragma: no cover — surfaced via response
+        resp["status"] = STATUS_ERROR_PIPELINE
+        resp["message"] = f"{type(exc).__name__}: {exc}"
+        resp["warnings"].append(traceback.format_exc(limit=3))
+        resp["timing_ms"]["upstream_ms"] = (time.perf_counter() - t_up) * 1000.0
+        resp["timing_ms"]["total_ms"] = (time.perf_counter() - t_total) * 1000.0
+        return resp
+    resp["timing_ms"]["upstream_ms"] = (time.perf_counter() - t_up) * 1000.0
+
+    if not upstream:
+        resp["status"] = STATUS_ERROR_PIPELINE
+        resp["message"] = "Upstream pipeline returned no result."
+        resp["timing_ms"]["total_ms"] = (time.perf_counter() - t_total) * 1000.0
+        return resp
+
+    _attach_upstream_fields(resp, upstream)
+    cut_csv = _resolve_artefacts(resp, video_path)
+
+    # Ugly recording → early return, no scoring.
+    if upstream.get("pipeline_stopped") or upstream.get("recording_quality") == "UGLY":
+        resp["status"] = STATUS_REJECTED_UGLY
+        resp["message"] = upstream.get(
+            "reason", "Recording rejected by quality gate."
+        )
+        resp["timing_ms"]["total_ms"] = (time.perf_counter() - t_total) * 1000.0
+        return resp
+
+    # ---- Stage 6: A15 scoring on the cut 3D CSV ----
+    if cut_csv is None:
+        resp["status"] = STATUS_ERROR_PIPELINE
+        resp["message"] = "Cut 3D CSV not produced by the upstream stage."
+        resp["timing_ms"]["total_ms"] = (time.perf_counter() - t_total) * 1000.0
+        return resp
+
+    t_score = time.perf_counter()
+    try:
+        import pandas as pd
+        from A15.inference import A15_C, predict_score
+
+        df = pd.read_csv(cut_csv)
+        if len(df) < A15_C:
+            resp["status"] = STATUS_ERROR_TOO_SHORT
+            resp["message"] = (
+                f"Only {len(df)} cut frames available; scorer needs {A15_C}."
+            )
+        else:
+            score, band, nn_ms = predict_score(df)
+            resp["score"]["value"] = round(score, 4)
+            resp["score"]["band"] = band
+            resp["timing_ms"]["scorer_nn_ms"] = round(nn_ms, 2)
+    except Exception as exc:
+        resp["status"] = STATUS_ERROR_SCORER
+        resp["message"] = f"Scorer failed: {type(exc).__name__}: {exc}"
+        resp["warnings"].append(traceback.format_exc(limit=3))
+    resp["timing_ms"]["scorer_total_ms"] = round(
+        (time.perf_counter() - t_score) * 1000.0, 2
+    )
+    resp["timing_ms"]["upstream_ms"] = round(resp["timing_ms"]["upstream_ms"], 2)
+    resp["timing_ms"]["total_ms"] = round(
+        (time.perf_counter() - t_total) * 1000.0, 2
+    )
+    return resp
+
+
+def run_pipeline_2d(
+    video_path: Optional[str],
+    quality_threshold: float = 0.6,
+) -> Dict[str, Any]:
+    """Reserved 2D-only alternative endpoint.
+
+    Not implemented yet — kept as a named slot so the UI / clients can probe
+    its existence and the schema stays stable when the 2D path lands. Returns
+    a well-formed response with ``status = ERROR_PIPELINE`` and an explanatory
+    message.
+    """
+    resp = _base_response(
+        STATUS_ERROR_PIPELINE,
+        variant="2D",
+        message="2D alternative not implemented yet — see A16_Report.ipynb.",
+    )
+    return resp
+
+
+def run_pipeline(
+    video_path: Optional[str],
+    quality_threshold: float = 0.6,
+    variant: str = "3D",
+) -> Dict[str, Any]:
+    """Dispatcher — pick the 2D or 3D alternative."""
+    variant = (variant or "3D").upper()
+    if variant == "2D":
+        return run_pipeline_2d(video_path, quality_threshold=quality_threshold)
+    return run_pipeline_3d(video_path, quality_threshold=quality_threshold)
+
+
+__all__ = [
+    "A16_RESPONSE_VERSION",
+    "STATUS_OK",
+    "STATUS_REJECTED_UGLY",
+    "STATUS_ERROR_NO_VIDEO",
+    "STATUS_ERROR_PIPELINE",
+    "STATUS_ERROR_SCORER",
+    "STATUS_ERROR_TOO_SHORT",
+    "run_pipeline",
+    "run_pipeline_3d",
+    "run_pipeline_2d",
+]

A16/service/ui.py

@@ -0,0 +1,127 @@
+"""Gradio UI tab for the A16 final unified endpoint."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Tuple
+
+from A16.service.endpoint import (
+    STATUS_OK,
+    STATUS_REJECTED_UGLY,
+    run_pipeline_3d,
+)
+
+
+def _format_summary(resp: Dict[str, Any]) -> str:
+    """Render the response as a human-readable Markdown block."""
+    rec = resp["recording"]
+    seg = resp["segment"]
+    cls = resp["classification"]
+    sc = resp["score"]
+    t = resp["timing_ms"]
+
+    lines = [
+        f"### A16 Final Endpoint — {resp['variant']} variant",
+        f"**Status:** `{resp['status']}`",
+    ]
+    if resp.get("message"):
+        lines.append(f"**Message:** {resp['message']}")
+
+    lines += [
+        "",
+        "#### Recording quality (ugly gate)",
+        f"- Label: **{rec['quality_label']}**",
+        f"- Confidence: **{rec['quality_confidence']}** "
+        f"(threshold {rec['threshold']})",
+        "",
+        "#### Exercise segment (start/stop cut)",
+        f"- Frames: **{seg['start_frame']} → {seg['stop_frame']}** "
+        f"of {seg['total_frames']}",
+        f"- Duration: **{seg['duration_frames']} frames "
+        f"≈ {seg['duration_sec']} s**",
+        "",
+        "#### Good / Bad classification",
+        f"- Label: **{cls['label']}**",
+        f"- Confidence: **{cls['confidence']}**",
+        "",
+        "#### Score (0–4, lower is better)",
+        f"- Score: **{sc['value']}**",
+        f"- Band: **{sc['band']}**",
+        "",
+        "#### Timing (ms)",
+        f"- Upstream (pose + 3D + cut + ugly/good-bad): **{t['upstream_ms']}**",
+        f"- Scorer total: **{t['scorer_total_ms']}**  "
+        f"(NN only: **{t['scorer_nn_ms']}**)",
+        f"- End-to-end total: **{t['total_ms']}**",
+    ]
+    if resp.get("warnings"):
+        lines += ["", "#### Warnings"]
+        for w in resp["warnings"]:
+            lines.append(f"- {w.splitlines()[0] if w else ''}")
+    return "\n".join(lines)
+
+
+def _status_badge(resp: Dict[str, Any]) -> str:
+    """Compact one-line status string for the textbox."""
+    if resp["status"] == STATUS_OK:
+        return f"OK — score {resp['score']['value']} ({resp['score']['band']})"
+    if resp["status"] == STATUS_REJECTED_UGLY:
+        return (
+            f"REJECTED — ugly recording "
+            f"(conf {resp['recording']['quality_confidence']})"
+        )
+    return f"{resp['status']} — {resp.get('message', '')}"
+
+
+def run_a16_tab(
+    video_path: str,
+    quality_threshold: float,
+) -> Tuple[str, str, Any, Dict[str, Any]]:
+    """Gradio callback for the A16 tab.
+
+    Returns ``(status_text, summary_markdown, skeleton_video, full_json)``.
+    """
+    resp = run_pipeline_3d(video_path, quality_threshold=quality_threshold)
+    skeleton_video = resp["artefacts"].get("skeleton_mp4")
+    return _status_badge(resp), _format_summary(resp), skeleton_video, resp
+
+
+def build_a16_tab(gr):
+    """Build the A16 Gradio tab. ``gr`` is the imported ``gradio`` module.
+
+    Kept as a builder so [app.py](../../app.py) can import and mount the tab
+    without re-implementing the layout.
+    """
+    with gr.TabItem("A16 Final Endpoint"):
+        gr.Markdown(
+            """
+            ## A16 — Final unified endpoint (3D alternative)
+
+            Single video upload runs the full Part-II chain:
+            **pose → PoseNet→Kinect 2D → 2D→3D → start/stop cut →
+            ugly/good-bad → 0–4 score**.
+
+            A 2D-only alternative is reserved on the same response schema
+            (see `A16.service.endpoint.run_pipeline_2d`).
+            """
+        )
+
+        with gr.Row():
+            with gr.Column():
+                a16_video = gr.Video(label="Upload exercise video")
+                a16_threshold = gr.Slider(
+                    minimum=0.1, maximum=0.9, value=0.6, step=0.05,
+                    label="Recording quality threshold",
+                )
+                a16_run = gr.Button("Run A16 endpoint", variant="primary")
+
+            with gr.Column():
+                a16_status = gr.Textbox(label="Status", interactive=False)
+                a16_summary = gr.Markdown()
+                a16_video_out = gr.Video(label="3D skeleton animation")
+                a16_json = gr.JSON(label="Full response (A16 schema)")
+
+        a16_run.click(
+            fn=run_a16_tab,
+            inputs=[a16_video, a16_threshold],
+            outputs=[a16_status, a16_summary, a16_video_out, a16_json],
+        )

A16/tests/test_endpoint.py

@@ -0,0 +1,199 @@
+"""Unit tests for the A16 unified endpoint.
+
+These tests are intentionally **model-free**: the upstream
+``ExercisePipeline`` and the A15 scorer are monkey-patched so CI does not
+need GPU / large model artefacts. They lock in the public response schema
+and the error/ugly branches.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parent.parent.parent
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from A16.service import endpoint as ep  # noqa: E402
+
+
+# ---------- helpers ----------------------------------------------------------
+
+class _FakePipeline:
+    """Stand-in for :class:`ExercisePipeline` — returns a canned result."""
+
+    def __init__(self, result, *args, **kwargs):
+        self._result = result
+
+    def process_video(self, video_path):  # noqa: D401 — signature match
+        return self._result
+
+    def close(self):
+        pass
+
+
+def _install_fake_pipeline(monkeypatch, result):
+    """Replace the upstream pipeline with a fake returning ``result``."""
+    import types
+
+    fake_module = types.ModuleType("exercise_pipeline")
+
+    class _Factory(_FakePipeline):
+        def __init__(self, *args, **kwargs):
+            super().__init__(result)
+
+    fake_module.ExercisePipeline = _Factory
+    monkeypatch.setitem(sys.modules, "exercise_pipeline", fake_module)
+
+
+# ---------- schema -----------------------------------------------------------
+
+REQUIRED_TOP_LEVEL_KEYS = {
+    "schema_version",
+    "endpoint",
+    "variant",
+    "status",
+    "message",
+    "recording",
+    "segment",
+    "classification",
+    "score",
+    "artefacts",
+    "timing_ms",
+    "warnings",
+}
+
+REQUIRED_TIMING_KEYS = {
+    "upstream_ms", "scorer_nn_ms", "scorer_total_ms", "total_ms",
+}
+
+
+def _assert_schema(resp):
+    assert isinstance(resp, dict)
+    missing = REQUIRED_TOP_LEVEL_KEYS - set(resp.keys())
+    assert not missing, f"missing top-level keys: {missing}"
+    assert resp["endpoint"] == "A16"
+    assert resp["schema_version"] == ep.A16_RESPONSE_VERSION
+    assert resp["variant"] in {"2D", "3D"}
+    assert REQUIRED_TIMING_KEYS <= set(resp["timing_ms"].keys())
+    for section in ("recording", "segment", "classification", "score", "artefacts"):
+        assert isinstance(resp[section], dict)
+    assert isinstance(resp["warnings"], list)
+
+
+# ---------- tests ------------------------------------------------------------
+
+class TestSchema:
+
+    def test_no_video_returns_well_formed_error(self):
+        resp = ep.run_pipeline_3d(None)
+        _assert_schema(resp)
+        assert resp["status"] == ep.STATUS_ERROR_NO_VIDEO
+
+    def test_2d_alternative_returns_well_formed_placeholder(self):
+        resp = ep.run_pipeline_2d("dummy.mp4")
+        _assert_schema(resp)
+        assert resp["variant"] == "2D"
+        # 2D not implemented yet — must surface as structured error, not raise.
+        assert resp["status"] == ep.STATUS_ERROR_PIPELINE
+
+    def test_dispatcher_routes_variant(self):
+        assert ep.run_pipeline(None, variant="2D")["variant"] == "2D"
+        assert ep.run_pipeline(None, variant="3D")["variant"] == "3D"
+
+
+class TestUglyPath:
+
+    def test_ugly_recording_short_circuits(self, monkeypatch):
+        ugly_upstream = {
+            "video": "x.mp4",
+            "total_frames": 30,
+            "recording_quality": "UGLY",
+            "recording_confidence": 0.31,
+            "recording_threshold": 0.6,
+            "pipeline_stopped": True,
+            "reason": "Poor recording quality.",
+        }
+        _install_fake_pipeline(monkeypatch, ugly_upstream)
+        resp = ep.run_pipeline_3d("does-not-need-to-exist.mp4")
+        _assert_schema(resp)
+        assert resp["status"] == ep.STATUS_REJECTED_UGLY
+        assert resp["recording"]["quality_label"] == "UGLY"
+        assert resp["recording"]["quality_confidence"] == 0.31
+        # No score should have been computed on the ugly branch.
+        assert resp["score"]["value"] is None
+        assert resp["timing_ms"]["scorer_nn_ms"] == 0.0
+
+
+class TestHappyPath:
+
+    def test_full_pipeline_with_mocked_scorer(self, monkeypatch, tmp_path):
+        # Fake cut CSV with 10 frames and the expected 13-joint xyz columns.
+        import pandas as pd
+        from A15.inference import A15_JOINTS, A15_C
+
+        cols = [f"{j}_{ax}" for j in A15_JOINTS for ax in ("x", "y", "z")]
+        df = pd.DataFrame(
+            [[0.0] * len(cols) for _ in range(A15_C)],
+            columns=cols,
+        )
+        cut_csv = tmp_path / "demo_cut_3d_points.csv"
+        df.to_csv(cut_csv, index=False)
+
+        # Point the endpoint's artefact resolution at our tmp dir.
+        monkeypatch.setattr(ep, "OUTPUTS_DIR", tmp_path)
+
+        good_upstream = {
+            "video": "demo.mp4",
+            "total_frames": 90,
+            "start_frame": 10,
+            "stop_frame": 70,
+            "exercise_frames": 61,
+            "exercise_duration_sec": 2.03,
+            "quality_label": "GOOD",
+            "quality_confidence": 0.87,
+            "recording_quality": "GOOD",
+            "recording_confidence": 0.78,
+            "recording_threshold": 0.6,
+        }
+        _install_fake_pipeline(monkeypatch, good_upstream)
+
+        # Mock the A15 scorer so we don't load Keras / joblib in CI.
+        import A15.inference as inf
+        monkeypatch.setattr(
+            inf, "predict_score", lambda d: (0.42, "GREEN — acceptable form (0-1)", 1.5)
+        )
+
+        # `_resolve_artefacts` uses the video stem; mirror it in tmp.
+        video_path = str(tmp_path / "demo.mp4")
+        # Need the upstream stem-prefixed CSV to exist where _resolve_artefacts looks.
+        (tmp_path / "demo_cut_3d_points.csv").write_text(cut_csv.read_text())
+
+        resp = ep.run_pipeline_3d(video_path)
+        _assert_schema(resp)
+        assert resp["status"] == ep.STATUS_OK
+        assert resp["classification"]["label"] == "GOOD"
+        assert resp["segment"]["start_frame"] == 10
+        assert resp["segment"]["stop_frame"] == 70
+        assert resp["score"]["value"] == pytest.approx(0.42)
+        assert "GREEN" in resp["score"]["band"]
+        assert resp["timing_ms"]["scorer_nn_ms"] == 1.5
+        assert resp["timing_ms"]["total_ms"] >= 0
+
+
+class TestBandMapping:
+
+    @pytest.mark.parametrize("score,prefix", [
+        (0.0, "GREEN"),
+        (0.99, "GREEN"),
+        (1.0, "AMBER"),
+        (1.99, "AMBER"),
+        (2.0, "RED"),
+        (4.0, "RED"),
+    ])
+    def test_score_band_boundaries(self, score, prefix):
+        from A15.inference import score_band
+        assert score_band(score).startswith(prefix)

app.py

@@ -4,6 +4,7 @@ from A8.pose_estimator import MoveNetPoseEstimator
 from A12.pose_interpolator import smooth_pose_sequence
 #http://127.0.0.1:7860from A12.service.ui import run_a12_tab
 from A12.service.ui import run_a12_video_tab
+from A16.service.ui import build_a16_tab
 from exercise_pipeline import ExercisePipeline
 import json
 import csv
@@ -751,6 +752,9 @@ with gr.Blocks(title="MoveNet Pose Estimation") as demo:
                 outputs=[a15_band, a15_score, a15_timing, a15_json],
             )
 
+        # A16 Final unified endpoint (capstone)
+        build_a16_tab(gr)
+
 
     # Example section
     with gr.Accordion("ℹ️ Information", open=False):

pytest.ini

@@ -1,5 +1,5 @@
 [pytest]
-testpaths = A4
+testpaths = A4 A16
 
 python_files = test_*.py
 python_classes = Test*