Revert repo to stable version c1edb3 (Issue rollback)

A12/service/__init__.py

@@ -1,4 +1,6 @@
-from .pipeline import run_video_pipeline, validate_video
-from .ui import run_a12_video_tab
+"""A12 service endpoint package."""
 
-__all__ = ["run_video_pipeline", "validate_video", "run_a12_video_tab"]
+from A12.service.model_service import predict_pose_csv, safe_predict_pose_csv
+from A12.service.ui import run_a12_tab
+
+__all__ = ["predict_pose_csv", "safe_predict_pose_csv", "run_a12_tab"]

A12/service/contracts.py

@@ -0,0 +1,134 @@
+"""Input/output contracts for the A12 Gradio service tab.
+
+The selected endpoint flavour is a Gradio tab inside app.py.  The endpoint
+accepts a pose-feature CSV, validates that it has the feature columns used by
+Rasa's A12 classifiers, and returns a structured prediction dictionary.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Dict, Iterable, List, Tuple
+
+import pandas as pd
+
+JOINTS: List[str] = [
+    "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",
+]
+
+KINECT_COLS: List[str] = [axis for joint in JOINTS for axis in (f"{joint}_x", f"{joint}_y", f"{joint}_z")]
+POSENET_COLS: List[str] = [axis for joint in JOINTS for axis in (f"{joint}_x", f"{joint}_y")]
+
+EXTRA_COLS: List[str] = [
+    "left_hand_to_left_shoulder",
+    "right_hand_to_right_shoulder",
+    "left_hand_to_left_hip",
+    "right_hand_to_right_hip",
+    "left_elbow_to_left_shoulder",
+    "right_elbow_to_right_shoulder",
+    "head_to_hip",
+    "head_vx",
+    "head_vy",
+    "head_vz",
+    "head_speed",
+    "left_hand_vx",
+    "left_hand_vy",
+    "left_hand_vz",
+    "left_hand_speed",
+    "right_hand_vx",
+    "right_hand_vy",
+    "right_hand_vz",
+    "right_hand_speed",
+    "head_ax",
+    "head_ay",
+    "head_az",
+    "head_accel",
+    "left_hand_ax",
+    "left_hand_ay",
+    "left_hand_az",
+    "left_hand_accel",
+    "right_hand_ax",
+    "right_hand_ay",
+    "right_hand_az",
+    "right_hand_accel",
+]
+
+FEATURES_BY_PROBLEM: Dict[str, List[str]] = {
+    "A": KINECT_COLS + EXTRA_COLS,
+    "B": POSENET_COLS + EXTRA_COLS,
+}
+
+LABEL_NAMES = ["non-exercise", "exercise"]
+
+
+def normalize_problem(problem: str) -> str:
+    """Return canonical problem name A/B or raise ValueError."""
+    value = str(problem).strip().upper()
+    if value.startswith("A"):
+        return "A"
+    if value.startswith("B"):
+        return "B"
+    raise ValueError("Problem must be 'A' or 'B'.")
+
+
+def read_pose_csv(csv_path: str | Path) -> pd.DataFrame:
+    """Read a pose feature CSV and strip whitespace from column names."""
+    if not csv_path:
+        raise ValueError("Please upload a pose CSV file.")
+
+    path = Path(csv_path)
+    if path.suffix.lower() != ".csv":
+        raise ValueError("The uploaded file must be a .csv pose-feature file.")
+    if not path.exists():
+        raise ValueError(f"CSV file not found: {path}")
+
+    df = pd.read_csv(path)
+    df.columns = df.columns.str.strip()
+    if df.empty:
+        raise ValueError("The pose CSV is empty.")
+    return df
+
+
+def missing_columns(df: pd.DataFrame, expected: Iterable[str]) -> List[str]:
+    """Return expected columns that are not present in *df*."""
+    return [column for column in expected if column not in df.columns]
+
+
+def validate_pose_dataframe(df: pd.DataFrame, problem: str) -> Tuple[pd.DataFrame, List[str]]:
+    """Validate and return numeric feature matrix as a DataFrame.
+
+    The current A12 saved scalers expect all base pose columns plus engineered
+    columns from A12_classifier.py.  This function fails fast with a useful
+    message instead of silently filling missing model features.
+    """
+    problem_key = normalize_problem(problem)
+    expected = FEATURES_BY_PROBLEM[problem_key]
+    missing = missing_columns(df, expected)
+    if missing:
+        preview = ", ".join(missing[:8])
+        suffix = "..." if len(missing) > 8 else ""
+        raise ValueError(
+            f"CSV is missing {len(missing)} required columns for Problem {problem_key}: "
+            f"{preview}{suffix}"
+        )
+
+    features = df[expected].apply(pd.to_numeric, errors="coerce")
+    if features.isna().any().any():
+        bad = features.columns[features.isna().any()].tolist()[:8]
+        raise ValueError(
+            "Pose CSV contains non-numeric or missing values in required columns: "
+            + ", ".join(bad)
+        )
+    return features, expected

A12/service/model_service.py

@@ -0,0 +1,159 @@
+"""Prediction service for A12 start/stop exercise classifiers."""
+
+from __future__ import annotations
+
+import time
+from functools import lru_cache
+from pathlib import Path
+from typing import Any, Dict
+
+import joblib
+import numpy as np
+
+from A12.service.contracts import LABEL_NAMES, normalize_problem, read_pose_csv, validate_pose_dataframe
+
+MODEL_VERSION = "A12 Dense classifiers from A12_results"
+
+MODEL_FILES = {
+    "A": {
+        "name": "A_Kinect_Dense_relu_adam_bs64",
+        "weights": "A_Kinect_Dense_relu_adam_bs64.weights.h5",
+        "scaler": "A_Kinect_Dense_relu_adam_bs64_scaler.pkl",
+    },
+    "B": {
+        "name": "B_PoseNet_Dense_relu_adam_bs64",
+        "weights": "B_PoseNet_Dense_relu_adam_bs64.weights.h5",
+        "scaler": "B_PoseNet_Dense_relu_adam_bs64_scaler.pkl",
+    },
+}
+
+
+def _project_root() -> Path:
+    return Path(__file__).resolve().parents[2]
+
+
+def _model_dir() -> Path:
+    """Prefer A12/A12_results, but also support A12_results at repo root."""
+    root = _project_root()
+    candidates = [root / "A12" / "A12_results", root / "A12_results"]
+    for candidate in candidates:
+        if candidate.exists():
+            return candidate
+    return candidates[0]
+
+
+def _build_dense_model(input_dim: int):
+    """Build the same dense architecture used by A12_classifier.py."""
+    import tensorflow as tf
+    from tensorflow import keras
+    from tensorflow.keras import layers, regularizers
+
+    # Keeping seed stable avoids warnings about unseeded initialisers before
+    # loading trained weights.
+    tf.random.set_seed(42)
+    inputs = keras.Input(shape=(input_dim,), name="input")
+    x = layers.Dense(
+        128,
+        activation="relu",
+        kernel_regularizer=regularizers.l2(1e-4),
+        name="dense_1",
+    )(inputs)
+    x = layers.Dropout(0.2, name="drop_1")(x)
+    x = layers.Dense(
+        64,
+        activation="relu",
+        kernel_regularizer=regularizers.l2(1e-4),
+        name="dense_2",
+    )(x)
+    x = layers.Dropout(0.2, name="drop_2")(x)
+    outputs = layers.Dense(2, activation="softmax", name="output")(x)
+    return keras.Model(inputs, outputs, name="Dense")
+
+
+@lru_cache(maxsize=2)
+def load_classifier(problem: str) -> Dict[str, Any]:
+    """Load model weights and scaler for Problem A or B."""
+    problem_key = normalize_problem(problem)
+    info = MODEL_FILES[problem_key]
+    directory = _model_dir()
+    scaler_path = directory / info["scaler"]
+    weights_path = directory / info["weights"]
+
+    if not scaler_path.exists() or not weights_path.exists():
+        raise FileNotFoundError(
+            "Missing A12 model files. Expected files in A12/A12_results/: "
+            f"{info['scaler']} and {info['weights']}"
+        )
+
+    scaler = joblib.load(scaler_path)
+    input_dim = int(getattr(scaler, "n_features_in_"))
+    model = _build_dense_model(input_dim)
+    model.load_weights(weights_path)
+    return {"model": model, "scaler": scaler, "model_name": info["name"]}
+
+
+def predict_pose_csv(csv_path: str, problem: str = "B") -> Dict[str, Any]:
+    """Predict exercise/non-exercise frames from an uploaded pose CSV."""
+    started = time.perf_counter()
+    problem_key = normalize_problem(problem)
+    df = read_pose_csv(csv_path)
+    features, feature_names = validate_pose_dataframe(df, problem_key)
+    bundle = load_classifier(problem_key)
+
+    scaler = bundle["scaler"]
+    model = bundle["model"]
+    x_scaled = scaler.transform(features.values.astype(np.float32))
+    probabilities = model.predict(x_scaled, verbose=0)
+    frame_predictions = np.argmax(probabilities, axis=1)
+    exercise_confidence = probabilities[:, 1]
+
+    exercise_ratio = float(np.mean(frame_predictions == 1))
+    overall_label_index = int(exercise_ratio >= 0.5)
+    overall_confidence = float(np.mean(exercise_confidence))
+
+    frame_preview = []
+    for idx in range(min(10, len(frame_predictions))):
+        frame_preview.append(
+            {
+                "frame_index": idx,
+                "label": LABEL_NAMES[int(frame_predictions[idx])],
+                "confidence": float(np.max(probabilities[idx])),
+                "probabilities": {
+                    LABEL_NAMES[0]: float(probabilities[idx, 0]),
+                    LABEL_NAMES[1]: float(probabilities[idx, 1]),
+                },
+            }
+        )
+
+    return {
+        "status": "ok",
+        "endpoint": "Gradio tab inside app.py",
+        "problem": problem_key,
+        "model_name": bundle["model_name"],
+        "model_version": MODEL_VERSION,
+        "input_contract": "Pose feature CSV with the same feature columns used by A12_classifier.py.",
+        "metadata": {
+            "rows": int(len(df)),
+            "features": int(len(feature_names)),
+            "inference_time_ms": round((time.perf_counter() - started) * 1000, 2),
+        },
+        "prediction": {
+            "label": LABEL_NAMES[overall_label_index],
+            "confidence": overall_confidence,
+            "exercise_frame_ratio": exercise_ratio,
+        },
+        "frame_preview": frame_preview,
+    }
+
+
+def safe_predict_pose_csv(csv_path: str | None, problem: str = "B") -> Dict[str, Any]:
+    """UI-safe wrapper that returns structured errors instead of crashing Gradio."""
+    try:
+        return predict_pose_csv(str(csv_path), problem)
+    except Exception as exc:  # Gradio should show JSON instead of runtime crash.
+        return {
+            "status": "error",
+            "endpoint": "Gradio tab inside app.py",
+            "problem": str(problem),
+            "message": str(exc),
+        }

A12/service/pipeline.py

@@ -1,285 +0,0 @@
-from __future__ import annotations
-
-import csv
-import json
-import math
-import os
-import time
-from datetime import datetime
-from pathlib import Path
-from typing import Any, Dict, List, Tuple
-
-import cv2
-import numpy as np
-
-from .schemas import ClassificationResult, PipelineMetadata, PipelineOutput
-
-try:
-    from A8.pose_estimator import MoveNetPoseEstimator
-except Exception:  # pragma: no cover - used only if A8 is unavailable during isolated tests
-    MoveNetPoseEstimator = None
-
-try:
-    from A12.pose_interpolator import smooth_pose_sequence
-except Exception:  # pragma: no cover
-    smooth_pose_sequence = None
-
-KEYPOINT_NAMES = [
-    "nose", "left_eye", "right_eye", "left_ear", "right_ear",
-    "left_shoulder", "right_shoulder", "left_elbow", "right_elbow",
-    "left_wrist", "right_wrist", "left_hip", "right_hip",
-    "left_knee", "right_knee", "left_ankle", "right_ankle",
-]
-
-# 13-joint subset used by the classification data in A13.
-CLASSIFIER_JOINTS = [
-    "left_shoulder", "right_shoulder", "left_elbow", "right_elbow", "left_wrist", "right_wrist",
-    "left_hip", "right_hip", "left_knee", "right_knee", "left_ankle", "right_ankle", "nose",
-]
-
-SKELETON_EDGES = [
-    ("left_shoulder", "right_shoulder"), ("left_shoulder", "left_elbow"), ("left_elbow", "left_wrist"),
-    ("right_shoulder", "right_elbow"), ("right_elbow", "right_wrist"), ("left_shoulder", "left_hip"),
-    ("right_shoulder", "right_hip"), ("left_hip", "right_hip"), ("left_hip", "left_knee"),
-    ("left_knee", "left_ankle"), ("right_hip", "right_knee"), ("right_knee", "right_ankle"),
-]
-
-
-def validate_video(video_path: str | None) -> Path:
-    if not video_path:
-        raise ValueError("A video file is required.")
-    path = Path(video_path)
-    if not path.exists():
-        raise ValueError(f"Video file does not exist: {video_path}")
-    if path.suffix.lower() not in {".mp4", ".mov", ".avi", ".mkv", ".webm", ".m4v"}:
-        raise ValueError("Unsupported video type. Use mp4, mov, avi, mkv, webm, or m4v.")
-    cap = cv2.VideoCapture(str(path))
-    ok = cap.isOpened()
-    cap.release()
-    if not ok:
-        raise ValueError("OpenCV could not open this video. Try re-encoding it as mp4.")
-    return path
-
-
-def _pose_result_to_frame_dict(pose_result: Dict[str, Any], frame_id: int, timestamp_s: float) -> Dict[str, Any]:
-    keypoints = pose_result.get("keypoints", {}) or {}
-    out = []
-    for name in KEYPOINT_NAMES:
-        kp = keypoints.get(name, {}) or {}
-        out.append({
-            "name": name,
-            "x": kp.get("x"),
-            "y": kp.get("y"),
-            "z": 0.0,
-            "score": kp.get("confidence", kp.get("score", 0.0)),
-        })
-    return {
-        "frame_id": frame_id,
-        "timestamp": timestamp_s,
-        "poses": [{"pose_id": 0, "keypoints": out}],
-        "inference_time_ms": pose_result.get("inference_time_ms", 0.0),
-    }
-
-
-def _get_pose_estimator():
-    if MoveNetPoseEstimator is None:
-        raise RuntimeError("A8.pose_estimator.MoveNetPoseEstimator could not be imported.")
-    return MoveNetPoseEstimator(model_name="lightning")
-
-
-def _detect_motion_window(frames: List[Dict[str, Any]], min_window: int = 10) -> Tuple[int, int]:
-    """Cut leading/trailing frames by motion energy of 13 classifier joints."""
-    if len(frames) <= min_window:
-        return 0, max(0, len(frames) - 1)
-
-    coords = []
-    for f in frames:
-        kp_by_name = {kp["name"]: kp for kp in f["poses"][0]["keypoints"]}
-        row = []
-        for name in CLASSIFIER_JOINTS:
-            kp = kp_by_name.get(name, {})
-            x = kp.get("x")
-            y = kp.get("y")
-            row.append([np.nan if x is None else float(x), np.nan if y is None else float(y)])
-        coords.append(row)
-    arr = np.asarray(coords, dtype="float32")
-    arr = np.nan_to_num(arr, nan=np.nanmean(arr) if not np.isnan(arr).all() else 0.0)
-    velocity = np.linalg.norm(np.diff(arr, axis=0), axis=-1).mean(axis=1)
-    if len(velocity) == 0 or float(np.max(velocity)) == 0.0:
-        return 0, len(frames) - 1
-
-    threshold = max(float(np.percentile(velocity, 35)), float(np.max(velocity)) * 0.08)
-    active = np.where(velocity >= threshold)[0]
-    if active.size == 0:
-        return 0, len(frames) - 1
-
-    start = max(0, int(active[0]) - 2)
-    end = min(len(frames) - 1, int(active[-1]) + 3)
-    if end - start + 1 < min_window:
-        mid = (start + end) // 2
-        start = max(0, mid - min_window // 2)
-        end = min(len(frames) - 1, start + min_window - 1)
-    return start, end
-
-
-def _frames_to_classifier_sequence(frames: List[Dict[str, Any]], sequence_len: int = 10) -> np.ndarray:
-    """Return B-problem PoseNet-like data: shape (10, 13, 2)."""
-    if not frames:
-        raise ValueError("No pose frames were produced from the video.")
-    indices = np.linspace(0, len(frames) - 1, sequence_len, dtype=int)
-    sequence = np.zeros((sequence_len, len(CLASSIFIER_JOINTS), 2), dtype="float32")
-    for out_i, frame_i in enumerate(indices):
-        kp_by_name = {kp["name"]: kp for kp in frames[frame_i]["poses"][0]["keypoints"]}
-        for joint_i, name in enumerate(CLASSIFIER_JOINTS):
-            kp = kp_by_name.get(name, {})
-            sequence[out_i, joint_i, 0] = 0.0 if kp.get("x") is None else float(kp.get("x"))
-            sequence[out_i, joint_i, 1] = 0.0 if kp.get("y") is None else float(kp.get("y"))
-    # normalize per clip to make dimensions roughly model-friendly
-    max_abs = float(np.max(np.abs(sequence)))
-    if max_abs > 1.5:
-        sequence[:, :, 0] /= max_abs
-        sequence[:, :, 1] /= max_abs
-    return sequence
-
-
-def _classify(sequence: np.ndarray) -> ClassificationResult:
-    """Use a real persisted model if present; otherwise deterministic dummy classifier.
-
-    Replace this function with the Issue #10 champion wrapper when it is merged.
-    Expected future wrapper: A13.models.champion.predict_proba(sequence[None, ...]).
-    """
-    try:
-        from A13.models.champion import predict_good_probability  # type: ignore
-        good_prob = float(predict_good_probability(sequence))
-        mode = "champion_model"
-    except Exception:
-        # Deterministic fallback so app integration can be finished before #10/#11 is merged.
-        # Higher motion smoothness is treated as slightly better. This is not a scientific classifier.
-        velocity = np.linalg.norm(np.diff(sequence, axis=0), axis=-1).mean()
-        good_prob = float(np.clip(0.55 + (0.08 - velocity) * 0.8, 0.05, 0.95))
-        mode = "deterministic_dummy_until_issue_10_model_is_available"
-    label = "good" if good_prob >= 0.5 else "bad"
-    return ClassificationResult(
-        label=label,
-        is_good=label == "good",
-        confidence=max(good_prob, 1.0 - good_prob),
-        probabilities={"bad": 1.0 - good_prob, "good": good_prob, "mode": mode},
-    )
-
-
-def _write_keypoints_csv(frames: List[Dict[str, Any]], path: Path) -> None:
-    with path.open("w", newline="") as f:
-        writer = csv.writer(f)
-        writer.writerow(["frame_id", "timestamp", "joint", "x", "y", "z", "confidence"])
-        for frame in frames:
-            for kp in frame["poses"][0]["keypoints"]:
-                writer.writerow([
-                    frame["frame_id"], frame["timestamp"], kp["name"], kp.get("x"), kp.get("y"), kp.get("z", 0.0), kp.get("score", 0.0)
-                ])
-
-
-def _write_animation_json(frames: List[Dict[str, Any]], path: Path) -> None:
-    skeleton_frames = []
-    for frame in frames:
-        joints = {kp["name"]: {"x": kp.get("x"), "y": kp.get("y"), "z": kp.get("z", 0.0), "confidence": kp.get("score", 0.0)}
-                  for kp in frame["poses"][0]["keypoints"]}
-        skeleton_frames.append({"frame_id": frame["frame_id"], "timestamp": frame["timestamp"], "joints": joints})
-    path.write_text(json.dumps({"joint_names": KEYPOINT_NAMES, "edges": SKELETON_EDGES, "frames": skeleton_frames}, indent=2))
-
-
-def run_video_pipeline(
-    video_path: str | None,
-    confidence_threshold: float = 0.3,
-    smoothing_strategy: str = "exponential",
-    smoothing_method: str = "zscore",
-    output_dir: str | os.PathLike[str] = "pose_outputs",
-) -> PipelineOutput:
-    start_time = time.perf_counter()
-    warnings: List[str] = []
-    video = validate_video(video_path)
-    out_dir = Path(output_dir)
-    out_dir.mkdir(parents=True, exist_ok=True)
-
-    cap = cv2.VideoCapture(str(video))
-    fps = float(cap.get(cv2.CAP_PROP_FPS) or 30.0)
-    if fps <= 0 or fps > 240:
-        warnings.append(f"Invalid FPS reported by OpenCV ({fps}); using 30 FPS.")
-        fps = 30.0
-    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
-    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
-
-    pose_estimator = _get_pose_estimator()
-    original_frames: List[np.ndarray] = []
-    pose_frames: List[Dict[str, Any]] = []
-    frame_id = 0
-    while True:
-        ok, frame = cap.read()
-        if not ok:
-            break
-        original_frames.append(frame)
-        pose_result = pose_estimator.detect_pose(frame)
-        pose_frames.append(_pose_result_to_frame_dict(pose_result, frame_id, frame_id / fps))
-        frame_id += 1
-    cap.release()
-
-    if not pose_frames:
-        raise ValueError("The video contained no readable frames.")
-
-    if smooth_pose_sequence is not None:
-        try:
-            pose_frames = smooth_pose_sequence(
-                pose_frames,
-                strategy=smoothing_strategy,
-                outlier_method=smoothing_method,
-                outlier_threshold=3.0,
-                window_size=7,
-                min_confidence=0.2,
-            )
-        except Exception as exc:
-            warnings.append(f"Smoothing failed; using unsmoothed poses. Error: {exc}")
-    else:
-        warnings.append("A12.pose_interpolator could not be imported; using unsmoothed poses.")
-
-    cut_start, cut_end = _detect_motion_window(pose_frames)
-    cut_pose_frames = pose_frames[cut_start:cut_end + 1]
-    cut_video_frames = original_frames[cut_start:cut_end + 1]
-
-    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-    annotated_video = out_dir / f"a12_annotated_cut_{timestamp}.mp4"
-    keypoints_csv = out_dir / f"a12_keypoints_cut_{timestamp}.csv"
-    animation_json = out_dir / f"a12_animation_data_{timestamp}.json"
-
-    writer = cv2.VideoWriter(str(annotated_video), cv2.VideoWriter_fourcc(*"mp4v"), fps, (width, height))
-    for frame, pose_frame in zip(cut_video_frames, cut_pose_frames):
-        # Redraw via estimator on frame. If a future A11 pipeline returns annotated frames, replace this block.
-        pose_result = pose_estimator.detect_pose(frame)
-        annotated = pose_estimator.draw_keypoints(frame, pose_result, confidence_threshold=confidence_threshold)
-        writer.write(annotated)
-    writer.release()
-
-    _write_keypoints_csv(cut_pose_frames, keypoints_csv)
-    _write_animation_json(cut_pose_frames, animation_json)
-
-    sequence = _frames_to_classifier_sequence(cut_pose_frames)
-    classification = _classify(sequence)
-    classifier_mode = str(classification.probabilities.pop("mode", "unknown"))
-
-    elapsed_ms = (time.perf_counter() - start_time) * 1000.0
-    return PipelineOutput(
-        annotated_video_path=str(annotated_video),
-        animation_data_path=str(animation_json),
-        keypoints_csv_path=str(keypoints_csv),
-        classification=classification,
-        metadata=PipelineMetadata(
-            model_version="A12-pipeline-integration-v1",
-            inference_time_ms=elapsed_ms,
-            frame_count_original=len(pose_frames),
-            frame_count_cut=len(cut_pose_frames),
-            fps=fps,
-            cut_start_frame=cut_start,
-            cut_end_frame=cut_end,
-            smoothing_strategy=f"{smoothing_strategy}/{smoothing_method}",
-            classifier_mode=classifier_mode,
-        ),
-        warnings=warnings,
-    )

A12/service/schemas.py

@@ -1,45 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass, asdict
-from typing import Any, Dict, List, Optional
-
-
-@dataclass
-class PipelineMetadata:
-    model_version: str
-    inference_time_ms: float
-    frame_count_original: int
-    frame_count_cut: int
-    fps: float
-    cut_start_frame: int
-    cut_end_frame: int
-    smoothing_strategy: str
-    classifier_mode: str
-
-
-@dataclass
-class ClassificationResult:
-    label: str
-    is_good: bool
-    confidence: float
-    probabilities: Dict[str, float]
-
-
-@dataclass
-class PipelineOutput:
-    annotated_video_path: str
-    animation_data_path: str
-    keypoints_csv_path: str
-    classification: ClassificationResult
-    metadata: PipelineMetadata
-    warnings: List[str]
-
-    def to_json(self) -> Dict[str, Any]:
-        return {
-            "annotated_video_path": self.annotated_video_path,
-            "animation_data_path": self.animation_data_path,
-            "keypoints_csv_path": self.keypoints_csv_path,
-            "classification": asdict(self.classification),
-            "metadata": asdict(self.metadata),
-            "warnings": self.warnings,
-        }

A12/service/ui.py

@@ -1,44 +1,28 @@
+"""Small UI helpers for the A12 Gradio tab."""
+
 from __future__ import annotations
 
-import json
-from typing import Any, Dict, Tuple
+from typing import Any, Dict
 
-from .pipeline import run_video_pipeline, validate_video
+from A12.service.model_service import safe_predict_pose_csv
 
 
+def run_a12_tab(csv_file: str | None, problem: str) -> tuple[Dict[str, Any], str]:
+    """Run prediction and return JSON plus a concise Markdown summary."""
+    result = safe_predict_pose_csv(csv_file, problem)
+    if result.get("status") != "ok":
+        return result, f"### Prediction failed\n\n{result.get('message', 'Unknown error')}"
 
-def run_a12_video_tab(video_path: str, confidence_threshold: float, smoothing_strategy: str, smoothing_method: str):
-    """Gradio callback: video -> annotated video, animation json, keypoints csv, JSON, Markdown."""
-    try:
-        result = run_video_pipeline(
-            video_path=video_path,
-            confidence_threshold=confidence_threshold,
-            smoothing_strategy=smoothing_strategy,
-            smoothing_method=smoothing_method,
-        )
-        payload = result.to_json()
-        summary = f"""
-### A12 pipeline completed
+    prediction = result["prediction"]
+    metadata = result["metadata"]
+    summary = f"""### A12 prediction
 
-- **Classification:** `{payload['classification']['label']}`  
-- **Confidence:** `{payload['classification']['confidence']:.3f}`  
-- **Frames:** `{payload['metadata']['frame_count_cut']}` cut from `{payload['metadata']['frame_count_original']}` original frames  
-- **Cut window:** frames `{payload['metadata']['cut_start_frame']}` to `{payload['metadata']['cut_end_frame']}`  
-- **Classifier mode:** `{payload['metadata']['classifier_mode']}`
+- **Problem:** {result['problem']}
+- **Model:** {result['model_name']}
+- **Overall label:** {prediction['label']}
+- **Confidence:** {prediction['confidence']:.3f}
+- **Exercise frame ratio:** {prediction['exercise_frame_ratio']:.3f}
+- **Rows processed:** {metadata['rows']}
+- **Inference time:** {metadata['inference_time_ms']} ms
 """
-        if payload["warnings"]:
-            summary += "\n### Warnings\n" + "\n".join(f"- {w}" for w in payload["warnings"])
-        return result.annotated_video_path, result.animation_data_path, result.keypoints_csv_path, payload, summary
-    except Exception as exc:
-        return None, None, None, {"error": str(exc)}, f"### Error\n{exc}"
-
-
-# Backward-compatible function name used by the uploaded app.py A12 CSV prototype.
-def run_a12_tab(csv_path: str, problem: str = "B"):
-    if not csv_path:
-        return {"error": "Upload a CSV file first."}, "### Error\nUpload a CSV file first."
-    return {
-        "message": "This CSV endpoint was superseded by the Issue #12 video pipeline tab.",
-        "csv_path": csv_path,
-        "problem": problem,
-    }, "### CSV received\nThe Issue #12 implementation uses the video pipeline endpoint."
+    return result, summary

A12/tests/test_a12_service_contracts.py

@@ -0,0 +1,29 @@
+import pandas as pd
+import pytest
+
+from A12.service.contracts import FEATURES_BY_PROBLEM, normalize_problem, validate_pose_dataframe
+
+
+def test_normalize_problem_accepts_a_and_b():
+    assert normalize_problem("A") == "A"
+    assert normalize_problem("b") == "B"
+    assert normalize_problem("B - PoseNet") == "B"
+
+
+def test_normalize_problem_rejects_unknown_value():
+    with pytest.raises(ValueError):
+        normalize_problem("C")
+
+
+def test_validate_pose_dataframe_reports_missing_columns():
+    df = pd.DataFrame({"head_x": [1.0]})
+    with pytest.raises(ValueError, match="missing"):
+        validate_pose_dataframe(df, "B")
+
+
+def test_validate_pose_dataframe_accepts_problem_b_feature_schema():
+    df = pd.DataFrame({col: [0.5] for col in FEATURES_BY_PROBLEM["B"]})
+    features, names = validate_pose_dataframe(df, "B")
+    assert list(features.columns) == FEATURES_BY_PROBLEM["B"]
+    assert names == FEATURES_BY_PROBLEM["B"]
+    assert features.shape == (1, len(FEATURES_BY_PROBLEM["B"]))

A12/tests/test_a12_service_ui.py

@@ -0,0 +1,8 @@
+from A12.service.ui import run_a12_tab
+
+
+def test_run_a12_tab_returns_structured_error_for_missing_file():
+    result, summary = run_a12_tab(None, "B")
+    assert result["status"] == "error"
+    assert "message" in result
+    assert "Prediction failed" in summary

A12/tests/test_service_schema.py

@@ -1,30 +0,0 @@
-from A12.service.schemas import ClassificationResult, PipelineMetadata, PipelineOutput
-
-
-def test_pipeline_output_schema():
-    payload = PipelineOutput(
-        annotated_video_path="out.mp4",
-        animation_data_path="anim.json",
-        keypoints_csv_path="points.csv",
-        classification=ClassificationResult(
-            label="good",
-            is_good=True,
-            confidence=0.8,
-            probabilities={"good": 0.8, "bad": 0.2},
-        ),
-        metadata=PipelineMetadata(
-            model_version="test",
-            inference_time_ms=1.0,
-            frame_count_original=20,
-            frame_count_cut=10,
-            fps=30.0,
-            cut_start_frame=2,
-            cut_end_frame=11,
-            smoothing_strategy="exponential/zscore",
-            classifier_mode="dummy",
-        ),
-        warnings=[],
-    ).to_json()
-    assert payload["classification"]["label"] in {"good", "bad"}
-    assert set(payload["classification"]["probabilities"]) == {"good", "bad"}
-    assert payload["metadata"]["frame_count_cut"] == 10

A12/tests/test_service_validation.py

@@ -1,15 +0,0 @@
-import pytest
-
-from A12.service.pipeline import validate_video
-
-
-def test_validate_video_requires_input():
-    with pytest.raises(ValueError, match="required"):
-        validate_video(None)
-
-
-def test_validate_video_rejects_unknown_extension(tmp_path):
-    path = tmp_path / "not_a_video.txt"
-    path.write_text("hello")
-    with pytest.raises(ValueError, match="Unsupported"):
-        validate_video(str(path))

app.py

@@ -3,7 +3,6 @@ import gradio as gr
 from A8.pose_estimator import MoveNetPoseEstimator
 from A12.pose_interpolator import smooth_pose_sequence
 from A12.service.ui import run_a12_tab
-from A12.service.ui import run_a12_video_tab
 import json
 import csv
 import os
@@ -417,73 +416,43 @@ with gr.Blocks(title="MoveNet Pose Estimation") as demo:
                 outputs=[video_output, video_result]
             )
 
-        # A12 Video Pipeline Tab
-        with gr.TabItem("🧪 A12 Video Pipeline"):
+
+
+        # A12 Classifier Tab
+        with gr.TabItem("🧪 A12 Classifier"):
             gr.Markdown(
                 """
-                ### Issue #12: App development and pipeline integration
+                ### A12 Service Endpoint: Pose CSV classifier
 
-                Endpoint alternative chosen: **Gradio tab inside the existing app.py**.
-
-                **Input:** one video file.  
-                **Output:** annotated cut 2D video, 3D-animation data JSON, keypoints CSV,
-                and good/bad classification JSON.
+                Endpoint alternative chosen: **Gradio tab inside the existing app**.
+                This keeps the HuggingFace Space architecture simple and avoids a
+                separate REST/FastAPI service. Upload a pose-feature CSV exported
+                with the same feature schema used by the A12 classifier.
                 """
             )
-
             with gr.Row():
                 with gr.Column():
-                    a12_video_input = gr.Video(label="Input exercise video")
-                    a12_confidence = gr.Slider(
-                        minimum=0.0,
-                        maximum=1.0,
-                        value=0.3,
-                        step=0.05,
-                        label="Confidence threshold"
-                    )
-                    a12_smoothing_strategy = gr.Dropdown(
-                        choices=[
-                            "exponential",
-                            "moving_average",
-                            "gaussian",
-                            "median",
-                            "savitzky_golay",
-                            "kalman",
-                            "spline",
-                            "hybrid"
-                        ],
-                        value="exponential",
-                        label="Smoothing strategy",
+                    a12_csv_input = gr.File(
+                        label="Pose feature CSV",
+                        file_types=[".csv"],
+                        type="filepath",
                     )
-                    a12_smoothing_method = gr.Dropdown(
-                        choices=["zscore", "velocity", "none"],
-                        value="zscore",
-                        label="Outlier detection method",
+                    a12_problem_input = gr.Radio(
+                        choices=["A", "B"],
+                        value="B",
+                        label="Classifier problem",
+                        info="A = Kinect 3D features, B = PoseNet 2D features",
                     )
-                    a12_run_btn = gr.Button("Run A12 pipeline", variant="primary")
+                    a12_predict_btn = gr.Button("Run A12 classifier", variant="primary")
 
                 with gr.Column():
-                    a12_video_output = gr.Video(label="Annotated cut 2D video")
-                    a12_animation_file = gr.File(label="3D animation data JSON")
-                    a12_keypoints_file = gr.File(label="Cut keypoints CSV")
-                    a12_json_output = gr.JSON(label="Structured output")
-                    a12_summary = gr.Markdown()
-
-            a12_run_btn.click(
-                fn=run_a12_video_tab,
-                inputs=[
-                    a12_video_input,
-                    a12_confidence,
-                    a12_smoothing_strategy,
-                    a12_smoothing_method
-                ],
-                outputs=[
-                    a12_video_output,
-                    a12_animation_file,
-                    a12_keypoints_file,
-                    a12_json_output,
-                    a12_summary
-                ],
+                    a12_summary_output = gr.Markdown(label="Summary")
+                    a12_json_output = gr.JSON(label="Structured JSON output")
+
+            a12_predict_btn.click(
+                fn=run_a12_tab,
+                inputs=[a12_csv_input, a12_problem_input],
+                outputs=[a12_json_output, a12_summary_output],
             )
 
     # Example section