import logging.config import os import numpy as np import pandas as pd from scipy.integrate import simps logging.config.fileConfig('./src/logging.conf', disable_existing_loggers=False) logger = logging.getLogger(__name__) def return_first_or_nan(l): """ Returns the first value of a list if empty or returns nan. Used for getting dune/toe and crest values. :param l: :return: """ if len(l) == 0: return np.nan else: return l[0] def volume_change(df_profiles, df_profile_features, zone): """ Calculates how much the volume change there is between prestrom and post storm profiles. :param df_profiles: :param df_profile_features: :param zone: Either 'swash' or 'dune_face' :return: """ logger.info('Calculating change in beach volume in {} zone'.format(zone)) df_vol_changes = pd.DataFrame(index=df_profile_features.index) df_profiles = df_profiles.sort_index() sites = df_profiles.groupby(level=['site_id']) for site_id, df_site in sites: logger.debug('Calculating change in beach volume at {} in {} zone'.format(site_id, zone)) prestorm_dune_toe_x = df_profile_features.loc[df_profile_features.index == site_id].dune_toe_x.tolist() prestorm_dune_crest_x = df_profile_features.loc[df_profile_features.index == site_id].dune_crest_x.tolist() # We may not have a dune toe or crest defined, or there may be multiple defined. prestorm_dune_crest_x = return_first_or_nan(prestorm_dune_crest_x) prestorm_dune_toe_x = return_first_or_nan(prestorm_dune_toe_x) # If no dune to has been defined, Dlow = Dhigh. Refer to Sallenger (2000). if np.isnan(prestorm_dune_toe_x): prestorm_dune_toe_x = prestorm_dune_crest_x # Find last x coordinate where we have both prestorm and poststorm measurements. If we don't do this, # the prestorm and poststorm values are going to be calculated over different lengths. df_zone = df_site.dropna(subset=['z']) x_last_obs = min([max(df_zone.query("profile_type == '{}'".format(profile_type)).index.get_level_values('x')) for profile_type in ['prestorm', 'poststorm']]) # Where we want to measure pre and post storm volume is dependant on the zone selected if zone == 'swash': x_min = prestorm_dune_toe_x x_max = x_last_obs elif zone == 'dune_face': x_min = prestorm_dune_crest_x x_max = prestorm_dune_toe_x else: logger.warning('Zone argument not properly specified. Please check') x_min = None x_max = None # Now, compute the volume of sand between the x-coordinates prestorm_dune_toe_x and x_swash_last for both prestorm # and post storm profiles. prestorm_vol = beach_volume(x=df_zone.query("profile_type=='prestorm'").index.get_level_values('x'), z=df_zone.query("profile_type=='prestorm'").z, x_min=x_min, x_max=x_max) poststorm_vol = beach_volume(x=df_zone.query("profile_type=='poststorm'").index.get_level_values('x'), z=df_zone.query("profile_type=='poststorm'").z, x_min=x_min, x_max=x_max) df_vol_changes.loc[site_id, 'prestorm_{}_vol'.format(zone)] = prestorm_vol df_vol_changes.loc[site_id, 'poststorm_{}_vol'.format(zone)] = poststorm_vol df_vol_changes.loc[site_id, '{}_vol_change'.format(zone)] = prestorm_vol - poststorm_vol return df_vol_changes def beach_volume(x, z, x_min=np.NINF, x_max=np.inf): """ Returns the beach volume of a profile, calculated with Simpsons rule :param x: x-coordinates of beach profile :param z: z-coordinates of beach profile :param x_min: Minimum x-coordinate to consider when calculating volume :param x_max: Maximum x-coordinate to consider when calculating volume :return: """ profile_mask = [True if x_min < x_coord < x_max else False for x_coord in x] x_masked = np.array(x)[profile_mask] z_masked = np.array(z)[profile_mask] if len(x_masked) == 0 or len(z_masked) == 0: return np.nan else: return simps(z_masked, x_masked) def storm_regime(df_observed_impacts): """ Returns the dataframe with an additional column of storm impacts based on the Storm Impact Scale. Refer to Sallenger (2000) for details. :param df_observed_impacts: :return: """ logger.info('Getting observed storm regimes') df_observed_impacts.loc[df_observed_impacts.swash_vol_change < 3, 'storm_regime'] = 'swash' df_observed_impacts.loc[df_observed_impacts.dune_face_vol_change > 3, 'storm_regime'] = 'collision' return df_observed_impacts if __name__ == '__main__': logger.info('Importing existing data') data_folder = './data/interim' df_profiles = pd.read_csv(os.path.join(data_folder, 'profiles.csv'), index_col=[0, 1, 2]) df_profile_features = pd.read_csv(os.path.join(data_folder, 'profile_features.csv'), index_col=[0]) logger.info('Creating new dataframe for observed impacts') df_observed_impacts = pd.DataFrame(index=df_profile_features.index) logger.info('Getting pre/post storm volumes') df_swash_vol_changes = volume_change(df_profiles, df_profile_features, zone='swash') df_dune_face_vol_changes = volume_change(df_profiles, df_profile_features, zone='dune_face') df_observed_impacts = df_observed_impacts.join([df_swash_vol_changes, df_dune_face_vol_changes]) # Classify regime based on volume changes df_observed_impacts = storm_regime(df_observed_impacts) # Save dataframe to csv df_observed_impacts.to_csv(os.path.join(data_folder, 'impacts_observed.csv'))