manchester_accident_create_unified_dataset.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install convertbng numpy typing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import csv\n",
    "from typing import List, Dict\n",
    "from convertbng.util import convert_lonlat\n",
    "import numpy as np"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Unified dataset saved to ./data/unified_dataset.csv\n"
     ]
    }
   ],
   "source": [
    "class DatasetUnifier:\n",
    "    def __init__(self, acc_file: str, cas_file: str, veh_file: str) -> None:\n",
    "        self.acc_file = acc_file\n",
    "        self.cas_file = cas_file\n",
    "        self.veh_file = veh_file\n",
    "\n",
    "    def load_csv_data(self, file_path: str) -> List[Dict[str, str]]:\n",
    "        with open(file_path, \"r\") as file:\n",
    "            csv_reader = csv.DictReader(file)\n",
    "            return list(csv_reader)\n",
    "\n",
    "    # Convert Eastings and Northings to Longitude and Latitude\n",
    "    def convert_coordinates(self, eastings: List[float], northings: List[float]) -> List[Dict[str, float]]:\n",
    "        lons, lats = convert_lonlat(np.array(eastings), np.array(northings))\n",
    "        return [{\"Longitude\": lon, \"Latitude\": lat} for lon, lat in zip(lons, lats)]\n",
    "\n",
    "    def join_datasets(self) -> List[Dict[str, str]]:\n",
    "        acc_data = self.load_csv_data(self.acc_file)\n",
    "        cas_data = self.load_csv_data(self.cas_file)\n",
    "        veh_data = self.load_csv_data(self.veh_file)\n",
    "\n",
    "        eastings = [float(row[\"Easting\"]) for row in acc_data]\n",
    "        northings = [float(row[\"Northing\"]) for row in acc_data]\n",
    "        coordinates = self.convert_coordinates(eastings, northings)\n",
    "\n",
    "        unified_data = []\n",
    "        for acc_row, coord in zip(acc_data, coordinates):\n",
    "            acc_index = acc_row[\"Accident Index\"]\n",
    "            acc_year = acc_row[\"Year\"]\n",
    "            acc_row[\"Longitude\"] = coord[\"Longitude\"]\n",
    "            acc_row[\"Latitude\"] = coord[\"Latitude\"]\n",
    "            del acc_row[\"Easting\"]  # Remove \"Easting\" from the unified dataset\n",
    "            del acc_row[\"Northing\"]  # Remove \"Northing\" from the unified dataset\n",
    "\n",
    "            for cas_row in cas_data:\n",
    "                if cas_row[\"Accident Index\"] == acc_index and cas_row[\"Year\"] == acc_year:\n",
    "                    veh_ref_num = cas_row[\"VehicleReferenceNumber\"]\n",
    "                    for veh_row in veh_data:\n",
    "                        if (\n",
    "                            veh_row[\"Accident Index\"] == acc_index\n",
    "                            and veh_row[\"Year\"] == acc_year\n",
    "                            and veh_row[\"VehicleReferenceNumber\"] == veh_ref_num\n",
    "                        ):\n",
    "                            unified_row = {**acc_row, **cas_row, **veh_row}\n",
    "                            unified_data.append(unified_row)\n",
    "\n",
    "        return unified_data\n",
    "\n",
    "    def write_unified_dataset(self, output_file: str) -> None:\n",
    "        unified_data = self.join_datasets()\n",
    "        if unified_data:\n",
    "            fieldnames = list(unified_data[0].keys())\n",
    "            with open(output_file, \"w\", newline=\"\") as file:\n",
    "                csv_writer = csv.DictWriter(file, fieldnames=fieldnames)\n",
    "                csv_writer.writeheader()\n",
    "                csv_writer.writerows(unified_data)\n",
    "            print(f\"Unified dataset saved to {output_file}\")\n",
    "        else:\n",
    "            print(\"No data to write\")\n",
    "\n",
    "def main() -> None:\n",
    "    acc_file = \"./data/STATS19AccDataJan2010Dec2021forGMServers.csv\"\n",
    "    cas_file = \"./data/STATS19CasDataJan2010Dec2021forGMServers.csv\"\n",
    "    veh_file = \"./data/STATS19VehDataJan2010Dec2021forGMServers.csv\"\n",
    "    output_file = \"./data/unified_dataset.csv\"\n",
    "\n",
    "    dataset_unifier = DatasetUnifier(acc_file, cas_file, veh_file)\n",
    "    dataset_unifier.write_unified_dataset(output_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
}