python/ons_lahb_deaths.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ONS Local Authority + Health Board Deaths\n",
    "\n",
    "Created by Michael George (AKA Logiqx)\n",
    "\n",
    "Website: https://logiqx.github.io/covid-stats/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Imports\n",
    "\n",
    "Standard python libraries plus determination of projdir, basic printable class, etc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import time\n",
    "\n",
    "import csv\n",
    "from xlrd import open_workbook\n",
    "\n",
    "import ons_core\n",
    "import ons_download\n",
    "import common_core"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Configuration\n",
    "\n",
    "Data to download from the NHS statistical work area"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "ONS_LOOKUPS = \"ons-lookups\"\n",
    "ONS_LAHB_DEATHS = \"ons-lahb-deaths\"\n",
    "\n",
    "NATION_CODE_ENGLAND_WALES = 'K04000001'\n",
    "NATION_CODE_ENGLAND = 'E92000001'\n",
    "NATION_CODE_WALES = 'W92000004'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Convert Spreadsheets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    if workbookCache:\n",
    "        pass\n",
    "except:\n",
    "    workbookCache = {}\n",
    "\n",
    "\n",
    "def loadLocalAuthorityLookup(verbose=common_core.verbose):\n",
    "    '''Load local authority lookup into dictionary'''\n",
    "    \n",
    "    localAuthorityLookup = {}\n",
    "\n",
    "    csvPath = os.path.join(common_core.dataDir, ONS_LOOKUPS, \"csv\")\n",
    "    csvFn = os.path.join(csvPath, \"local_authority_region_2020_12.csv\")\n",
    "\n",
    "    if verbose:\n",
    "        print(f\"Loading {csvPath}...\")\n",
    "\n",
    "    with open(csvFn, 'r') as f:\n",
    "        reader = csv.reader(f, delimiter = ',')\n",
    "        \n",
    "        # Skip header\n",
    "        next(reader)\n",
    "        \n",
    "        # Process records\n",
    "        for row in reader:\n",
    "            localAuthority = row[0].strip()\n",
    "            region = row[1].strip()\n",
    "            \n",
    "            if region in common_core.regions:\n",
    "                localAuthorityLookup[localAuthority] = region\n",
    "            else:\n",
    "                raise RuntimeError(f\"Unrecognised region {region}\")\n",
    "            \n",
    "    return localAuthorityLookup\n",
    "\n",
    "\n",
    "def loadWorkbook(partName, verbose=common_core.verbose):\n",
    "    '''Load workbook into memory'''\n",
    "    \n",
    "    if partName in workbookCache:\n",
    "        if verbose:\n",
    "            print(f\"Using cached {partName}...\")\n",
    "        workbook = workbookCache[partName]\n",
    "    else:\n",
    "        fileName = os.path.join(common_core.dataDir, partName)\n",
    "        if verbose:\n",
    "            print(f\"Loading {partName}...\")\n",
    "            start = time.time()\n",
    "        workbook = open_workbook(fileName)\n",
    "        if verbose:\n",
    "            end = time.time()\n",
    "            print(f\"Took {(end - start):.2f} seconds...\")\n",
    "\n",
    "        workbookCache[partName] = workbook\n",
    "        \n",
    "    return workbook\n",
    "\n",
    "\n",
    "def updateRegionData(regionData, category, partName, localAuthorityLookup, verbose=common_core.verbose):\n",
    "    '''Process workbook for local authority data'''\n",
    "    \n",
    "    workbook = workbookCache[partName]\n",
    "\n",
    "    if '2020' in partName:\n",
    "        year = 2020\n",
    "    else:\n",
    "        year = 2021\n",
    "\n",
    "    sheetName = f'{category.capitalize()} - All data'\n",
    "    found = False\n",
    "\n",
    "    for sheet in workbook.sheets():\n",
    "        if sheet.name == sheetName:\n",
    "            for rowNo in range(5):\n",
    "                for colNo in range(sheet.ncols):\n",
    "                    if sheet.cell(rowNo, colNo).value == 'Area code':\n",
    "                        colNoAreaCode = colNo\n",
    "                        headerRowNo = rowNo\n",
    "                    elif sheet.cell(rowNo, colNo).value == 'Geography type':\n",
    "                        colNoGeographyType = colNo\n",
    "                    elif sheet.cell(rowNo, colNo).value == 'Cause of death':\n",
    "                        colNoCauseDeath = colNo\n",
    "                    elif sheet.cell(rowNo, colNo).value == 'Week number':\n",
    "                        colWeekNumber = colNo\n",
    "                    elif sheet.cell(rowNo, colNo).value == 'Number of deaths':\n",
    "                        colNumDeaths = colNo\n",
    "\n",
    "            for rowNo in range(headerRowNo + 1, sheet.nrows):\n",
    "                areaCode = sheet.cell(rowNo, colNoAreaCode).value\n",
    "                if areaCode.startswith('W'):\n",
    "                    regionCode = NATION_CODE_WALES\n",
    "                elif areaCode in localAuthorityLookup:\n",
    "                    regionCode = localAuthorityLookup[areaCode]\n",
    "                else:\n",
    "                    raise RuntimeError(f\"Unrecognised area code {areaCode}\")\n",
    "\n",
    "                geographyType = sheet.cell(rowNo, colNoGeographyType).value\n",
    "                if geographyType == 'Local Authority':\n",
    "                    skip = False\n",
    "                elif geographyType == 'Health Board':\n",
    "                    skip = True\n",
    "                else:\n",
    "                    raise RuntimeError(f\"Unrecognised geography type {geographyType}\")\n",
    "\n",
    "                if not skip:\n",
    "                    causeDeath = sheet.cell(rowNo, colNoCauseDeath).value\n",
    "                    if causeDeath == \"All causes\":\n",
    "                        metric = f\"total_{category}\"\n",
    "                    elif causeDeath == \"COVID 19\":\n",
    "                        metric = f\"covid_{category}\"\n",
    "                    else:\n",
    "                        raise RuntimeError(f\"Unrecognised cause of death {causeDeath}\")\n",
    "\n",
    "                    weekNumber = int(sheet.cell(rowNo, colWeekNumber).value)\n",
    "                    numDeaths = int(sheet.cell(rowNo, colNumDeaths).value)\n",
    "\n",
    "                    if regionCode not in regionData:\n",
    "                        regionData[regionCode] = {}\n",
    "\n",
    "                    if year not in regionData[regionCode]:\n",
    "                        regionData[regionCode][year] = {}\n",
    "\n",
    "                    if weekNumber not in regionData[regionCode][year]:\n",
    "                        regionData[regionCode][year][weekNumber] = {}\n",
    "\n",
    "                    if metric not in regionData[regionCode][year][weekNumber]:\n",
    "                        regionData[regionCode][year][weekNumber][metric] = numDeaths\n",
    "                    else:\n",
    "                        regionData[regionCode][year][weekNumber][metric] += numDeaths\n",
    "\n",
    "            found = True\n",
    "            \n",
    "    if not found:\n",
    "        raise RuntimeError(f\"Could not find {sheetName}\")\n",
    "\n",
    "    return regionData\n",
    "    \n",
    "\n",
    "def saveAreaData(regionType, regions, regionData):\n",
    "    '''Save region data in CSV format'''\n",
    "    \n",
    "    currWeekNumber = 1\n",
    "    currWeekEnded = ons_core.getWeekEnded(2020, currWeekNumber)\n",
    "\n",
    "    for regionCode in regionData:\n",
    "        if regionCode in regions:\n",
    "            regionName = regions[regionCode]\n",
    "            \n",
    "            csvPath = os.path.join(common_core.dataDir, ONS_LAHB_DEATHS, \"csv\", \"weekly\", regionType)\n",
    "            if not os.path.exists(csvPath):\n",
    "                os.makedirs(csvPath)\n",
    "            \n",
    "            csvFn = os.path.join(csvPath, common_core.getSafeName(regionName) + '.csv')\n",
    "\n",
    "            # Save data to CSV\n",
    "            with open(csvFn, 'w') as csvFile:\n",
    "                writer = csv.writer(csvFile)\n",
    "\n",
    "                colNames = [ons_core.WEEK_ENDED, ons_core.WEEK_NUMBER,\n",
    "                            ons_core.TOTAL_REGISTRATIONS, ons_core.TOTAL_OCCURRENCES,\n",
    "                            ons_core.COVID_REGISTRATIONS, ons_core.COVID_OCCURRENCES]\n",
    "                writer.writerow(colNames)\n",
    "\n",
    "                years = [*regionData[regionCode]]\n",
    "                years.sort()\n",
    "                \n",
    "                for year in years:\n",
    "                    weekNumbers = [*regionData[regionCode][year]]\n",
    "                    weekNumbers.sort()\n",
    "\n",
    "                    for weekNumber in weekNumbers:\n",
    "                        if weekNumber != currWeekNumber:\n",
    "                            currWeekNumber = weekNumber\n",
    "                            currWeekEnded = ons_core.getWeekEnded(year, currWeekNumber)\n",
    "\n",
    "                        weekEnded = regionData[regionCode][year][currWeekNumber]\n",
    "\n",
    "                        totalRegistrations = weekEnded[ons_core.TOTAL_REGISTRATIONS]\n",
    "                        totalOccurrences = weekEnded[ons_core.TOTAL_OCCURRENCES]\n",
    "                        covidRegistrations = weekEnded[ons_core.COVID_REGISTRATIONS]\n",
    "                        covidOccurrences = weekEnded[ons_core.COVID_OCCURRENCES]\n",
    "\n",
    "                        colValues = [currWeekEnded, currWeekNumber,\n",
    "                                     totalRegistrations, totalOccurrences,\n",
    "                                     covidRegistrations, covidOccurrences]\n",
    "\n",
    "                        writer.writerow(colValues)\n",
    "\n",
    "                        \n",
    "def calculateNationData(regionData):\n",
    "    '''Calculate national figures'''\n",
    "\n",
    "    nationData = {}\n",
    "\n",
    "    for regionCode in regionData:\n",
    "        if regionCode == NATION_CODE_WALES:\n",
    "            nationCode = NATION_CODE_WALES\n",
    "        else:\n",
    "            nationCode = NATION_CODE_ENGLAND\n",
    "\n",
    "        if NATION_CODE_ENGLAND_WALES not in nationData:\n",
    "            nationData[NATION_CODE_ENGLAND_WALES] = {}\n",
    "\n",
    "        if nationCode not in nationData:\n",
    "            nationData[nationCode] = {}\n",
    "\n",
    "        years = [*regionData[regionCode]]\n",
    "        years.sort()\n",
    "\n",
    "        for year in years:\n",
    "            if year not in nationData[NATION_CODE_ENGLAND_WALES]:\n",
    "                nationData[NATION_CODE_ENGLAND_WALES][year] = {}\n",
    "\n",
    "            if year not in nationData[nationCode]:\n",
    "                nationData[nationCode][year] = {}\n",
    "\n",
    "            weekNumbers = [*regionData[regionCode][year]]\n",
    "            weekNumbers.sort()\n",
    "\n",
    "            for weekNumber in weekNumbers:\n",
    "                if weekNumber not in nationData[NATION_CODE_ENGLAND_WALES][year]:\n",
    "                    nationData[NATION_CODE_ENGLAND_WALES][year][weekNumber] = \\\n",
    "                    {\n",
    "                        ons_core.TOTAL_REGISTRATIONS: 0,\n",
    "                        ons_core.TOTAL_OCCURRENCES: 0,\n",
    "                        ons_core.COVID_REGISTRATIONS: 0,\n",
    "                        ons_core.COVID_OCCURRENCES: 0\n",
    "                    }\n",
    "\n",
    "                if weekNumber not in nationData[nationCode][year]:\n",
    "                    nationData[nationCode][year][weekNumber] = \\\n",
    "                    {\n",
    "                        ons_core.TOTAL_REGISTRATIONS: 0,\n",
    "                        ons_core.TOTAL_OCCURRENCES: 0,\n",
    "                        ons_core.COVID_REGISTRATIONS: 0,\n",
    "                        ons_core.COVID_OCCURRENCES: 0\n",
    "                    }\n",
    "\n",
    "                weekEndedRegion = regionData[regionCode][year][weekNumber]\n",
    "\n",
    "                weekEndedNation = nationData[NATION_CODE_ENGLAND_WALES][year][weekNumber]\n",
    "                weekEndedNation[ons_core.TOTAL_REGISTRATIONS] += weekEndedRegion[ons_core.TOTAL_REGISTRATIONS]\n",
    "                weekEndedNation[ons_core.TOTAL_OCCURRENCES] += weekEndedRegion[ons_core.TOTAL_OCCURRENCES]\n",
    "                weekEndedNation[ons_core.COVID_REGISTRATIONS] += weekEndedRegion[ons_core.COVID_REGISTRATIONS]\n",
    "                weekEndedNation[ons_core.COVID_OCCURRENCES] += weekEndedRegion[ons_core.COVID_OCCURRENCES]\n",
    "\n",
    "                weekEndedNation = nationData[nationCode][year][weekNumber]\n",
    "                weekEndedNation[ons_core.TOTAL_REGISTRATIONS] += weekEndedRegion[ons_core.TOTAL_REGISTRATIONS]\n",
    "                weekEndedNation[ons_core.TOTAL_OCCURRENCES] += weekEndedRegion[ons_core.TOTAL_OCCURRENCES]\n",
    "                weekEndedNation[ons_core.COVID_REGISTRATIONS] += weekEndedRegion[ons_core.COVID_REGISTRATIONS]\n",
    "                weekEndedNation[ons_core.COVID_OCCURRENCES] += weekEndedRegion[ons_core.COVID_OCCURRENCES]\n",
    "\n",
    "    return nationData\n",
    "\n",
    "\n",
    "def main(verbose=common_core.verbose):\n",
    "    '''Do the main processing'''\n",
    "\n",
    "    localAuthorityLookup = loadLocalAuthorityLookup(verbose=verbose)\n",
    "    \n",
    "    partNames = ons_download.downloadLocalAuthorityDeaths(verbose=verbose)\n",
    "    regionData = {}\n",
    "    for partName in partNames:\n",
    "        workbook = loadWorkbook(partName, verbose=verbose)\n",
    "        regionData = updateRegionData(regionData, 'registrations', partName, localAuthorityLookup, verbose=verbose)\n",
    "        regionData = updateRegionData(regionData, 'occurrences', partName, localAuthorityLookup, verbose=verbose)\n",
    "        \n",
    "    nationData = calculateNationData(regionData)\n",
    "\n",
    "    saveAreaData(\"region\", common_core.regions, regionData)\n",
    "    saveAreaData(\"nation\", common_core.nations, nationData)\n",
    "    \n",
    "    if verbose:\n",
    "        print('All done!')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Running Interactively"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Loading /home/jovyan/work/covid-stats/data/ons-lookups/csv...\n",
      "Skipping download of ons-lahb-deaths/raw/weekly/lahbtables2021week11.xlsx...\n",
      "Skipping download of ons-lahb-deaths/raw/weekly/lahbtablesweek01to532020datawk112021.xlsx...\n",
      "Loading ons-lahb-deaths/raw/weekly/lahbtables2021week11.xlsx...\n",
      "Took 19.52 seconds...\n",
      "Loading ons-lahb-deaths/raw/weekly/lahbtablesweek01to532020datawk112021.xlsx...\n",
      "Took 138.55 seconds...\n",
      "All done!\n"
     ]
    }
   ],
   "source": [
    "if __name__ == '__main__':\n",
    "\n",
    "    main(verbose = True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}