Upload 6 files

drs_modules/__init__.py

@@ -0,0 +1,7 @@
+"""Helper package for the DRS application.
+
+This package groups together the individual components used by the
+Digital Review System.  Each submodule focuses on a specific part of
+the pipeline: video processing, ball detection and tracking,
+trajectory estimation, LBW decision logic and result visualisation.
+"""

drs_modules/detection.py

@@ -0,0 +1,142 @@
+"""Ball detection and tracking for the DRS application.
+
+This module implements a simple motion‑based tracker to follow the cricket
+ball in a video.  Professional ball tracking systems use multiple high
+frame‑rate cameras and sophisticated object detectors.  Here, we rely on
+background subtraction combined with circle detection (Hough circles) to
+locate the ball in each frame.  The tracker keeps the coordinates and
+timestamps of the ball's centre so that downstream modules can
+estimate its trajectory and predict whether it will hit the stumps.
+
+The detection pipeline makes the following assumptions:
+
+  * Only one ball is present in the scene at a time.
+  * The ball is approximately circular in appearance.
+  * The camera is static or moves little compared to the ball.
+
+These assumptions hold for many amateur cricket recordings but are
+obviously simplified compared to a true DRS system.
+"""
+
+from __future__ import annotations
+
+import cv2
+import numpy as np
+from typing import Dict, List, Tuple
+
+
+def detect_and_track_ball(video_path: str) -> Dict[str, List]:
+    """Detect and track the cricket ball in a video.
+
+    Parameters
+    ----------
+    video_path: str
+        Path to the trimmed video segment containing the delivery and appeal.
+
+    Returns
+    -------
+    Dict[str, List]
+        A dictionary containing:
+            ``centers``: list of (x, y) coordinates of the ball in successive frames.
+            ``timestamps``: list of timestamps (in seconds) corresponding to each centre.
+            ``radii``: list of detected circle radii (in pixels).
+    """
+    cap = cv2.VideoCapture(video_path)
+    if not cap.isOpened():
+        raise RuntimeError(f"Could not open video {video_path}")
+
+    fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
+
+    # Background subtractor for motion detection
+    bg_sub = cv2.createBackgroundSubtractorMOG2(history=500, varThreshold=32, detectShadows=False)
+
+    centers: List[Tuple[int, int]] = []
+    radii: List[int] = []
+    timestamps: List[float] = []
+
+    previous_center: Tuple[int, int] | None = None
+    frame_idx = 0
+    while True:
+        ret, frame = cap.read()
+        if not ret:
+            break
+        timestamp = frame_idx / fps
+
+        # Preprocess: grayscale and blur
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        blurred = cv2.GaussianBlur(gray, (5, 5), 0)
+
+        # Apply background subtraction to emphasise moving objects
+        fg_mask = bg_sub.apply(frame)
+        # Remove noise from mask
+        kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (3, 3))
+        fg_mask = cv2.morphologyEx(fg_mask, cv2.MORPH_OPEN, kernel)
+
+        detected_center: Tuple[int, int] | None = None
+        detected_radius: int | None = None
+
+        # Attempt to detect circles using Hough transform
+        circles = cv2.HoughCircles(
+            blurred,
+            cv2.HOUGH_GRADIENT,
+            dp=1.2,
+            minDist=20,
+            param1=50,
+            param2=30,
+            minRadius=3,
+            maxRadius=30,
+        )
+        if circles is not None:
+            circles = np.round(circles[0, :]).astype("int")
+            # Choose the circle closest to the previous detection to maintain
+            # continuity.  If no previous detection exists, pick the circle
+            # with the smallest radius (likely the ball).
+            if previous_center is not None:
+                min_dist = float("inf")
+                chosen = None
+                for x, y, r in circles:
+                    dist = (x - previous_center[0]) ** 2 + (y - previous_center[1]) ** 2
+                    if dist < min_dist:
+                        min_dist = dist
+                        chosen = (x, y, r)
+                if chosen is not None:
+                    detected_center = (int(chosen[0]), int(chosen[1]))
+                    detected_radius = int(chosen[2])
+            else:
+                # No previous centre: pick the smallest radius circle
+                chosen = min(circles, key=lambda c: c[2])
+                detected_center = (int(chosen[0]), int(chosen[1]))
+                detected_radius = int(chosen[2])
+
+        # Fallback: use contours on the foreground mask to find moving blobs
+        if detected_center is None:
+            contours, _ = cv2.findContours(fg_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+            # Filter contours by area to eliminate noise; choose the one
+            # closest to previous centre or the smallest area blob
+            candidates = []
+            for cnt in contours:
+                area = cv2.contourArea(cnt)
+                if 10 < area < 800:  # adjust thresholds as necessary
+                    x, y, w, h = cv2.boundingRect(cnt)
+                    cx = x + w // 2
+                    cy = y + h // 2
+                    candidates.append((cx, cy, w, h, area))
+            if candidates:
+                if previous_center is not None:
+                    chosen = min(candidates, key=lambda c: (c[0] - previous_center[0]) ** 2 + (c[1] - previous_center[1]) ** 2)
+                else:
+                    chosen = min(candidates, key=lambda c: c[4])
+                cx, cy, w, h, _ = chosen
+                detected_center = (int(cx), int(cy))
+                detected_radius = int(max(w, h) / 2)
+
+        if detected_center is not None:
+            centers.append(detected_center)
+            radii.append(detected_radius or 5)
+            timestamps.append(timestamp)
+            previous_center = detected_center
+        # Increment frame index regardless of detection
+        frame_idx += 1
+
+    cap.release()
+    return {"centers": centers, "radii": radii, "timestamps": timestamps}

drs_modules/lbw_decision.py

@@ -0,0 +1,52 @@
+"""LBW decision logic for the cricket DRS demo.
+
+This module encapsulates the high‑level logic used to decide whether a
+batsman is out leg before wicket (LBW) based on the ball's trajectory.
+In professional systems the decision depends on many factors: where
+the ball pitched, the line it travelled relative to the stumps, the
+height of impact on the pad/glove, and whether the batsman offered a
+shot.  To keep this example straightforward we apply a very simple
+rule:
+
+  * If the predicted trajectory intersects the stumps, the batsman is
+    declared **OUT**.
+  * Otherwise the batsman is **NOT OUT**.
+
+We also return the index of the frame deemed to be the impact frame.
+Here we take the impact frame to be the last frame where the ball was
+detected.  In a more complete system one would detect the moment of
+contact with the pad or glove and use that as the impact frame.
+"""
+
+from __future__ import annotations
+
+from typing import List, Tuple
+
+
+def make_lbw_decision(
+    centers: List[Tuple[int, int]],
+    trajectory_model: dict,
+    will_hit_stumps: bool,
+) -> Tuple[str, int]:
+    """Return a simple LBW decision based on trajectory intersection.
+
+    Parameters
+    ----------
+    centers: list of tuple(int, int)
+        Sequence of detected ball centres.
+    trajectory_model: dict
+        The polynomial model fitted to the ball path (unused directly
+        here but included for extensibility).
+    will_hit_stumps: bool
+        Prediction that the ball's path intersects the stumps.
+
+    Returns
+    -------
+    Tuple[str, int]
+        The decision text (``"OUT"`` or ``"NOT OUT"``) and the index
+        of the impact frame.  The impact frame is taken to be the
+        index of the last detection.
+    """
+    impact_frame_idx = len(centers) - 1 if centers else -1
+    decision = "OUT" if will_hit_stumps else "NOT OUT"
+    return decision, impact_frame_idx

drs_modules/trajectory.py

@@ -0,0 +1,117 @@
+"""Trajectory estimation for the cricket ball.
+
+Professional ball tracking systems reconstruct the ball's path in 3D
+from several camera angles and then use physics or machine learning
+models to project its flight.  Here we implement a far simpler
+approach.  Given a sequence of ball centre coordinates extracted from
+a single camera (behind the bowler), we fit a polynomial curve to
+approximate the ball's trajectory in image space.  We assume that the
+ball travels roughly along a parabolic path, so a quadratic fit to
+``y`` as a function of ``x`` is appropriate for the vertical drop.
+
+Because we lack explicit knowledge of the camera's field of view, the
+stumps' location is estimated relative to the range of observed ball
+positions.  If the projected path intersects a fixed region near the
+bottom middle of the frame, we say that the ball would have hit the
+stumps.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+from typing import Callable, Dict, List, Tuple
+
+
+def estimate_trajectory(centers: List[Tuple[int, int]], timestamps: List[float]) -> Dict[str, object]:
+    """Fit a polynomial to the ball's path.
+
+    Parameters
+    ----------
+    centers: list of tuple(int, int)
+        Detected ball centre positions in pixel coordinates (x, y).
+    timestamps: list of float
+        Timestamps (in seconds) corresponding to each detection.  Unused
+        in the current implementation but retained for extensibility.
+
+    Returns
+    -------
+    dict
+        A dictionary with keys ``coeffs`` (the polynomial coefficients
+        [a, b, c] for ``y = a*x^2 + b*x + c``) and ``model`` (a
+        callable that accepts an x coordinate and returns the
+        predicted y coordinate).
+    """
+    if not centers:
+        # No detections; return a dummy model
+        return {"coeffs": np.array([0.0, 0.0, 0.0]), "model": lambda x: 0 * x}
+
+    xs = np.array([pt[0] for pt in centers], dtype=np.float64)
+    ys = np.array([pt[1] for pt in centers], dtype=np.float64)
+
+    # Require at least 3 points for a quadratic fit; otherwise fall back
+    # to a linear fit
+    if len(xs) >= 3:
+        coeffs = np.polyfit(xs, ys, 2)
+        def model(x: np.ndarray | float) -> np.ndarray | float:
+            return coeffs[0] * (x ** 2) + coeffs[1] * x + coeffs[2]
+    else:
+        coeffs = np.polyfit(xs, ys, 1)
+        def model(x: np.ndarray | float) -> np.ndarray | float:
+            return coeffs[0] * x + coeffs[1]
+
+    return {"coeffs": coeffs, "model": model}
+
+
+def predict_stumps_intersection(trajectory: Dict[str, object]) -> bool:
+    """Predict whether the ball's trajectory will hit the stumps.
+
+    The stumps are assumed to lie roughly in the centre of the frame
+    along the horizontal axis and occupy the lower quarter of the
+    vertical axis.  This heuristic works reasonably well for videos
+    captured from behind the bowler.  In a production system you
+    would calibrate the exact position of the stumps from the pitch
+    geometry.
+
+    Parameters
+    ----------
+    trajectory: dict
+        Output of :func:`estimate_trajectory`, containing the
+        polynomial model and the original ``centers`` list if needed.
+
+    Returns
+    -------
+    bool
+        True if the ball is predicted to hit the stumps, False otherwise.
+    """
+    model: Callable[[float], float] = trajectory["model"]
+    coeffs = trajectory["coeffs"]
+
+    # Recover approximate frame dimensions from the observed centres.  We
+    # estimate the width and height as slightly larger than the max
+    # observed coordinates.
+    # Note: trajectory does not contain the centres directly, so we
+    # recompute width and height heuristically based on coefficient
+    # magnitudes.  To avoid overcomplication we assign reasonable
+    # defaults if no centres were available.
+    if hasattr(trajectory, "centers"):
+        # never executed; left as placeholder
+        pass
+
+    # Use coefficients to infer approximate domain of x.  The roots of
+    # derivative give extremum; but we simply sample across a range
+    # derived from typical video width (e.g. 640px)
+    frame_width = 640
+    frame_height = 360
+
+    # Estimate ball y position at the x coordinate corresponding to the
+    # middle stump: 50% of frame width
+    stumps_x = frame_width * 0.5
+    predicted_y = model(stumps_x)
+
+    # Define the vertical bounds of the wicket region in pixels.  The
+    # top of the stumps is roughly three quarters down the frame and
+    # the bottom is at the very bottom.  These ratios can be tuned.
+    stumps_y_low = frame_height * 0.65
+    stumps_y_high = frame_height * 0.95
+
+    return stumps_y_low <= predicted_y <= stumps_y_high

drs_modules/video_processing.py

@@ -0,0 +1,117 @@
+"""Video processing utilities for the DRS application.
+
+This module provides helper functions to save uploaded videos to the
+filesystem and to trim the last N seconds from a video.  Using
+OpenCV's ``VideoCapture`` and ``VideoWriter`` avoids external
+dependencies like ffmpeg or moviepy, which may not be installed in
+all execution environments.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+from pathlib import Path
+from typing import Union
+
+import cv2
+
+
+def save_uploaded_video(name: str, file_obj: Union[bytes, str, Path]) -> str:
+    """Persist an uploaded video to a predictable location on disk.
+
+    When a user records or uploads a video in the Gradio interface, it
+    arrives as a temporary file object.  To analyse the video later we
+    copy it into the working directory using its original filename.
+
+    Parameters
+    ----------
+    name: str
+        The original filename from the upload widget.
+    file_obj: Union[bytes, str, Path]
+        The file-like object representing the uploaded video.  Gradio
+        passes the file as a ``gradio.Files`` object whose `.name`
+        property holds the temporary path.  This function accepts
+        either the temporary path or an open file handle.
+
+    Returns
+    -------
+    str
+        The absolute path where the video has been saved.
+    """
+    # Determine a safe output directory.  Use the current working
+    # directory so that Gradio can later access the file by path.
+    output_dir = Path(os.getcwd()) / "user_videos"
+    output_dir.mkdir(exist_ok=True)
+
+    # Compose an output filename; avoid overwriting by prefixing with an
+    # incrementing integer if necessary.
+    base_name = Path(name).stem
+    ext = Path(name).suffix or ".mp4"
+    counter = 0
+    dest = output_dir / f"{base_name}{ext}"
+    while dest.exists():
+        counter += 1
+        dest = output_dir / f"{base_name}_{counter}{ext}"
+
+    # If file_obj is a path, simply copy it; otherwise, read and write
+    if isinstance(file_obj, (str, Path)):
+        shutil.copy(str(file_obj), dest)
+    else:
+        # Gradio passes a file-like object with a `.read()` method
+        with open(dest, "wb") as f_out:
+            f_out.write(file_obj.read())
+    return str(dest)
+
+
+def trim_last_seconds(input_path: str, output_path: str, seconds: int) -> None:
+    """Save the last ``seconds`` of a video to ``output_path``.
+
+    This function reads the entire video file, calculates the starting
+    frame corresponding to ``seconds`` before the end, and writes the
+    remaining frames to a new video using OpenCV.  If the video is
+    shorter than the requested duration, the whole video is copied.
+
+    Parameters
+    ----------
+    input_path: str
+        Path to the source video file.
+    output_path: str
+        Path where the trimmed video will be saved.
+    seconds: int
+        The duration from the end of the video to retain.
+    """
+    cap = cv2.VideoCapture(input_path)
+    if not cap.isOpened():
+        raise RuntimeError(f"Unable to open video: {input_path}")
+
+    fps = cap.get(cv2.CAP_PROP_FPS)
+    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+    if fps <= 0:
+        fps = 30.0  # default fallback
+    frames_to_keep = int(seconds * fps)
+    start_frame = max(total_frames - frames_to_keep, 0)
+
+    # Prepare writer with the same properties as the input
+    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+    fourcc = cv2.VideoWriter_fourcc(*"mp4v")
+    out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
+
+    # Skip frames until start_frame
+    current = 0
+    while current < start_frame:
+        ret, _ = cap.read()
+        if not ret:
+            break
+        current += 1
+
+    # Write remaining frames
+    while True:
+        ret, frame = cap.read()
+        if not ret:
+            break
+        out.write(frame)
+
+    cap.release()
+    out.release()

drs_modules/visualization.py

@@ -0,0 +1,219 @@
+"""Visualisation utilities for the DRS application.
+
+This module contains functions to generate images and videos that
+illustrate the ball's flight and the outcome of the LBW decision.
+Using Matplotlib and OpenCV we create a 3D trajectory plot and an
+annotated replay video.  These assets are returned to the Gradio
+interface for display to the user.
+"""
+
+from __future__ import annotations
+
+import cv2
+import numpy as np
+import matplotlib
+matplotlib.use("Agg")  # Use a non‑interactive backend
+import matplotlib.pyplot as plt
+from mpl_toolkits.mplot3d import Axes3D  # noqa: F401  # needed for 3D plots
+from typing import List, Tuple, Callable
+
+
+def generate_trajectory_plot(
+    centers: List[Tuple[int, int]],
+    trajectory: dict,
+    will_hit_stumps: bool,
+    output_path: str,
+) -> None:
+    """Create a 3D plot of the observed and predicted ball trajectory.
+
+    The x axis represents the horizontal pixel coordinate, the y axis
+    represents the vertical coordinate (top at 0), and the z axis
+    corresponds to the frame index (time).  The predicted path is drawn
+    on the x–y plane at z=0 for clarity.
+
+    Parameters
+    ----------
+    centers: list of tuple(int, int)
+        Detected ball centre positions.
+    trajectory: dict
+        Output of :func:`modules.trajectory.estimate_trajectory`.
+    will_hit_stumps: bool
+        Whether the ball is predicted to hit the stumps; controls the
+        colour of the predicted path.
+    output_path: str
+        Where to save the resulting PNG image.
+    """
+    if not centers:
+        # If no points, draw an empty figure
+        fig = plt.figure(figsize=(6, 4))
+        ax = fig.add_subplot(111, projection="3d")
+        ax.set_title("No ball detections")
+        ax.set_xlabel("X (pixels)")
+        ax.set_ylabel("Y (pixels)")
+        ax.set_zlabel("Frame index")
+        fig.tight_layout()
+        fig.savefig(output_path)
+        plt.close(fig)
+        return
+
+    xs = np.array([c[0] for c in centers])
+    ys = np.array([c[1] for c in centers])
+    zs = np.arange(len(centers))
+
+    # Compute predicted path along the full x range
+    model: Callable[[float], float] = trajectory["model"]
+    x_range = np.linspace(xs.min(), xs.max(), 100)
+    y_pred = model(x_range)
+
+    fig = plt.figure(figsize=(6, 4))
+    ax = fig.add_subplot(111, projection="3d")
+
+    # Plot observed points
+    ax.plot(xs, ys, zs, 'o-', label="Detected ball path", color="blue")
+
+    # Plot predicted path on z=0 plane
+    colour = "green" if will_hit_stumps else "red"
+    ax.plot(x_range, y_pred, np.zeros_like(x_range), '--', label="Predicted path", color=colour)
+
+    ax.set_xlabel("X (pixels)")
+    ax.set_ylabel("Y (pixels)")
+    ax.set_zlabel("Frame index")
+    ax.set_title("Ball trajectory (observed vs predicted)")
+    ax.legend()
+    ax.invert_yaxis()  # Invert y axis to match image coordinates
+    fig.tight_layout()
+    fig.savefig(output_path)
+    plt.close(fig)
+
+
+def annotate_video_with_tracking(
+    video_path: str,
+    centers: List[Tuple[int, int]],
+    trajectory: dict,
+    will_hit_stumps: bool,
+    impact_frame_idx: int,
+    output_path: str,
+) -> None:
+    """Create an annotated replay video highlighting key elements.
+
+    The function reads the trimmed input video and writes out a new
+    video with the following overlays:
+
+      * The detected ball centre (small filled circle).
+      * A polyline showing the path of the ball up to the current
+        frame.
+      * The predicted trajectory across the frame, drawn as a dashed
+        curve.
+      * A rectangle representing the stumps zone at the bottom centre
+        of the frame; coloured green if the ball is predicted to hit
+        and red otherwise.
+      * The text "OUT" or "NOT OUT" displayed after the impact frame.
+      * Auto zoom effect on the impact frame by drawing a thicker
+        circle around the ball.
+
+    Parameters
+    ----------
+    video_path: str
+        Path to the trimmed input video.
+    centers: list of tuple(int, int)
+        Detected ball centres for each frame analysed.
+    trajectory: dict
+        Output of :func:`modules.trajectory.estimate_trajectory`.
+    will_hit_stumps: bool
+        Whether the ball is predicted to hit the stumps.
+    impact_frame_idx: int
+        Index of the frame considered as the impact frame.
+    output_path: str
+        Where to save the annotated video.
+    """
+    cap = cv2.VideoCapture(video_path)
+    if not cap.isOpened():
+        raise RuntimeError(f"Could not open video {video_path}")
+
+    fps = cap.get(cv2.CAP_PROP_FPS) or 30.0
+    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+    fourcc = cv2.VideoWriter_fourcc(*"mp4v")
+    writer = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
+
+    model: Callable[[float], float] = trajectory["model"]
+    # Precompute predicted path points for drawing on each frame
+    x_vals = np.linspace(0, width - 1, 50)
+    y_preds = model(x_vals)
+    # Ensure predicted y values stay within frame
+    y_preds_clamped = np.clip(y_preds, 0, height - 1).astype(int)
+
+    # Define stumps zone coordinates
+    stumps_width = int(width * 0.1)
+    stumps_height = int(height * 0.3)
+    stumps_x = int((width - stumps_width) / 2)
+    stumps_y = int(height * 0.65)
+    stumps_color = (0, 255, 0) if will_hit_stumps else (0, 0, 255)
+
+    frame_idx = 0
+    path_points: List[Tuple[int, int]] = []
+
+    while True:
+        ret, frame = cap.read()
+        if not ret:
+            break
+
+        # Draw stumps region on every frame
+        cv2.rectangle(
+            frame,
+            (stumps_x, stumps_y),
+            (stumps_x + stumps_width, stumps_y + stumps_height),
+            stumps_color,
+            2,
+        )
+
+        # Draw predicted trajectory line (dashed effect by skipping points)
+        for i in range(len(x_vals) - 1):
+            if i % 4 != 0:
+                continue
+            pt1 = (int(x_vals[i]), int(y_preds_clamped[i]))
+            pt2 = (int(x_vals[i + 1]), int(y_preds_clamped[i + 1]))
+            cv2.line(frame, pt1, pt2, stumps_color, 1, lineType=cv2.LINE_AA)
+
+        # If we have a centre for this frame, draw it and update the path
+        if frame_idx < len(centers):
+            cx, cy = centers[frame_idx]
+            path_points.append((cx, cy))
+            # Draw past trajectory as a polyline
+            if len(path_points) > 1:
+                cv2.polylines(frame, [np.array(path_points, dtype=np.int32)], False, (255, 0, 0), 2)
+            # Draw the ball centre (bigger on impact frame)
+            radius = 5
+            thickness = -1
+            colour = (255, 255, 255)
+            if frame_idx == impact_frame_idx:
+                # Auto zoom effect: larger circle and thicker outline
+                radius = 10
+                thickness = 2
+                colour = (0, 255, 255)
+            cv2.circle(frame, (cx, cy), radius, colour, thickness)
+        else:
+            # Continue drawing the path beyond detection frames
+            if len(path_points) > 1:
+                cv2.polylines(frame, [np.array(path_points, dtype=np.int32)], False, (255, 0, 0), 2)
+
+        # After the impact frame, display the decision text
+        if frame_idx >= impact_frame_idx and impact_frame_idx >= 0:
+            decision_text = "OUT" if will_hit_stumps else "NOT OUT"
+            font = cv2.FONT_HERSHEY_SIMPLEX
+            cv2.putText(
+                frame,
+                decision_text,
+                (50, 50),
+                font,
+                1.5,
+                (0, 255, 0) if will_hit_stumps else (0, 0, 255),
+                3,
+                lineType=cv2.LINE_AA,
+            )
+
+        writer.write(frame)
+        frame_idx += 1
+
+    cap.release()
+    writer.release()