video2visData

A12/service/__init__.py

@@ -1,6 +1,4 @@
-"""A12 service endpoint package."""
+from .pipeline import run_video_pipeline, validate_video
+from .ui import 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"]
+__all__ = ["run_video_pipeline", "validate_video", "run_a12_video_tab"]

A12/service/contracts.py

@@ -1,134 +0,0 @@
-"""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

@@ -1,159 +0,0 @@
-"""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

@@ -0,0 +1,285 @@
+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

@@ -0,0 +1,45 @@
+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,28 +1,44 @@
-"""Small UI helpers for the A12 Gradio tab."""
-
 from __future__ import annotations
 
-from typing import Any, Dict
+import json
+from typing import Any, Dict, Tuple
 
-from A12.service.model_service import safe_predict_pose_csv
+from .pipeline import run_video_pipeline, validate_video
 
 
-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')}"
 
-    prediction = result["prediction"]
-    metadata = result["metadata"]
-    summary = f"""### A12 prediction
+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
 
-- **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
+- **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']}`
 """
-    return result, summary
+        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."

A12/tests/test_a12_service_contracts.py

@@ -1,29 +0,0 @@
-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

@@ -1,8 +0,0 @@
-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

@@ -0,0 +1,30 @@
+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

@@ -0,0 +1,15 @@
+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,6 +3,7 @@ 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
@@ -416,43 +417,73 @@ with gr.Blocks(title="MoveNet Pose Estimation") as demo:
                 outputs=[video_output, video_result]
             )
 
-
-
-        # A12 Classifier Tab
-        with gr.TabItem("🧪 A12 Classifier"):
+        # A12 Video Pipeline Tab
+        with gr.TabItem("🧪 A12 Video Pipeline"):
             gr.Markdown(
                 """
-                ### A12 Service Endpoint: Pose CSV classifier
+                ### Issue #12: App development and pipeline integration
 
-                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.
+                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.
                 """
             )
+
             with gr.Row():
                 with gr.Column():
-                    a12_csv_input = gr.File(
-                        label="Pose feature CSV",
-                        file_types=[".csv"],
-                        type="filepath",
+                    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_problem_input = gr.Radio(
-                        choices=["A", "B"],
-                        value="B",
-                        label="Classifier problem",
-                        info="A = Kinect 3D features, B = PoseNet 2D features",
+                    a12_smoothing_method = gr.Dropdown(
+                        choices=["zscore", "velocity", "none"],
+                        value="zscore",
+                        label="Outlier detection method",
                     )
-                    a12_predict_btn = gr.Button("Run A12 classifier", variant="primary")
+                    a12_run_btn = gr.Button("Run A12 pipeline", variant="primary")
 
                 with gr.Column():
-                    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],
+                    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
+                ],
             )
 
     # Example section