postcode_to_british_national_grid.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# UK Postcode to British National Grid Mapper\n",
    "\n",
    "This tool links each postcode to British National Grid Reference numbers. Before running this notebook, please download following data to `data` folder into their own directories:\n",
    "\n",
    "- British National Grid from [here](https://github.com/OrdnanceSurvey/OS-British-National-Grids)\n",
    "- Latest National Statistics Postcode Lookup [here](https://geoportal.statistics.gov.uk/search?sort=-modified&tags=prd_nspl) (for [February 2022](https://geoportal.statistics.gov.uk/datasets/national-statistics-postcode-lookup-february-2022/about))\n",
    "\n",
    "For convenience, British National Grid repository added as a submodule. To get it, you can either clone this repository with `--recurse-submodules` flag or execute the following commands to pull it after cloning.\n",
    "\n",
    "```bash\n",
    "git submodule init\n",
    "git submodule update\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import geopandas as gpd\n",
    "from pathlib import Path\n",
    "\n",
    "DATA = Path(\"data\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ONSPD_VERSION = \"FEB_2022\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ONSPD_PATH = DATA/f'ONSPD_{ONSPD_VERSION}_UK'/'Data'/f'ONSPD_{ONSPD_VERSION}_UK.csv'\n",
    "BNG_PATH = DATA / \"OS-British-National-Grids\" / \"os_bng_grids.gpkg\"\n",
    "\n",
    "assert ONSPD_PATH.exists(), (f\"ONSPD file not found: {ONSPD_PATH}. \\n\"\n",
    "\"Please download it from: \"\n",
    "\"https://geoportal.statistics.gov.uk/search?sort=-modified&tags=prd_nspl.\"\n",
    "\" and decompress under the 'data' directory.\\n\"\n",
    "\"If you already downloaded, please check ONSPD_VERSION and the file path.\")\n",
    "\n",
    "assert BNG_PATH.with_suffix('.7z').exists(), (\n",
    "    f\"British National Grid file not found: {BNG_PATH}. \\n\"\n",
    "    \"Please run the following command to download it:\\n\"\n",
    "    \"git submodule init\\n\"\n",
    "    \"git submodule update\"\n",
    ")\n",
    "\n",
    "print(\n",
    "    \"Decompressing British National Grid file. \"\n",
    "    \"You can comment this line if it's already decompressed.\"\n",
    ")\n",
    "\n",
    "!tar -xJf {BNG_PATH.with_suffix('.7z').absolute()} -C {BNG_PATH.parent.absolute()}\n",
    "\n",
    "print(\"Decompression completed.\")\n",
    "\n",
    "assert BNG_PATH.exists(), (\n",
    "    f\"British National Grid cannot be decompressed. Please decompress the \"\n",
    "    f\"{BNG_PATH.with_suffix('.7z').name} file manually\"\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## British National Grid Hierarchy\n",
    "\n",
    "Read all the layers from British National Grid file and create a hierarchy whilst keeping 1km grid as the base geometry."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# reference: https://stackoverflow.com/a/57971376/1360267\n",
    "bng_refs = {\n",
    "    layername: gpd.read_file(BNG_PATH, layer=layername)[[\"tile_name\", \"geometry\"]]\n",
    "    .rename(columns={\"tile_name\": layername})\n",
    "    .set_index(layername)\n",
    "    for layername in gpd.io.file.fiona.listlayers(BNG_PATH)\n",
    "}\n",
    "\n",
    "# create BNG hierarchy\n",
    "bng_key_order = sorted(bng_refs.keys(), key=lambda x: int(x.replace(\"km_grid\", \"\")))\n",
    "\n",
    "# merge grids with higher levels starting from lowest resolution\n",
    "bng_hierarchy = bng_refs[bng_key_order[0]].copy()\n",
    "for key in bng_key_order[1:]:\n",
    "    bng_hierarchy = bng_hierarchy.sjoin(\n",
    "        bng_refs[key], how=\"left\", predicate=\"within\"\n",
    "    ).rename(\n",
    "        columns={\"index_right\": key}\n",
    "    )  # protect level name\n",
    "# construct the index hierarchy\n",
    "bng_hierarchy = bng_hierarchy.reset_index().set_index(bng_key_order[::-1])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# read postcode and construct geometry from easting and northings\n",
    "# also keep lat,lon in WGS84 projection in case needed later\n",
    "postcode_gpd = (\n",
    "    pd.read_csv(\n",
    "        ONSPD_PATH,\n",
    "        usecols=[\"pcd\", \"oseast1m\", \"osnrth1m\", \"lat\", \"long\"],\n",
    "    )\n",
    "    .rename(\n",
    "        columns={\n",
    "            \"pcd\": \"Postcode\",\n",
    "            \"long\": \"lon\",\n",
    "        }\n",
    "    )\n",
    "    .pipe(\n",
    "        lambda df: gpd.GeoDataFrame(\n",
    "            df,\n",
    "            geometry=gpd.points_from_xy(\n",
    "                x=df[\"oseast1m\"],\n",
    "                y=df[\"osnrth1m\"],\n",
    "                crs=\"EPSG:27700\",\n",
    "            ),\n",
    "        ).set_index(\"Postcode\")\n",
    "    )\n",
    "    .drop(columns=[\"oseast1m\", \"osnrth1m\"])\n",
    ")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# join postcode coordinates with BNG hierarchy\n",
    "postcode_to_national_grid_gdf = postcode_gpd.sjoin(\n",
    "    bng_hierarchy, how=\"left\", predicate=\"within\"\n",
    ")\n",
    "# recover grid level names\n",
    "postcode_to_national_grid_gdf.columns = (\n",
    "    postcode_gpd.columns.to_list() + bng_hierarchy.index.names\n",
    ")\n",
    "\n",
    "# sort columns in the hierarchical order\n",
    "postcode_to_national_grid_gdf = postcode_to_national_grid_gdf.loc[\n",
    "    :, bng_hierarchy.index.names + postcode_gpd.columns.to_list()\n",
    "]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# convert it to pandas dataframe and set national grid reference types to categorical\n",
    "postcode_to_national_grid_df = (\n",
    "    pd.DataFrame(postcode_to_national_grid_gdf.drop(columns=\"geometry\")).pipe(\n",
    "        lambda df: df.assign(\n",
    "            **{\n",
    "                # convert grid reference values to categorical\n",
    "                grid_ref: df[grid_ref].astype(\"category\")\n",
    "                for grid_ref in bng_hierarchy.index.names\n",
    "            }\n",
    "        )\n",
    "    )\n",
    "    # drop rows with missing values - i.e. unused postcodes\n",
    "    .dropna(how=\"any\")\n",
    ")\n",
    "\n",
    "postcode_to_national_grid_df.to_parquet(\n",
    "    DATA / \"postcode_to_national_grid_references.parquet\", compression=\"gzip\"\n",
    ")\n",
    "\n",
    "postcode_to_national_grid_df.to_csv(\n",
    "    DATA / \"postcode_to_national_grid_references.csv.gz\", compression=\"gzip\"\n",
    ")\n"
   ]
  }
 ],
 "metadata": {
  "interpreter": {
   "hash": "84faba2c3d73f72195a31da7dbd07f0b601f44c80958afd4e19c4d66e3eef367"
  },
  "kernelspec": {
   "display_name": "Python 3.9.7 ('py39')",
   "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.10.2"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}