02_munge/src/munge_noaa_nerrs.py
def process_data_to_csv(site, site_raw_datafiles, column_mapping, flags_to_drop, agg_level, prop_obs_required, fill_anom_w_nos):
    '''
    process raw data text files into clean csvs, including:
        dropping unwanted flags
        converting datetime column to datetime format
        converting all data columns to numeric type
        removing metadata columns so that only datetime and data columns remain 
    '''
    for raw_datafile in site_raw_datafiles:
        year_df = read_data(raw_datafile)
        if raw_datafile == site_raw_datafiles[0]:
            combined_df = year_df.copy()
        else:
            combined_df = pd.concat([combined_df, year_df], ignore_index=True)

    # filter out data that we don't want
    col_values_accepted = config['col_values_accepted']
    for col in col_values_accepted.keys():
        combined_df = combined_df[combined_df[col].isin(col_values_accepted[col])]
        combined_df.drop(col, axis=1, inplace=True)

    # replace all flagged data we want to remove with NaN
    vars_to_keep = list(column_mapping.keys())
    vars_to_keep.remove('DatetimeStamp')
    for var in vars_to_keep:
        flag_col = f'F_{var}'
        combined_df[var] = np.where(combined_df[flag_col].isin(flags_to_drop), np.nan, combined_df[var])

    # drop any columns we don't want
    combined_df = combined_df[column_mapping.keys()]

    # map column names
    combined_df.rename(columns=column_mapping, inplace=True)

    # convert datetime column to datetime type
    combined_df['datetime'] = combined_df['datetime'].astype('datetime64')

    # make all other columns numeric
    cols = combined_df.columns.drop('datetime')
    combined_df[cols] = combined_df[cols].apply(pd.to_numeric, errors='coerce')

    # aggregate data to specified timestep
    combined_df = utils.process_to_timestep(combined_df, column_mapping.values(), agg_level, prop_obs_required)

    # drop any columns with no data
    combined_df.dropna(axis=1, how='all', inplace=True)
    
    # there are some anomalous values below 980
    combined_df.loc[combined_df['air_pressure'] < 980,'air_pressure'] = np.nan
    
    if fill_anom_w_nos:
        try:
            noaa_nos_file = os.listdir('02_munge/out/daily_summaries')
            noaa_nos = pd.read_csv(os.path.join('02_munge','out','daily_summaries',noaa_nos_file[0]), index_col = 'datetime')
            combined_df['air_pressure'] = combined_df['air_pressure'].fillna(noaa_nos['air_pressure'])
        except:
            print('No NOAA NOS data to fill NERR air pressure record')
    
    #fill gaps in temperature, air pressure, wind speed and wind direction
    # all < 0.5 %, precipitation has more like 10% missing so we won't fill that
    combined_df['temperature'] = fill_gaps(combined_df['temperature'])
    combined_df['air_pressure'] = fill_gaps(combined_df['air_pressure'])
    combined_df['wind_speed'] = fill_gaps(combined_df['wind_speed'])
    combined_df['wind_direction'] = fill_gaps(combined_df['wind_direction'])
    combined_df['wind_speed_direction'] = -1*combined_df['wind_speed']*np.cos(combined_df['wind_direction']*(np.pi/180))

    # save pre-processed data
    data_outfile_csv = os.path.join('.', '02_munge', 'out', agg_level, 'noaa_nerrs_delsjmet.csv')
    combined_df.to_csv(data_outfile_csv, index=True)

    if write_location == 'S3':
        print('uploading to s3')
        s3_client.upload_file(data_outfile_csv, s3_bucket, utils.local_to_s3_pathname(data_outfile_csv))

    return combined_df

    process raw data text files into clean csvs, including:
        dropping unwanted flags
        converting datetime column to datetime format
        converting all data columns to numeric type
        removing metadata columns so that only datetime and data columns remain 
    
process raw data text files into clean csvs, including:
dropping unwanted flags
converting datetime column to datetime format
converting all data columns to numeric type
removing metadata columns so that only datetime and data columns remain
process raw data text files into clean csvs including : dropping unwanted flags converting datetime column to datetime format converting all data columns to numeric type removing metadata columns so that only datetime and data columns remain
def process_data_to_csv(site, site_raw_datafiles, column_mapping, flags_to_drop, agg_level, prop_obs_required, fill_anom_w_nos):
    for raw_datafile in site_raw_datafiles:
        year_df = read_data(raw_datafile)
        if raw_datafile == site_raw_datafiles[0]:
            combined_df = year_df.copy()
        else:
            combined_df = pd.concat([combined_df, year_df], ignore_index=True)
    col_values_accepted = config['col_values_accepted']
    for col in col_values_accepted.keys():
        combined_df = combined_df[combined_df[col].isin(col_values_accepted[col])]
        combined_df.drop(col, axis=1, inplace=True)
    vars_to_keep = list(column_mapping.keys())
    vars_to_keep.remove('DatetimeStamp')
    for var in vars_to_keep:
        flag_col = f'F_{var}'
        combined_df[var] = np.where(combined_df[flag_col].isin(flags_to_drop), np.nan, combined_df[var])
    combined_df = combined_df[column_mapping.keys()]
    combined_df.rename(columns=column_mapping, inplace=True)
    combined_df['datetime'] = combined_df['datetime'].astype('datetime64')
    cols = combined_df.columns.drop('datetime')
    combined_df[cols] = combined_df[cols].apply(pd.to_numeric, errors='coerce')
    combined_df = utils.process_to_timestep(combined_df, column_mapping.values(), agg_level, prop_obs_required)
    combined_df.dropna(axis=1, how='all', inplace=True)
    combined_df.loc[combined_df['air_pressure'] < 980,'air_pressure'] = np.nan
    if fill_anom_w_nos:
        try:
            noaa_nos_file = os.listdir('02_munge/out/daily_summaries')
            noaa_nos = pd.read_csv(os.path.join('02_munge','out','daily_summaries',noaa_nos_file[0]), index_col = 'datetime')
            combined_df['air_pressure'] = combined_df['air_pressure'].fillna(noaa_nos['air_pressure'])
        except:
            print('No NOAA NOS data to fill NERR air pressure record')
    combined_df['temperature'] = fill_gaps(combined_df['temperature'])
    combined_df['air_pressure'] = fill_gaps(combined_df['air_pressure'])
    combined_df['wind_speed'] = fill_gaps(combined_df['wind_speed'])
    combined_df['wind_direction'] = fill_gaps(combined_df['wind_direction'])
    combined_df['wind_speed_direction'] = -1*combined_df['wind_speed']*np.cos(combined_df['wind_direction']*(np.pi/180))
    data_outfile_csv = os.path.join('.', '02_munge', 'out', agg_level, 'noaa_nerrs_delsjmet.csv')
    combined_df.to_csv(data_outfile_csv, index=True)
    if write_location == 'S3':
        print('uploading to s3')
        s3_client.upload_file(data_outfile_csv, s3_bucket, utils.local_to_s3_pathname(data_outfile_csv))
    return combined_df
def process_data_to_csv ( site , site_raw_datafiles , column_mapping , flags_to_drop , agg_level , prop_obs_required , fill_anom_w_nos ) : for raw_datafile in site_raw_datafiles : year_df = read_data ( raw_datafile ) if raw_datafile == site_raw_datafiles [ 0 ] : combined_df = year_df . copy ( ) else : combined_df = pd . concat ( [ combined_df , year_df ] , ignore_index = True ) col_values_accepted = config [ 'col_values_accepted' ] for col in col_values_accepted . keys ( ) : combined_df = combined_df [ combined_df [ col ] . isin ( col_values_accepted [ col ] ) ] combined_df . drop ( col , axis = 1 , inplace = True ) vars_to_keep = list ( column_mapping . keys ( ) ) vars_to_keep . remove ( 'DatetimeStamp' ) for var in vars_to_keep : flag_col = f'F_{var}' combined_df [ var ] = np . where ( combined_df [ flag_col ] . isin ( flags_to_drop ) , np . nan , combined_df [ var ] ) combined_df = combined_df [ column_mapping . keys ( ) ] combined_df . rename ( columns = column_mapping , inplace = True ) combined_df [ 'datetime' ] = combined_df [ 'datetime' ] . astype ( 'datetime64' ) cols = combined_df . columns . drop ( 'datetime' ) combined_df [ cols ] = combined_df [ cols ] . apply ( pd . to_numeric , errors = 'coerce' ) combined_df = utils . process_to_timestep ( combined_df , column_mapping . values ( ) , agg_level , prop_obs_required ) combined_df . dropna ( axis = 1 , how = 'all' , inplace = True ) combined_df . loc [ combined_df [ 'air_pressure' ] < 980 , 'air_pressure' ] = np . nan if fill_anom_w_nos : try : noaa_nos_file = os . listdir ( '02_munge/out/daily_summaries' ) noaa_nos = pd . read_csv ( os . path . join ( '02_munge' , 'out' , 'daily_summaries' , noaa_nos_file [ 0 ] ) , index_col = 'datetime' ) combined_df [ 'air_pressure' ] = combined_df [ 'air_pressure' ] . fillna ( noaa_nos [ 'air_pressure' ] ) except : print ( 'No NOAA NOS data to fill NERR air pressure record' ) combined_df [ 'temperature' ] = fill_gaps ( combined_df [ 'temperature' ] ) combined_df [ 'air_pressure' ] = fill_gaps ( combined_df [ 'air_pressure' ] ) combined_df [ 'wind_speed' ] = fill_gaps ( combined_df [ 'wind_speed' ] ) combined_df [ 'wind_direction' ] = fill_gaps ( combined_df [ 'wind_direction' ] ) combined_df [ 'wind_speed_direction' ] = - 1 * combined_df [ 'wind_speed' ] * np . cos ( combined_df [ 'wind_direction' ] * ( np . pi / 180 ) ) data_outfile_csv = os . path . join ( '.' , '02_munge' , 'out' , agg_level , 'noaa_nerrs_delsjmet.csv' ) combined_df . to_csv ( data_outfile_csv , index = True ) if write_location == 'S3' : print ( 'uploading to s3' ) s3_client . upload_file ( data_outfile_csv , s3_bucket , utils . local_to_s3_pathname ( data_outfile_csv ) ) return combined_df
process raw data text files into clean csvs, including:
dropping unwanted flags
converting datetime column to datetime format
converting all data columns to numeric type
removing metadata columns so that only datetime and data columns remain
process raw data text files into clean csvs including : dropping unwanted flags converting datetime column to datetime format converting all data columns to numeric type removing metadata columns so that only datetime and data columns remain
'''
    process raw data text files into clean csvs, including:
        dropping unwanted flags
        converting datetime column to datetime format
        converting all data columns to numeric type
        removing metadata columns so that only datetime and data columns remain 
    ''' # filter out data that we don't want # replace all flagged data we want to remove with NaN # drop any columns we don't want # map column names # convert datetime column to datetime type # make all other columns numeric # aggregate data to specified timestep # drop any columns with no data # there are some anomalous values below 980 #fill gaps in temperature, air pressure, wind speed and wind direction # all < 0.5 %, precipitation has more like 10% missing so we won't fill that # save pre-processed data
site
site_raw_datafiles
column_mapping
flags_to_drop
agg_level
prop_obs_required
fill_anom_w_nos