"""Dagster assets for cross-dataset glue tables.
This module defines assets that build the association tables linking related records
across multiple datasets, including FERC, EIA, and EPA CAMD. Put asset definitions here
when they create or refine those crosswalk-style tables and supporting relationships,
rather than the domain-specific transforms for any one source dataset.
"""
import collections
from itertools import repeat
import dagster as dg
import networkx as nx
import pandas as pd
import pudl.glue.ferc1_eia
import pudl.glue.ferc714
import pudl.helpers
import pudl.logging_helpers
from pudl.metadata.classes import DataSource, Package
from pudl.metadata.dtypes import apply_pudl_dtypes
[docs]
logger = pudl.logging_helpers.get_logger(__name__)
@dg.multi_asset(
outs={
table_name: dg.AssetOut(io_manager_key="parquet_io_manager", is_required=False)
for table_name in Package.get_etl_group_tables("glue")
# do not load core_epa__assn_eia_epacamd glue assets bc they are stand-alone assets below.
if "core_epa__assn_eia_epacamd" not in table_name
},
can_subset=True,
required_resource_keys={"datastore", "global_data_config"},
)
[docs]
def create_glue_tables(context):
"""Extract, transform and load CSVs for the FERC-EIA Glue tables.
Args:
context: dagster keyword that provides access to resources and config.
core_eia__entity_generators: Static generator attributes compiled from across the EIA-860 and EIA-923 data.
core_eia__entity_boilers: core_eia__entity_boilers.
Returns:
A dictionary of DataFrames whose keys are the names of the corresponding
database table.
"""
# TODO 2024-09-23: double check if these settings are actually
# doing anything for the FERC-EIA glue... doesn't look like it.
pudl_data_config = context.resources.global_data_config.pudl
# grab the glue tables for ferc1 & eia
glue_dfs = pudl.glue.ferc1_eia.glue(
ferc1=pudl_data_config.glue.ferc1,
eia=pudl_data_config.glue.eia,
)
# these 714 glue tables are so easy to build, it doesn't seem worth it
# to not build/load them if we are not etl-ing 714
glue_dfs = glue_dfs | pudl.glue.ferc714.glue()
# Ensure they are sorted so they match up with the asset outs
glue_dfs = dict(sorted(glue_dfs.items()))
selected_outputs = set(context.selected_output_names)
return (
dg.Output(output_name=table_name, value=df)
for table_name, df in glue_dfs.items()
if table_name in selected_outputs
)
#####################
# EPA - EIA Crosswalk
#####################
@dg.asset(required_resource_keys={"datastore"})
[docs]
def raw_pudl__assn_eia_epacamd(context) -> pd.DataFrame:
"""Extract the EPACAMD-EIA Crosswalk from the Datastore."""
logger.info("Extracting the EPACAMD-EIA crosswalk from Zenodo")
csv_map = {2018: "camd-eia-crosswalk-master/epa_eia_crosswalk.csv"} | {
year: f"camd-eia-crosswalk-latest-{year}/epa_eia_crosswalk.csv"
for year in DataSource.from_id("epacamd_eia").working_partitions["years"]
if year != 2018
}
ds = context.resources.datastore
year_matches = []
for year, csv_path in csv_map.items():
with ds.get_zipfile_resource("epacamd_eia", year=year).open(csv_path) as f:
df = pd.read_csv(f)
df["report_year"] = year
year_matches.append(df)
return pd.concat(year_matches, ignore_index=True)
@dg.asset(
required_resource_keys={"global_data_config"}, io_manager_key="parquet_io_manager"
)
[docs]
def core_epa__assn_eia_epacamd(
context,
raw_pudl__assn_eia_epacamd: pd.DataFrame,
core_eia__entity_generators: pd.DataFrame,
core_eia__entity_boilers: pd.DataFrame,
) -> pd.DataFrame:
"""Clean up the EPACAMD-EIA Crosswalk file.
In its raw form, the crosswalk contains many fields. The transform process removes
descriptive fields like state, location, facility name, capacity, operating status,
and fuel type that can be found by linking this dataset to others already in the
database. We're primarily concerned with linking ids in this table, not including
any other plant information.
The raw file contains several fields with the prefix ``MOD``. These fields are used
to create a single ID for matches with different spelling. For example,
``EIA_PLANT_ID`` (``plant_id_eia`` in PUDL) has the generator ID ``CTG5`` in EPA and
``GT5`` in EIA. The ``MOD`` columns for both of these generators
(``MOD_CAMD_GENERATOR_ID``, ``MOD_EIA_GENERATOR_ID_GEN``) converts that value to 5.
The ``MATCH_TYPE_GEN``, ``MATCH_TYPE_BOILER``, and ``PLANT_ID_CHANGE_FLAG`` fields
indicate whether these ``MOD`` fields contain new information. Because we're not
concerned with creating a new, modified ID field for either EPA or EIA data, we
don't need these ``MOD`` or ``MATCH_TYPE`` columns in our final output. We just care
which EPA value maps to which EIA value.
In terms of cleaning, we implement the standard column name changes: lower-case, no
special characters, and underscores instead of spaces. We also rename some of the
columns for clarity and to match how they appear in the tables you will merge with.
Besides standardizing datatypes (again for merge compatibility) the only meaningful
data alteration we employ here is removing leading zeros from numeric strings on
the ``generator_id`` and ``emissions_unit_id_epa`` fields. This is because the same
function is used to clean those same fields in all the other tables in which they
appear. In order to merge properly, we need to clean the values in the crosswalk the
same way. Lastly, we drop all rows without ``EIA_PLANT_ID`` (``plant_id_eia``)
values because that means that they are unmatched and do not provide any useful
information to users.
It's important to note that the crosswalk is kept intact (and not separated into
smaller reference tables) because the relationship between the ids is not 1:1. For
example, you can't isolate the plant_id fields, drop duplicates, and call it a day.
The plant discrepancies depend on which generator ids it's referring to. This goes
for all fields. Be careful, and do some due diligence before eliminating columns.
We talk more about the complexities regarding EPA "units" in our :doc:`Data Source
documentation page for EPACEMS </data_sources/epacems>`.
In it's original format, the crosswalk is a static file - however, we manually
run the crosswalk code for each year of EIA data, adding the report_date field
to the crosswalk. The plant_id_eia and generator_id fields are foreign keys from an
annualized table. If the fast ETL is run (on one year of data) the test will break
because the crosswalk tables with ``plant_id_eia`` and ``generator_id`` contain
values from various years. To keep the crosswalk in alignment with the available eia
data, we'll restrict it based on the generator entity table which has
``plant_id_eia`` and ``generator_id`` so long as it's not using the full suite of
available years. If it is, we don't want to restrict the crosswalk so we can get
warnings and errors from any foreign key discrepancies. This isn't an ideal
solution, but it works for now.
Args:
context: dagster keyword that provides access to resources and config. For this
asset, this determines whether the years from the Eia860DataConfig object
match the EIA860 working partitions. This indicates whether or not to
restrict the crosswalk data so the tests don't fail on foreign key
restraints.
raw_pudl__assn_eia_epacamd: The result of running this module's extract() function.
core_eia__entity_generators: The core_eia__entity_generator table.
core_eia__entity_boilers: The core_eia__entity_boilerstable.
Returns:
A dictionary containing the cleaned EPACAMD-EIA crosswalk DataFrame.
"""
logger.info("Transforming the EPACAMD-EIA crosswalk")
column_rename = {
"report_year": "report_year",
"camd_plant_id": "plant_id_epa",
"camd_unit_id": "emissions_unit_id_epa",
"camd_generator_id": "generator_id_epa",
"eia_plant_id": "plant_id_eia",
"eia_boiler_id": "boiler_id", # Eventually change to boiler_id_eia
"eia_generator_id": "generator_id", # Eventually change to generator_id_eia
}
# Basic column rename, selection, and dtype alignment.
crosswalk_clean = (
raw_pudl__assn_eia_epacamd.pipe(pudl.helpers.simplify_columns)
.rename(columns=column_rename)
.filter(list(column_rename.values()))
.pipe(
pudl.helpers.remove_leading_zeros_from_numeric_strings,
col_name="generator_id",
)
.pipe(
pudl.helpers.remove_leading_zeros_from_numeric_strings, col_name="boiler_id"
)
.pipe(
pudl.helpers.remove_leading_zeros_from_numeric_strings,
col_name="emissions_unit_id_epa",
)
.pipe(apply_pudl_dtypes, field_namespace="eia")
.dropna(subset=["plant_id_eia"])
.pipe(correct_epa_eia_plant_id_mapping)
)
pudl_data_config = context.resources.global_data_config.pudl
processing_all_eia_years = (
pudl_data_config.eia.eia860.years
== pudl_data_config.eia.eia860.data_source.working_partitions["years"]
)
# Restrict crosswalk for tests if running fast etl
if not processing_all_eia_years:
logger.info(
"Selected subset of available EIA years--restricting EPACAMD-EIA Crosswalk \
to chosen subset of EIA years"
)
crosswalk_clean = pd.merge(
crosswalk_clean,
core_eia__entity_generators[["plant_id_eia", "generator_id"]],
on=["plant_id_eia", "generator_id"],
how="inner",
validate="m:1",
)
crosswalk_clean = pd.merge(
crosswalk_clean,
core_eia__entity_boilers[["plant_id_eia", "boiler_id"]],
on=["plant_id_eia", "boiler_id"],
how="inner",
validate="m:1",
)
# TODO: Add manual crosswalk cleanup from @grgmiller
return crosswalk_clean
@dg.asset
[docs]
def _core_epa__assn_eia_epacamd_unique(
core_epa__assn_eia_epacamd: pd.DataFrame,
) -> pd.DataFrame:
"""Intermediate asset that contains all unique core_epa__assn_eia_epacamd matches.
The core_epa__assn_eia_epacamd asset contains crosswalk matches from 2018 through
the latest full year of EIA 860 data. This means there are many duplicate matches
found from both years. Several downstream assets expect these matches to be unique,
so this asset will drop duplicates to serve as the input to those downstream assets.
This asset, however, will not itself be written to the PUDL DB. This asset will also
address conflicting matches by taking the match from the most recent year.
Args:
core_epa__assn_eia_epacamd: Cleaned crosswalk with duplicate matches.
Returns:
Cleaned crosswalk with duplicates removed.
"""
# Drop fully duplicated matches
core_epa__assn_eia_epacamd = core_epa__assn_eia_epacamd.drop_duplicates(
subset=core_epa__assn_eia_epacamd.columns.difference(["report_year"])
)
# Find mismatches where there are different plant_id_eia values between years for
# the same plant_id_epa and emissions_unit_id_epa value. Keep the value from
# the most recent year.
one_to_many = (
core_epa__assn_eia_epacamd.sort_values("report_year")
.groupby(["plant_id_epa", "emissions_unit_id_epa"])
.filter(
lambda x: (
x.plant_id_eia.nunique() > 1 # noqa: PD101
and x.report_year.nunique() > 1 # noqa: PD101
)
)
)
logger.info(f"The following crosswalk matches are duplicated: \n{one_to_many}")
if (
not one_to_many.empty
): # When running the fast ETL, there are no rows of data here.
# Assert some expectations about the duplicated matches
assert len(one_to_many) <= 8, (
f"{len(one_to_many)} rows found with changes in matches over time."
)
assert one_to_many.plant_id_eia.unique() == 63628
# Check there are only two years of data, so we can drop one below
assert one_to_many.report_year.unique() == (2018, 2019)
# For this one plant, we drop the first year of data and keep 2019 records.
return core_epa__assn_eia_epacamd.drop(
one_to_many[one_to_many.report_year == 2018].index
).drop(["report_year"], axis=1)
[docs]
def correct_epa_eia_plant_id_mapping(df: pd.DataFrame) -> pd.DataFrame:
"""Manually correct one plant ID.
The EPA's power sector data crosswalk incorrectly maps plant_id_epa 55248 to
plant_id_eia 55248, when it should be mapped to id 2847.
"""
df.loc[df["plant_id_eia"] == 55248, "plant_id_eia"] = 2847
return df
##############################
# Add EPACAMD-EIA Sub-Plant ID
##############################
@dg.asset(io_manager_key="parquet_io_manager")
[docs]
def core_epa__assn_eia_epacamd_subplant_ids(
_core_epa__assn_eia_epacamd_unique: pd.DataFrame,
core_eia860__scd_generators: pd.DataFrame,
_core_epacems__emissions_unit_ids: pd.DataFrame,
core_eia860__assn_boiler_generator: pd.DataFrame,
) -> pd.DataFrame:
"""Groups units and generators into unique subplant groups.
This takes :func:`_core_epa__assn_eia_epacamd_unique` as an input so this asset
doesn't have to deal with duplicate matches that may be present in the
:func:`core_epa__assn_eia_epacamd` asset due to its use of multiple years of raw
crosswalk outputs.
This function consists of two primary parts:
#. Augment the EPA CAMD:EIA crosswalk with all IDs from EIA and EPA CAMD. Fill in
key IDs when possible. Because the published crosswalk was only meant to map
CAMD units to EIA generators, it is missing a large number of subplant_ids for
generators that do not report to CEMS. Before applying this function to the
subplant crosswalk, the crosswalk must be completed with all generators by outer
merging in the complete list of generators from EIA-860. This dataframe also
contains the complete list of ``unit_id_pudl`` mappings that will be necessary.
#. :func:`make_subplant_ids`: Use graph analysis to identify distinct groupings of
EPA units, EIA generators, and EIA-860 boiler-generator ``unit_id_pudl``
associations, based on 1:1, 1:m, m:1, or m:m relationships.
Returns:
table of cems_ids and with subplant_id added
"""
# Ensure ALL relevant IDs are present. Basically just merge in all the IDs
# Later note: As of April 2023, there is an experimental augmentation of the
# unit_id_pudl living in pudl.output.eia.assign_unit_ids. It is currently non-
# functioning (#2535) but when it is, ensure that it gets plugged into the dag
# BEFORE this step so the subplant IDs can benefit from the more fleshed out units
epacamd_eia_complete = (
augment_crosswalk_with_generators_eia860(
_core_epa__assn_eia_epacamd_unique, core_eia860__scd_generators
)
.pipe(augment_crosswalk_with_epacamd_ids, _core_epacems__emissions_unit_ids)
.pipe(augment_crosswalk_with_bga_eia860, core_eia860__assn_boiler_generator)
)
# use graph analysis to identify subplants. unit_id_pudl connectivity is folded
# directly into the graph (see _prep_for_networkx), so the resulting subplant_id
# already reflects any generators unit_id_pudl says are connected, even where the
# EPA crosswalk doesn't connect them directly.
subplant_ids_updated = make_subplant_ids(epacamd_eia_complete).pipe(
apply_pudl_dtypes, "glue"
)
logger.info(
"After making the networkx generated subplant_ids, "
f"{subplant_ids_updated.subplant_id.notnull().sum() / len(subplant_ids_updated):.1%} of"
" records have a subplant_id."
)
# The subplant_ids from make_subplant_ids are zero-indexed, which results in
# less intuitive alignments between subplant_ids and generator_ids. Increment all
# subplant_ids by 1 *before* applying manual overrides below, so those overrides
# can be expressed in terms of the final, user-facing subplant_id values rather
# than needing to track this internal zero-indexed implementation detail.
subplant_ids_updated["subplant_id"] = subplant_ids_updated["subplant_id"] + 1
subplant_ids_updated = manually_update_subplant_id(subplant_ids_updated)
# check for duplicates in sudo-PKs. These are not the actual PKs because there are
# some nulls in generator_id, so this won't be checked during the db construction
if (
epacamd_eia_dupe_mask := subplant_ids_updated.duplicated(
subset=[
"plant_id_epa",
"emissions_unit_id_epa",
"plant_id_eia",
"generator_id",
"subplant_id",
]
)
).any():
raise AssertionError(
"Duplicates found in pseudo primary keys of EPA CAMD/EIA subplant ID table "
"when none are expected. Duplicates found: \n"
f"{subplant_ids_updated[epacamd_eia_dupe_mask]}"
)
return subplant_ids_updated
[docs]
def augment_crosswalk_with_generators_eia860(
crosswalk_clean: pd.DataFrame, core_eia860__scd_generators: pd.DataFrame
) -> pd.DataFrame:
"""Merge any plants that are missing from the EPA crosswalk but appear in EIA-860.
Args:
crosswalk_clean: transformed EPA CEMS-EIA crosswalk.
core_eia860__scd_generators: EIA860 generators table.
"""
crosswalk_clean = crosswalk_clean.merge(
core_eia860__scd_generators[["plant_id_eia", "generator_id"]].drop_duplicates(),
how="outer",
on=["plant_id_eia", "generator_id"],
validate="m:1",
)
crosswalk_clean["plant_id_epa"] = crosswalk_clean["plant_id_epa"].fillna(
crosswalk_clean["plant_id_eia"]
)
return crosswalk_clean
[docs]
def augment_crosswalk_with_epacamd_ids(
crosswalk_clean: pd.DataFrame, _core_epacems__emissions_unit_ids: pd.DataFrame
) -> pd.DataFrame:
"""Merge all EPA CAMD IDs into the crosswalk."""
return crosswalk_clean.assign(
emissions_unit_id_epa=lambda x: x.emissions_unit_id_epa.fillna(x.generator_id)
).merge(
_core_epacems__emissions_unit_ids[
["plant_id_eia", "emissions_unit_id_epa"]
].drop_duplicates(),
how="outer",
on=["plant_id_eia", "emissions_unit_id_epa"],
validate="m:m",
)
[docs]
def augment_crosswalk_with_bga_eia860(
crosswalk_clean: pd.DataFrame, core_eia860__assn_boiler_generator: pd.DataFrame
) -> pd.DataFrame:
"""Merge all EIA Unit IDs into the crosswalk."""
return crosswalk_clean.merge(
core_eia860__assn_boiler_generator[
["plant_id_eia", "generator_id", "unit_id_pudl"]
].drop_duplicates(),
how="outer",
on=["plant_id_eia", "generator_id"],
validate="m:1",
)
######################
# Networkx subplant_id
######################
[docs]
def _prep_for_networkx(crosswalk: pd.DataFrame) -> pd.DataFrame:
"""Build composite node identities for combustors, generators, and EIA units.
Node identities are plain ``(type_tag, plant_id_eia, natural_id)`` tuples rather
than numeric surrogates. networkx accepts any hashable object as a node, so
there's no need to invent a global integer ID space -- crosswalk associations
never connect two different plants, so embedding ``plant_id_eia`` directly in
each node is enough to keep every plant's subgraph naturally disjoint from every
other plant's, without ever computing a plant-spanning "global" ID.
``type_tag`` (``"combustor"``/``"generator"``/``"unit"``) keeps otherwise-identical
IDs from colliding, e.g. a generator and a boiler that both happen to be named
``"1"`` at the same plant, which is a common EIA naming convention.
A row missing its natural id (e.g. an EPA unit that never matched an EIA
generator) is keyed by its own row position instead of left null, so two such
rows are never accidentally treated as sharing an id.
Args:
crosswalk: The :ref:`core_epa__assn_eia_epacamd` crosswalk, augmented with
``unit_id_pudl`` (see :func:`augment_crosswalk_with_bga_eia860`).
Returns:
A copy of :ref:`core_epa__assn_eia_epacamd` crosswalk with new node-identity
columns ``combustor_node``, ``generator_node``, and ``unit_node``.
"""
prepped = crosswalk.reset_index(drop=True)
row_position = prepped.index.to_series()
prepped["combustor_node"] = list(
zip(
repeat("combustor", len(prepped)),
prepped["plant_id_eia"],
prepped["emissions_unit_id_epa"].fillna(row_position),
strict=True,
)
)
prepped["generator_node"] = list(
zip(
repeat("generator", len(prepped)),
prepped["plant_id_eia"],
prepped["generator_id"].fillna(row_position),
strict=True,
)
)
# unit_id_pudl is left null where missing rather than falling back to a
# per-row sentinel: rows with a null unit_id_pudl never get a unit_node edge
# added at all (see _subplant_ids_from_prepped_crosswalk), so there's no
# collision risk to guard against here.
prepped["unit_node"] = list(
zip(
repeat("unit", len(prepped)),
prepped["plant_id_eia"],
prepped["unit_id_pudl"],
strict=True,
)
)
return prepped
[docs]
def _subplant_ids_from_prepped_crosswalk(prepped: pd.DataFrame) -> pd.DataFrame:
"""Use networkx graph analysis to assign a per-plant ``subplant_id``.
Args:
prepped: ``core_epa__assn_eia_epacamd`` crosswalk passed through
:func:`_prep_for_networkx`
Returns:
A copy of ``core_epa__assn_eia_epacamd`` crosswalk plus a new ``subplant_id``
column. ``subplant_id`` is already 0-indexed and contiguous within each
``plant_id_eia`` -- no separate global-to-composite-ID conversion step is
needed, since node identities already embed ``plant_id_eia`` (see
:func:`_prep_for_networkx`) and crosswalk edges never connect two plants.
"""
graph = nx.from_pandas_edgelist(
prepped,
source="combustor_node",
target="generator_node",
edge_attr=True,
)
edge_list = nx.to_pandas_edgelist(
graph, source="combustor_node", target="generator_node"
)
# nx.Graph silently collapses multiple edges between the same pair of nodes into
# one, keeping only the last edge's attributes. If prepped contains a duplicated
# (combustor_node, generator_node) pair -- e.g. because upstream deduplication of
# the crosswalk failed to catch it -- that row disappears with no warning.
if len(edge_list) != len(prepped):
raise AssertionError(
"Expected make_subplant_ids to preserve row count, but networkx's graph "
f"construction collapsed {len(prepped) - len(edge_list)} duplicate "
"(combustor_node, generator_node) edge(s). This likely means the "
"crosswalk contains a duplicated EPA unit/EIA generator association "
f"within a plant. Input had {len(prepped)} rows, output has "
f"{len(edge_list)}."
)
# Fold in unit_id_pudl connectivity via a separate copy of the graph, purely to
# determine which combustor/generator components should be merged. These
# synthetic generator<->unit edges must never appear in the crosswalk edge list
# returned above, since they don't correspond to a crosswalk row.
connectivity_graph = graph.copy()
unit_edges = prepped.loc[
prepped["unit_id_pudl"].notna(), ["generator_node", "unit_node"]
].drop_duplicates()
connectivity_graph.add_edges_from(unit_edges.itertuples(index=False, name=None))
# Assign subplant_id directly, per plant, as each connected component is
# discovered -- no plant-spanning "global" ID is ever created. Every node in a
# component shares one plant_id_eia (node identities embed it, and crosswalk
# edges never span plants), so pulling it off any single node in the component
# is enough to know which plant's counter to use.
next_subplant_id: dict[int, int] = collections.defaultdict(int)
subplant_id_by_node = {}
for node_set in nx.connected_components(connectivity_graph):
assert nx.algorithms.bipartite.is_bipartite(
connectivity_graph.subgraph(node_set)
), f"non-bipartite node_set={node_set}"
# networkx types graph nodes as opaque Hashable; ours are always the
# (type_tag, plant_id_eia, natural_id) tuples built in _prep_for_networkx.
plant_id_eia = next(iter(node_set))[1] # type: ignore[bad-index]
subplant_id = next_subplant_id[plant_id_eia]
next_subplant_id[plant_id_eia] += 1
subplant_id_by_node.update(dict.fromkeys(node_set, subplant_id))
# Safe to map through a plain dict here: every node embeds a real plant_id_eia
# and, for rows missing a natural id, a per-row sentinel (see
# _prep_for_networkx), so no two distinct nodes are ever null/equal-looking in a
# way that could make pandas silently collapse them.
edge_list["subplant_id"] = edge_list["combustor_node"].map(subplant_id_by_node)
return edge_list
[docs]
def make_subplant_ids(crosswalk: pd.DataFrame) -> pd.DataFrame:
"""Identify sub-plants in the EPA/EIA crosswalk graph.
In graph analysis terminology, the crosswalk is a list of edges between nodes (EPA
emissions units and EIA generators) in a bipartite graph. A third node type, EIA-860
boiler-generator ``unit_id_pudl`` groupings, is folded into the same graph as well,
connecting generators that share a ``unit_id_pudl`` even where the EPA crosswalk
doesn't connect them directly (see :func:`_subplant_ids_from_prepped_crosswalk`).
The networkx python package provides functions to analyze this graph and extract
disjoint subgraphs (groups of units, generators, and unit_id_pudl groupings that are
connected to each other). These are the distinct power plants. To avoid a name
collision with plant_id, we term these collections 'subplants', and identify them
with a subplant_id that is unique within each plant_id. Subplants are thus
identified with the composite key (plant_id_eia, subplant_id).
Through this analysis, we found that 56% of plant_ids contain multiple distinct
subplants, and 11% contain subplants with different technology types, such as a gas
boiler and gas turbine (not in a combined cycle).
Any row filtering should be done before this step if desired. Rows that are
duplicates with respect to the EPA unit/EIA generator association -- e.g. two
crosswalk rows that only differ in ``generator_id_epa`` or ``boiler_id``, neither
of which appears in the output -- are collapsed here explicitly, since silently
leaving that to the graph construction below would make its row-count guarantee
(see :func:`_subplant_ids_from_prepped_crosswalk`) fail on real data.
Note that sub-plant ids should be used in conjunction with ``plant_id_eia`` rather
than ``plant_id_epa`` because the former is more granular and integrated into CEMS
during the transform process.
Args:
crosswalk: The core_epa__assn_eia_epacamd crosswalk
Returns:
An edge list connecting EPA units to EIA generators, with connected pieces
issued a subplant_id
"""
crosswalk = crosswalk.drop_duplicates(
subset=[
"plant_id_eia",
"plant_id_epa",
"emissions_unit_id_epa",
"generator_id",
"unit_id_pudl",
]
)
edge_list = _prep_for_networkx(crosswalk)
edge_list = _subplant_ids_from_prepped_crosswalk(edge_list)
return edge_list
#############################
# Augmentation of subplant_id
#############################
[docs]
def manually_update_subplant_id(subplant_crosswalk: pd.DataFrame) -> pd.DataFrame:
"""Manually update the subplant_id for ``plant_id_eia`` 1391.
This function lumps all records within ``plant_id_eia`` 1391 into the same
``subplant_id`` group. See `comment <https://github.com/singularity-energy/open-grid-emissions/pull/142#issuecomment-1186579260>_`
for explanation of why.
"""
# set all generators in plant 1391 to the same subplant
subplant_crosswalk.loc[
subplant_crosswalk["plant_id_eia"] == 1391, "subplant_id"
] = 1
return subplant_crosswalk
