From c20699e0e6c50dc1b299480124c01c7c23f64062 Mon Sep 17 00:00:00 2001 From: Ben Young Date: Tue, 9 Jun 2026 08:54:39 -0400 Subject: [PATCH 1/2] refactor calculate_electricity_by_source for performance --- electricitylci/generation.py | 117 ++++++++++++++++++++--------------- 1 file changed, 66 insertions(+), 51 deletions(-) diff --git a/electricitylci/generation.py b/electricitylci/generation.py index 2c946f4..99629cb 100644 --- a/electricitylci/generation.py +++ b/electricitylci/generation.py @@ -163,6 +163,41 @@ def _calc_geom_params(p_series): return (d['mu_g'], d['sigma_g']) +def _source_string_for_group(s, source_limit=None): + """Build underscore-joined source string for a grouped Source series.""" + sources = sorted(s.dropna().unique()) + if source_limit is not None and len(sources) > source_limit: + return np.nan + if len(sources) == 0: + return np.nan + return "_".join(sources) + + +def _source_table_from_groupby(df, keys, source_limit=None): + """Faster replacement for groupby(Source).apply(_combine_sources).""" + logging.info( + " grouping %d rows by %s", + len(df), + keys if isinstance(keys, str) else list(keys), + ) + t0 = datetime.now() + out = ( + df.groupby(keys, sort=False)["Source"] + .agg(lambda s: _source_string_for_group(s, source_limit=source_limit)) + .rename("source_string") + .reset_index() + ) + out["source_list"] = out["source_string"].apply( + lambda x: x.split("_") if isinstance(x, str) else np.nan + ) + logging.info( + " finished in %s (%d groups)", + datetime.now() - t0, + len(out), + ) + return out + + def _combine_sources(p_series, df, cols, source_limit=None): """Take the sources from a groupby.apply and return a list that contains one column containing a list of the sources and another @@ -749,69 +784,52 @@ def calculate_electricity_by_source(db, subregion="BA"): db_cols = list(db_powerplant.columns) + ['source_list', 'source_string'] db_powerplant = pd.DataFrame(columns=db_cols) else: - # This is a pretty expensive process when we have to start looking - # at each flow generated in each compartment for each balancing - # authority area. To hopefully speed this up, we'll group by FlowName - # and Compartment and look and try to eliminate flows where all - # sources are single entities. - combine_source_by_flow = lambda x: _combine_sources( - x, db, ["FlowName", "Compartment"], 1 + # Pass 1: flows with a single inventory source across all plants. + logging.info( + "Assigning source labels (pass 1/2: FlowName + Compartment)" ) - # Find all single-source flows (all multiple sources are nans) - source_df = pd.DataFrame( - db_powerplant.groupby(["FlowName", "Compartment"])[ - ["Source"]].apply(combine_source_by_flow), - columns=["source_list"], + source_df = _source_table_from_groupby( + db_powerplant, + ["FlowName", "Compartment"], + source_limit=1, ) - source_df[["source_list", "source_string"]] = pd.DataFrame( - source_df["source_list"].values.tolist(), - index=source_df.index - ) - source_df.reset_index(inplace=True) old_index = db_powerplant.index db_powerplant = db_powerplant.merge( right=source_df, - left_on=["FlowName", "Compartment"], - right_on=["FlowName", "Compartment"], + on=["FlowName", "Compartment"], how="left", ) db_powerplant.index = old_index - # Filter out single flows; leaving only multi-flows + # Pass 2: rows where source mix varies by region/fuel/stage. db_multiple_sources = db_powerplant.loc[ - db_powerplant["source_string"].isna(), :].copy() + db_powerplant["source_string"].isna(), : + ].copy() if len(db_multiple_sources) > 0: - combine_source_lambda = lambda x: _combine_sources( - x, db_multiple_sources, groupby_cols - ) - # HOTFIX: it doesn't make sense to groupby a different group; - # it gives different results from the first-pass filter; - # changed to match criteria above. [2023-12-19; TWD] - # HOTFIX undone [2024-08-13; MBJ] - source_df = pd.DataFrame( - db_multiple_sources.groupby(groupby_cols)[ - ["Source"]].apply(combine_source_lambda), - columns=["source_list"], + logging.info( + "Assigning source labels (pass 2/2: %d rows need " + "region-level source resolution)", + len(db_multiple_sources), ) - source_df[["source_list", "source_string"]] = pd.DataFrame( - source_df["source_list"].values.tolist(), - index=source_df.index + source_df = _source_table_from_groupby( + db_multiple_sources, + groupby_cols, + source_limit=None, ) - source_df.reset_index(inplace=True) - db_multiple_sources.drop( - columns=["source_list", "source_string"], inplace=True + db_multiple_sources = db_multiple_sources.drop( + columns=["source_list", "source_string"], + errors="ignore", ) old_index = db_multiple_sources.index db_multiple_sources = db_multiple_sources.merge( right=source_df, - left_on=groupby_cols, - right_on=groupby_cols, + on=groupby_cols, how="left", ) db_multiple_sources.index = old_index db_powerplant.loc[ db_powerplant["source_string"].isna(), - ["source_string", "source_list"] + ["source_string", "source_list"], ] = db_multiple_sources[["source_string", "source_list"]] unique_source_lists = list(db_powerplant["source_string"].unique()) unique_source_lists = [x for x in unique_source_lists if str(x) != "nan"] @@ -820,16 +838,13 @@ def calculate_electricity_by_source(db, subregion="BA"): # used as proxies for Canadian generation. In those cases the electricity # generation will be equal to the Electricity already in the dataframe. elec_sum_lists = list() - for src in unique_source_lists: - logging.info(f"Calculating electricity for {src}") - db["temp_src"] = src - src_filter = [ - a in b - for a, b in zip( - db["Source"].values.tolist(), db["temp_src"].values.tolist() - ) - ] - sub_db = db.loc[src_filter, :].copy() + n_src = len(unique_source_lists) + for i, src in enumerate(unique_source_lists, start=1): + logging.info( + "Calculating electricity for %s (%d/%d)", src, i, n_src + ) + src_parts = src.split("_") + sub_db = db[db["Source"].isin(src_parts)].copy() sub_db.drop_duplicates(subset=fuel_agg + ["eGRID_ID","Year"], inplace=True) # HOTFIX: fix pandas futurewarning syntax [2024-03-08; TWD] sub_db_group = sub_db.groupby(elec_groupby_cols, as_index=False).agg( From faba841cecdc3415ad866aaf829a5d077e8e6d56 Mon Sep 17 00:00:00 2001 From: Ben Young Date: Tue, 9 Jun 2026 08:57:35 -0400 Subject: [PATCH 2/2] avoid MemoryErorr in upstream flow mapping by splitting and deduplicating --- electricitylci/combinator.py | 65 +++++++++++++++++++++++++++++++----- 1 file changed, 56 insertions(+), 9 deletions(-) diff --git a/electricitylci/combinator.py b/electricitylci/combinator.py index 44efb7d..fabf291 100644 --- a/electricitylci/combinator.py +++ b/electricitylci/combinator.py @@ -51,6 +51,49 @@ # FUNCTIONS ############################################################################## +def _build_flowmapping_small_for_merge( + flow_mapping: pd.DataFrame, + elci_flow_mapping: pd.DataFrame, +) -> pd.DataFrame: + """Build merge keys for upstream flow mapping without row blow-up. + + Issue #274 appends the full FEDEFL flowlist. Many keys then have both an + eLCI row and a flowlist self-map with a different TargetFlowName. Merging + upstream on (SourceFlowName, SourceFlowContext) alone multiplies rows and + triggers ArrayMemoryError on large upstream inventories. + + For keys with intentional eLCI 1→many splits (e.g. sand and gravel → + Sand + Gravel), retain every eLCI target row. For all other keys, keep one + mapping row per merge key (eLCI wins because it is concatenated first). + """ + merge_keys = ["SourceFlowName", "SourceFlowContext"] + elci = elci_flow_mapping.copy() + elci["SourceFlowName"] = elci["SourceFlowName"].str.lower() + elci_triple = elci[merge_keys + ["TargetFlowName"]].drop_duplicates() + split_keys = ( + elci.groupby(merge_keys, observed=True) + .size() + .reset_index(name="n") + .query("n > 1")[merge_keys] + ) + + fm_idx = flow_mapping.reset_index().rename(columns={"index": "orig_fm_index"}) + + if split_keys.empty: + return fm_idx.drop_duplicates(subset=merge_keys, keep="first")[ + merge_keys + ["orig_fm_index"] + ] + + tagged = fm_idx.merge(split_keys.assign(_split=True), on=merge_keys, how="left") + split_part = tagged[tagged["_split"].eq(True)].merge( + elci_triple, on=merge_keys + ["TargetFlowName"], how="inner" + )[merge_keys + ["orig_fm_index"]] + nonsplit_part = tagged[tagged["_split"].isna()].drop_duplicates( + subset=merge_keys, keep="first" + )[merge_keys + ["orig_fm_index"]] + return pd.concat([split_part, nonsplit_part], ignore_index=True) + + def add_fuel_inputs(gen_df, upstream_df, upstream_dict): """Convert the upstream emissions database to fuel inputs and add them to the generator data frame. @@ -260,12 +303,13 @@ def concat_map_upstream_databases(eia_gen_year, *arg, **kwargs): # standard units (e.g., kg, MJ, m2*a). Note that 'SourceFlowContext' is # already in lowercase letters, which is why no change happens below. logging.info("Creating flow mapping database") - flow_mapping = fedefl.get_flowmapping('eLCI') + elci_flow_mapping = fedefl.get_flowmapping('eLCI') + flow_mapping = elci_flow_mapping - # as hotfix for https://github.com/NETL-RIC/ElectricityLCI/issues/274 - # append full flowlist to the flow mapping file (dropping duplicates) - # to catch any other mappings of flows that use the same name as already - # in the flow list + # Issue #274: append full flowlist, then dedupe on + # (SourceFlowName, SourceFlowContext, TargetFlowName) so intentional eLCI + # 1:many splits (e.g. sand and gravel) are preserved. Piecewise merge below + # avoids ArrayMemoryError. flowlist = (fedefl.get_flows() .filter(['Flowable', 'Context', 'Unit', 'Flow UUID']) .assign(SourceFlowName = lambda x: x['Flowable']) @@ -372,10 +416,13 @@ def concat_map_upstream_databases(eia_gen_year, *arg, **kwargs): # dataframe to a slice of upstream_df_grp, so that we only copy a small # dataframe during the merge. Then we can used the matched indeces to assign # column values later on. - flowmapping_small = ( - flow_mapping[["SourceFlowName", "SourceFlowContext"]] - .reset_index() - .rename(columns={"index": "orig_fm_index"}) + flowmapping_small = _build_flowmapping_small_for_merge( + flow_mapping, elci_flow_mapping + ) + logging.info( + "Flow mapping merge keys: %d (from %d mapping rows)", + len(flowmapping_small), + len(flow_mapping), ) upstream_df_grp_small = ( upstream_df_grp[["FlowName", "Compartment_path"]]