Skip to article frontmatterSkip to article content
Site not loading correctly?

This may be due to an incorrect BASE_URL configuration. See the MyST Documentation for reference.

ReadMe

ReadMe

This notebook is used to compute infrastructure exposure to a set of hazards for a given country and subnational division (admin1 or admin2). It takes as input population and hazard raster files, infrastructure locations and admin boundaries. Output is 7 different spreadsheets (.csv and .xlsx): 1 with all hazards and exposure per infrastructure location; 6 (1 per hazard and current/future) for aggregate number of infrastructure facilities exposed per admin subdivision

Users only need to indicate country 3-letters iso_code (e.g., ‘UKR’ or ‘VNM’), admin level (‘admin2’,‘admin1’) and hazard list. Users can also choose to use default GADM subnational boundaries (instead of a custom subdivision) by setting use_gadm_boundaries = True.

This notebook should be executed after country data preparation (country_data_preparation.ipynb)

All file paths and custom parameters are defined on file constants.py

Custom Functions

def process_all_hazards(infrastructure_df,iso_code, admin_level, hazard_list, infrastructure_layer):
    """
    Loops over all hazard in hazard_list and compute infrastructure exposure to each one of them.
    If no raster is available for a given country / hazard, print a warning message. This
    happens when a country is not exposed to a specific hazard (commun for cyclone).

    """

    infrastructure_id = GLOBAL_INFRASTRUCTURE_UNIQUE_ID[infrastructure_layer]
    
    admin_pcode = REF_ADMIN_PCODE[admin_level]
    infrastructure_exposure_df = pd.DataFrame(infrastructure_df[[admin_pcode, infrastructure_id]])
    infrastructure_exposure_df['ISO'] = iso_code

    ############################
    for hazard in hazard_list:

        col_name = hazard

        exposure_df = pd.DataFrame(infrastructure_df[infrastructure_id])
        exposure_df[col_name] = 'no_exposure'
        #Loop over all return period for a given hazard (defined on file constants.py)
        for return_period in HAZARD_RETURN_PERIOD[hazard]:  
            try:
                new_rp_df = process_infrastructure_exposure_single_hazard(infrastructure_df.copy(),hazard, iso_code, return_period,infrastructure_layer)
                exposure_df = exposure_df.merge(new_rp_df, on = infrastructure_id, how = 'left')
                #Output value is the lower return period to which the infrastructure is exposed
                exposure_df.loc[(exposure_df[col_name] == 'no_exposure') & (exposure_df['value_rp'] == 1),col_name] = str(return_period) + 'yr'
                exposure_df.drop(columns = 'value_rp', inplace = True)
            except:
                #If no raster is available for a given country / hazard, print a warning message.
                print('No exposure for hazard ' + hazard + ' - RP-' + str(return_period) + ' on country ' + iso_code) 

        infrastructure_exposure_df = infrastructure_exposure_df.merge(exposure_df, on = infrastructure_id)
        
    ############################

    
    # Create country-specific output path and export data on csv and xlsx formats 
    infrastructure_exposure_df = infrastructure_exposure_df.round(2)
    free_text = infrastructure_layer + '-facilities'
    OUTPUT_PATH = ANALYSIS_OUTPUT_PATH.replace('wrl',iso_code.lower()).replace('XXX',free_text)
    infrastructure_exposure_df.to_excel(OUTPUT_PATH.replace('.csv','.xlsx'), index=False, sheet_name = free_text)
    print(OUTPUT_PATH)

    return(infrastructure_exposure_df)



def aggregate_infrastructure_exposure(infrastructure_exposure_df, admin_df, admin_level, hazard, infrastructure_layer):
    """
    Computes number of facilities exposed for a given hazard and admin level.
    """

    admin_infrastructure_df = admin_df.drop(columns = ['geometry']).copy()
    admin_pcode = REF_ADMIN_PCODE[admin_level]
      
    ###################################
    infrastructure_hazard_df = infrastructure_exposure_df[[admin_pcode, hazard]].copy()


    ###################################
    #Initiate the DataFrame using all possible exposure level (return periods 
    #for cyclone, flood and earthquake). Initialise all values to 0.
    hazard_col_list = hazard_columns_list([hazard])
    for col in hazard_col_list:
        admin_infrastructure_df[col] = 0
        
    ###################################
    infrastructure_hazard_df = infrastructure_hazard_df.value_counts().reset_index()
    infrastructure_hazard_df.rename(columns = {hazard : 'hazard_value'}, inplace = True)
    infrastructure_hazard_df['hazard_value'] = str(hazard) + '_' + infrastructure_hazard_df['hazard_value']
    infrastructure_hazard_df = infrastructure_hazard_df.pivot(index=admin_pcode, columns='hazard_value', values='count').reset_index()
 
    admin_infrastructure_df = admin_infrastructure_df.merge(infrastructure_hazard_df, on = admin_pcode, how = 'left', suffixes=('','_new'))
    admin_infrastructure_df.fillna(0, inplace = True)

    ###################################
    for col in hazard_col_list:
        if col + '_new' in  admin_infrastructure_df.columns:
            admin_infrastructure_df[col] = admin_infrastructure_df[col + '_new']
            admin_infrastructure_df.drop(columns = [col + '_new'], inplace = True)

    ###################################
    admin_infrastructure_df['num_facilities'] = admin_infrastructure_df[hazard_col_list].sum(axis = 1)
    admin_infrastructure_df[hazard + '_exposure'] = admin_infrastructure_df['num_facilities'] - admin_infrastructure_df[hazard + '_no_exposure']
            
    ###################################
    admin_infrastructure_df = admin_infrastructure_df.round(2)
    free_text = infrastructure_layer + '-facilities-' + admin_level + '-' + hazard.replace('_','-')
    sheet_name_text = infrastructure_layer + '-facilities-' + hazard.replace('_','-') 
    OUTPUT_PATH = ANALYSIS_OUTPUT_PATH.replace('wrl',iso_code.lower()).replace('XXX',free_text)
    admin_infrastructure_df.to_excel(OUTPUT_PATH.replace('.csv','.xlsx'), index=False, sheet_name = sheet_name_text)
    print(OUTPUT_PATH)

    return(admin_infrastructure_df)

Run code

import pandas as pd

from constants import *
from utils import *

########################## USER input ####################
iso_code = 'MOZ'
hazard_list = ['flood_current', 'flood_future', 'cyclone_current', 'cyclone_future', 'earthquake']
infrastructure_list = ['health']
admin_level_list = ['admin1', 'admin2']
use_gadm_boundaries = True
############################################################


for admin_level in admin_level_list:
    ##Load admin boundary and infrastructure location data
    admin_df = load_admin_data(use_gadm_boundaries, iso_code, admin_level)

    for infrastructure_layer in infrastructure_list:
        infrastructure_df = load_infrastructure_data(iso_code, admin_df, admin_level, infrastructure_layer)
        ##Process exposure at infrastructure location level
        infrastructure_exposure_df = process_all_hazards(infrastructure_df, iso_code, admin_level, hazard_list, infrastructure_layer)
        ##Aggregate number of exposed infrastructure facilities per admin subdivision
        for hazard in hazard_list:
            admin_infrastructure_df = aggregate_infrastructure_exposure(infrastructure_exposure_df.copy(), admin_df, admin_level, hazard, infrastructure_layer)
Python
ReadMe
Python
ReadMe