File size: 4,265 Bytes
7a58ba4 8862f2a f68e74f 8862f2a f68e74f 8862f2a f68e74f 8862f2a f68e74f 8862f2a f68e74f 8862f2a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 |
import gradio as gr
import os
import tempfile
from Fight_detec_func import fight_detec
from objec_detect_yolo import detection
import time # Added for unique temp file names
def analyze_video(video_file_obj):
if video_file_obj is None:
return {"Error": "No video file uploaded."}
temp_dir = tempfile.mkdtemp()
# Create a unique filename within the temp dir
input_filename = os.path.basename(video_file_obj.name)
base, ext = os.path.splitext(input_filename)
unique_suffix = str(int(time.time() * 1000)) # Add timestamp for uniqueness
safe_base = "".join(c if c.isalnum() or c in ('-', '_') else '_' for c in base) # Sanitize name
video_path = os.path.join(temp_dir, f"{safe_base}_{unique_suffix}{ext}")
try:
# Gradio file object has '.name' attribute with the path to temp copy
# Copy it to ensure control over the path and name if needed downstream
with open(video_path, 'wb') as f_dst, open(video_file_obj.name, 'rb') as f_src:
f_dst.write(f_src.read())
print(f"Processing video: {video_path}")
# Run detection functions
# fight_detec returns (result_string, prediction_score)
# detection returns (set_of_labels, output_video_path)
fight_status, _ = fight_detec(video_path, debug=False)
detected_objects_set, annotated_video_path = detection(video_path)
# Format results
# Convert set to sorted list for consistent JSON output
detected_objects_list = sorted(list(detected_objects_set))
print(f"Fight Status: {fight_status}")
print(f"Detected Objects: {detected_objects_list}")
# Note: annotated_video_path points to a file saved in the 'results' directory
# within the Space container, but we are not returning it via the UI here.
results = {
"Fight Detection": fight_status,
"Detected Objects": detected_objects_list
}
except Exception as e:
print(f"Error during processing: {e}")
results = {"Error": f"Processing failed: {str(e)}"}
finally:
# Clean up the specific temp file and directory
if 'video_path' in locals() and os.path.exists(video_path):
try:
os.remove(video_path)
print(f"Removed temp video file: {video_path}")
except OSError as e:
print(f"Error removing temp file {video_path}: {e}")
if 'temp_dir' in locals() and os.path.exists(temp_dir):
try:
# Clean up the results dir created by objec_detect_yolo if it's inside temp_dir
results_dir_path = os.path.join(temp_dir, "results")
if os.path.exists(results_dir_path) and os.path.isdir(results_dir_path):
# Remove files inside results dir first
for item in os.listdir(results_dir_path):
item_path = os.path.join(results_dir_path, item)
if os.path.isfile(item_path):
os.remove(item_path)
os.rmdir(results_dir_path) # Now remove empty results dir
print(f"Removed temp results directory: {results_dir_path}")
os.rmdir(temp_dir) # Attempt to remove the main temp dir
print(f"Removed temp directory: {temp_dir}")
except OSError as e:
# Might fail if other files are present or dir not empty
print(f"Error removing temp directory {temp_dir} or its contents: {e}")
return results
# Interface Definition
iface = gr.Interface(
fn=analyze_video,
inputs=gr.Video(label="Upload Video"), # Source can be 'upload' or 'webcam'
outputs=gr.JSON(label="Detection Results"),
title="Fight and Object Detection Analysis",
description="Upload a video (< 1 min recommended) to detect potential fights and specific objects (Fire, Gun, Knife, Smoke, License_Plate). Results appear as JSON.",
allow_flagging='never',
examples=[
# Add paths to example videos if you upload them to the HF repo
# e.g., ["example_fight.mp4"], ["example_normal_gun.mp4"]
]
)
# Launch the interface
if __name__ == "__main__":
iface.launch()
|