45 new variable statistics final energy by carriernatural gas

pypsa_validation_processing/configs/mapping.default.yaml

@@ -6,6 +6,7 @@
 # Primary Energy|Coal: primary_energy_coal
 
 Final Energy [by Carrier]|Electricity: Final_Energy_by_Carrier__Electricity
+Final Energy [by Carrier]|Natural Gas: Final_Energy_by_Carrier__Natural_Gas
 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

pypsa_validation_processing/statistics_functions.py

@@ -26,9 +26,10 @@ def <function_name>(network: pypsa.Network) -> pd.Series:
 import pandas as pd
 import numpy as np
 import pypsa
-from pypsa_validation_processing.utils import statistics_kwargs as kwargs
 from pypsa_validation_processing.utils import (
     statistics_kwargs_for_filtering as kwargs_filtering,
+    statistics_kwargs as kwargs,
+    UNITS_MAPPING,
 )
 from pypsa_validation_processing.utils import (
     statistics_grouping_index,
@@ -133,6 +134,91 @@ def Final_Energy_by_Carrier__Electricity(
     return result
 
 
+def Final_Energy_by_Carrier__Natural_Gas(
+    n: pypsa.Network,
+    aggregate_per_year: bool = True,
+) -> pd.Series | pd.DataFrame:
+    """IAMC variable Final Energy[by Carrier]|Gas"""
+    # get fraction fossil-gas non-fossil-gas
+    # non-fossil-gas-production per region
+    non_fossil_gas_prod = n.statistics.supply(
+        bus_carrier="gas",
+        carrier=[
+            "Sabatier",
+            "biogas to gas",
+            "biogas to gas CC",
+            "BioSNG",
+            "BioSNG CC",
+        ],
+        at_port="bus1",
+        components="Link",
+        **kwargs,
+    )
+
+    # all gas-usage per region
+    all_gas = n.statistics.withdrawal(
+        bus_carrier="gas", components="Link", **kwargs_filtering
+    )
+    all_gas.index.get_level_values("carrier").unique()
+    forbidden_parts = ["pipeline"]
+    forbidden_pattern = "|".join(re.escape(part) for part in forbidden_parts)
+
+    total_gas_usage_carriers = all_gas.index.get_level_values("carrier").astype(str)
+    forbidden_mask = total_gas_usage_carriers.str.contains(
+        forbidden_pattern, case=False, regex=True
+    )
+    total_gas_usage = all_gas[~forbidden_mask]
+    total_gas_usage = total_gas_usage.groupby(kwargs["groupby"]).sum()
+
+    # fraction of usage and production values
+    non_fossil_fraction = non_fossil_gas_prod / total_gas_usage
+    non_fossil_fraction = non_fossil_fraction.replace([np.inf, -np.inf], np.nan)
+    non_fossil_fraction = non_fossil_fraction.clip(upper=1)
+    non_fossil_fraction = non_fossil_fraction.groupby(kwargs["groupby"]).mean()
+    non_fossil_fraction = non_fossil_fraction.rename(index=UNITS_MAPPING)
+
+    # Final Energy|Residential and Commercial|Natural Gas - urban decentral gas boiler
+    rescom = n.statistics.withdrawal(
+        bus_carrier="gas",
+        carrier=["urban decentral gas boiler", "rural gas boiler"],
+        components="Link",
+        **kwargs,
+    )
+
+    # Final Energy|Industry|Natural Gas - gas for industry
+    industry = n.statistics.withdrawal(
+        bus_carrier="gas for industry",
+        carrier=["gas for industry", "as for industry CC"],
+        components="Load",
+        **kwargs,
+    )
+
+    # get fraction of non-energetic use in industry
+    # data from Eurostat energy balance for 2024 | EU27 | in TWh
+    # online available (used 2026-04-28): https://ec.europa.eu/eurostat/cache/visualisations/energy-balances/enbal.html?geo=EU27_2020&unit=KTOE&language=EN&year=&fuel=fuelMainFuel&siec=TOTAL&details=1&chartOptions=0&stacking=normal&chartBal=&chart=&full=0&chartBalText=&order=DESC&siecs=&dataset=nrg_bal_c&decimals=0&agregates=0&share=false&fuelList=fuelElectricity%2CfuelCombustible%2CfuelNonCombustible%2CfuelOtherPetroleum%2CfuelMainPetroleum%2CfuelOil%2CfuelOtherFossil%2CfuelFossil%2CfuelCoal%2CfuelMainFuel
+    # Final consumption - energy use: 7 217 049
+    # Final consumption - non-energy use (fuels not combusted): 458 572
+    fc_energy = 7217049
+    fc_noenergy = 458572
+    energy_fraction_industry_gas = fc_energy / (fc_energy + fc_noenergy)
+    industry = industry.mul(energy_fraction_industry_gas)
+
+    series_list = [rescom, industry]
+    series_list = [series for series in series_list if not series.empty]
+
+    if not series_list:
+        return pd.Series(
+            dtype=float,
+            index=pd.MultiIndex.from_tuples([], names=kwargs["groupby"]),
+        )
+
+    total = pd.concat(series_list)
+    total = total.rename(index=UNITS_MAPPING).groupby(kwargs["groupby"]).sum()
+    non_fossil_fraction = non_fossil_fraction.reindex_like(total).fillna(0)
+    result = total.mul(1 - non_fossil_fraction, axis=0)
+    return result
+
+
 def Final_Energy_by_Sector__Transportation(
     n: pypsa.Network,
     aggregate_per_year: bool = True,

pypsa_validation_processing/utils.py

@@ -128,6 +128,7 @@
     "land transport": "MWh",
     "t_co2": "t",
     "": "",
+    "MWh": "MWh",
 }
 
 ## standards for statistics-functions

tests/test_statistics_functions.py

@@ -7,6 +7,7 @@
 
 from pypsa_validation_processing.statistics_functions import (
     Final_Energy_by_Carrier__Electricity,
+    Final_Energy_by_Carrier__Natural_Gas,
     Final_Energy_by_Sector__Industry,
     Final_Energy_by_Sector__Agriculture,
     Final_Energy_by_Sector__Transportation,
@@ -219,6 +220,181 @@ def test_issues_expected_withdrawal_queries(self):
         assert calls[4]["components"] == "Link"
 
 
+# ---------------------------------------------------------------------------
+# Tests for Final_Energy_by_Carrier__Natural_Gas
+# ---------------------------------------------------------------------------
+
+
+class TestFinalEnergyByCarrierNaturalGas:
+    """Test suite for Final_Energy_by_Carrier__Natural_Gas function."""
+
+    class _NaturalGasStatisticsAccessor:
+        """Minimal accessor to verify natural-gas extraction behavior."""
+
+        def __init__(self, scenario: str = "mixed"):
+            self.scenario = scenario
+
+        @staticmethod
+        def _empty_series(groupby: list[str]) -> pd.Series:
+            return pd.Series(
+                dtype=float,
+                index=pd.MultiIndex.from_tuples([], names=groupby),
+            )
+
+        @staticmethod
+        def _series_from_groupby(
+            groupby: list[str],
+            values: list[float],
+            location: str = "AT1",
+            unit: str = "MWh_LHV",
+        ) -> 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 supply(
+            self,
+            bus_carrier: str | None = None,
+            carrier: list[str] | str | None = None,
+            at_port: str | None = None,
+            components: str | list[str] | None = None,
+            groupby: list[str] | None = None,
+            nice_names: bool | None = None,
+            **_: object,
+        ) -> pd.Series:
+            if groupby is None:
+                groupby = ["location", "unit"]
+
+            if (
+                bus_carrier == "gas"
+                and isinstance(carrier, list)
+                and components == "Link"
+                and at_port == "bus1"
+            ):
+                if self.scenario in ("no_gas", "no_renewable"):
+                    return self._empty_series(groupby)
+                if self.scenario == "no_fossil":
+                    return self._series_from_groupby(groupby, [100.0])
+                return self._series_from_groupby(groupby, [20.0])
+
+            return self._empty_series(groupby)
+
+        def withdrawal(
+            self,
+            bus_carrier: str | None = None,
+            carrier: list[str] | str | None = None,
+            components: str | list[str] | None = None,
+            groupby: list[str] | None = None,
+            nice_names: bool | None = None,
+            **_: object,
+        ) -> pd.Series:
+            if groupby is None:
+                groupby = ["location", "unit"]
+
+            if (
+                bus_carrier == "gas"
+                and components == "Link"
+                and groupby == ["name", "bus", "carrier", "location", "unit"]
+            ):
+                if self.scenario == "no_gas":
+                    return self._empty_series(groupby)
+
+                index = pd.MultiIndex.from_tuples(
+                    [
+                        (
+                            "gas_link",
+                            "AT1 gas",
+                            "urban decentral gas boiler",
+                            "AT1",
+                            "MWh_LHV",
+                        ),
+                        (
+                            "pipeline_link",
+                            "AT1 gas",
+                            "gas pipeline",
+                            "AT1",
+                            "MWh_LHV",
+                        ),
+                    ],
+                    names=groupby,
+                )
+                return pd.Series([100.0, 900.0], index=index, dtype=float)
+
+            if (
+                bus_carrier == "gas"
+                and carrier == ["urban decentral gas boiler", "rural gas boiler"]
+                and components == "Link"
+            ):
+                if self.scenario == "no_gas":
+                    return self._empty_series(groupby)
+                return self._series_from_groupby(groupby, [40.0])
+
+            if (
+                bus_carrier == "gas for industry"
+                and carrier == ["gas for industry", "as for industry CC"]
+                and components == "Load"
+            ):
+                if self.scenario == "no_gas":
+                    return self._empty_series(groupby)
+                return self._series_from_groupby(groupby, [60.0])
+
+            return self._empty_series(groupby)
+
+    class _NaturalGasNetwork:
+        """Minimal network exposing the custom natural-gas statistics accessor."""
+
+        def __init__(self, scenario: str = "mixed"):
+            self.statistics = (
+                TestFinalEnergyByCarrierNaturalGas._NaturalGasStatisticsAccessor(
+                    scenario=scenario
+                )
+            )
+
+    def _natural_gas_network(self, scenario: str = "mixed"):
+        return self._NaturalGasNetwork(scenario=scenario)
+
+    def test_returns_series(self):
+        """Function returns a Series with expected output format."""
+        result = Final_Energy_by_Carrier__Natural_Gas(self._natural_gas_network())
+        assert isinstance(result, pd.Series)
+        assert isinstance(result.index, pd.MultiIndex)
+        assert result.index.names == ["location", "unit"]
+
+    def test_filters_pipeline_from_total_gas_usage(self):
+        """Pipeline carriers are excluded when building the non-fossil share denominator."""
+        result = Final_Energy_by_Carrier__Natural_Gas(
+            self._natural_gas_network("mixed")
+        )
+        # total=100, non-fossil share=20/100, result=100*(1-0.2)=80
+        assert result.loc[("AT1", "MWh")] == 80.0
+
+    def test_edge_case_no_gas_returns_empty_series(self):
+        """No gas usage and no gas demand should return an empty result."""
+        result = Final_Energy_by_Carrier__Natural_Gas(
+            self._natural_gas_network("no_gas")
+        )
+        assert isinstance(result, pd.Series)
+        assert isinstance(result.index, pd.MultiIndex)
+        assert result.index.names == ["location", "unit"]
+        assert result.empty
+
+    def test_edge_case_no_fossil_gas_returns_zero(self):
+        """If all gas is renewable, fossil natural gas final energy must be zero."""
+        result = Final_Energy_by_Carrier__Natural_Gas(
+            self._natural_gas_network("no_fossil")
+        )
+        assert result.loc[("AT1", "MWh")] == 0.0
+
+    def test_edge_case_no_renewable_gas_returns_total(self):
+        """If there is no renewable gas production, all demand counts as fossil gas."""
+        result = Final_Energy_by_Carrier__Natural_Gas(
+            self._natural_gas_network("no_renewable")
+        )
+        assert result.loc[("AT1", "MWh")] == 100.0
+
+
 # ---------------------------------------------------------------------------
 # Tests for Final_Energy_by_Sector__Transportation
 # ---------------------------------------------------------------------------