Eia176 wide table

src/pudl/transform/eia176.py

@@ -0,0 +1,76 @@
+"""Module to perform data cleaning functions on EIA176 data tables."""
+
+import warnings
+
+import pandas as pd
+from dagster import ExperimentalWarning, asset, asset_check
+
+from pudl.logging_helpers import get_logger
+
+logger = get_logger(__name__)
+
+# Asset Checks are still Experimental, silence the warning since we use them
+# everywhere.
+warnings.filterwarnings("ignore", category=ExperimentalWarning)
+
+
+@asset
+def _core_eia176__data(raw_eia176__data: pd.DataFrame) -> pd.DataFrame:
+    """Take entity-attribute-value rows and convert to rows with primary key and one column per variable."""
+    raw_eia176__data["variable_name"] = (
+        raw_eia176__data["line"] + "_" + raw_eia176__data["atype"]
+    )
+    primary_key = ["report_year", "area", "id"]
+    variable_names = list(raw_eia176__data.groupby("variable_name").count().index)
+    wide_table = pd.DataFrame(columns=primary_key + variable_names)
+
+    granular_data = raw_eia176__data[
+        (raw_eia176__data["company"] != " Total of All Companies")
+    ]
+    for report_year, area, id_ in granular_data.groupby(primary_key).count().index:
+        # Get the data corresponding to one completed form EIA-176
+        form_data = granular_data[
+            (granular_data["report_year"] == report_year)
+            & (granular_data["area"] == area)
+            & (granular_data["id"] == id_)
+        ]
+
+        wide_row = {"report_year": report_year, "area": area, "id": id_}
+
+        # Translate each piece of data entered into the form into its own column
+        for record in form_data.iterrows():
+            form_row = record[1]
+            wide_row[form_row["variable_name"]] = form_row["value"]
+
+        wide_table.loc[len(wide_table.index)] = wide_row
+
+    return wide_table
+
+
+@asset_check(asset=_core_eia176__data, blocking=True)
+def validate_totals():
+    """Compare reported and calculated totals for different geographical aggregates, report any differences."""
+
+
+def _compare_totals(
+    reported_totals: pd.DataFrame,
+    calculated_totals: pd.DataFrame,
+    groupby_cols: list[str],
+) -> pd.DataFrame:
+    """Compare two dataframes representing reporting and calculated totals."""
+    reset_calculated = (
+        calculated_totals.sort_values(by=groupby_cols)
+        .reset_index()[groupby_cols + ["value"]]
+        .round(2)
+    )
+
+    reset_reported = (
+        reported_totals.sort_values(by=groupby_cols)
+        .reset_index()[groupby_cols + ["value"]]
+        .fillna(0)
+    )
+
+    return reset_calculated.compare(reset_reported)
+
+
+# TODO: Reasonable boundaries -- in a script/notebook in the 'validate' directory? How are those executed?

test/unit/transform/eia176_test.py

@@ -0,0 +1,87 @@
+import pandas as pd
+
+from pudl.transform.eia176 import _compare_totals, _core_eia176__data
+
+
+def get_test_df():
+    col_names = ["area", "atype", "company", "id", "line", "report_year", "value"]
+    df = pd.DataFrame(columns=col_names)
+    df.loc[0] = [
+        "New Mexico",
+        "VL",
+        "ZIA NATURAL GAS",
+        "17635019NM",
+        "1010",
+        "2022",
+        2013231.0,
+    ]
+    df.loc[1] = [
+        "New Mexico",
+        "VL",
+        " Total of All Companies",
+        "17635019NM",
+        "1010",
+        "2022",
+        2013231.0,
+    ]
+
+    return df
+
+
+def test_core_eia176__data():
+    eav_model = get_test_df()
+
+    wide_table = _core_eia176__data(eav_model)
+    assert wide_table.shape == (1, 4)
+    row = wide_table.loc[0]
+    assert list(row.index) == ["report_year", "area", "id", "1010_VL"]
+    assert list(row.values) == ["2022", "New Mexico", "17635019NM", 2013231.0]
+
+
+def test_compare_totals_no_diff():
+    states_df = get_test_df()
+    states_df.loc[2] = [
+        "New Mexico",
+        "VL",
+        "ANOTHER COMPANY",
+        "12345679NM",
+        "1010",
+        "2022",
+        1.0,
+    ]
+    states_df.loc[1, "value"] += 1
+
+    reported_state_totals = states_df[states_df["company"] == " Total of All Companies"]
+    state_companies_df = states_df[states_df["company"] != " Total of All Companies"]
+    state_level_cols = ["report_year", "area", "line", "atype"]
+    calculated_state_totals = state_companies_df.groupby(state_level_cols).sum()
+    assert _compare_totals(
+        reported_state_totals, calculated_state_totals, state_level_cols
+    ).empty
+
+
+# TODO: Test on specific details here
+def test_compare_totals_diff():
+    states_df = get_test_df()
+    states_df.loc[2] = [
+        "New Mexico",
+        "VL",
+        "ANOTHER COMPANY",
+        "12345679NM",
+        "1010",
+        "2022",
+        1.0,
+    ]
+
+    reported_state_totals = states_df[states_df["company"] == " Total of All Companies"]
+    state_companies_df = states_df[states_df["company"] != " Total of All Companies"]
+    state_level_cols = ["report_year", "area", "line", "atype"]
+    calculated_state_totals = state_companies_df.groupby(state_level_cols).sum()
+    assert not _compare_totals(
+        reported_state_totals, calculated_state_totals, state_level_cols
+    ).empty
+
+
+# TODO: Implement, if we can even unit-test a function annotated as an asset check
+def test_validate__totals():
+    pass