{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.12.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceType":"datasetVersion","sourceId":15438184,"datasetId":9876164,"databundleVersionId":16357926}],"dockerImageVersionId":31328,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# **YOLO Car Accident Detection**","metadata":{}},{"cell_type":"markdown","source":"This code document details a computer vision system designed to identify and analyze traffic accidents from video recordings. By leveraging the YOLOv8 object detection model, the program tracks various vehicles and records their speed, trajectory, and behavioral changes over time. The script calculates specific metrics, such as sudden deceleration and sharp changes in direction, to distinguish between the vehicle that initiated a collision and the one that was struck. Users can customize the analysis timeframe and visual output to better examine the moments immediately surrounding an impact. Ultimately, the source serves as a specialized tool for automated accident reconstruction and vehicle behavior monitoring.","metadata":{}},{"cell_type":"code","source":"!pip install ultralytics","metadata":{"trusted":true,"_kg_hide-output":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n# ★ CHANGE ONLY THESE PARAMETERS ★\nVIDEO_PATH = \"/kaggle/input/datasets/unidpro/car-accident-video/clip_04.mp4\"\nBEFORE_SEC = 1.0   # Seconds before the estimated collision time\nAFTER_SEC  = 1.0   # Seconds after the estimated collision time\nCOLS       = 3     # Number of images to display horizontally\n# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\nimport cv2, numpy as np, matplotlib.pyplot as plt\nimport matplotlib.gridspec as gridspec, time, math\nfrom dataclasses import dataclass, field\nfrom collections import deque\nfrom IPython.display import Image, display\nfrom ultralytics import YOLO\n\nHISTORY_LEN = 60; MIN_TRACK_FRAMES = 4\nVEHICLE_CLASSES = {2: \"car\", 3: \"motorcycle\", 5: \"bus\", 7: \"truck\"}\n\n@dataclass\nclass VehicleTrack:\n    track_id: int; cls_name: str\n    centers:    deque = field(default_factory=lambda: deque(maxlen=HISTORY_LEN))\n    speeds_kmh: deque = field(default_factory=lambda: deque(maxlen=HISTORY_LEN))\n    timestamps: deque = field(default_factory=lambda: deque(maxlen=HISTORY_LEN))\n    alert: bool = False; alert_time: float = 0.0\n    def add(self, cx, cy, spd, t):\n        self.centers.append((cx, cy)); self.speeds_kmh.append(spd); self.timestamps.append(t)\n\nclass TrafficSpeedTracker:\n    def __init__(self, video_path, scale=0.05, thresh=25.0):\n        self.video_path = video_path; self.scale = scale; self.thresh = thresh\n        self.cap = cv2.VideoCapture(video_path)\n        self.fps = self.cap.get(cv2.CAP_PROP_FPS) or 30.0\n        self.total = int(self.cap.get(cv2.CAP_PROP_FRAME_COUNT))\n        print(f\"[INFO] {video_path}  FPS={self.fps:.2f}  frames={self.total}\")\n        self.model = YOLO(\"yolov8n.pt\")\n        self.tracks = {}; self.finished_tracks = {}; self.alerts = []\n\n    def run(self):\n        print(\"[INFO] Processing started...\")\n        t0 = time.time(); fi = 0\n        while True:\n            ret, frame = self.cap.read()\n            if not ret: break\n            t = fi / self.fps\n            res = self.model.track(frame, persist=True, tracker=\"bytetrack.yaml\",\n                                 classes=list(VEHICLE_CLASSES.keys()),\n                                 verbose=False, conf=0.25, iou=0.45)\n            active = set()\n            if res[0].boxes.id is not None:\n                for box, cls_id, tid in zip(\n                    res[0].boxes.xyxy.cpu().numpy(),\n                    res[0].boxes.cls.cpu().numpy().astype(int),\n                    res[0].boxes.id.cpu().numpy().astype(int)):\n                    cls_name = VEHICLE_CLASSES.get(cls_id, \"vehicle\")\n                    x1, y1, x2, y2 = box; cx, cy = (x1 + x2) / 2, (y1 + y2) / 2\n                    active.add(tid)\n                    if tid not in self.tracks: self.tracks[tid] = VehicleTrack(tid, cls_name)\n                    trk = self.tracks[tid]\n                    spd = 0.0\n                    if trk.centers:\n                        px, py = trk.centers[-1]\n                        spd = np.hypot(cx - px, cy - py) * self.scale * self.fps * 3.6\n                    trk.add(cx, cy, spd, t)\n                    if len(trk.speeds_kmh) >= 3:\n                        recent = list(trk.speeds_kmh)[-3:]\n                        decel = (recent[-1] - recent[0]) / (2 / self.fps)\n                        if decel < -self.thresh and not trk.alert:\n                            trk.alert = True; trk.alert_time = t\n                            self.alerts.append(dict(track_id=tid, cls=cls_name,\n                                time_s=t, speed_kmh=float(np.mean(recent)),\n                                decel_kmhps=decel, position=(int(cx), int(cy))))\n            for tid in set(self.tracks) - active:\n                trk = self.tracks.pop(tid)\n                if len(trk.speeds_kmh) >= MIN_TRACK_FRAMES: self.finished_tracks[tid] = trk\n            fi += 1\n            if fi % 100 == 0: print(f\"  {fi}/{self.total}  {fi / (time.time() - t0):.1f}fps\")\n        for tid, trk in list(self.tracks.items()):\n            if len(trk.speeds_kmh) >= MIN_TRACK_FRAMES: self.finished_tracks[tid] = trk\n        self.cap.release()\n        print(f\"[INFO] Completed: {fi} frames | Tracks={len(self.finished_tracks)} | Alerts={len(self.alerts)}\")\n\ndef calc_direction_change(trk, window=3):\n    centers = list(trk.centers); times = list(trk.timestamps)\n    if len(centers) < window * 2 + 1: return 0.0, None\n    max_angle = 0.0; max_t = None\n    for i in range(window, len(centers) - window):\n        vbx = centers[i][0] - centers[i - window][0]; vby = centers[i][1] - centers[i - window][1]\n        vax = centers[i + window][0] - centers[i][0]; vay = centers[i + window][1] - centers[i][1]\n        nb, na = np.hypot(vbx, vby), np.hypot(vax, vay)\n        if nb < 1 or na < 1: continue\n        cos = np.clip((vbx * vax + vby * vay) / (nb * na), -1, 1)\n        ang = float(np.degrees(np.arccos(cos)))\n        if ang > max_angle: max_angle = ang; max_t = times[i]\n    return max_angle, max_t\n\ndef analyze_vehicles(tracker):\n    all_tracks = {**tracker.finished_tracks, **tracker.tracks}; results = []\n    for tid, trk in all_tracks.items():\n        if len(trk.speeds_kmh) < 4: continue\n        speeds = np.array(list(trk.speeds_kmh)); times = np.array(list(trk.timestamps))\n        max_spd = float(np.max(speeds)); duration = float(times[-1] - times[0])\n        last_spd = float(np.mean(speeds[-3:])) if len(speeds) >= 3 else speeds[-1]\n        decel_ratio = (max_spd - last_spd) / max(max_spd, 1.0)\n        max_angle, angle_t = calc_direction_change(trk)\n        stopped = speeds < 5.0; max_consec = cur = 0; stop_t = None\n        for i, s in enumerate(stopped):\n            if s:\n                cur += 1\n                if cur == 1: _st = times[i]\n                if cur > max_consec: max_consec = cur; stop_t = _st\n            else: cur = 0\n        h_score = max_spd * (1.0 - decel_ratio) * (1.0 / max(duration, 0.2)) if max_spd > 20 and duration < 5 else 0\n        v_score = max_angle * max_spd if max_angle > 45 and max_spd > 3 else (max_consec * max_spd * 0.5 if max_consec >= 8 and max_spd > 5 else 0)\n        role = \"normal\"; score = 0\n        if h_score > v_score and h_score > 50: role = \"hitting\"; score = h_score\n        elif v_score > 100:                    role = \"hit\";     score = v_score\n        results.append(dict(track_id=tid, cls=trk.cls_name, role=role, score=score,\n            h_score=h_score, v_score=v_score, max_spd=max_spd, decel_ratio=decel_ratio,\n            duration=duration, n_frames=len(speeds), t_start=float(times[0]), t_end=float(times[-1]),\n            max_angle=max_angle, angle_t=angle_t, stop_t=stop_t, consec_stop=max_consec, trk=trk))\n    \n    hitting = sorted([r for r in results if r[\"role\"] == \"hitting\"], key=lambda x: -x[\"score\"])\n    hit     = sorted([r for r in results if r[\"role\"] == \"hit\"],     key=lambda x: -x[\"score\"])\n    \n    print(\"=== HITTER CANDIDATES (Aggressor) ===\")\n    for r in hitting[:5]:\n        print(f\"  ID={r['track_id']:>3} {r['cls']:>6}  max={r['max_spd']:5.1f}km/h  \"\n              f\"decelR={r['decel_ratio']:.2f}  dur={r['duration']:.1f}s  \"\n              f\"t={r['t_start']:.1f}~{r['t_end']:.1f}  score={r['score']:.0f}\")\n    \n    print(\"=== VICTIM CANDIDATES (Hit) ===\")\n    for r in hit[:5]:\n        print(f\"  ID={r['track_id']:>3} {r['cls']:>6}  max={r['max_spd']:5.1f}km/h  \"\n              f\"angle={r['max_angle']:.1f}deg  consec={r['consec_stop']}f  \"\n              f\"t={r['t_start']:.1f}~{r['t_end']:.1f}  score={r['score']:.0f}\")\n    return hitting, hit, results","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ── Display All Frames (Neutral Version) ──────────────────────────────────\ndef show_all_frames(tracker, vehicle_a, vehicle_b, results,\n                    before_sec=BEFORE_SEC, after_sec=AFTER_SEC,\n                    cols=COLS, save_path=\"collision_analysis_neutral.png\"):\n    all_tracks = {**tracker.finished_tracks, **tracker.tracks}; fps = tracker.fps\n\n    # Use the identified event time regardless of role\n    event_t = vehicle_b[\"angle_t\"] if (vehicle_b and vehicle_b[\"angle_t\"]) else vehicle_a[\"t_end\"]\n    f_start = max(0, int((event_t - before_sec) * fps))\n    f_end   = min(tracker.total - 1, int((event_t + after_sec) * fps))\n    all_frame_indices = list(range(f_start, f_end + 1))\n    n = len(all_frame_indices)\n    rows = math.ceil(n / cols)\n\n    thumb_w, thumb_h = 320, 180\n    fig = plt.figure(figsize=(cols * 3.2, rows * 2.0 + 3.0), facecolor=\"#0d1117\")\n    gs = gridspec.GridSpec(rows + 1, cols, figure=fig, hspace=0.35, wspace=0.03, height_ratios=[1] * rows + [0.5])\n\n    cap = cv2.VideoCapture(tracker.video_path)\n    cap.set(cv2.CAP_PROP_POS_FRAMES, f_start)\n\n    for idx, fi in enumerate(all_frame_indices):\n        ret, frame = cap.read()\n        if not ret: break\n\n        t = fi / fps\n        img = cv2.resize(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB), (thumb_w, thumb_h))\n        h, w = img.shape[:2]\n        scale_x = w / frame.shape[1]; scale_y = h / frame.shape[0]\n\n        for tid, trk in all_tracks.items():\n            if not trk.timestamps: continue\n            tarr = np.array(list(trk.timestamps)); ni = int(np.argmin(np.abs(tarr - t)))\n            if abs(tarr[ni] - t) > 0.5: continue\n            \n            cx = list(trk.centers)[ni][0] * scale_x\n            cy = list(trk.centers)[ni][1] * scale_y\n            spd = list(trk.speeds_kmh)[ni]\n            \n            is_a = (tid == vehicle_a[\"track_id\"])\n            is_b = (vehicle_b and tid == vehicle_b[\"track_id\"])\n            \n            # Neutral coloring: Orange for A, Cyan for B\n            if is_a:   col, thick, bs, label, fs = (255, 165, 0), 2, 28, f\"Veh A:{tid}\", 0.45\n            elif is_b: col, thick, bs, label, fs = (0, 255, 255), 2, 28, f\"Veh B:{tid}\", 0.45\n            else:      col, thick, bs, label, fs = (140, 140, 140), 1, 18, f\"{tid}\", 0.32\n            \n            x1 = max(0, int(cx - bs)); y1 = max(0, int(cy - bs * 1.3))\n            x2 = min(w, int(cx + bs)); y2 = min(h, int(cy + bs * 0.7))\n            cv2.rectangle(img, (x1, y1), (x2, y2), col, thick)\n            (lw, lh), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, fs, 1)\n            cv2.rectangle(img, (x1, y1 - lh - 4), (x1 + lw + 2, y1), col, -1)\n            cv2.putText(img, label, (x1 + 1, y1 - 2), cv2.FONT_HERSHEY_SIMPLEX, fs, (0, 0, 0), 1)\n\n        ax = fig.add_subplot(gs[idx // cols, idx % cols])\n        ax.imshow(img); ax.axis(\"off\")\n        rel = t - event_t; sign = f\"+{rel:.2f}\" if rel >= 0 else f\"{rel:.2f}\"\n        ax.set_title(f\"{sign}s\", color=\"#8b949e\", fontsize=6.5, pad=1.5)\n\n    cap.release()\n\n    # Speed Graph (Bottom) - Neutral labels\n    ax_s = fig.add_subplot(gs[rows, :]); ax_s.set_facecolor(\"#161b22\")\n    for info, color, lbl in [(vehicle_a, \"#ffa500\", \"Vehicle A\"), (vehicle_b, \"#00ffff\", \"Vehicle B\") if vehicle_b else (None, None, None)]:\n        if info is None: continue\n        trk = info[\"trk\"]; times = np.array(list(trk.timestamps)); spds = np.array(list(trk.speeds_kmh))\n        sm = np.convolve(spds, np.ones(2) / 2, mode=\"same\")\n        ax_s.plot(times, spds, color=color, alpha=0.2, lw=0.8)\n        ax_s.plot(times, sm, color=color, lw=2, label=f\"{lbl} (ID={info['track_id']})\")\n        ax_s.fill_between(times, sm, alpha=0.1, color=color)\n    \n    ax_s.axvline(event_t, color=\"#ffffff\", lw=1.5, linestyle=\"--\", label=\"Estimated Impact\")\n    ax_s.set_ylabel(\"Speed (km/h)\", color=\"#8b949e\", fontsize=8)\n    ax_s.legend(fontsize=8, facecolor=\"#21262d\", labelcolor=\"white\")\n    ax_s.set_title(\"Speed Profiles of Involved Vehicles\", color=\"white\", fontsize=9)\n\n    fig.suptitle(f\"Collision Event Analysis: Vehicle {vehicle_a['track_id']} & {vehicle_b['track_id'] if vehicle_b else 'Unknown'}\", \n                 fontsize=11, color=\"white\", fontweight=\"bold\")\n\n    plt.savefig(save_path, dpi=110, bbox_inches=\"tight\", facecolor=fig.get_facecolor())\n    plt.show()\n    plt.close(fig)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n# Execution (Neutral Analysis)\n# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\ntracker = TrafficSpeedTracker(VIDEO_PATH, scale=0.05, thresh=25.0)\ntracker.run()\n\n# The detection identifies vehicles with significant kinematic changes\ncandidates_a, candidates_b, results = analyze_vehicles(tracker)\n\n# Assign to neutral labels (Vehicle A and Vehicle B)\n# Note: We avoid \"Hitter/Victim\" as fault depends on traffic signals/rules\nveh_a = candidates_a[0] if candidates_a else None\nveh_b = candidates_b[0] if candidates_b else None\n\n# ★ Manually override IDs if the automated detection picks the wrong pair ★\n# veh_a = next(r for r in results if r[\"track_id\"] == 31)\n# veh_b = next(r for r in results if r[\"track_id\"] == 24)\n\nif veh_a:\n    # Generate the summarized grid for the two primary involved vehicles\n    show_all_frames(tracker, veh_a, veh_b, results,\n                    before_sec=BEFORE_SEC, after_sec=AFTER_SEC,\n                    cols=COLS, save_path=\"collision_analysis_neutral.png\")\nelse:\n    print(\"[WARN] No primary involved vehicles detected for analysis.\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"Based on the log provided, here is a professional summary of the accident analysis.\n\n---\n\n## **Analysis Summary: Vehicle Collision Detection**\n\nThe AI pipeline successfully processed 179 frames of the dashcam footage to detect and analyze a vehicle-to-vehicle collision. The system utilized **YOLOv8n** for object detection and the **LAP (Linear Assignment Problem)** solver for robust multi-object tracking.\n\n### **1. Identified Involved Vehicles**\nThe system flagged two primary vehicles involved in the incident based on kinematic anomalies (sudden changes in velocity and orientation):\n\n* **Vehicle A (ID 49 - Car):** * **Max Speed:** 59.7 km/h.\n    * **Behavior:** Exhibited a sharp deceleration ratio of **0.53** within a 0.1s duration (t=2.7s~2.8s). This suggests a significant impact or emergency braking maneuver.\n* **Vehicle B (ID 60 - Car):** * **Max Speed:** 20.6 km/h.\n    * **Behavior:** Experienced a violent angular displacement of **49.6 degrees** over 4 consecutive frames (t=3.6s~4.1s). This high rotation score (1023) indicates the vehicle was likely spun or overturned by the impact.\n\n### **2. Collision Timeline**\n* **Estimated Impact:** **t=3.94s**.\n* **Analysis Window:** The system captured a 2.0-second window surrounding the event (from frame 88 to 147) for detailed review.\n\n### **3. Neutral Technical Conclusion**\nWhile the internal scoring initially categorized the vehicles as \"Hitter\" and \"Victim\" based on momentum changes, the final report treats them as **Involved Vehicles A and B**. A definitive determination of fault cannot be made through kinematics alone, as external factors such as **traffic signal status** and right-of-way regulations are critical to the legal assessment of the accident.\n\n\n\n---\n\n### **Key Metrics Table**\n\n| Vehicle ID | Type | Peak Speed | Key Event | Confidence Score |\n| :--- | :--- | :--- | :--- | :--- |\n| **ID 49** | Car | 59.7 km/h | Rapid Deceleration (0.53) | 140 |\n| **ID 60** | Car | 20.6 km/h | Sudden Rotation (49.6°) | 1023 |","metadata":{}},{"cell_type":"markdown","source":"","metadata":{}}]}