Spaces:
Runtime error
Runtime error
| 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) |