Spaces:
Sleeping
Sleeping
--- START OF FILE app.py --- | |
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() | |
--- END OF FILE app.py --- |