Upload 7 files

app.py

@@ -0,0 +1,197 @@
+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., [email protected]"
+            )
+            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., [email protected]"
+            )
+        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()

cctv_agent.py

@@ -0,0 +1,68 @@
+### START OF cctv_agent.py ###
+import os
+import cv2
+from ultralytics import YOLO
+from langchain.agents import tool
+
+@tool("cctv_detection_tool", return_direct=True)
+def cctv_detection_tool(rtsp_url: str, model_path: str = None, output_dir: str = "detections", frame_skip: int = 30, conf: float = 0.75) -> str:
+    """
+    RTSP URL'si verilen CCTV kamerasına bağlanarak YOLO ile canlı nesne tespiti yapar.
+    
+    Args:
+        rtsp_url (str): CCTV RTSP bağlantı URL'si.
+        model_path (str, optional): YOLO model dosyasının yolu. Belirtilmezse yolo_agent.py içindeki default (last.pt) kullanılır.
+        output_dir (str): Tespit çıktı dosyalarının kaydedileceği dizin.
+        frame_skip (int): İşlenecek kareler arasında atlanacak kare sayısı.
+        conf (float): Tespit için güven eşiği.
+        
+    Returns:
+        str: İşlemin sonucuna dair özet mesaj.
+    """
+    if model_path is None:
+        model_path = os.path.join(os.path.dirname(__file__), "yolo", "last.pt")
+    cap = cv2.VideoCapture(rtsp_url)
+    if not cap.isOpened():
+        return f"RTSP akışı açılamadı: {rtsp_url}"
+    os.makedirs(output_dir, exist_ok=True)
+    output_txt = os.path.join(output_dir, "cctv_detections.txt")
+    ftxt = open(output_txt, "w")
+    try:
+        model = YOLO(model_path)
+    except Exception as e:
+        return f"YOLO modeli yüklenemedi: {e}"
+    frame_count = 0
+    saved_image_index = 1
+    while cap.isOpened():
+        ret, frame = cap.read()
+        if not ret:
+            break
+        results = model(frame, conf=conf)
+        detections = results[0].boxes.data.cpu().numpy() if len(results) > 0 else []
+        valid_detections = [det for det in detections if int(det[5]) in [0, 1, 2, 3, 4, 5]]
+        if valid_detections:
+            for det in valid_detections:
+                x1, y1, x2, y2, conf_score, cls_ = det
+                class_id = int(cls_)
+                if class_id in [0, 1, 2]:
+                    class_label = "Danger"
+                elif class_id in [3, 4, 5]:
+                    class_label = model.names.get(class_id, f"Class {class_id}")
+                else:
+                    class_label = f"Class {class_id}"
+                cv2.rectangle(frame, (int(x1), int(y1)), (int(x2), int(y2)), (0, 0, 255), 3)
+                ftxt.write(f"Frame {frame_count}: {class_label} at ({int(x1)}, {int(y1)}, {int(x2)}, {int(y2)})\n")
+            # Örnek olarak bir kareyi kaydet
+            output_frame_path = os.path.join(output_dir, f"cctv_{saved_image_index}.png")
+            cv2.imwrite(output_frame_path, frame)
+            saved_image_index += 1
+        frame_count += frame_skip
+        cap.set(cv2.CAP_PROP_POS_FRAMES, frame_count)
+        # Demo amacıyla belirli sayıda kare işlendikten sonra döngüden çıkıyoruz
+        if frame_count > 300:
+            break
+    ftxt.close()
+    cap.release()
+    cv2.destroyAllWindows()
+    return f"CCTV tespiti tamamlandı. Çıktılar '{output_dir}' dizininde; log: '{output_txt}'."
+### END OF cctv_agent.py ###

deepseek_agent.py

@@ -0,0 +1,126 @@
+import os
+import re
+from langchain.tools import tool
+from langchain_ollama import ChatOllama
+
+@tool("analysis_tool", return_direct=True)
+def analysis_tool(detections_file: str) -> str:
+    """
+    Reads the YOLO detection log (detections_file) and uses the DeepSeek model via ChatOllama
+    to produce an email-style summary in plain text with:
+
+    - Number of 'Danger' detections
+    - Brief explanation
+    - General comment/warning
+
+    The final text is written to 'analysis.txt', no chain-of-thought or placeholders.
+    """
+    if not os.path.exists(detections_file):
+        return f"detections_file not found: {detections_file}"
+
+    # Read YOLO detection log
+    with open(detections_file, "r") as f:
+        detections_text = f.read().strip()
+
+    # Count Danger occurrences
+    manual_count = detections_text.count("Danger")
+
+    # Build system and human prompts
+    system_prompt = (
+        "You are an AI that summarizes YOLO detection logs. "
+        "Return only the final summary with 3 bullet lines:\n"
+        "1) - Number of 'Danger' detections: X\n"
+        "2) - Brief explanation: ...\n"
+        "3) - General comment/warning: ...\n"
+        "No chain of thought, no subject, no email headers. Output ONLY the body lines in plain text."
+    )
+
+    human_prompt = f"""
+YOLO detection log:
+{detections_text}
+
+Please produce exactly 3 bullet lines:
+- Number of 'Danger' detections: <number>
+- Brief explanation: <one line summary>
+- General comment/warning: <safety note>
+"""
+
+    llm = ChatOllama(
+        model="deepseek-r1:1.5b",
+        temperature=0.0,
+        base_url="http://localhost:11434",
+    )
+
+    messages = [
+        ("system", system_prompt),
+        ("human", human_prompt),
+    ]
+
+    # Get raw model output
+    ai_msg = llm.invoke(messages)
+    raw_output = ai_msg.content.strip()
+
+    # Remove <think> blocks
+    raw_output = re.sub(r"<think>.*?</think>", "", raw_output, flags=re.DOTALL)
+
+    # Split lines
+    lines = [line.strip() for line in raw_output.splitlines() if line.strip()]
+
+    # We'll look for the bullet lines
+    danger_idx = None
+    explain_idx = None
+    comment_idx = None
+
+    for i, line in enumerate(lines):
+        lower_line = line.lower()
+        if "number of 'danger' detections:" in lower_line:
+            danger_idx = i
+        elif "brief explanation:" in lower_line:
+            explain_idx = i
+        elif "general comment/warning:" in lower_line:
+            comment_idx = i
+
+    final_lines = lines[:]
+
+    # 1) If Danger line doesn't exist, append fallback
+    if danger_idx is None:
+        final_lines.append(f"- Number of 'Danger' detections: {manual_count}")
+    else:
+        # If it exists but has placeholder <number>, replace it
+        line_text = final_lines[danger_idx]
+        if "<number>" in line_text:
+            new_line = line_text.replace("<number>", str(manual_count))
+            final_lines[danger_idx] = new_line
+
+    # 2) Explanation line fallback
+    if explain_idx is None:
+        if manual_count == 0:
+            final_lines.append("- Brief explanation: No Danger found.")
+        else:
+            final_lines.append("- Brief explanation: Potential risks detected.")
+    else:
+        # If it has <one line summary> placeholder
+        if "<one line summary>" in final_lines[explain_idx]:
+            final_lines[explain_idx] = final_lines[explain_idx].replace(
+                "<one line summary>",
+                "Potential risks detected." if manual_count else "No Danger found."
+            )
+
+    # 3) Comment line fallback
+    if comment_idx is None:
+        final_lines.append("- General comment/warning: Review and investigate for safety issues.")
+    else:
+        if "<safety note>" in final_lines[comment_idx]:
+            final_lines[comment_idx] = final_lines[comment_idx].replace(
+                "<safety note>", "Please monitor closely."  # or custom
+            )
+
+    # Join final lines
+    final_output = "\n".join(final_lines)
+
+    # Save to analysis.txt
+    analysis_path = os.path.join(os.path.dirname(detections_file), "analysis.txt")
+    with open(analysis_path, "w") as out_f:
+        out_f.write(final_output)
+
+    return f"Analysis complete. Summary saved to '{analysis_path}'."

last.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:3c7f69ad83e71c4a19ffbcdff30fcabbbbee4ef335033c8fb24b35186d0f38ce
+size 5470931

mail_agent.py

@@ -0,0 +1,64 @@
+from langchain.tools import tool
+import os
+import smtplib
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+from email.mime.base import MIMEBase
+from email import encoders
+
+@tool("mail_tool", return_direct=True)
+def mail_tool(input_data: dict) -> str:
+    """
+    Expects a dict parameter with the following keys:
+      - analysis_file: Path to the analysis.txt file.
+      - image_file: Path to an image file (e.g., 1.png) to attach.
+      - sender_email: The email address from which to send the email.
+      - sender_password: The password or app-specific password for the sender email.
+      - recipient_email: The email address to send the email to.
+    
+    This function reads the analysis report, attaches the image, and sends an email using the provided credentials.
+    """
+    analysis_file = input_data.get("analysis_file", "")
+    image_file = input_data.get("image_file", "")
+    sender_email = input_data.get("sender_email", "")
+    sender_password = input_data.get("sender_password", "")
+    recipient_email = input_data.get("recipient_email", "")
+
+    if not os.path.exists(analysis_file):
+        return f"Analysis file not found: {analysis_file}"
+
+    with open(analysis_file, "r") as f:
+        email_body = f.read().strip()
+
+    SMTP_SERVER = "smtp.gmail.com"
+    SMTP_PORT = 587
+
+    msg = MIMEMultipart()
+    msg["Subject"] = "DeepSeek Analysis Report"
+    msg["From"] = sender_email
+    msg["To"] = recipient_email
+    msg.attach(MIMEText(email_body, "plain"))
+
+    if image_file and os.path.isfile(image_file):
+        try:
+            with open(image_file, "rb") as attachment:
+                part = MIMEBase("application", "octet-stream")
+                part.set_payload(attachment.read())
+            encoders.encode_base64(part)
+            part.add_header("Content-Disposition", f'attachment; filename={os.path.basename(image_file)}')
+            msg.attach(part)
+        except Exception as e:
+            return f"Error attaching {image_file}: {e}"
+    else:
+        return f"No image file found at '{image_file}'. Email not sent."
+
+    try:
+        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
+            server.ehlo()
+            server.starttls()
+            server.login(sender_email, sender_password)
+            server.send_message(msg)
+    except Exception as e:
+        return f"Failed to send email: {str(e)}"
+
+    return f"Email sent successfully to {recipient_email}."

requirements.txt

@@ -0,0 +1,8 @@
+langchain
+langchain_ollama
+ultralytics
+opencv-python
+transformers
+accelerate
+torch
+torchvision

yolo_agent.py

@@ -0,0 +1,160 @@
+"""
+LangChain YOLO Agent
+---------------------
+
+This project provides a YOLO-based object detection tool integrated with LangChain.
+Users can upload any video to analyze its contents, generate object detection logs,
+and visualize detections with bounding boxes.
+
+Steps:
+1) Install dependencies: `pip install langchain openai ultralytics opencv-python`
+2) Add this file (`yolo_agent.py`) to your project.
+3) Ensure that the YOLO model file (`last.pt`) is available in the working directory.
+4) Use the provided functions to analyze uploaded videos dynamically.
+"""
+
+import os
+import cv2
+import shutil
+from langchain.agents import Tool, tool
+from ultralytics import YOLO
+
+UPLOAD_FOLDER = "uploads"
+os.makedirs(UPLOAD_FOLDER, exist_ok=True)
+
+MODEL_PATH = os.path.join(os.path.dirname(__file__), "last.pt")
+
+def detect_with_yolo(
+    video_path: str,
+    model_path: str = MODEL_PATH,
+    output_dir: str = "detections",
+    frame_skip: int = 7,
+    conf: float = 0.75,
+) -> str:
+    """
+    Runs YOLO detection on the given video.
+    - Detects only class_id 0..5 (Danger / Handgun / Knife, etc.)
+    - Draws red bounding boxes
+    - Saves logs to a text file
+    - Saves detected frames as sequential PNG images (1.png, 2.png, 3.png, ...)
+    """
+
+    if not os.path.exists(video_path):
+        return f"Video not found: {video_path}"
+
+    try:
+        model = YOLO(model_path)
+    except Exception as e:
+        return f"Failed to load model: {e}"
+
+    cap = cv2.VideoCapture(video_path)
+    if not cap.isOpened():
+        return f"Cannot open video: {video_path}"
+
+    os.makedirs(output_dir, exist_ok=True)
+    output_txt = os.path.join(output_dir, "detections.txt")
+
+    frame_count = 0
+    saved_image_index = 1  # Sıralı kaydetmek için sayaç
+
+    with open(output_txt, "w") as ftxt:
+        while cap.isOpened():
+            ret, frame = cap.read()
+            if not ret:
+                break
+
+            results = model(frame, conf=conf)
+            detections = (
+                results[0].boxes.data.cpu().numpy() if len(results) > 0 else []
+            )
+
+            valid_detections = [det for det in detections if int(det[5]) in [0, 1, 2, 3, 4, 5]]
+
+            if len(valid_detections) > 0:
+                for det in valid_detections:
+                    x1, y1, x2, y2, conf_score, cls_ = det
+                    class_id = int(cls_)
+
+                    if class_id in [0, 1, 2]:
+                        class_label = "Danger"
+                    elif class_id in [3, 4, 5]:
+                        # Alternatif class isimleri
+                        class_label = model.names.get(class_id, f"Class {class_id}")
+                    else:
+                        class_label = f"Class {class_id}"
+
+                    cv2.rectangle(
+                        frame,
+                        (int(x1), int(y1)),
+                        (int(x2), int(y2)),
+                        (0, 0, 255),
+                        3,
+                    )
+
+                    (w, h), _ = cv2.getTextSize(class_label, cv2.FONT_HERSHEY_COMPLEX, 0.8, 2)
+                    label_x1 = int(x1)
+                    label_y2 = int(y1)
+                    label_y1 = label_y2 - h - 10
+                    label_x2 = label_x1 + w + 10
+
+                    cv2.rectangle(
+                        frame,
+                        (label_x1, label_y1),
+                        (label_x2, label_y2),
+                        (0, 0, 255),
+                        cv2.FILLED,
+                    )
+
+                    cv2.putText(
+                        frame,
+                        class_label,
+                        (label_x1 + 5, label_y1 + h + 5),
+                        cv2.FONT_HERSHEY_COMPLEX,
+                        0.85,
+                        (255, 255, 255),
+                        2,
+                        cv2.LINE_AA,
+                    )
+
+                    ftxt.write(
+                        f"Frame {frame_count}: {class_label} at ({int(x1)}, {int(y1)}, {int(x2)}, {int(y2)})\n"
+                    )
+
+                # Kaydedilecek resim ismi -> 1.png, 2.png, 3.png
+                output_frame_path = os.path.join(output_dir, f"{saved_image_index}.png")
+                cv2.imwrite(output_frame_path, frame)
+                saved_image_index += 1
+
+            frame_count += frame_skip
+            cap.set(cv2.CAP_PROP_POS_FRAMES, frame_count)
+
+    cap.release()
+    cv2.destroyAllWindows()
+
+    return f"Processing complete. Outputs saved in '{output_dir}' and '{output_txt}'."
+
+@tool("video_detection_tool", return_direct=True)
+def video_detection_tool(video) -> str:
+    """
+    Handles video uploads dynamically and runs YOLO detection.
+    Expects that the input 'video' is either a file-like object with a 'name' attribute
+    or a string representing the file path.
+    """
+    if not os.path.exists(UPLOAD_FOLDER):
+        os.makedirs(UPLOAD_FOLDER, exist_ok=True)
+
+    # Eğer video bir dosya nesnesiyse video.name kullan, aksi halde doğrudan video string'idir.
+    try:
+        video_name = video.name
+    except AttributeError:
+        video_name = video
+
+    video_path = os.path.join(UPLOAD_FOLDER, os.path.basename(video_name))
+    if os.path.abspath(video_name) != os.path.abspath(video_path):
+        shutil.copy(video_name, video_path)
+    return detect_with_yolo(video_path)
+
+if __name__ == "__main__":
+    print("LangChain YOLO Agent Ready!")
+
+__all__ = ["video_detection_tool"]