|
import cv2 |
|
import numpy as np |
|
import torch |
|
from ultralytics import YOLO |
|
import gradio as gr |
|
from scipy.interpolate import interp1d |
|
import uuid |
|
import os |
|
|
|
|
|
model = YOLO("best.pt") |
|
|
|
|
|
STUMPS_WIDTH = 0.2286 |
|
BALL_DIAMETER = 0.073 |
|
FRAME_RATE = 30 |
|
|
|
def process_video(video_path): |
|
|
|
cap = cv2.VideoCapture(video_path) |
|
frames = [] |
|
ball_positions = [] |
|
|
|
while cap.isOpened(): |
|
ret, frame = cap.read() |
|
if not ret: |
|
break |
|
frames.append(frame.copy()) |
|
|
|
results = model.predict(frame, conf=0.5) |
|
for detection in results[0].boxes: |
|
if detection.cls == 0: |
|
x1, y1, x2, y2 = detection.xyxy[0].cpu().numpy() |
|
ball_positions.append([(x1 + x2) / 2, (y1 + y2) / 2]) |
|
|
|
cv2.rectangle(frame, (int(x1), int(y1)), (int(x2), int(y2)), (0, 255, 0), 2) |
|
frames[-1] = frame |
|
cap.release() |
|
|
|
return frames, ball_positions |
|
|
|
def estimate_trajectory(ball_positions, frames): |
|
|
|
if len(ball_positions) < 2: |
|
return None, None |
|
|
|
x_coords = [pos[0] for pos in ball_positions] |
|
y_coords = [pos[1] for pos in ball_positions] |
|
times = np.arange(len(ball_positions)) / FRAME_RATE |
|
|
|
|
|
try: |
|
fx = interp1d(times, x_coords, kind='linear', fill_value="extrapolate") |
|
fy = interp1d(times, y_coords, kind='quadratic', fill_value="extrapolate") |
|
except: |
|
return None, None |
|
|
|
|
|
t_future = np.linspace(times[-1], times[-1] + 0.5, 10) |
|
x_future = fx(t_future) |
|
y_future = fy(t_future) |
|
|
|
return list(zip(x_future, y_future)), t_future |
|
|
|
def lbw_decision(ball_positions, trajectory, frames): |
|
|
|
if not trajectory or len(ball_positions) < 2: |
|
return "Not enough data", None |
|
|
|
|
|
frame_height, frame_width = frames[0].shape[:2] |
|
stumps_x = frame_width / 2 |
|
stumps_y = frame_height * 0.9 |
|
stumps_width_pixels = frame_width * (STUMPS_WIDTH / 3.0) |
|
|
|
|
|
pitch_x, pitch_y = ball_positions[0] |
|
if pitch_x < stumps_x - stumps_width_pixels / 2 or pitch_x > stumps_x + stumps_width_pixels / 2: |
|
return "Not Out (Pitched outside line)", None |
|
|
|
|
|
impact_x, impact_y = ball_positions[-1] |
|
if impact_x < stumps_x - stumps_width_pixels / 2 or impact_x > stumps_x + stumps_width_pixels / 2: |
|
return "Not Out (Impact outside line)", None |
|
|
|
|
|
for x, y in trajectory: |
|
if abs(x - stumps_x) < stumps_width_pixels / 2 and abs(y - stumps_y) < frame_height * 0.1: |
|
return "Out", trajectory |
|
return "Not Out (Missing stumps)", trajectory |
|
|
|
def generate_slow_motion(frames, trajectory, output_path): |
|
|
|
fourcc = cv2.VideoWriter_fourcc(*'mp4v') |
|
out = cv2.VideoWriter(output_path, fourcc, FRAME_RATE / 2, (frames[0].shape[1], frames[0].shape[0])) |
|
|
|
for frame in frames: |
|
if trajectory: |
|
for x, y in trajectory: |
|
cv2.circle(frame, (int(x), int(y)), 5, (255, 0, 0), -1) |
|
out.write(frame) |
|
out.write(frame) |
|
out.release() |
|
return output_path |
|
|
|
def drs_review(video): |
|
|
|
if not os.path.exists(video): |
|
return "Error: Video file not found", None |
|
frames, ball_positions = process_video(video) |
|
trajectory, _ = estimate_trajectory(ball_positions, frames) |
|
decision, trajectory = lbw_decision(ball_positions, trajectory, frames) |
|
|
|
|
|
output_path = f"output_{uuid.uuid4()}.mp4" |
|
slow_motion_path = generate_slow_motion(frames, trajectory, output_path) |
|
|
|
return decision, slow_motion_path |
|
|
|
|
|
iface = gr.Interface( |
|
fn=drs_review, |
|
inputs=gr.Video(label="Upload Video Clip"), |
|
outputs=[ |
|
gr.Textbox(label="DRS Decision"), |
|
gr.Video(label="Slow-Motion Replay with Ball Detection and Trajectory") |
|
], |
|
title="AI-Powered DRS for LBW in Local Cricket", |
|
description="Upload a video clip of a cricket delivery to get an LBW decision and slow-motion replay showing ball detection and trajectory." |
|
) |
|
|
|
if __name__ == "__main__": |
|
iface.launch() |