app.py

"""
Digital Review System (DRS) application for LBW decisions
========================================================

This application provides a simplified demonstration of how a cricket‑style
digital review system (DRS) could be implemented using open source
computer vision tools.  It is not a complete Hawk‑Eye replacement, but
illustrates the key steps in building such a system: capturing video,
detecting and tracking the ball, estimating its flight trajectory,
analysing whether it would have hit the stumps, estimating speed and
generating a replay with annotations.  A Gradio interface ties these
components together to provide an easy way to record a match, appeal
for an LBW decision and review the result.

The app has two main pages:

  • **Live Match Recording** – allows the user to upload or record match video.
    The video is stored on disk and can be analysed later.

  • **LBW Review** – analyses the last few seconds of the recorded video
    whenever an appeal is made.  It performs ball tracking, trajectory
    estimation and stumps intersection checks to predict whether the
    batsman is out or not.  An annotated replay and a 3D trajectory
    visualisation are returned along with speed and impact information.

The implementation relies on simple background subtraction and circle
detection rather than proprietary tracking systems.  It assumes a
single static camera behind the bowler and a fairly unobstructed view
of the pitch.  See the individual modules in the ``modules`` package
for more details on each processing step.

Note: because this space is intended to run on Hugging Face, file
paths and heavy downloads are avoided wherever possible.  The code is
fully self contained and uses only packages available in this runtime.
"""

from __future__ import annotations

import os
import shutil
import tempfile
from pathlib import Path
from typing import Any, Dict, Tuple

import gradio as gr

from drs_modules.modules.video_processing import trim_last_seconds, save_uploaded_video
from drs_modules.modules.detection import detect_and_track_ball
from drs_modules.modules.trajectory import estimate_trajectory, predict_stumps_intersection
from drs_modules.modules.lbw_decision import make_lbw_decision
from drs_modules.modules.visualization import (
    generate_trajectory_plot,
    annotate_video_with_tracking,
)


def analyse_appeal(video_path: str, review_seconds: int = 8) -> Tuple[str, Dict[str, Any]]:
    """Analyse the last few seconds of a match video and return DRS results.

    Parameters
    ----------
    video_path: str
        Path to the full match video recorded on the Live Match Recording page.
    review_seconds: int, optional
        Number of seconds from the end of the video to analyse.  Defaults to 8.

    Returns
    -------
    Tuple[str, Dict[str, Any]]
        A message summarising the decision and a dictionary with the
        underlying data for display (decision text, ball speed, impact
        frame number, annotated video path and trajectory plot path).
    """
    # Create a temporary directory to hold intermediate files
    temp_dir = tempfile.mkdtemp()
    trimmed_path = os.path.join(temp_dir, "trimmed.mp4")

    # Step 1: Trim the last N seconds of the input video
    trim_last_seconds(video_path, trimmed_path, review_seconds)

    # Step 2: Detect and track the ball through the trimmed segment
    tracking_data = detect_and_track_ball(trimmed_path)

    # Step 3: Estimate the ball's trajectory (2D for simplicity) and predict
    # whether it will hit the stumps
    trajectory_model = estimate_trajectory(tracking_data["centers"], tracking_data["timestamps"])
    will_hit_stumps = predict_stumps_intersection(trajectory_model)

    # Step 4: Make a decision based on trajectory and impact detection
    decision, impact_frame_idx = make_lbw_decision(
        tracking_data["centers"],
        trajectory_model,
        will_hit_stumps,
    )

    # Step 5: Calculate ball speed (pixels per second scaled to km/h)
    total_distance_px = 0.0
    for i in range(1, len(tracking_data["centers"])):
        cx0, cy0 = tracking_data["centers"][i - 1]
        cx1, cy1 = tracking_data["centers"][i]
        total_distance_px += ((cx1 - cx0) ** 2 + (cy1 - cy0) ** 2) ** 0.5
    # Duration of captured frames
    duration = tracking_data["timestamps"][-1] - tracking_data["timestamps"][0]
    if duration <= 0:
        speed_kmh = 0.0
    else:
        # Convert pixel distance per second to km/h using an assumed scale
        pixels_per_metre = 50.0
        speed_mps = (total_distance_px / pixels_per_metre) / duration
        speed_kmh = speed_mps * 3.6

    # Step 6: Generate annotated replay video and trajectory plot
    annotated_video_path = os.path.join(temp_dir, "annotated.mp4")
    annotate_video_with_tracking(
        trimmed_path,
        tracking_data["centers"],
        trajectory_model,
        will_hit_stumps,
        impact_frame_idx,
        annotated_video_path,
    )
    plot_path = os.path.join(temp_dir, "trajectory_plot.png")
    generate_trajectory_plot(
        tracking_data["centers"], trajectory_model, will_hit_stumps, plot_path
    )

    # Compose the message and result dictionary
    decision_message = f"Decision: {decision}"
    result = {
        "decision": decision,
        "ball_speed_kmh": round(speed_kmh, 2),
        "impact_frame_index": impact_frame_idx,
        "annotated_video": annotated_video_path,
        "trajectory_plot": plot_path,
    }

    return decision_message, result


def build_interface() -> gr.Blocks:
    """Construct the Gradio interface with multiple pages."""
    with gr.Blocks(title="Cricket LBW DRS Demo") as demo:
        gr.Markdown(
            """# Digital Review System (LBW)

            This demo illustrates how a simplified digital review system can be
            implemented using computer vision techniques.  You can record or
            upload match footage, and when an appeal occurs, the system will
            analyse the last few seconds to decide whether the batsman is **OUT**
            or **NOT OUT**.  Alongside the decision you will receive an
            annotated replay, a 3D trajectory plot and an estimate of the ball
            speed.
            """
        )

        with gr.Tab("Live Match Recording"):
            video_input = gr.Video(
                label="Record or upload match video",
                sources=["upload", "webcam"],
                # Do not specify `type` because some versions of Gradio
                # reject that argument. The file path is available via
                # video_file.name in the callback.
            )
            out_video_path = gr.State()

            def on_video_upload(video_file):
                if video_file is None:
                    return None
                save_path = save_uploaded_video(video_file.name, video_file)
                return save_path

            video_input.change(
                fn=on_video_upload,
                inputs=[video_input],
                outputs=[out_video_path],
            )

            gr.Markdown(
                """
                After recording or uploading a video, switch to the **LBW Review**
                tab and press **Analyse Appeal** to review the last 8 seconds.
                """
            )

        with gr.Tab("LBW Review"):
            with gr.Row():
                analyse_button = gr.Button("Analyse Appeal")
                review_seconds = gr.Number(
                    value=8, label="Seconds to review", minimum=2, maximum=20
                )
            decision_output = gr.Textbox(label="Decision", lines=1)
            ball_speed_output = gr.Textbox(
                label="Ball speed (km/h)", lines=1, interactive=False
            )
            impact_frame_output = gr.Textbox(
                label="Impact frame index", lines=1, interactive=False
            )
            annotated_video_output = gr.Video(
                label="Annotated replay video"
            )
            trajectory_plot_output = gr.Image(
                label="3D Trajectory plot"
            )

            def on_analyse(_):
                video_path = out_video_path.value
                if not video_path or not os.path.exists(video_path):
                    return (
                        "Please record or upload a video in the first tab.",
                        None,
                        None,
                        None,
                        None,
                    )
                message, result = analyse_appeal(video_path, int(review_seconds.value))
                return (
                    message,
                    str(result["ball_speed_kmh"]),
                    str(result["impact_frame_index"]),
                    result["annotated_video"],
                    result["trajectory_plot"],
                )

            analyse_button.click(
                fn=on_analyse,
                inputs=[analyse_button],
                outputs=[
                    decision_output,
                    ball_speed_output,
                    impact_frame_output,
                    annotated_video_output,
                    trajectory_plot_output,
                ],
            )

    return demo


if __name__ == "__main__":
    demo = build_interface()
    demo.launch()