diff --git a/pypsa_validation_processing/configs/mapping.default.yaml b/pypsa_validation_processing/configs/mapping.default.yaml index 7a77c3b..f080c0b 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 Carrier]|District Heat: Final_Energy_by_Carrier__District_Heat Final Energy [by Sector]|Transportation: Final_Energy_by_Sector__Transportation Final Energy [by Sector]|Industry: Final_Energy_by_Sector__Industry diff --git a/pypsa_validation_processing/statistics_functions.py b/pypsa_validation_processing/statistics_functions.py index a355dd7..1afcf53 100644 --- a/pypsa_validation_processing/statistics_functions.py +++ b/pypsa_validation_processing/statistics_functions.py @@ -353,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/tests/test_statistics_functions.py b/tests/test_statistics_functions.py index 560c934..ec1b4d2 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__District_Heat, @@ -222,6 +223,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 # ---------------------------------------------------------------------------