{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":92399,"databundleVersionId":11038207,"sourceType":"competition"},{"sourceId":11187573,"sourceType":"datasetVersion","datasetId":6983904}],"dockerImageVersionId":30918,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import shutil\n\n# Delete the entire folder and its contents\nshutil.rmtree('/kaggle/working/output_frames')# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load\n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\nimport os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        print(os.path.join(dirname, filename))\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-03-27T21:20:36.626251Z","iopub.execute_input":"2025-03-27T21:20:36.626649Z","iopub.status.idle":"2025-03-27T21:20:36.661299Z","shell.execute_reply.started":"2025-03-27T21:20:36.626616Z","shell.execute_reply":"2025-03-27T21:20:36.659662Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import shutil\n\n# Delete the entire folder and its contents\nshutil.rmtree('/kaggle/working')","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-27T21:20:41.079326Z","iopub.execute_input":"2025-03-27T21:20:41.079697Z","iopub.status.idle":"2025-03-27T21:20:41.113313Z","shell.execute_reply.started":"2025-03-27T21:20:41.079668Z","shell.execute_reply":"2025-03-27T21:20:41.111848Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!pip install ultralytics","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-02T19:25:53.423055Z","iopub.execute_input":"2025-04-02T19:25:53.423545Z","iopub.status.idle":"2025-04-02T19:26:00.966979Z","shell.execute_reply.started":"2025-04-02T19:25:53.423506Z","shell.execute_reply":"2025-04-02T19:26:00.965482Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport cv2\nimport numpy as np\nimport matplotlib.pyplot as plt\nimport matplotlib.animation as animation\nfrom tqdm import tqdm\nfrom ultralytics import YOLO\n\ndef extract_frames(video_path, output_folder):\n    \"\"\"\n    Extract frames from a video and save them as images.\n    \n    Parameters:\n    -----------\n    video_path : str\n        Path to the video file.\n    output_folder : str\n        Folder where frames will be saved.\n    \"\"\"\n    if not os.path.exists(output_folder):\n        os.makedirs(output_folder)\n    \n    cap = cv2.VideoCapture(video_path)\n    frame_count = 0\n    while True:\n        ret, frame = cap.read()\n        if not ret:\n            break\n        \n        frame_filename = os.path.join(output_folder, f'{frame_count:04d}.png')\n        cv2.imwrite(frame_filename, frame)\n        frame_count += 1\n    \n    cap.release()\n    print(f\"Extracted {frame_count} frames from {video_path}\")\n\ndef detect_objects_in_frames(model, input_folder, output_folder, conf_threshold=0.2):\n    \"\"\"\n    Use the YOLOv8 model to detect objects in each frame in the input folder.\n    \n    Parameters:\n    -----------\n    model : YOLO\n        The pre-trained YOLOv8 model.\n    input_folder : str\n        Folder containing frames to detect objects in.\n    output_folder : str\n        Folder to save the detected frames.\n    conf_threshold : float, optional\n        Confidence threshold for object detection (default: 0.2)\n    \"\"\"\n    if not os.path.exists(output_folder):\n        os.makedirs(output_folder)\n    \n    frames = sorted([f for f in os.listdir(input_folder) if f.endswith('.png')])\n    for frame_filename in tqdm(frames, desc=\"Detecting Objects\", unit=\"frame\"):\n        frame_path = os.path.join(input_folder, frame_filename)\n        frame = cv2.imread(frame_path)\n        \n        # Run object detection on the frame\n        results = model(frame, conf=conf_threshold)\n        \n        # Annotate the frame with detection results\n        annotated_frame = results[0].plot()  # Plot the detected objects\n        \n        # Save the annotated frame\n        annotated_frame_path = os.path.join(output_folder, frame_filename)\n        cv2.imwrite(annotated_frame_path, annotated_frame)\n    \n    print(f\"Object detection complete. Annotated frames saved to {output_folder}\")\n\ndef create_animation(ims):\n    \"\"\"\n    Create an animation from a list of image frames.\n    \n    Parameters:\n    -----------\n    ims : list or numpy array\n        List of image frames.\n    \n    Returns:\n    --------\n    matplotlib animation object\n    \"\"\"\n    fig = plt.figure(figsize=(10, 6))\n    im = plt.imshow(cv2.cvtColor(ims[0], cv2.COLOR_BGR2RGB))\n    plt.axis('off')\n    plt.close()\n\n    def animate_func(i):\n        im.set_array(cv2.cvtColor(ims[i], cv2.COLOR_BGR2RGB))\n        return [im]\n    \n    return animation.FuncAnimation(fig, animate_func, frames=len(ims), interval=1000//3)\n\ndef main():\n    # Path to the video file\n    video_path = '/kaggle/input/nexar-collision-prediction/test/00092.mp4'\n    \n    # Define folders to store extracted frames and annotated frames\n    frames_folder = '/kaggle/working/frames'\n    annotated_frames_folder = '/kaggle/working/annotated_frames'\n    \n    # Extract frames from the video\n    extract_frames(video_path, frames_folder)\n    \n    # Load the YOLOv8 model\n    model = YOLO('yolov8x.pt')\n    \n    # Run object detection on extracted frames\n    detect_objects_in_frames(model, frames_folder, annotated_frames_folder)\n    \n    # Load the annotated frames to create animation\n    annotated_frames = sorted([cv2.imread(os.path.join(annotated_frames_folder, f)) for f in os.listdir(annotated_frames_folder) if f.endswith('.png')])\n    \n    # Create and display the animation\n    anim = create_animation(np.array(annotated_frames))\n    from IPython.display import HTML\n    HTML(anim.to_jshtml())\n\nif __name__ == '__main__':\n    main()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-02T19:27:30.088324Z","iopub.execute_input":"2025-04-02T19:27:30.088793Z","iopub.status.idle":"2025-04-02T19:37:32.078330Z","shell.execute_reply.started":"2025-04-02T19:27:30.088757Z","shell.execute_reply":"2025-04-02T19:37:32.076717Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import cv2\nimport os\n\n# Path to the directory containing the annotated frames\ninput_frames_dir = '/kaggle/working/annotated_frames'\noutput_video_path = '/kaggle/working/annotated_video.mp4'\n\n# Get sorted list of frames\nframes = sorted([os.path.join(input_frames_dir, f) for f in os.listdir(input_frames_dir) if f.endswith('.png')])\n\n# Read the first frame to get the frame size\nframe = cv2.imread(frames[0])\nheight, width, layers = frame.shape\n\n# Create a VideoWriter object to write the frames into a video\nfourcc = cv2.VideoWriter_fourcc(*'mp4v')  # Codec for .mp4 format\nvideo_writer = cv2.VideoWriter(output_video_path, fourcc, 30.0, (width, height))  # 30 fps\n\n# Add frames to the video\nfor frame_path in frames:\n    frame = cv2.imread(frame_path)\n    video_writer.write(frame)\n\n# Release the VideoWriter object\nvideo_writer.release()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-02T19:37:39.489254Z","iopub.execute_input":"2025-04-02T19:37:39.489831Z","iopub.status.idle":"2025-04-02T19:37:49.005482Z","shell.execute_reply.started":"2025-04-02T19:37:39.489786Z","shell.execute_reply":"2025-04-02T19:37:49.004060Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport cv2\n\n# Path to annotated frames folder\nannotated_frames_folder = '/kaggle/working/annotated_frames'\n\n# Get a sorted list of valid image paths\nframe_paths = sorted([os.path.join(annotated_frames_folder, f) for f in os.listdir(annotated_frames_folder) if f.endswith('.png')])\n\n# Read images while ensuring no None values\nannotated_frames = []\nfor path in frame_paths:\n    img = cv2.imread(path)\n    if img is not None:\n        annotated_frames.append(img)  # Append only valid images\n\n# Check if any valid images were loaded\nif not annotated_frames:\n    raise ValueError(\"Error: No valid annotated frames found. Check if the folder contains PNG files.\")\n\nprint(f\"Loaded {len(annotated_frames)} annotated frames successfully!\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-02T19:37:52.369932Z","iopub.execute_input":"2025-04-02T19:37:52.370352Z","iopub.status.idle":"2025-04-02T19:37:59.814116Z","shell.execute_reply.started":"2025-04-02T19:37:52.370322Z","shell.execute_reply":"2025-04-02T19:37:59.812915Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"###LOGIC1\n\nimport os\nimport cv2\nimport numpy as np\n\n# Path settings\nannotated_frames_folder = '/kaggle/working/annotated_frames'\noutput_video_path = '/kaggle/working/traffic_caution_collision_video.mp4'\n\n# Get sorted list of valid image paths\nframe_paths = sorted([os.path.join(annotated_frames_folder, f) for f in os.listdir(annotated_frames_folder) if f.endswith('.png')])\n\n# Load frames\nannotated_frames = []\nfor path in frame_paths:\n    img = cv2.imread(path)\n    if img is not None:\n        annotated_frames.append(img)\n\nif not annotated_frames:\n    raise ValueError(\"Error: No valid annotated frames found. Check if the folder contains PNG files.\")\n\nprint(f\"Loaded {len(annotated_frames)} annotated_frames successfully!\")\n\n# Get frame dimensions from first frame\nheight, width, layers = annotated_frames[0].shape\n\n# Initialize video writer\nfourcc = cv2.VideoWriter_fourcc(*'mp4v')\nvideo_writer = cv2.VideoWriter(output_video_path, fourcc, 30.0, (width, height))\n\n# Function to check proximity in heavy traffic\ndef check_dashcam_proximity(frame, caution_area=20000, collision_area=100000, \n                          caution_bottom=0.15, collision_bottom=0.02, debug=False):\n    \"\"\"\n    Adjusted for heavy traffic:\n    - Caution: For typical close proximity in traffic\n    - Collision Possible: Only when vehicles are extremely close (near contact)\n    \"\"\"\n    # Placeholder contour detection (replace with your bounding box data)\n    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)\n    _, thresh = cv2.threshold(gray, 127, 255, cv2.THRESH_BINARY)\n    contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)\n    \n    caution = False\n    collision = False\n    for contour in contours:\n        area = cv2.contourArea(contour)\n        if area > 500:  # Filter small noise\n            x, y, w, h = cv2.boundingRect(contour)\n            box_area = w * h\n            bottom_y = y + h\n            \n            # Caution conditions (lenient for traffic)\n            caution_large = box_area > caution_area\n            caution_close = bottom_y > height * (1 - caution_bottom)\n            \n            # Collision conditions (very strict)\n            collision_large = box_area > collision_area\n            collision_close = bottom_y > height * (1 - collision_bottom)\n            \n            if debug:\n                print(f\"Box: Area={box_area}, Bottom_Y={bottom_y}, Frame_Height={height}\")\n                print(f\"Caution: Large={caution_large}, Close={caution_close}\")\n                print(f\"Collision: Large={collision_large}, Close={collision_close}\")\n            \n            # Collision check (highest priority)\n            if collision_large and collision_close:  # Both conditions for collision\n                collision = True\n                cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 0, 255), 4)\n                cv2.putText(frame, \"COLLISION POSSIBLE!\", (50, 50),\n                           cv2.FONT_HERSHEY_SIMPLEX, 1.2, (0, 0, 255), 3)\n                if debug:\n                    print(\"Collision possible triggered!\")\n            \n            # Caution check (only if no collision)\n            elif caution_large or caution_close:\n                caution = True\n                cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 255, 255), 2)\n                cv2.putText(frame, \"CAUTION: Vehicle Close!\", (50, 100),\n                           cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 2)\n                if debug:\n                    print(\"Caution triggered!\")\n    \n    return frame, caution, collision\n\n# Process and write frames\ncaution_count = 0\ncollision_count = 0\nfor i, frame in enumerate(annotated_frames):\n    # Check proximity with debug for first few frames\n    debug = i < 5\n    processed_frame, caution_detected, collision_detected = check_dashcam_proximity(\n        frame.copy(),\n        caution_area=20000,         # Suitable for traffic proximity\n        collision_area=100000,      # Very large for collision\n        caution_bottom=0.15,        # Bottom 15% for caution in traffic\n        collision_bottom=0.02,      # Bottom 2% for collision\n        debug=debug\n    )\n    \n    if collision_detected:\n        collision_count += 1\n        cv2.rectangle(processed_frame, (0, 0), (width, height), (0, 0, 255), 10)  # Red border\n    elif caution_detected:\n        caution_count += 1\n        cv2.rectangle(processed_frame, (0, 0), (width, height), (0, 255, 255), 10)  # Yellow border\n    \n    # Write frame to video\n    video_writer.write(processed_frame)\n\n# Release video writer and print summary\nvideo_writer.release()\nprint(f\"Video processing complete.\")\nprint(f\"Caution frames: {caution_count} ({(caution_count / len(annotated_frames)) * 100:.2f}%)\")\nprint(f\"Collision possible frames: {collision_count} ({(collision_count / len(annotated_frames)) * 100:.2f}%)\")\nprint(f\"Output saved to: {output_video_path}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-02T19:38:08.031867Z","iopub.execute_input":"2025-04-02T19:38:08.032346Z","iopub.status.idle":"2025-04-02T19:38:20.183164Z","shell.execute_reply.started":"2025-04-02T19:38:08.032310Z","shell.execute_reply":"2025-04-02T19:38:20.181983Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"###LOGIC2\n\nimport os\nimport cv2\nimport numpy as np\nfrom scipy.spatial import distance\n\n# Path settings\nannotated_frames_folder = '/kaggle/working/annotated_frames'\noutput_video_path = '/kaggle/working/hybrid_collision_video_v2.mp4'\n\n# Get sorted list of valid image paths\nframe_paths = sorted([os.path.join(annotated_frames_folder, f) for f in os.listdir(annotated_frames_folder) if f.endswith('.png')])\n\n# Load frames\nannotated_frames = []\nfor path in frame_paths:\n    img = cv2.imread(path)\n    if img is not None:\n        annotated_frames.append(img)\n\nif not annotated_frames:\n    raise ValueError(\"Error: No valid annotated_frames found. Check if the folder contains PNG files.\")\n\nprint(f\"Loaded {len(annotated_frames)} annotated_frames successfully!\")\n\n# Get frame dimensions from first frame\nheight, width, layers = annotated_frames[0].shape\n\n# Initialize video writer\nfourcc = cv2.VideoWriter_fourcc(*'mp4v')\nvideo_writer = cv2.VideoWriter(output_video_path, fourcc, 30.0, (width, height))\n\n# Function for hybrid spatial-temporal collision detection\ndef check_hybrid_proximity(frame, prev_centroids, frame_num, caution_dist=100, collision_dist=50, \n                          closure_rate_threshold=10, debug=False):\n    \"\"\"\n    Adjusted hybrid model:\n    - Relaxed collision_dist and closure_rate_threshold\n    - Improved centroid tracking with stricter matching\n    \"\"\"\n    dashcam_pos = (width // 2, height)\n    \n    # Placeholder contour detection (replace with bounding box data)\n    gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)\n    _, thresh = cv2.threshold(gray, 127, 255, cv2.THRESH_BINARY)\n    contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)\n    \n    # Get current centroids\n    current_centroids = {}\n    for contour in contours:\n        area = cv2.contourArea(contour)\n        if area > 500:\n            x, y, w, h = cv2.boundingRect(contour)\n            centroid_x = x + w // 2\n            centroid_y = y + h // 2\n            centroid_id = f\"{x}_{y}\"\n            current_centroids[centroid_id] = (centroid_x, centroid_y)\n            cv2.rectangle(frame, (x, y), (x + w, y + h), (0, 255, 0), 2)\n    \n    caution = False\n    collision = False\n    for curr_id, curr_centroid in current_centroids.items():\n        dist_to_dashcam = distance.euclidean(dashcam_pos, curr_centroid)\n        \n        # Temporal check with improved matching\n        closure_rate = None\n        if prev_centroids and frame_num > 0:\n            # Match centroids within a reasonable radius (e.g., 50 pixels) to avoid mismatches\n            prev_dists = {prev_id: distance.euclidean(prev_centroid, curr_centroid) \n                         for prev_id, prev_centroid in prev_centroids.items()}\n            if prev_dists:\n                closest_prev_id = min(prev_dists, key=prev_dists.get)\n                if prev_dists[closest_prev_id] < 50:  # Only match if close enough\n                    prev_dist_to_dashcam = distance.euclidean(dashcam_pos, prev_centroids[closest_prev_id])\n                    closure_rate = prev_dist_to_dashcam - dist_to_dashcam\n        \n        if debug:\n            print(f\"Frame {frame_num}, Centroid {curr_centroid}: Dist={dist_to_dashcam:.2f}, \"\n                  f\"Closure Rate={closure_rate if closure_rate is not None else 'N/A'}\")\n        \n        # Collision: Relaxed distance and closure rate\n        if (dist_to_dashcam < collision_dist and \n            (closure_rate is not None and closure_rate > closure_rate_threshold)):\n            collision = True\n            cv2.circle(frame, curr_centroid, 5, (0, 0, 255), -1)\n            cv2.putText(frame, \"COLLISION POSSIBLE!\", (50, 50),\n                       cv2.FONT_HERSHEY_SIMPLEX, 1.2, (0, 0, 255), 3)\n            if debug:\n                print(\"Collision possible triggered!\")\n        \n        # Caution: Moderately close\n        elif dist_to_dashcam < caution_dist:\n            caution = True\n            cv2.circle(frame, curr_centroid, 5, (0, 255, 255), -1)\n            cv2.putText(frame, \"CAUTION: Vehicle Close!\", (50, 100),\n                       cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 255, 255), 2)\n            if debug:\n                print(\"Caution triggered!\")\n    \n    cv2.circle(frame, dashcam_pos, 5, (255, 0, 0), -1)\n    return frame, caution, collision, current_centroids\n\n# Process and write frames\ncaution_count = 0\ncollision_count = 0\nprev_centroids = None\nfor i, frame in enumerate(annotated_frames):\n    debug = i < 10  # Extended debug for first 10 frames\n    processed_frame, caution_detected, collision_detected, current_centroids = check_hybrid_proximity(\n        frame.copy(),\n        prev_centroids,\n        frame_num=i,\n        caution_dist=80,           # Unchanged\n        collision_dist=40,          # Relaxed from 30 to 50\n        closure_rate_threshold=15,  # Relaxed from 20 to 10\n        debug=debug\n    )\n    \n    if collision_detected:\n        collision_count += 1\n        cv2.rectangle(processed_frame, (0, 0), (width, height), (0, 0, 255), 10)\n    elif caution_detected:\n        caution_count += 1\n        cv2.rectangle(processed_frame, (0, 0), (width, height), (0, 255, 255), 10)\n    \n    video_writer.write(processed_frame)\n    prev_centroids = current_centroids\n\n# Release video writer and print summary\nvideo_writer.release()\nprint(f\"Video processing complete.\")\nprint(f\"Caution frames: {caution_count} ({(caution_count / len(annotated_frames)) * 100:.2f}%)\")\nprint(f\"Collision possible frames: {collision_count} ({(collision_count / len(annotated_frames)) * 100:.2f}%)\")\nprint(f\"Output saved to: {output_video_path}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-04-02T19:39:56.744992Z","iopub.execute_input":"2025-04-02T19:39:56.745721Z","iopub.status.idle":"2025-04-02T19:40:08.249990Z","shell.execute_reply.started":"2025-04-02T19:39:56.745673Z","shell.execute_reply":"2025-04-02T19:40:08.248460Z"}},"outputs":[],"execution_count":null}]}