diff --git a/pypsa_validation_processing/statistics_functions.py b/pypsa_validation_processing/statistics_functions.py index 51b803e..b7afb71 100644 --- a/pypsa_validation_processing/statistics_functions.py +++ b/pypsa_validation_processing/statistics_functions.py @@ -20,6 +20,7 @@ def (network: pypsa.Network) -> pd.Series: """ from __future__ import annotations +import re from functools import reduce from pathlib import Path import pandas as pd @@ -39,10 +40,11 @@ def Final_Energy_by_Carrier__Electricity( n: pypsa.Network, aggregate_per_year: bool = True, ) -> pd.Series | pd.DataFrame: - """Extract electricity final energy from a PyPSA Network. + """Extract electricity-carrier final energy from a PyPSA Network. - Returns the total electricity consumption (excluding transmission / - distribution losses) + Returns the total final energy demand supplied by electricity across + agriculture, residential and commercial, transportation, and industry + and DAC. Parameters ---------- @@ -50,7 +52,7 @@ def Final_Energy_by_Carrier__Electricity( PyPSA network to process. aggregate_per_year : bool, optional If ``True`` (default), aggregate over all snapshots and return a - :class:`pandas.Series`. If ``False``, return a + :class:`pandas.Series`. If ``False``, return a :class:`pandas.DataFrame` with snapshots as columns. Returns @@ -65,30 +67,71 @@ def Final_Energy_by_Carrier__Electricity( Notes ----- - Extracts all withdrawals from elec network. low_voltage is included in AC withdrawal. - Remove discharger afterwards, as battery-connecting links have different carrier names. + Combines electricity withdrawals from the following contributions: + agriculture electricity loads, residential/commercial low-voltage loads + (excluding dedicated industry/agriculture/charger/distribution categories, + BEV charging) industry electricity loads, and DAC electricity demand + change in home battery is of magnitude 1e-9 compared to electricity demand. + Concerning heat: rural air heat pump, rural ground heat pump, rural resisive + heater and urban decentral heatings are included. Central heatings using + electricity as fuel are NOT included. """ - # withdrawal from electricity including low_voltage - res = abs( - n.statistics.energy_balance( - bus_carrier="AC", - groupby=["carrier", "location", "unit"], - groupby_time=aggregate_per_year, - ) + # get Final Energy|Agriculture|Electricity + agri = n.statistics.withdrawal( + carrier=["agriculture electricity", "agriculture machinery electric"], + components="Load", + aggregate_time=aggregate_per_year, + **kwargs, ) - # as battery is Store, discharger-link needs to be evaluated separately. - res_storage = n.statistics.energy_balance( - bus_carrier="AC", - groupby=["carrier", "location", "unit"], - carrier=["battery discharger"], - groupby_time=aggregate_per_year, + + # get Final Energy|Residential and Commercial|Electricity + lv = n.statistics.withdrawal( + bus_carrier="low voltage", aggregate_time=aggregate_per_year, **kwargs_filtering ) - return ( - pd.concat([res, res_storage.mul(-1)], axis=0) - .groupby(["location", "unit"]) - .sum() + forbidden_parts = [ + "urban central", + "industry", + "agriculture", + "charger", + "distribution", + ] + lv_carriers = lv.index.get_level_values("carrier").astype(str) + forbidden_pattern = "|".join(re.escape(part) for part in forbidden_parts) + forbidden_mask = lv_carriers.str.contains(forbidden_pattern, case=False, regex=True) + rescom = lv[~forbidden_mask].groupby(kwargs["groupby"]).sum() + + # get Final Energy|Transportation|Electricity + transpo = n.statistics.withdrawal( + bus_carrier="low voltage", + carrier="BEV charger", + components="Link", + aggregate_time=aggregate_per_year, + **kwargs, + ) + + # get Final Energy|Industry|Electricity + industry = n.statistics.withdrawal( + carrier="industry electricity", + components="Load", + aggregate_time=aggregate_per_year, + **kwargs, ) + # get electricity load from DAC + dac = n.statistics.withdrawal( + bus_carrier="AC", + carrier="DAC", + components="Link", + aggregate_time=aggregate_per_year, + **kwargs, + ) + + series_list = [agri, rescom, transpo, industry, dac] + series_list = [series for series in series_list if not series.empty] + + result = pd.concat(series_list) + return result + def Final_Energy_by_Sector__Transportation( n: pypsa.Network, diff --git a/tests/test_statistics_functions.py b/tests/test_statistics_functions.py index 25923fe..a73c3fd 100644 --- a/tests/test_statistics_functions.py +++ b/tests/test_statistics_functions.py @@ -23,6 +23,112 @@ class TestFinalEnergyByCarrierElectricity: """Test suite for Final_Energy_by_Carrier__Electricity function.""" + class _ElectricityStatisticsAccessor: + """Minimal accessor to verify electricity-carrier extraction behavior.""" + + def __init__(self, empty_dac: bool = False): + self.empty_dac = empty_dac + self.calls: list[dict] = [] + + @staticmethod + def _series_from_groupby( + groupby: list[str], + values: list[float], + location: str = "AT1", + unit: str = "MWh_el", + ) -> pd.Series: + index = pd.MultiIndex.from_tuples( + [tuple({"location": location, "unit": unit}[k] for k in groupby)], + names=groupby, + ) + return pd.Series(values, index=index, dtype=float) + + def withdrawal( + self, + bus_carrier: str | None = None, + carrier: list[str] | str | None = None, + components: str | list[str] | None = None, + aggregate_time: bool = True, + groupby: list[str] | None = None, + nice_names: bool | None = None, + **_: object, + ) -> pd.Series | pd.DataFrame: + self.calls.append( + { + "bus_carrier": bus_carrier, + "carrier": carrier, + "components": components, + "aggregate_time": aggregate_time, + "groupby": groupby, + "nice_names": nice_names, + } + ) + + if groupby is None: + groupby = ["location", "unit"] + + if carrier == ["agriculture electricity", "agriculture machinery electric"]: + return self._series_from_groupby(groupby, [10.0]) + + if bus_carrier == "low voltage" and components is None: + index = pd.MultiIndex.from_tuples( + [ + ("n1", "AT1 bus", "household demand", "AT1", "MWh_el"), + ("n2", "AT1 bus", "rural air heat pump", "AT1", "MWh_el"), + ("n3", "AT1 bus", "industry electricity", "AT1", "MWh_el"), + ("n4", "AT1 bus", "agriculture electricity", "AT1", "MWh_el"), + ("n5", "AT1 bus", "BEV charger", "AT1", "MWh_el"), + ("n6", "AT1 bus", "distribution losses", "AT1", "MWh_el"), + ( + "n7", + "AT1 bus", + "urban central resistive heater", + "AT1", + "MWh_el", + ), + ], + names=["name", "bus", "carrier", "location", "unit"], + ) + return pd.Series( + [7.0, 8.0, 100.0, 200.0, 300.0, 400.0, 500.0], index=index + ) + + if ( + bus_carrier == "low voltage" + and carrier == "BEV charger" + and components == "Link" + ): + return self._series_from_groupby(groupby, [40.0]) + + if carrier == "industry electricity" and components == "Load": + return self._series_from_groupby(groupby, [50.0]) + + if bus_carrier == "AC" and carrier == "DAC" and components == "Link": + if self.empty_dac: + return pd.Series( + dtype=float, + index=pd.MultiIndex.from_tuples([], names=groupby), + ) + return self._series_from_groupby(groupby, [60.0]) + + return pd.Series( + dtype=float, + index=pd.MultiIndex.from_tuples([], names=groupby), + ) + + class _ElectricityNetwork: + """Minimal network exposing the custom electricity statistics accessor.""" + + def __init__(self, empty_dac: bool = False): + self.statistics = ( + TestFinalEnergyByCarrierElectricity._ElectricityStatisticsAccessor( + empty_dac=empty_dac + ) + ) + + def _electricity_network(self, empty_dac: bool = False): + return self._ElectricityNetwork(empty_dac=empty_dac) + def test_returns_series(self, mock_network: MockPyPSANetwork): """Test that the function returns a pandas Series.""" result = Final_Energy_by_Carrier__Electricity(mock_network) @@ -62,6 +168,56 @@ def test_multiple_networks(self, mock_network_collection: MockNetworkCollection) assert result.index.names == ["location", "unit"] assert len(result) > 0 + def test_filters_forbidden_low_voltage_carriers(self): + """Residential/commercial low-voltage sum excludes forbidden carrier patterns.""" + result = Final_Energy_by_Carrier__Electricity(self._electricity_network()) + # Only household demand (7) and rural air heat pump (8) must remain from LV. + assert 15.0 in result.values + assert 100.0 not in result.values + assert 200.0 not in result.values + assert 300.0 not in result.values + assert 400.0 not in result.values + assert 500.0 not in result.values + + def test_includes_all_non_empty_contributions_in_order(self): + """Result concatenates agriculture, res/com, transport, industry, and DAC.""" + result = Final_Energy_by_Carrier__Electricity(self._electricity_network()) + assert list(result.values) == [10.0, 15.0, 40.0, 50.0, 60.0] + + def test_ignores_empty_contributions_before_concat(self): + """Empty contribution series are removed before concatenation.""" + result = Final_Energy_by_Carrier__Electricity( + self._electricity_network(empty_dac=True) + ) + assert list(result.values) == [10.0, 15.0, 40.0, 50.0] + + def test_issues_expected_withdrawal_queries(self): + """Function requests the expected carrier/component combinations.""" + network = self._electricity_network() + _ = Final_Energy_by_Carrier__Electricity(network) + calls = network.statistics.calls + assert len(calls) == 5 + + assert calls[0]["carrier"] == [ + "agriculture electricity", + "agriculture machinery electric", + ] + assert calls[0]["components"] == "Load" + + assert calls[1]["bus_carrier"] == "low voltage" + assert calls[1]["components"] is None + + assert calls[2]["bus_carrier"] == "low voltage" + assert calls[2]["carrier"] == "BEV charger" + assert calls[2]["components"] == "Link" + + assert calls[3]["carrier"] == "industry electricity" + assert calls[3]["components"] == "Load" + + assert calls[4]["bus_carrier"] == "AC" + assert calls[4]["carrier"] == "DAC" + assert calls[4]["components"] == "Link" + # --------------------------------------------------------------------------- # Tests for Final_Energy_by_Sector__Transportation