manchester_accident_analysis.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install collections-extended folium scikit-learn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import csv\n",
    "import folium\n",
    "from sklearn.cluster import DBSCAN\n",
    "from collections import Counter"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Map saved to ./data/severity_1_accident_map.html\n",
      "977 accidents with severity 1 in Manchester.\n",
      "Top High-Risk Areas:\n",
      "Cluster 0: 10191 severe accidents\n",
      "  Latitude: 53.501460, Longitude: -2.275125\n",
      "Cluster 1: 101 severe accidents\n",
      "  Latitude: 53.592716, Longitude: -2.549845\n",
      "Cluster 4: 37 severe accidents\n",
      "  Latitude: 53.411820, Longitude: -2.435307\n",
      "Cluster 3: 30 severe accidents\n",
      "  Latitude: 53.585847, Longitude: -2.042128\n",
      "Top Contributing Factors:\n",
      "RoadSurface:\n",
      "- 1: 7391 accidents\n",
      "- 2: 3057 accidents\n",
      "- 4: 139 accidents\n",
      "\n",
      "LightingCondition:\n",
      "- 1: 5959 accidents\n",
      "- 4: 3501 accidents\n",
      "- 2: 508 accidents\n",
      "\n",
      "WeatherCondition:\n",
      "- 1: 8524 accidents\n",
      "- 2: 1333 accidents\n",
      "- 9: 314 accidents\n",
      "\n",
      "JunctionDetail:\n",
      "- 3: 4118 accidents\n",
      "- 0: 3531 accidents\n",
      "- 6: 1896 accidents\n",
      "\n"
     ]
    }
   ],
   "source": [
    "def render_data_on_map(unified_csv_file: str, output_map_file: str) -> None:\n",
    "    # Read the unified dataset and extract the accidents with severity 1\n",
    "    data = []\n",
    "    with open(unified_csv_file, \"r\") as file:\n",
    "        csv_reader = csv.DictReader(file)\n",
    "        data = [row for row in csv_reader if row[\"Severity\"] == \"1\"]\n",
    "\n",
    "    if not data:\n",
    "        print(\"No accidents with severity 1 found in the dataset.\")\n",
    "        return\n",
    "\n",
    "    latitudes = [float(row[\"Latitude\"]) for row in data]\n",
    "    longitudes = [float(row[\"Longitude\"]) for row in data]\n",
    "\n",
    "    mean_lat = sum(latitudes) / len(latitudes)\n",
    "    mean_lon = sum(longitudes) / len(longitudes)\n",
    "\n",
    "    map_center = [mean_lat, mean_lon]\n",
    "    accident_map = folium.Map(location=map_center, zoom_start=6)\n",
    "\n",
    "    # Add markers for each accident location with severity 1\n",
    "    for row in data:\n",
    "        lat = float(row[\"Latitude\"])\n",
    "        lon = float(row[\"Longitude\"])\n",
    "        severity = row[\"Severity\"]\n",
    "\n",
    "        marker_color = \"red\"  # Use red color for severity 1 accidents\n",
    "\n",
    "        folium.Marker(\n",
    "            location=[lat, lon],\n",
    "            popup=f\"Severity: {severity}\",\n",
    "            icon=folium.Icon(color=marker_color)\n",
    "        ).add_to(accident_map)\n",
    "\n",
    "    # Save the map as an HTML file\n",
    "    accident_map.save(output_map_file)\n",
    "    print(f\"Map saved to {output_map_file}\")\n",
    "\n",
    "    # Print the severity count message\n",
    "    severity_count = len(data)\n",
    "    print(f\"{severity_count} accidents with severity 1 in Manchester.\")\n",
    "\n",
    "def analyze_high_risk_areas(unified_csv_file: str) -> None:\n",
    "    # Read the unified dataset and extract the accidents with severity 1 or 2\n",
    "    data = []\n",
    "    with open(unified_csv_file, \"r\") as file:\n",
    "        csv_reader = csv.DictReader(file)\n",
    "        data = [row for row in csv_reader if row[\"Severity\"] in [\"1\", \"2\"]]\n",
    "\n",
    "    # Extract latitude and longitude coordinates\n",
    "    coordinates = [[float(row[\"Latitude\"]), float(row[\"Longitude\"])] for row in data]\n",
    "\n",
    "    # Perform clustering analysis using DBSCAN\n",
    "    dbscan = DBSCAN(eps=0.01, min_samples=5)\n",
    "    clusters = dbscan.fit_predict(coordinates)\n",
    "\n",
    "    # Count the number of accidents in each cluster\n",
    "    cluster_counts = Counter(clusters)\n",
    "\n",
    "    # Calculate the center coordinates of each cluster\n",
    "    cluster_centers = {}\n",
    "    for idx, cluster_id in enumerate(clusters):\n",
    "        if cluster_id != -1:  # Exclude noise points\n",
    "            if cluster_id not in cluster_centers:\n",
    "                cluster_centers[cluster_id] = [coordinates[idx]]\n",
    "            else:\n",
    "                cluster_centers[cluster_id].append(coordinates[idx])\n",
    "\n",
    "    for cluster_id, center_coords in cluster_centers.items():\n",
    "        latitudes = [coord[0] for coord in center_coords]\n",
    "        longitudes = [coord[1] for coord in center_coords]\n",
    "        cluster_centers[cluster_id] = [sum(latitudes) / len(latitudes), sum(longitudes) / len(longitudes)]\n",
    "\n",
    "    # Print out the top high-risk areas with their coordinates and the number of severe accidents in each area\n",
    "    print(\"Top High-Risk Areas:\")\n",
    "    for cluster_id, count in cluster_counts.most_common(5):\n",
    "        if cluster_id != -1:  # Exclude noise points\n",
    "            center_lat, center_lon = cluster_centers[cluster_id]\n",
    "            print(f\"Cluster {cluster_id}: {count} severe accidents\")\n",
    "            print(f\"  Latitude: {center_lat:.6f}, Longitude: {center_lon:.6f}\")\n",
    "\n",
    "def analyze_contributing_factors(unified_csv_file: str) -> None:\n",
    "    # Read the unified dataset and extract the accidents with severity 1 or 2\n",
    "    data = []\n",
    "    with open(unified_csv_file, \"r\") as file:\n",
    "        csv_reader = csv.DictReader(file)\n",
    "        data = [row for row in csv_reader if row[\"Severity\"] in [\"1\", \"2\"]]\n",
    "\n",
    "    # Analyze the contributing factors and calculate their frequency\n",
    "    contributing_factors = [\n",
    "        \"RoadSurface\", \"LightingCondition\", \"WeatherCondition\", \"JunctionDetail\"\n",
    "    ]\n",
    "    factor_counts = {factor: Counter() for factor in contributing_factors}\n",
    "\n",
    "    for row in data:\n",
    "        for factor in contributing_factors:\n",
    "            factor_counts[factor][row[factor]] += 1\n",
    "\n",
    "    # Print out the top contributing factors associated with severe accidents\n",
    "    print(\"Top Contributing Factors:\")\n",
    "    for factor, counts in factor_counts.items():\n",
    "        print(f\"{factor}:\")\n",
    "        for value, count in counts.most_common(3):\n",
    "            print(f\"- {value}: {count} accidents\")\n",
    "        print()\n",
    "\n",
    "def main() -> None:\n",
    "    unified_dataset_file = \"./data/unified_dataset.csv\"\n",
    "    output_map_file = \"./data/severity_1_accident_map.html\"\n",
    "\n",
    "    render_data_on_map(unified_dataset_file, output_map_file)\n",
    "    analyze_high_risk_areas(unified_dataset_file)\n",
    "    analyze_contributing_factors(unified_dataset_file)\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    main()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}