Spaces:
Sleeping
Sleeping
| import os | |
| import sys | |
| import gradio as gr | |
| from yolo_agent import video_detection_tool | |
| from deepseek_agent import analysis_tool | |
| from mail_agent import mail_tool | |
| from cctv_agent import cctv_detection_tool | |
| def cli_pipeline(): | |
| """ | |
| """ | |
| video_path = input("Enter the full path to the video file: ").strip() | |
| if not os.path.exists(video_path): | |
| print("The specified file was not found!") | |
| return | |
| # 1. YOLO Detection | |
| with open(video_path, 'rb') as video_file: | |
| detection_result = video_detection_tool({"video": video_file}) | |
| print("Detection Result:\n", detection_result) | |
| # 2. DeepSeek Analysis | |
| detections_folder = "detections" | |
| detections_file = os.path.join(detections_folder, "detections.txt") | |
| if not os.path.exists(detections_file): | |
| print("Detections file not found! Please check the detection process first.") | |
| return | |
| analysis_result = analysis_tool(detections_file) | |
| print("Analysis Result:\n", analysis_result) | |
| # 3. Email Sending (optional) | |
| analysis_file = os.path.join(detections_folder, "analysis.txt") | |
| if not os.path.exists(analysis_file): | |
| print("analysis.txt not found; email cannot be sent.") | |
| return | |
| image_path = os.path.join(detections_folder, "1.png") | |
| if not os.path.exists(image_path): | |
| print("1.png not found; email will not be sent.") | |
| return | |
| sender_email = input("Enter the sender email address: ").strip() | |
| sender_password = input("Enter the email application password: ").strip() | |
| recipient_email = input("Enter the recipient email address: ").strip() | |
| mail_result = mail_tool({ | |
| "input_data": { | |
| "analysis_file": analysis_file, | |
| "image_file": image_path, | |
| "sender_email": sender_email, | |
| "sender_password": sender_password, | |
| "recipient_email": recipient_email | |
| } | |
| }) | |
| print("Email Result:\n", mail_result) | |
| def process_input(mode, video_file, rtsp_url, model_selection, user_id, sender_email, sender_password, recipient_email): | |
| """ | |
| Processes inputs from the Gradio interface: | |
| - In "Upload Video" mode: runs YOLO detection via video_detection_tool. | |
| - In "Connect to CCTV" mode: runs detection via cctv_detection_tool using the provided RTSP URL. | |
| In video mode, if detection and analysis are successful, it uses the email settings | |
| provided by the user to send an email. | |
| """ | |
| if mode == "Upload Video": | |
| if video_file is None: | |
| return "Please upload a video file." | |
| detection_result = video_detection_tool({"video": video_file}) | |
| detections_file = os.path.join("detections", "detections.txt") | |
| if not os.path.exists(detections_file): | |
| return detection_result + "\n\nDetections file not found." | |
| analysis_result = analysis_tool(detections_file) | |
| image_path = os.path.join("detections", "1.png") | |
| if os.path.exists(image_path): | |
| if not sender_email or not sender_password or not recipient_email: | |
| mail_part = "Missing email information for sending the email." | |
| else: | |
| mail_part = mail_tool({ | |
| "input_data": { | |
| "analysis_file": os.path.join("detections", "analysis.txt"), | |
| "image_file": image_path, | |
| "sender_email": sender_email, | |
| "sender_password": sender_password, | |
| "recipient_email": recipient_email | |
| } | |
| }) | |
| return f"{detection_result}\n\n{analysis_result}\n\n{mail_part}" | |
| else: | |
| return f"{detection_result}\n\n{analysis_result}\n\n1.png not found, email not sent." | |
| elif mode == "Connect to CCTV": | |
| if not rtsp_url or rtsp_url.strip() == "": | |
| return "Please enter a valid RTSP URL." | |
| detection_result = cctv_detection_tool(rtsp_url, model_selection) | |
| return detection_result | |
| else: | |
| return "Invalid mode selected." | |
| def launch_gradio_interface(): | |
| """ | |
| Launches the Gradio interface. | |
| Users can choose either "Upload Video" or "Connect to CCTV" mode and provide the necessary inputs | |
| to perform detection, analysis, and optionally email sending. | |
| """ | |
| mode_choices = ["Upload Video", "Connect to CCTV"] | |
| model_choices = ["last.pt"] | |
| with gr.Blocks(css=".gradio-container { font-family: 'Segoe UI', sans-serif; }") as demo: | |
| gr.Markdown("## YOLO Detection: Video or CCTV Stream\n" | |
| "In this interface, you can either upload your local video or connect via a CCTV RTSP stream " | |
| "to perform object detection.\n" | |
| "**Additionally:** DeepSeek analysis and email sending are integrated.") | |
| gr.Markdown("### 1) Mode Selection") | |
| mode = gr.Radio(mode_choices, label="How would you like to perform the detection?", value="Upload Video") | |
| gr.Markdown("---") | |
| with gr.Row(): | |
| with gr.Column(): | |
| gr.Markdown("**Local Video File**") | |
| video_input = gr.Video( | |
| label="Upload Video", | |
| sources="upload", | |
| visible=True | |
| ) | |
| with gr.Column(): | |
| gr.Markdown("**CCTV Stream (RTSP)**") | |
| rtsp_input = gr.Textbox( | |
| label="RTSP URL", | |
| placeholder="rtsp://username:password@ip:port/stream", | |
| visible=False | |
| ) | |
| gr.Markdown("---") | |
| with gr.Row(): | |
| model_selection_input = gr.Dropdown( | |
| model_choices, | |
| label="YOLO Model Selection", | |
| value="last.pt", | |
| info="Select the YOLO model you wish to use." | |
| ) | |
| user_id = gr.Textbox( | |
| label="User ID (Optional)", | |
| placeholder="Enter your user ID..." | |
| ) | |
| gr.Markdown("### Email Information (for sending results)") | |
| with gr.Row(): | |
| sender_email = gr.Textbox( | |
| label="Sender Email Address", | |
| placeholder="e.g., sender@example.com" | |
| ) | |
| sender_password = gr.Textbox( | |
| label="Email (app) Password", | |
| placeholder="Enter your password", | |
| type="password" | |
| ) | |
| recipient_email = gr.Textbox( | |
| label="Recipient Email Address", | |
| placeholder="e.g., recipient@example.com" | |
| ) | |
| gr.Markdown("---") | |
| output_text = gr.Textbox( | |
| label="Output", | |
| interactive=False, | |
| placeholder="Detection, analysis, and email output will be shown here..." | |
| ) | |
| def update_visibility(selected_mode): | |
| if selected_mode == "Upload Video": | |
| return gr.update(visible=True), gr.update(visible=False) | |
| else: | |
| return gr.update(visible=False), gr.update(visible=True) | |
| mode.change(fn=update_visibility, inputs=mode, outputs=[video_input, rtsp_input]) | |
| submit_btn = gr.Button("Start Detection", variant="primary") | |
| submit_btn.click(fn=process_input, | |
| inputs=[mode, video_input, rtsp_input, model_selection_input, user_id, sender_email, sender_password, recipient_email], | |
| outputs=output_text) | |
| gr.Markdown("---\n" | |
| "bahakizil\n") | |
| demo.launch(server_name="0.0.0.0", server_port=7860) | |
| def main(): | |
| cli_pipeline() | |
| if __name__ == "__main__": | |
| if len(sys.argv) > 1 and sys.argv[1] == "--cli": | |
| main() | |
| else: | |
| launch_gradio_interface() | |