diff --git a/.github/workflows/compare_single_FBA.yml b/.github/workflows/compare_single_FBA.yml index d1f91fe3c..36377a232 100644 --- a/.github/workflows/compare_single_FBA.yml +++ b/.github/workflows/compare_single_FBA.yml @@ -25,7 +25,7 @@ jobs: - name: Set up Python uses: actions/setup-python@v3 with: - python-version: "3.10" + python-version: "3.12" - name: Update pip run: | diff --git a/.github/workflows/compare_single_FBS.yml b/.github/workflows/compare_single_FBS.yml index bf6c26608..2110cd5e1 100644 --- a/.github/workflows/compare_single_FBS.yml +++ b/.github/workflows/compare_single_FBS.yml @@ -21,7 +21,7 @@ jobs: - name: Set up Python uses: actions/setup-python@v3 with: - python-version: "3.10" + python-version: "3.12" - name: Update pip run: | diff --git a/.github/workflows/generate_FBA.yml b/.github/workflows/generate_FBA.yml index 41b92ce1a..d6f621f71 100644 --- a/.github/workflows/generate_FBA.yml +++ b/.github/workflows/generate_FBA.yml @@ -25,7 +25,7 @@ jobs: - name: Set up Python uses: actions/setup-python@v3 with: - python-version: "3.10" + python-version: "3.12" - name: Update pip run: | diff --git a/.github/workflows/python-app.yml b/.github/workflows/python-app.yml index b97edbea1..13da50277 100644 --- a/.github/workflows/python-app.yml +++ b/.github/workflows/python-app.yml @@ -31,7 +31,7 @@ jobs: fail-fast: false matrix: os: [ubuntu-latest, windows-latest, macos-latest] - py-version: ['3.9', '3.10', '3.11'] + py-version: ['3.10', '3.11', '3.12'] steps: - uses: actions/checkout@v3 @@ -74,7 +74,7 @@ jobs: - name: Set up Python uses: actions/setup-python@v3 with: - python-version: "3.10" + python-version: "3.12" - name: Update pip & install testing pkgs run: | diff --git a/.github/workflows/test_methods.yml b/.github/workflows/test_methods.yml index 2f0e32ac7..424bc8970 100644 --- a/.github/workflows/test_methods.yml +++ b/.github/workflows/test_methods.yml @@ -19,7 +19,7 @@ jobs: - name: Set up Python uses: actions/setup-python@v3 with: - python-version: "3.10" + python-version: "3.12" - name: Update pip & install testing pkgs run: | @@ -47,7 +47,7 @@ jobs: - name: Set up Python uses: actions/setup-python@v3 with: - python-version: "3.10" + python-version: "3.12" - name: Update pip & install testing pkgs run: | diff --git a/docs/user_guide/user_guide.md b/docs/user_guide/user_guide.md index 35a4e5294..04aa34d67 100644 --- a/docs/user_guide/user_guide.md +++ b/docs/user_guide/user_guide.md @@ -327,7 +327,7 @@ To attribute a primary data source to target level sectors, identify the primary specified attribution data source - equal: Equally attribute parent values to child values until reach target sector length -- inheritance: Assign parent values to all child values. Usefull in +- inheritance: Assign parent values to all child values. Useful in situations where value is a rate, such as kg/m2. ### Recursive vs Sequential Attribution diff --git a/flowsa/data/unit_conversion.csv b/flowsa/data/unit_conversion.csv index 9376be144..b3db8ea59 100644 --- a/flowsa/data/unit_conversion.csv +++ b/flowsa/data/unit_conversion.csv @@ -13,6 +13,7 @@ TON,kg,907.185 tons,kg,907.185 short tons,kg,907.185 Thousands of Tons,kg,907185 +million short tons,kg,907184740 MT,kg,1000 Metric Tons,kg,1000 LB,kg,0.45359 @@ -20,7 +21,9 @@ ACRE FEET / ACRE,kg/m2,304.8000006 gal/USD,kg/USD,3.79 Quadrillion Btu,MJ,1.06E+12 Trillion Btu,MJ,1055055900 +trillion Btu,MJ,1055055900 Tbtu,MJ,1055055900 +million kWh,MJ,3600000 MT CO2e,MMT CO2e,0.000001 employees,p,1 Thousand USD,USD,1000 diff --git a/flowsa/data_source_scripts/EIA_CBECS_Land.py b/flowsa/data_source_scripts/EIA_CBECS_Land.py index b9c9ad753..e8e740ed4 100644 --- a/flowsa/data_source_scripts/EIA_CBECS_Land.py +++ b/flowsa/data_source_scripts/EIA_CBECS_Land.py @@ -405,7 +405,7 @@ def disaggregate_eia_cbecs_vacant_and_other(df_load): df_nvno = df_nvno[df_nvno['ActivityConsumedBy'] != 'All buildings'] df_nvno = df_nvno.groupby( ['Unit', 'Location', 'LocationSystem', 'Year', 'Description'], - as_index=False).agg({'FlowAmount': sum}) + as_index=False).agg({'FlowAmount': "sum"}) df_act = df_nvno.rename(columns={'FlowAmount': 'NonVacantNonOther'}) # merge df and subtract to determine FlowAmount to allocate to diff --git a/flowsa/data_source_scripts/EIA_MECS.py b/flowsa/data_source_scripts/EIA_MECS.py index a7b654e4c..bbfcfcf0d 100644 --- a/flowsa/data_source_scripts/EIA_MECS.py +++ b/flowsa/data_source_scripts/EIA_MECS.py @@ -509,8 +509,8 @@ def update_regions_to_states(fba: FlowByActivity, fba = ( fba .drop(columns=['group_id', 'group_total']) - .reset_index(drop=True).reset_index() - .rename(columns={'index': 'group_id'}) + .reset_index(drop=True) + .reset_index(names='group_id') .assign(group_total=fba.FlowAmount) ) diff --git a/flowsa/data_source_scripts/EPA_GHGI.py b/flowsa/data_source_scripts/EPA_GHGI.py index 5fc0d926a..6b73f0594 100644 --- a/flowsa/data_source_scripts/EPA_GHGI.py +++ b/flowsa/data_source_scripts/EPA_GHGI.py @@ -40,9 +40,9 @@ ANNEX_ENERGY_TABLES = ["A-" + str(x) for x in list(range(4,16))] DROP_COLS = ["Unnamed: 0"] + list(pd.date_range( - start="1990", end="2010", freq='Y').year.astype(str)) + start="1990", end="2010", freq='YE').year.astype(str)) -YEARS = list(pd.date_range(start="2010", end="2024", freq='Y').year.astype(str)) +YEARS = list(pd.date_range(start="2010", end="2024", freq='YE').year.astype(str)) def ghg_url_helper(*, build_url, config, **_): @@ -496,20 +496,15 @@ def ghg_parse(*, df_list, year, config, **_): # set suppressed values to 0 but mark as suppressed # otherwise set non-numeric to nan try: - df = (df.assign( - Suppressed = np.where(df.FlowAmount.str.strip() == "+", "+", - np.nan), - FlowAmount = pd.Series( - np.where(df.FlowAmount.str.strip() == "+", 0, - df.FlowAmount.str.replace(',','')))) - ) - df = (df.assign( - FlowAmount = np.where(pd.to_numeric( - df.FlowAmount, errors='coerce').isnull(), - np.nan, pd.to_numeric( - df.FlowAmount, errors='coerce'))) - .dropna(subset='FlowAmount') - ) + df['Suppressed'] = (df['FlowAmount'] + .astype(str).str.strip().eq('+') + .replace({True: '+', False: np.nan}) + .infer_objects(copy=False) + ) + df['FlowAmount'] = df['FlowAmount'].astype(str).str.replace(',', '').infer_objects(copy=False) + df['FlowAmount'] = df['FlowAmount'].replace('+', '0').infer_objects(copy=False) + df['FlowAmount'] = pd.to_numeric(df['FlowAmount'], errors='coerce') + df = df.dropna(subset='FlowAmount') except AttributeError: # if no string in FlowAmount, then proceed df = df.dropna(subset='FlowAmount') diff --git a/flowsa/data_source_scripts/USDA_ERS_MLU.py b/flowsa/data_source_scripts/USDA_ERS_MLU.py index 5246646c5..34d9d7981 100644 --- a/flowsa/data_source_scripts/USDA_ERS_MLU.py +++ b/flowsa/data_source_scripts/USDA_ERS_MLU.py @@ -201,7 +201,7 @@ def allocate_usda_ers_mlu_land_in_urban_areas( df_non_urban_transport_area[['Location', 'Unit', 'FlowAmount']] non_urban_transport_area_sum = df_non_urban_transport_area.groupby( ['Location', 'Unit'], as_index=False).agg( - {'FlowAmount': sum}).rename(columns={'FlowAmount': 'NonTransport'}) + {'FlowAmount': "sum"}).rename(columns={'FlowAmount': 'NonTransport'}) # compare units compare_df_units(fba, df_non_urban_transport_area) # calculate total urban transportation by subtracting @@ -232,7 +232,7 @@ def allocate_usda_ers_mlu_land_in_urban_areas( air_rail_area = air_rail_area[['Location', 'Unit', 'FlowAmount']] air_rail_area_sum = air_rail_area.groupby( ['Location', 'Unit'], as_index=False).agg( - {'FlowAmount': sum}).rename(columns={'FlowAmount': 'AirRail'}) + {'FlowAmount': "sum"}).rename(columns={'FlowAmount': 'AirRail'}) df_highway = df_transport.merge(air_rail_area_sum, how='left') df_highway = df_highway.assign( @@ -256,8 +256,8 @@ def allocate_usda_ers_mlu_land_in_urban_areas( allocated_urban_areas_df .drop(columns=['group_id', 'group_total', 'rurl_res', 'total_area'], errors='ignore') - .reset_index(drop=True).reset_index() - .rename(columns={'index': 'group_id'}) + .reset_index(drop=True) + .reset_index(names='group_id') .assign(group_total=allocated_urban_areas_df.FlowAmount) ) @@ -317,7 +317,7 @@ def allocate_usda_ers_mlu_land_in_rural_transportation_areas( air_rail_area = air_rail_area[['Location', 'Unit', 'FlowAmount']] air_rail_area_sum = air_rail_area.groupby( ['Location', 'Unit'], as_index=False).agg( - {'FlowAmount': sum}).rename(columns={'FlowAmount': 'AirRail'}) + {'FlowAmount': "sum"}).rename(columns={'FlowAmount': 'AirRail'}) # compare units compare_df_units(fba, air_rail_area) @@ -341,8 +341,8 @@ def allocate_usda_ers_mlu_land_in_rural_transportation_areas( allocated_rural_trans = ( allocated_rural_trans .drop(columns=['group_id', 'group_total']) - .reset_index(drop=True).reset_index() - .rename(columns={'index': 'group_id'}) + .reset_index(drop=True) + .reset_index(names='group_id') .assign(group_total=allocated_rural_trans.FlowAmount) ) diff --git a/flowsa/data_source_scripts/USGS_NWIS_WU.py b/flowsa/data_source_scripts/USGS_NWIS_WU.py index b5db767d3..a86c50df6 100644 --- a/flowsa/data_source_scripts/USGS_NWIS_WU.py +++ b/flowsa/data_source_scripts/USGS_NWIS_WU.py @@ -610,6 +610,13 @@ def subset_and_merge_irrigation_types(df: FlowByActivity): "ActivityConsumedBy_y": "Crop_ACB", "Description_x": 'Description', "Description_y": "Crop_Description"}) + + # ensure activity cols are object, not float, so not converted to 0 + col = ['ActivityProducedBy', 'Golf_APB', 'Crop_APB'] + for c in col: + df_m2[c] = df_m2[c].astype(str) + + # fill na and sum crop and golf for col in df_m2: if df_m2[col].dtype in ("int", "float"): diff --git a/flowsa/data_source_scripts/stewiFBS.py b/flowsa/data_source_scripts/stewiFBS.py index 9a9bb36c6..eb9020fc4 100644 --- a/flowsa/data_source_scripts/stewiFBS.py +++ b/flowsa/data_source_scripts/stewiFBS.py @@ -338,6 +338,7 @@ def prepare_stewi_fbs(df_load, config) -> 'FlowBySector': .reset_index(drop=True), full_name=config.get('full_name'), config=config, + convert_df_to_flowby=True ).prepare_fbs() fbs.config.update({'data_format': 'FBS'}) diff --git a/flowsa/dataclean.py b/flowsa/dataclean.py index 976609f85..ecedb0ed3 100644 --- a/flowsa/dataclean.py +++ b/flowsa/dataclean.py @@ -49,21 +49,24 @@ def add_missing_flow_by_fields(flowby_partial_df, flowbyfields): flowby_partial_df[col] = np.nan # convert all None, 'nan' to np.nan flowby_partial_df = (flowby_partial_df - .replace('None', np.nan) - .replace('nan', np.nan)) + .replace({'None': np.nan, 'nan': np.nan}) + .infer_objects(copy=False) + ) # convert data types to match those defined in flow_by_activity_fields for k, v in flowbyfields.items(): if k in flowby_partial_df.columns: flowby_partial_df[k] = \ flowby_partial_df[k].astype(v[0]['dtype']) if v[0]['dtype'] in ['string', 'str', 'object']: - flowby_partial_df[k] = flowby_partial_df[k].fillna(np.nan) + flowby_partial_df[k] = flowby_partial_df[k].fillna(np.nan).infer_objects(copy=False) else: - flowby_partial_df[k] = flowby_partial_df[k].fillna(0) + flowby_partial_df[k] = flowby_partial_df[k].fillna(0).infer_objects(copy=False) # convert all None, 'nan' to np.nan - flowby_partial_df = (flowby_partial_df - .replace('None', np.nan) - .replace('nan', np.nan)) + with pd.option_context('future.no_silent_downcasting', True): + flowby_partial_df = (flowby_partial_df + .replace({'None': np.nan, 'nan': np.nan}) + .infer_objects(copy=False) + ) # Resort it so order is correct cols = [e for e in flowbyfields.keys() if e in flowby_partial_df.columns] flowby_df = flowby_partial_df[cols] @@ -99,7 +102,7 @@ def standardize_units(df): .merge(conversion_table, how='left', left_on='Unit', right_on='old_unit') .assign(Unit=lambda x: x.new_unit.mask(x.new_unit.isna(), x.Unit), - conversion_factor=lambda x: x.conversion_factor.fillna(1), + conversion_factor=lambda x: x.conversion_factor.fillna(1).infer_objects(copy=False), FlowAmount=lambda x: x.FlowAmount * x.conversion_factor) .drop(columns=['old_unit', 'new_unit', 'conversion_factor']) ) diff --git a/flowsa/flowby.py b/flowsa/flowby.py index b0fe3f076..50d4f0ba4 100644 --- a/flowsa/flowby.py +++ b/flowsa/flowby.py @@ -67,7 +67,7 @@ def get_flowby_from_config( external_data_path=external_data_path ) elif config.get('data_format') == 'FBS_outside_flowsa': - return FlowBySector( + return FlowBySector( # todo: add convert_df_to_flowby=True? config['FBS_datapull_fxn']( config=config, external_config_path=external_config_path, @@ -97,6 +97,8 @@ def __init__( self, data: pd.DataFrame or '_FlowBy' = None, *args, + # needs to initially be false bc otherwise all internal pandas fxns (.assign(), .copy()) trigger the code to run + convert_df_to_flowby: bool = False, add_missing_columns: bool = True, fields: dict = None, column_order: List[str] = None, @@ -105,7 +107,7 @@ def __init__( ) -> None: ''' Extends pandas DataFrame. Attaches metadata if provided as kwargs and - ensures that all columns described in flowby_config.yaml are present + ensures that all columns described in flowby_config.yaml are present and of the correct datatype. All args and kwargs not specified above or in FBA/FBS metadata are @@ -124,9 +126,9 @@ def __init__( self.__annotations__.get(attribute, None)())) ) - + # only runs if truly a pandas df and not retriggered due to .copy() or .assign() if isinstance(data, pd.DataFrame) and fields is not None: - if add_missing_columns: + if convert_df_to_flowby and add_missing_columns: data = data.assign(**{field: None for field in fields if field not in data.columns}) @@ -144,12 +146,16 @@ def __init__( for field, dtype in fields.items() if dtype == 'object' } - data = (data - .fillna(fill_na_dict) - .replace(null_string_dict) - .astype(fields)) + # avoid warning: "Downcasting object dtype arrays on .fillna, .ffill, .bfill is deprecated + # and will change in a future version" + with pd.option_context('future.no_silent_downcasting', True): + data = (data + .fillna(fill_na_dict).infer_objects(copy=False) + .replace(null_string_dict).infer_objects(copy=False) + .astype(fields) + ) - if isinstance(data, pd.DataFrame) and column_order is not None: + if isinstance(data, pd.DataFrame) and column_order is not None and convert_df_to_flowby: data = data[[c for c in column_order if c in data.columns] + [c for c in data.columns if c not in column_order]] super().__init__(data, *args, **kwargs) @@ -264,7 +270,7 @@ def _getFlowBy( else: log.error(f'{file_metadata.name_data} {file_metadata.category} ' f'could not be found locally, downloaded, or generated') - fb = cls(df, full_name=full_name or '', config=config or {}) + fb = cls(df, full_name=full_name or '', config=config or {}, convert_df_to_flowby=True) return fb def convert_daily_to_annual(self: FB) -> FB: @@ -289,21 +295,25 @@ def standardize_units(self: FB, year: int = None) -> FB: conversion_table = pd.concat([ pd.read_csv(settings.datapath / 'unit_conversion.csv'), - pd.Series({'old_unit': 'Canadian Dollar', - 'new_unit': 'USD', - 'conversion_factor': 1 / exchange_rate}).to_frame().T + pd.DataFrame([{ + 'old_unit': 'Canadian Dollar', + 'new_unit': 'USD', + 'conversion_factor': 1 / exchange_rate + }]) ]) - - standardized = ( - self - .assign(Unit=self.Unit.str.strip()) - .merge(conversion_table, how='left', - left_on='Unit', right_on='old_unit') - .assign(Unit=lambda x: x.new_unit.mask(x.new_unit.isna(), x.Unit), - conversion_factor=lambda x: x.conversion_factor.fillna(1), - FlowAmount=lambda x: x.FlowAmount * x.conversion_factor) - .drop(columns=['old_unit', 'new_unit', 'conversion_factor']) - ) + # avoid warning: "Downcasting object dtype arrays on .fillna, .ffill, .bfill is deprecated + # and will change in a future version" + with pd.option_context('future.no_silent_downcasting', True): + standardized = ( + self + .assign(Unit=self.Unit.str.strip()) + .merge(conversion_table, how='left', + left_on='Unit', right_on='old_unit') + .assign(Unit=lambda x: x.new_unit.mask(x.new_unit.isna(), x.Unit), + conversion_factor=lambda x: x.conversion_factor.fillna(1), + FlowAmount=lambda x: x.FlowAmount * x.conversion_factor) + .drop(columns=['old_unit', 'new_unit', 'conversion_factor']) + ) standardized_units = list(conversion_table.new_unit.unique()) @@ -492,7 +502,7 @@ def select_by_fields( if k not in ['Activity', 'Sector'] } - filtered_fb = self + filtered_fb = self.copy() for field, values in special_fields.items(): check_values = ([*values.keys(), *values.values()] if isinstance(values, dict) else values) @@ -527,6 +537,7 @@ def select_by_fields( replaced_fb = ( filtered_fb .replace(replace_dict) + .infer_objects(copy=False) .drop(columns=['PrimaryActivity', 'PrimarySector'], errors='ignore') .reset_index(drop=True) @@ -591,7 +602,7 @@ def aggregate_flowby( **{f'_{c}_weights': fb.FlowAmount * fb[c].notnull() for c in columns_to_average}) .groupby(columns_to_group_by, dropna=False) - .agg(sum) + .agg("sum") .reset_index() ) aggregated = ( @@ -657,8 +668,8 @@ def attribute_flows_to_sectors( validate = True grouped: 'FB' = ( self - .reset_index(drop=True).reset_index() - .rename(columns={'index': 'group_id'}) + .reset_index(drop=True) + .reset_index(names='group_id') .assign(group_total=lambda x: x.FlowAmount) ) if len(grouped) == 0: @@ -848,7 +859,8 @@ def activity_sets(self) -> List['FB']: if k not in ['activity_sets', 'clean_fba_before_activity_sets'] and not k.startswith('_')} - parent_df = self.reset_index().rename(columns={'index': 'row'}) + parent_df = (self + .reset_index(names='row')) child_df_list = [] assigned_rows = set() @@ -1379,6 +1391,11 @@ def equally_attribute(self: 'FB') -> 'FB': groupby_cols = ['group_id', 'Location'] for rank in ['Primary', 'Secondary']: + # continue if values are all np.nan + if fba[f'{rank}Sector'].isna().all(): + groupby_cols.append(f'{rank}Sector') + continue + fba = ( fba .merge(naics_key, how='left', left_on=f'{rank}Sector', @@ -1603,7 +1620,7 @@ def astype(self: FB, *args, **kwargs) -> FB: metadata = {attribute: self.__getattr__(attribute) for attribute in self._metadata} df = pd.DataFrame(self).astype(*args, **kwargs) - fb = type(self)(df, add_missing_columns=False, **metadata) + fb = type(self)(df, convert_df_to_flowby=True, add_missing_columns=False, **metadata) return fb diff --git a/flowsa/flowbyactivity.py b/flowsa/flowbyactivity.py index 61e283d69..8349be9e2 100644 --- a/flowsa/flowbyactivity.py +++ b/flowsa/flowbyactivity.py @@ -322,15 +322,15 @@ def convert_to_geoscale( fba_at_source_geoscale = ( fba_with_reporting_levels - .assign(source_geoscale=( - fba_with_reporting_levels[reporting_level_columns] - .max(axis='columns'))) - # ^^^ max() with axis='columns' takes max along rows + .assign(source_geoscale= + fba_with_reporting_levels[reporting_level_columns].apply( + lambda row: max((v for v in row if isinstance(v, geo.scale)), default=np.nan), + axis=1)) .query('geoscale == source_geoscale') .drop(columns=(['geoscale', *geoscale_name_columns, *reporting_level_columns])) - ) + ).reset_index(drop=True) if len(fba_at_source_geoscale.source_geoscale.unique()) > 1: log.warning(f"{fba_at_source_geoscale.full_name} has multiple " @@ -369,7 +369,7 @@ def convert_to_geoscale( # ^^^ TODO: Rewrite validation to use fb metadata ) - return fba_at_target_geoscale + return fba_at_target_geoscale.reset_index(drop=True) def map_to_sectors( self: 'FlowByActivity', @@ -533,10 +533,13 @@ def map_to_sectors( log.warning('Activities in %s are not mapped to sectors: %s', not_mapped.full_name, sorted( set(not_mapped.ActivityProducedBy.dropna()).union(set(not_mapped.ActivityConsumedBy.dropna())))) - # drop all NA data - fba_w_naics = (fba_w_naics - .dropna(subset=['SectorProducedBy', 'SectorConsumedBy'], how='all') - .assign(SectorSourceName=f'NAICS_{target_year}_Code') + # drop all NA data and clean up df + fba_w_naics = fba_w_naics[ + ~(fba_w_naics['SectorProducedBy'].isna() & fba_w_naics['SectorConsumedBy'].isna()) + ] + + fba_w_naics2 = (fba_w_naics + .assign(SectorSourceName=f'NAICS_{target_year}_Code') .drop(columns=['TechnologicalCorrelation_x', 'TechnologicalCorrelation_y', 'DataReliability_x', 'DataReliability_y', 'DataCollection_x', 'DataCollection_y'], @@ -544,7 +547,7 @@ def map_to_sectors( .reset_index(drop=True) ) - return fba_w_naics + return fba_w_naics2 def prepare_fbs( @@ -584,7 +587,7 @@ def prepare_fbs( .reset_index(drop=True) ) except ValueError: - return FlowBySector(pd.DataFrame()) + return FlowBySector(pd.DataFrame(), convert_df_to_flowby=True) log.info(f'Processing FlowBySector for {self.full_name}') # Primary FlowBySector generation approach: return FlowBySector( @@ -603,7 +606,8 @@ def prepare_fbs( download_sources_ok=download_sources_ok) # recursive call to prepare_fbs .drop(columns=drop_cols) .aggregate_flowby() - .function_socket('clean_fbs_after_aggregation') + .function_socket('clean_fbs_after_aggregation'), + convert_df_to_flowby=True ) def activity_sets(self) -> List['FlowByActivity']: @@ -626,7 +630,9 @@ def activity_sets(self) -> List['FlowByActivity']: if k not in ['activity_sets', 'clean_fba_before_activity_sets'] and not k.startswith('_')} - parent_fba = self.reset_index().rename(columns={'index': 'row'}) + parent_fba = (self + .reset_index(names='row') + ) child_fba_list = [] assigned_rows = set() diff --git a/flowsa/flowbyclean.py b/flowsa/flowbyclean.py index fd35d22c4..8015458cb 100644 --- a/flowsa/flowbyclean.py +++ b/flowsa/flowbyclean.py @@ -56,7 +56,7 @@ def load_prepare_clean_source( **config}, download_sources_ok=download_sources_ok ).prepare_fbs(download_sources_ok=download_sources_ok) - return clean_fbs + return clean_fbs.reset_index(drop=True) def weighted_average( @@ -86,31 +86,33 @@ def weighted_average( else 'Location'], right_on=['PrimarySector', 'Location'], suffixes=[None, '_other']) - .fillna({'FlowAmount_other': fba['FlowAmount']}) ) + merged['FlowAmount_other'] = merged['FlowAmount_other'].mask( + merged['FlowAmount_other'] == 0, merged['FlowAmount']) + # drop rows where flow is 0 merged = merged[merged['FlowAmount'] != 0] # replace terms for original, replacement in fba.config.get( 'replacement_dictionary').items(): - merged = merged.replace({original: replacement}) - - wt_flow = (merged - .groupby(['Class', 'Flowable', 'Unit', - 'FlowType', 'ActivityProducedBy', - 'ActivityConsumedBy', 'Context', 'Location', - 'LocationSystem', 'Year', 'MeasureofSpread', - 'Spread', 'DistributionType', 'Min', 'Max', - 'DataReliability', 'DataCollection', - 'SectorProducedBy', 'ProducedBySectorType', - 'SectorConsumedBy', 'ConsumedBySectorType', - 'SectorSourceName'], - dropna=False) - .apply(lambda x: np.average(x['FlowAmount'], - weights=x['FlowAmount_other'])) - .drop(columns='FlowAmount') # original flowamounts - .reset_index(name='FlowAmount') # new, weighted flows - ) + with pd.option_context('future.no_silent_downcasting', True): + merged = (merged + .replace({original: replacement}) + .infer_objects(copy=False) + ) + + wt_flow = ( + merged + .groupby(['Class', 'Flowable', 'Unit', 'FlowType', 'ActivityProducedBy', + 'ActivityConsumedBy', 'Context', 'Location', 'LocationSystem', + 'Year', 'MeasureofSpread', 'Spread', 'DistributionType', 'Min', + 'Max', 'DataReliability', 'DataCollection', 'SectorProducedBy', + 'ProducedBySectorType', 'SectorConsumedBy', 'ConsumedBySectorType', + 'SectorSourceName'], dropna=False)[['FlowAmount', 'FlowAmount_other']] + .apply(lambda x: np.average(x['FlowAmount'], weights=x['FlowAmount_other'])) + .reset_index(name='FlowAmount') + ) + # set attributes todo: revise above code so don't lose attributes attributes_to_save = { attr: getattr(fba, attr) for attr in fba._metadata + ['_metadata'] @@ -120,8 +122,8 @@ def weighted_average( # reset dropped information wt_flow = (wt_flow - .reset_index(drop=True).reset_index() - .rename(columns={'index': 'group_id'}) + .reset_index(drop=True) + .reset_index(names='group_id') .assign(group_total=wt_flow.FlowAmount) ) @@ -147,7 +149,7 @@ def substitute_nonexistent_values( state_geo = pd.concat([ (geo.filtered_fips(fb.config['geoscale'])[['FIPS']] .assign(Location=location.US_FIPS)) - ]) + ]).reset_index(drop=True) other = (other .merge(state_geo) @@ -155,6 +157,7 @@ def substitute_nonexistent_values( .rename(columns={'FIPS': 'Location'}) ) + # todo: revise these check merge cols, expand merged = (fb .merge(other, on=list(other.select_dtypes( @@ -179,8 +182,8 @@ def substitute_nonexistent_values( merged = (merged .drop(merged.filter(regex='_y').columns, axis=1) .drop(columns=['group_id']) - .reset_index(drop=True).reset_index() - .rename(columns={'index': 'group_id'}) + .reset_index(drop=True) + .reset_index(names='group_id') .assign(group_total=merged.FlowAmount) ) @@ -291,7 +294,7 @@ def estimate_suppressed_sectors_equal_attribution( right_on='source_naics') .assign(location=fba3.Location, category=fba3.FlowName) - .replace({'FlowAmount': {0: np.nan} #, + # .replace({'FlowAmount': {0: np.nan} #, # col: {'1125 & 1129': '112X', # '11193 & 11194 & 11199': '1119X', # '31-33': '3X', @@ -302,7 +305,7 @@ def estimate_suppressed_sectors_equal_attribution( # '48': '4Y', '49': '4Y'}, # 'n4': {'1125': '112X', '1129': '112X'}, # 'n5': {'11193': '1119X', '11194': '1119X', '11199': '1119X'} - }) + # }) .dropna(subset='source_naics') .drop(columns='source_naics') ) @@ -311,16 +314,16 @@ def estimate_suppressed_sectors_equal_attribution( 'location', 'category'], verify_integrity=True) def fill_suppressed( - flows, level: int, activity + flows, level: int, activity ): parent = flows[flows[activity].str.len() == level] children = flows[flows[activity].str.len() == level + 1] - null_children = children[children['FlowAmount'].isna()] + null_children = children[children['flow_suppressed']] if null_children.empty or parent.empty: return flows else: - value = max(parent['Unattributed'][0] / len(null_children), 0) + value = max(parent['Unattributed'].iloc[0] / len(null_children), 0) # update the null children by adding the unattributed data to # the attributed data null_children = ( @@ -328,11 +331,17 @@ def fill_suppressed( .assign(FlowAmount=value+null_children['Attributed']) .assign(Unattributed=value) ) - flows.update(null_children) + flows.loc[null_children.index, ['FlowAmount', 'Unattributed']] = null_children[['FlowAmount', 'Unattributed']] return flows unsuppressed = indexed.copy() + # replace 0 values with np.nan for suppressed data to be estimated + unsuppressed['FlowAmount'] = unsuppressed['FlowAmount'].mask(unsuppressed['FlowAmount'] == 0) + unsuppressed['flow_suppressed'] = unsuppressed['FlowAmount'].isna() + + # loop through sector lengths, estimating suppressed data for level in [2, 3, 4, 5, 6]: + log.info(f"Estimating suppressed data at sector level {level}") groupcols = (["{}{}".format("n", i) for i in range(2, level+1)] + ['location', 'category']) unsuppressed = (unsuppressed @@ -340,11 +349,12 @@ def fill_suppressed( .apply(fill_suppressed, level, col) ) unsuppressed['Year'] = unsuppressed['Year'].astype('int') + aggregated = ( unsuppressed .reset_index(drop=True) .fillna({'FlowAmount': 0}) - .drop(columns=['Unattributed', 'Attributed']) + .drop(columns=['Unattributed', 'Attributed', 'flow_suppressed']) # .replace({col: {'3X': '31-33', # '4X': '44-45', # '4Y': '48-49'}}) @@ -394,8 +404,8 @@ def attribute_national_to_states(fba: FlowByActivity, **_) -> FlowByActivity: fba = ( fba .drop(columns=['group_id', 'group_total']) - .reset_index(drop=True).reset_index() - .rename(columns={'index': 'group_id'}) + .reset_index(drop=True) + .reset_index(names='group_id') .assign(group_total=fba.FlowAmount) ) @@ -541,7 +551,7 @@ def define_parentincompletechild_descendants( fba = (fba .drop(columns='group_total') .merge((fba.groupby('group_id') - .agg({'FlowAmount':sum}) + .agg({'FlowAmount': "sum"}) .rename(columns={'FlowAmount': 'group_total'}) ), on='group_id', how='left', validate='m:1') @@ -612,8 +622,8 @@ def proxy_sector_data( # break each sector into separate line fba3 = (fba2 .explode(col) - .reset_index(drop=True).reset_index() - .rename(columns={'index': 'group_id'}) + .reset_index(drop=True) + .reset_index(names='group_id') ) return fba3 diff --git a/flowsa/flowbysector.py b/flowsa/flowbysector.py index 722323516..f38ec8d62 100644 --- a/flowsa/flowbysector.py +++ b/flowsa/flowbysector.py @@ -301,7 +301,7 @@ def prepare_fbs( .reset_index(drop=True) ) except ValueError: - return FlowBySector(pd.DataFrame()) + return FlowBySector(pd.DataFrame(), convert_df_to_flowby=True) return ( self .function_socket('clean_fbs') diff --git a/flowsa/literature_values.py b/flowsa/literature_values.py index ee18e6f7d..8550aeec1 100644 --- a/flowsa/literature_values.py +++ b/flowsa/literature_values.py @@ -15,7 +15,7 @@ def get_Canadian_to_USD_exchange_rate(year): """ Return exchange rate (Canadian $/USD) - From https://www.federalreserve.gov/releases/h10/current/ on 09/07/2020 + From https://www.federalreserve.gov/releases/h10/current/ on 10/28/2025 :param year: str, year of exchange rate to return :return: number, value of exchange rate for year """ @@ -38,7 +38,12 @@ def get_Canadian_to_USD_exchange_rate(year): 2016: 1.3243, 2017: 1.2984, 2018: 1.2957, - 2019: 1.3269 + 2019: 1.3269, + 2020: 1.3422, + 2021: 1.2533, + 2022: 1.3014, + 2023: 1.3494, + 2024: 1.3699 }) exchange_rate = er.get(year, np.nan) diff --git a/flowsa/naics.py b/flowsa/naics.py index 145fd5233..4f5901db5 100644 --- a/flowsa/naics.py +++ b/flowsa/naics.py @@ -149,12 +149,14 @@ def subset_sector_key(flowbyactivity, activitycol, sector_source_year, primary_s # for situations where an activity can be listed in both columns for different circumstances subset_cols = ['Class', 'Flowable', 'Context', 'ActivityProducedBy', 'ActivityConsumedBy', 'DataReliability', 'DataCollection'] - if "DataReliability" not in flowbyactivity.columns: - subset_cols = ['Class', 'Flowable', 'Context', 'ActivityProducedBy', 'ActivityConsumedBy'] + # list DQI columns in df + dqi = [col for col in ['DataReliability', 'DataCollection'] if col in flowbyactivity.columns] + # Drop missing DQI columns from subset list + subset_cols = [col for col in subset_cols if col not in ['DataReliability', 'DataCollection'] or col in dqi] # ensure dq column decimals do not cause errors with dropping duplicates, without this statement, rows often # duplicated - flowbyactivity.loc[:, ['DataReliability', 'DataCollection']] = ( - flowbyactivity.loc[:, ['DataReliability', 'DataCollection']].round(decimals=5)) + if dqi: + flowbyactivity.loc[:, dqi] = (flowbyactivity.loc[:, dqi].round(decimals=5)) flowbyactivity = flowbyactivity[subset_cols].drop_duplicates() primary_sector_key_2 = pd.DataFrame(flowbyactivity.merge( @@ -172,10 +174,12 @@ def drop_parent_sectors(sector_key): is_parent = lambda x: any(sector != x and sector.startswith(x) for sector in sector_list) return sector_key[~sector_key['source_naics'].astype(str).apply(is_parent)] - primary_sector_key_2 = primary_sector_key_2.groupby(['Class', 'Flowable', 'Context'], - group_keys=False, - dropna=False - ).apply(drop_parent_sectors) + # todo: check futurewarning dataframegroupby.apply fix working as expected + primary_sector_key_2 = (primary_sector_key_2 + .groupby(['Class', 'Flowable', 'Context'], + group_keys=False, dropna=False)[primary_sector_key_2.columns.tolist()] + .apply(drop_parent_sectors) + ) # modify dqi scores for data reliability and collection based on mapping if "DataReliability" in flowbyactivity.columns: @@ -224,10 +228,11 @@ def subset_target_sectors_by_source_sectors(group): if flowbyactivity.config.get('sector_hierarchy') == 'parent-incompleteChild': df_remaining_mapped = df_remaining.copy() else: + # todo: check impact of changing code to remove future warning df_remaining_mapped = (df_remaining - .groupby(group_cols, dropna=False) + .groupby(group_cols, dropna=False, group_keys=False)[df_remaining.columns.tolist()] .apply(subset_target_sectors_by_source_sectors) - .reset_index(drop=True) + # .reset_index(drop=True) ) mapping = pd.concat([df_keep, df_remaining_mapped], ignore_index=True) @@ -583,6 +588,13 @@ def convert_naics_year(df_load, targetsectorsourcename, sectorsourcename, # methods, so assigning schema as None activity_schema = "None" + # however, need to ensure that these NAICS are formatted correctly - stewi data are at times imported + # with some NAICS values including decimals that do not get mapped correctly (ex. '311712.0') + for col in column_headers: + if col in df_load.columns: + df_load[col] = (df_load[col] + .apply(lambda x: x.split(".")[0] if isinstance(x, str) else x)) + if "NAICS" in activity_schema and "ActivityProducedBy" in df_load.columns: column_headers += ['ActivityProducedBy', 'ActivityConsumedBy'] diff --git a/flowsa/validation.py b/flowsa/validation.py index 912eeaa15..93a9e4311 100644 --- a/flowsa/validation.py +++ b/flowsa/validation.py @@ -77,7 +77,7 @@ def calculate_flowamount_diff_between_dfs(dfa_load, dfb_load): 'Unit', 'geoscale'] dfagg = dfs.groupby( agg_cols, dropna=False, as_index=False).agg( - {'FlowAmount_Original': sum, 'FlowAmount_Modified': sum}) + {'FlowAmount_Original': "sum", 'FlowAmount_Modified': "sum"}) # column calculating difference dfagg['FlowAmount_Difference'] = \ dfagg['FlowAmount_Modified'] - dfagg['FlowAmount_Original'] @@ -95,7 +95,7 @@ def calculate_flowamount_diff_between_dfs(dfa_load, dfb_load): 'FlowAmount_Difference', 'Percent_Increase']) dfagg4 = dfagg3.groupby(flowcols + ['Unit', 'geoscale'], dropna=False, as_index=False).agg( - {'FlowAmount_Original': sum, 'FlowAmount_Modified': sum}) + {'FlowAmount_Original': "sum", 'FlowAmount_Modified': "sum"}) # column calculating difference dfagg4['FlowAmount_Difference'] = \ dfagg4['FlowAmount_Modified'] - dfagg4['FlowAmount_Original'] @@ -295,10 +295,10 @@ def compare_FBS_results(fbs1, fbs2, ignore_metasources=False, return df_m -def compare_FBS(df1, df2, ignore_metasources=False): - "Assess differences between two FBS dataframes." - df1 = df1.rename(columns={'FlowAmount': 'FlowAmount_fbs1'}) - df2 = df2.rename(columns={'FlowAmount': 'FlowAmount_fbs2'}) +def compare_FBS(df1_load, df2_load, ignore_metasources=False): + """Assess differences between two FBS dataframes.""" + df1 = pd.DataFrame(df1_load.rename(columns={'FlowAmount': 'FlowAmount_fbs1'})) + df2 = pd.DataFrame(df2_load.rename(columns={'FlowAmount': 'FlowAmount_fbs2'})) merge_cols = [c for c in df2.select_dtypes(include=[ 'object', 'int']).columns if c not in dq_fields] if ignore_metasources: @@ -312,13 +312,7 @@ def compare_FBS(df1, df2, ignore_metasources=False): except ValueError: pass - # aggregate dfs before merge - might have duplicate sectors due to - # dropping metasources/attribution sources - df1 = (df1.groupby(merge_cols, dropna=False) - .agg({'FlowAmount_fbs1': 'sum'}).reset_index()) - df2 = (df2.groupby(merge_cols, dropna=False) - .agg({'FlowAmount_fbs2': 'sum'}).reset_index()) - # convert sector columns to object to avoid valueErrors + # convert sector columns to object to avoid valueErrors and df clean up cols = ['SectorProducedBy', 'SectorConsumedBy'] for c in cols: df1[c] = df1[c].astype(str) @@ -327,12 +321,23 @@ def compare_FBS(df1, df2, ignore_metasources=False): df1 = df1.drop(columns=c, errors='ignore') df2 = df2.drop(columns=c, errors='ignore') merge_cols = [x for x in merge_cols if x != c] + # convert all np.nan in the string type merge cols to empty strings, to ensure correct merge + fill_cols = [c for c in merge_cols if df2[c].dtype == 'object'] + df1[fill_cols] = df1[fill_cols].replace(['nan', np.nan], '') + df2[fill_cols] = df2[fill_cols].replace(['nan', np.nan], '') + + # subset dfs + df1_sub = df1[merge_cols + ['FlowAmount_fbs1']] + df2_sub = df2[merge_cols + ['FlowAmount_fbs2']] + + # aggregate dfs before merge - might have duplicate sectors due to + # dropping metasources/attribution sources + df1_sub = df1_sub.groupby(merge_cols, dropna=False).agg({'FlowAmount_fbs1': 'sum'}).reset_index() + df2_sub = df2_sub.groupby(merge_cols, dropna=False).agg({'FlowAmount_fbs2': 'sum'}).reset_index() + # check units - # compare_df_units(df1, df2) - df_m = pd.DataFrame( - pd.merge(df1[merge_cols + ['FlowAmount_fbs1']], - df2[merge_cols + ['FlowAmount_fbs2']], - how='outer')) + # compare_df_units(df1_sub, df2_sub) + df_m = pd.merge(df1_sub, df2_sub,how='outer') df_m = df_m.assign(FlowAmount_diff=df_m['FlowAmount_fbs2'] .fillna(0) - df_m['FlowAmount_fbs1'].fillna(0)) df_m = df_m.assign( @@ -373,6 +378,8 @@ def compare_single_FBS_against_remote(m, outdir=diffpath, inplace=True) if len(df) > 0: print(f"Saving differences in {m} to csv") + # maintain leading 0s in location col + df.Location = df.Location.apply('="{}"'.format) df.to_csv(f"{outdir}/{m}_diff.csv", index=False) else: print(f"***No differences found in {m}***") @@ -402,6 +409,8 @@ def compare_single_FBA_against_remote(source, year, outdir=diffpath, inplace=True) if len(df) > 0: print(f"Saving differences in {source} {year} to csv") + # maintain leading 0s in location col + df.Location = df.Location.apply('="{}"'.format) df.to_csv(f"{outdir}/{source}_{year}_diff.csv", index=False) else: print(f"***No differences found in {source} {year}***") diff --git a/requirements.txt b/requirements.txt index 5c7ede698..442cea6d1 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,22 +1,22 @@ git+https://github.com/USEPA/fedelemflowlist.git@develop#egg=fedelemflowlist git+https://github.com/USEPA/esupy.git@develop#egg=esupy git+https://github.com/USEPA/standardizedinventories.git@develop#egg=StEWI -appdirs >= 1.4.3 # Storing user data -bibtexparser>=1.2.0 # Generates bibtex +appdirs >= 1.4.4 # Storing user data +bibtexparser>=1.4.3 # Generates bibtex kaleido == 0.1.0.post1; platform_system == "Windows" # export plotly.py figures as static images kaleido == 0.2.0; platform_system=="Linux" or platform_system=="Darwin" -matplotlib >= 3.4.3 # Plotting -numpy>=1.20.1, <2.0.0 # Library used for arrays -openpyxl >= 3.0.7 # Extract data from xlsm files -pandas>=1.4.0, <2.1.0 # Powerful data structures for data analysis, time series, and statistics. -pip>=9 # The PyPA recommended tool for installing Python packages. -plotly >= 5.10.0 # Plotting -pycountry >= 19.8.18 # ISO country codes -python-dotenv >= 0.19.1 # Reads .env files -pyyaml>=5.3 # Yaml for python -requests >=2.22.0 # Web service calls +matplotlib >= 3.10.3 # Plotting +numpy>=2.2.6 # Library used for arrays +openpyxl >= 3.1.5 # Extract data from xlsm files +pandas>=2.3.0 # Powerful data structures for data analysis, time series, and statistics. +pip>=25.1.1 # The PyPA recommended tool for installing Python packages. +plotly >= 6.1.2 # Plotting +pycountry >= 24.6.1 # ISO country codes +python-dotenv >= 1.1.0 # Reads .env files +pyyaml>=6.0.2 # Yaml for python +requests >=2.32.4 # Web service calls requests_ftp == 0.3.1 # Requests implementation for FTP -seaborn >= 0.11.2 # Plotting -setuptools>=41 # Fully-featured library designed to facilitate packaging Python projects. -tabula-py >= 2.1.1 # PDF reader +seaborn >= 0.13.2 # Plotting +setuptools>=80.9.0 # Fully-featured library designed to facilitate packaging Python projects. +tabula-py >= 2.10.0 # PDF reader xlrd >= 2.0.1 # Extract data from xls files diff --git a/setup.py b/setup.py index 1b250b1e5..cf6742f9f 100644 --- a/setup.py +++ b/setup.py @@ -10,29 +10,29 @@ packages=find_packages(), package_dir={'flowsa': 'flowsa'}, include_package_data=True, - python_requires=">=3.9", + python_requires=">=3.10", install_requires=[ 'fedelemflowlist @ git+https://github.com/USEPA/fedelemflowlist.git@develop#egg=fedelemflowlist', 'esupy @ git+https://github.com/USEPA/esupy.git@develop#egg=esupy', 'StEWI @ git+https://github.com/USEPA/standardizedinventories.git@develop#egg=StEWI', - 'appdirs>=1.4.3', - 'bibtexparser>=1.2.0', + 'appdirs>=1.4.4', + 'bibtexparser>=1.4.3', "kaleido==0.1.0.post1;platform_system=='Windows'", "kaleido==0.2.0;platform_system=='Linux' or platform_system=='Darwin'", - 'matplotlib>=3.4.3', - 'numpy>=1.20.1, <2.0.0', - 'openpyxl>=3.0.7', - 'pandas>=1.4.0, <2.1.0', - 'pip>=9', - 'plotly>=5.10.0 ', - 'pycountry>=19.8.18', - 'python-dotenv >= 0.19.1', - 'pyyaml>=5.3', - 'requests>=2.22.0', + 'matplotlib>=3.10.3', + 'numpy>=2.2.6', + 'openpyxl>=3.1.5', + 'pandas>=2.3.0', + 'pip>=25.1.1', + 'plotly>= 6.1.2', + 'pycountry>=24.6.1', + 'python-dotenv >= 1.1.0', + 'pyyaml>=6.0.2', + 'requests>=2.32.4', 'requests_ftp==0.3.1', - 'seaborn>=0.11.2', - 'setuptools>=41', - 'tabula-py>=2.1.1', + 'seaborn>=0.13.2', + 'setuptools>=80.9.0', + 'tabula-py>=2.10.0', 'xlrd>=2.0.1' ], url='https://github.com/USEPA/FLOWSA',