import gradio as gr import cv2 import numpy as np from PIL import Image from pathlib import Path # Assuming find_target_in_source, reid_model, yolo_model, and transform are defined globally from previous cells def process_video_with_reid_gradio(target_image: Image.Image, video1_filepath: str, video1_gps: str, video2_filepath: str, video2_gps: str, video3_filepath: str, video3_gps: str): """ Processes three videos frame by frame, performs human detection and re-identification based on a target image, and saves annotated output videos. Includes GPS location for each video input and reports the location if a match is found. """ video_inputs = [ {"filepath": video1_filepath, "gps": video1_gps, "name": "Camera 1"}, {"filepath": video2_filepath, "gps": video2_gps, "name": "Camera 2"}, {"filepath": video3_filepath, "gps": video3_gps, "name": "Camera 3"} ] if target_image is None or any(v["filepath"] is None for v in video_inputs): return [None] * 3, "Please provide both a target image and all source videos.", "Error: Please provide both a target image and all source videos." output_videos = [] all_summaries = [] concise_summaries = [] for video_info in video_inputs: video_filepath = video_info["filepath"] camera_name = video_info["name"] camera_gps = video_info["gps"] video_filename = Path(video_filepath).name print(f"Processing video from {camera_name} ({video_filename})") # 1. Load the video cap = cv2.VideoCapture(video_filepath) if not cap.isOpened(): all_summaries.append(f"Error: Could not open video file {camera_name} ({video_filename})") concise_summaries.append(f"{camera_name}: Error opening video file.") output_videos.append(None) continue # Get video properties fps = int(cap.get(cv2.CAP_PROP_FPS)) width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) frame_size = (width, height) processed_frames = [] video_summary_lines = [] frame_count = 0 found_match_in_video = False match_time_sec = -1 print("Processing video frames...") # 2. Loop through each frame while True: ret, frame = cap.read() if not ret: break # Convert OpenCV BGR frame to PIL RGB image frame_pil = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)) # 3. Call find_target_in_source annotated_frame_rgb, summary = find_target_in_source(target_image, frame_pil) # Check if any match was found in this frame if "MATCH" in summary and not found_match_in_video: found_match_in_video = True match_time_sec = frame_count / fps if fps > 0 else 0 # Store the results processed_frames.append(annotated_frame_rgb) video_summary_lines.append(f"Frame {frame_count + 1}: {summary}") frame_count += 1 # 4. Release the video capture object cap.release() print("Finished processing frames.") if not processed_frames: all_summaries.append(f"No frames were processed from {camera_name} ({video_filename}).") concise_summaries.append(f"{camera_name}: No frames processed.") output_videos.append(None) continue # 5. Create output video output_video_path = f'/content/gradio_output_{Path(video_filepath).stem}.mp4' fourcc = cv2.VideoWriter_fourcc(*'mp4v') out = cv2.VideoWriter(output_video_path, fourcc, fps, frame_size) if not out.isOpened(): fourcc = cv2.VideoWriter_fourcc(*'XVID') out = cv2.VideoWriter(output_video_path, fourcc, fps, frame_size) if not out.isOpened(): all_summaries.append(f"Error: Could not create video writer object for {camera_name} ({video_filename}).") concise_summaries.append(f"{camera_name}: Error creating output video.") output_videos.append(None) continue print(f"Writing {len(processed_frames)} frames to {output_video_path}...") # 6. Write frames to video for frame_rgb in processed_frames: frame_bgr = cv2.cvtColor(frame_rgb, cv2.COLOR_RGB2BGR) out.write(frame_bgr) # 7. Release the VideoWriter out.release() print(f"Output video created successfully for {camera_name} ({video_filename}).") # 8. Combine summaries for this video, including camera name combined_video_summary = f"--- Summary for {camera_name} ({video_filename}) ---\n" + "\n".join(video_summary_lines) all_summaries.append(combined_video_summary) # 9. Create concise summary for this video if found_match_in_video: concise_summaries.append(f"{camera_name} (GPS: {camera_gps}): Suspect found at approximately {match_time_sec:.2f} seconds.") else: concise_summaries.append(f"{camera_name}: Suspect is not found.") output_videos.append(output_video_path) # 10. Combine all summaries combined_detailed_summary = "\n\n".join(all_summaries) combined_concise_summary = "\n".join(concise_summaries) # 11. Return output video paths and combined summaries return output_videos, combined_detailed_summary, combined_concise_summary # # Create a Gradio Interface video_reid_iface = gr.Interface( fn=process_video_with_reid_gradio, inputs=[ gr.Image(type="pil", label="Target Person Image"), gr.Video(label="Source Video 1"), gr.Text(label="Video 1 GPS Location (e.g., Latitude, Longitude)"), # Added GPS input gr.Video(label="Source Video 2"), gr.Text(label="Video 2 GPS Location (e.g., Latitude, Longitude)"), # Added GPS input gr.Video(label="Source Video 3"), gr.Text(label="Video 3 GPS Location (e.g., Latitude, Longitude)") # Added GPS input ], outputs=[ gr.Gallery(label="Detection Results Videos"), gr.Text(label="Detailed Summary of Results Per Frame Per Video", lines=10), # Increased lines for bigger box gr.Text(label="Concise Summary (Suspect Found/Not Found)") ], title="Person Re-identification in Multiple Videos", description="Upload an image of a target person and exactly 3 videos with their GPS locations. The system will detect all people in each video frame and check if any of them match the target, indicating which video the match was found in and its GPS location." ) # Launch the Gradio interface video_reid_iface.launch(share=True, debug=True)12. Create a Gradio Interface (Interface definition moved to a new cell) # 13. Launch the Gradio interface (Launch call moved to a new cell)