From ac972ae844dffccaddd441c019400a4c3513b81b Mon Sep 17 00:00:00 2001 From: Max Nutz Date: Mon, 29 Jun 2026 08:58:37 +0200 Subject: [PATCH 1/2] Introduce first version of Statistics-function using new helper-function in utils --- .../configs/mapping.default.yaml | 1 + .../statistics_functions.py | 80 +++++++++++++++++++ pypsa_validation_processing/utils.py | 13 +++ 3 files changed, 94 insertions(+) diff --git a/pypsa_validation_processing/configs/mapping.default.yaml b/pypsa_validation_processing/configs/mapping.default.yaml index 8566d48..73e6201 100644 --- a/pypsa_validation_processing/configs/mapping.default.yaml +++ b/pypsa_validation_processing/configs/mapping.default.yaml @@ -9,6 +9,7 @@ Final Energy [by Carrier]|Electricity: Final_Energy_by_Carrier__Electricity Final Energy [by Carrier]|Coal: Final_Energy_by_Carrier__Coal Final Energy [by Carrier]|Natural Gas: Final_Energy_by_Carrier__Natural_Gas Final Energy [by Carrier]|Oil: Final_Energy_by_Carrier__Oil +Final Energy [by Carrier]|Ambient Heat: Final_Energy_by_Carrier__Ambient_Heat Final Energy [by Sector]|Transportation: Final_Energy_by_Sector__Transportation Final Energy [by Sector]|Industry: Final_Energy_by_Sector__Industry Final Energy [by Sector]|Agriculture: Final_Energy_by_Sector__Agriculture diff --git a/pypsa_validation_processing/statistics_functions.py b/pypsa_validation_processing/statistics_functions.py index d664969..6e96305 100644 --- a/pypsa_validation_processing/statistics_functions.py +++ b/pypsa_validation_processing/statistics_functions.py @@ -30,6 +30,7 @@ def (network: pypsa.Network) -> pd.Series: statistics_kwargs_for_filtering as kwargs_filtering, statistics_kwargs as kwargs, UNITS_MAPPING, + remap_unit_index, ) from pypsa_validation_processing.utils import ( get_energy_totals_domestic_share, @@ -352,6 +353,85 @@ def Final_Energy_by_Carrier__Oil( return fossil_oil +def Final_Energy_by_Carrier__Ambient_Heat( + n: pypsa.Network, + aggregate_per_year: bool = True, +) -> pd.Series | pd.DataFrame: + """Extract ambient heat consumption of heat pumps and solar-thermal + from a PyPSA-Network. + + Parameters + ---------- + n : pypsa.Network + 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.DataFrame` with snapshots as columns. + + Returns + ------- + pd.Series | pd.DataFrame + Pandas Series (``aggregate_per_year=True``) or DataFrame + (``aggregate_per_year=False``) with MultiIndex including + ``location`` and ``unit``. + Returns data at regional level as provided by the PyPSA network. + Country-level aggregation is handled by + Network_Processor._aggregate_to_country() if configured. + + Notes + ----- + Heat pump CHP is temperature dependent. So the comparison of input + to heat output brings the surplus, as efficiency of these links + is set to 1.0. + """ + ambient_technologies = [ + "urban central air heat pump", + "urban decentral air heat pump", + "rural air heat pump", + "rural ground heat pump", + ] + # heat pumps + # restrict evaluation on heat output + e_out = n.statistics.energy_balance( + bus_carrier=["urban central heat", "urban decentral heat", "rural heat"], + carrier=ambient_technologies, + components="Link", + groupby_time=aggregate_per_year, + **kwargs, + ) + # restrict evaluation on electricity input - no elec to elec, so no at_port needed. + e_in = n.statistics.energy_balance( + bus_carrier="low voltage", + carrier=ambient_technologies, + components="Link", + groupby_time=aggregate_per_year, + **kwargs, + ) + + e_out = remap_unit_index(e_out) + e_in = remap_unit_index(e_in) + res_heat_pump = e_out.add(e_in, fill_value=0).groupby(kwargs["groupby"]).sum() + + # solar thermal (NaN values for non-available input during nighttime.) + solar_thermal_carriers = [ + "urban central solar thermal", + "rural solar thermal", + "urban decentral solar thermal", + ] + st = n.statistics.supply( + bus_carrier=["urban central heat", "urban decentral heat", "rural heat"], + carrier=solar_thermal_carriers, + components="Generator", + groupby_time=aggregate_per_year, + **kwargs, + ).replace(np.nan, 0) + res_series = [res_heat_pump, st] + res_series = [series for series in res_series if not series.empty] + res = pd.concat(res_series) + return res + + def Final_Energy_by_Carrier__Natural_Gas( n: pypsa.Network, aggregate_per_year: bool = True, diff --git a/pypsa_validation_processing/utils.py b/pypsa_validation_processing/utils.py index 1dc4132..7463ed4 100644 --- a/pypsa_validation_processing/utils.py +++ b/pypsa_validation_processing/utils.py @@ -130,6 +130,19 @@ "MWh": "MWh", } + +def remap_unit_index( + series: pd.Series, unit_mapping: dict = UNITS_MAPPING +) -> pd.Series: + """Remaps unit index to standard value for statistics with need for + direct comparison of input-output flows.""" + idx = series.index.to_frame(index=False) + idx["unit"] = idx["unit"].replace(unit_mapping) + out = series.copy() + out.index = pd.MultiIndex.from_frame(idx, names=series.index.names) + return out + + ## standards for statistics-functions # standardize kwargs for pypsa-statistics statements statistics_kwargs = { From a7ec3621d7997f64ed9bf1d4444e82508b31d155 Mon Sep 17 00:00:00 2001 From: Max Nutz Date: Mon, 29 Jun 2026 09:05:41 +0200 Subject: [PATCH 2/2] add testing routine for newly introduced function --- tests/test_statistics_functions.py | 220 +++++++++++++++++++++++++++++ 1 file changed, 220 insertions(+) diff --git a/tests/test_statistics_functions.py b/tests/test_statistics_functions.py index f8cb81f..bc10417 100644 --- a/tests/test_statistics_functions.py +++ b/tests/test_statistics_functions.py @@ -6,6 +6,7 @@ import pytest from pypsa_validation_processing.statistics_functions import ( + Final_Energy_by_Carrier__Ambient_Heat, Final_Energy_by_Carrier__Electricity, Final_Energy_by_Carrier__Coal, Final_Energy_by_Carrier__Natural_Gas, @@ -221,6 +222,225 @@ def test_issues_expected_withdrawal_queries(self): assert calls[4]["carrier"] == "DAC" assert calls[4]["components"] == "Link" +# --------------------------------------------------------------------------- +# Tests for Final_Energy_by_Carrier__Ambient_Heat +# --------------------------------------------------------------------------- + + +class TestFinalEnergyByCarrierAmbientHeat: + """Test suite for Final_Energy_by_Carrier__Ambient_Heat function.""" + + class _AmbientHeatStatisticsAccessor: + """Minimal accessor to verify ambient-heat balance behavior.""" + + ambient_carriers = [ + "urban central air heat pump", + "urban decentral air heat pump", + "rural air heat pump", + "rural ground heat pump", + ] + + def __init__(self, scenario: str = "balanced"): + self.scenario = scenario + self.calls: list[dict[str, object]] = [] + + @staticmethod + def _series_from_groupby( + groupby: list[str], + values: list[float], + location: str = "AT1", + unit: str = "MWh_th", + ) -> pd.Series: + index = pd.MultiIndex.from_tuples( + [tuple({"location": location, "unit": unit}[key] for key in groupby)], + names=groupby, + ) + return pd.Series(values, index=index, dtype=float) + + @staticmethod + def _dataframe_from_groupby( + groupby: list[str], + values: list[float], + location: str = "AT1", + unit: str = "MWh_th", + ) -> pd.DataFrame: + index = pd.MultiIndex.from_tuples( + [tuple({"location": location, "unit": unit}[key] for key in groupby)], + names=groupby, + ) + columns = pd.DatetimeIndex( + pd.to_datetime(["2019-01-01", "2019-01-02"]), name="snapshot" + ) + return pd.DataFrame([values[: len(columns)]], index=index, columns=columns) + + def energy_balance( + self, + bus_carrier: list[str] | str | None = None, + carrier: list[str] | str | None = None, + components: str | list[str] | None = None, + groupby: list[str] | None = None, + groupby_time: bool = True, + **_: object, + ) -> pd.Series | pd.DataFrame: + if groupby is None: + groupby = ["location", "unit"] + + self.calls.append( + { + "bus_carrier": bus_carrier, + "carrier": carrier, + "components": components, + "groupby": groupby, + "groupby_time": groupby_time, + } + ) + + if not (carrier == self.ambient_carriers and components == "Link"): + return pd.Series( + dtype=float, + index=pd.MultiIndex.from_tuples([], names=groupby), + ) + + if groupby_time: + if self.scenario in {"zero_heat", "zero_electric"}: + values = [0.0] + elif bus_carrier == [ + "urban central heat", + "urban decentral heat", + "rural heat", + ]: + values = [10.0] + else: + values = [-10.0] + return self._series_from_groupby(groupby, values) + + if self.scenario in {"zero_heat", "zero_electric"}: + values = [0.0, 0.0] + elif bus_carrier == [ + "urban central heat", + "urban decentral heat", + "rural heat", + ]: + values = [10.0, 10.0] + else: + values = [-10.0, -10.0] + return self._dataframe_from_groupby(groupby, values) + + def supply( + self, + bus_carrier: list[str] | str | None = None, + carrier: list[str] | str | None = None, + components: str | list[str] | None = None, + groupby: list[str] | None = None, + groupby_time: bool = True, + **_: object, + ) -> pd.Series | pd.DataFrame: + if groupby is None: + groupby = ["location", "unit"] + + self.calls.append( + { + "bus_carrier": bus_carrier, + "carrier": carrier, + "components": components, + "groupby": groupby, + "groupby_time": groupby_time, + } + ) + + if not ( + bus_carrier + == ["urban central heat", "urban decentral heat", "rural heat"] + and carrier + == [ + "urban central solar thermal", + "rural solar thermal", + "urban decentral solar thermal", + ] + and components == "Generator" + ): + return pd.Series( + dtype=float, + index=pd.MultiIndex.from_tuples([], names=groupby), + ) + + if groupby_time: + values = ( + [0.0] if self.scenario in {"zero_heat", "zero_electric"} else [5.0] + ) + return self._series_from_groupby(groupby, values, unit="MWh_th") + + values = ( + [0.0, 0.0] + if self.scenario in {"zero_heat", "zero_electric"} + else [5.0, 5.0] + ) + return self._dataframe_from_groupby(groupby, values, unit="MWh_th") + + class _AmbientHeatNetwork: + """Minimal network object exposing only the statistics accessor.""" + + def __init__(self, scenario: str = "balanced"): + self.statistics = ( + TestFinalEnergyByCarrierAmbientHeat._AmbientHeatStatisticsAccessor( + scenario=scenario + ) + ) + + def _ambient_heat_network(self, scenario: str = "balanced"): + return self._AmbientHeatNetwork(scenario=scenario) + + def test_returns_series_and_uses_expected_queries(self): + """Function returns a Series and queries both heat and electricity sides.""" + network = self._ambient_heat_network("balanced") + + result = Final_Energy_by_Carrier__Ambient_Heat(network) + + assert isinstance(result, pd.Series) + assert isinstance(result.index, pd.MultiIndex) + assert result.index.names == ["location", "unit"] + assert result.loc[("AT1", "MWh")] == pytest.approx(0.0) + assert result.loc[("AT1", "MWh_th")] == pytest.approx(5.0) + assert len(network.statistics.calls) == 3 + assert network.statistics.calls[0]["bus_carrier"] == [ + "urban central heat", + "urban decentral heat", + "rural heat", + ] + assert network.statistics.calls[1]["bus_carrier"] == "low voltage" + assert network.statistics.calls[2]["components"] == "Generator" + + def test_zero_heat_output_also_has_zero_electric_input(self): + """If the heat-output side is zero, the electricity-input side is zero too.""" + result = Final_Energy_by_Carrier__Ambient_Heat( + self._ambient_heat_network("zero_heat") + ) + + assert isinstance(result, pd.Series) + assert result.loc[("AT1", "MWh")] == pytest.approx(0.0) + + def test_zero_electric_input_also_has_zero_heat_output(self): + """If the electricity-input side is zero, the heat-output side is zero too.""" + result = Final_Energy_by_Carrier__Ambient_Heat( + self._ambient_heat_network("zero_electric") + ) + + assert isinstance(result, pd.Series) + assert result.loc[("AT1", "MWh")] == pytest.approx(0.0) + + def test_returns_dataframe_for_aggregate_per_year_false(self): + """aggregate_per_year=False returns a DataFrame with snapshot columns.""" + result = Final_Energy_by_Carrier__Ambient_Heat( + self._ambient_heat_network("balanced"), + aggregate_per_year=False, + ) + + assert isinstance(result, pd.DataFrame) + assert isinstance(result.index, pd.MultiIndex) + assert result.index.names == ["location", "unit"] + assert isinstance(result.columns, pd.DatetimeIndex) + + # --------------------------------------------------------------------------- # Tests for Final_Energy_by_Carrier__Oil # ---------------------------------------------------------------------------