Skip to content

Improve performance #336

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bl-young wants to merge 2 commits into
base: development
Choose a base branch
from
Open

Improve performance #336

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c20699e
refactor calculate_electricity_by_source for performance
bl-young Jun 9, 2026
faba841
avoid MemoryErorr in upstream flow mapping by splitting and deduplica…
bl-young Jun 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 56 additions & 9 deletions electricitylci/combinator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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'])
Expand Down Expand Up @@ -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"]]
Expand Down
117 changes: 66 additions & 51 deletions electricitylci/generation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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"]
Expand All @@ -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(
Expand Down
Loading