Debug Final Energy [by Sector]|Transportation

README.md

@@ -127,7 +127,9 @@ def <function_name>(
     ...
 ```
 
-If a variable-specific function needs additional settings, an optional `config: dict` argument can be added and is passed automatically by the processor when present.
+If a variable-specific function needs additional settings, optional function parameters can be added. Currently, these include: 
+- `config: dict`: dict of the configuration 
+- `energy_totals: Path`: Path to the file energy_totals, needed to calculate domestic-to-international ratios. This path is currently set to `self.network_results_path / "resources" / "energy_totals.csv"`
 
 Return format rules:
 

pypsa_validation_processing/class_definitions.py

@@ -288,6 +288,11 @@ def _execute_function_for_variable(
             kwargs["config"] = config
         if "aggregate_per_year" in params:
             kwargs["aggregate_per_year"] = self.aggregate_per_year
+        if "energy_totals" in params:
+            kwargs["energy_totals"] = (
+                self.network_results_path / "resources" / "energy_totals.csv"
+            )
+
         return func(n, **kwargs)
 
     def _aggregate_to_country(self, result: pd.DataFrame) -> pd.DataFrame:

pypsa_validation_processing/statistics_functions.py

@@ -20,9 +20,19 @@ def <function_name>(network: pypsa.Network) -> pd.Series:
 """
 
 from __future__ import annotations
-
+from functools import reduce
+from pathlib import Path
 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,
+)
+from pypsa_validation_processing.utils import (
+    statistics_grouping_index,
+    get_energy_totals_domestic_share,
+)
 
 
 def Final_Energy_by_Carrier__Electricity(
@@ -83,56 +93,132 @@ def Final_Energy_by_Carrier__Electricity(
 def Final_Energy_by_Sector__Transportation(
     n: pypsa.Network,
     aggregate_per_year: bool = True,
+    energy_totals: Path | None = None,
 ) -> pd.Series | pd.DataFrame:
     """Extract transportation-sector final energy from a PyPSA Network.
 
-    Returns the total energy consumed by the transportation sector (excluding
-    transmission / distribution losses)
+    Returns the total energy consumed by the Transportation sector
+    (excluding transmission / distribution losses) across the 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.Series`. If ``False``, return a
         :class:`pandas.DataFrame` with snapshots as columns.
+    energy_totals : pathlib.Path | None, optional
+        Path to an energy totals file used to determine domestic shares for
+        aviation and navigation liquid fuels. If ``None``, default domestic
+        shares from :func:`get_energy_totals_domestic_share` are used.
 
     Returns
     -------
     pd.Series | pd.DataFrame
         Pandas Series (``aggregate_per_year=True``) or DataFrame
-        (``aggregate_per_year=False``) with MultiIndex of ``location`` and
-        ``unit``.
+        (``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
     -----
-    Includes all transportation-relevant carriers for component Load. Vehicle to Grid
-    does not need to be evaluated, as evaluation is restricted to Load-Components only.
+    Sums transportation final energy from electricity, hydrogen, and liquid
+    fuels:
+    - Electricity demand from BEV charging loads.
+    - Additional EV charging losses computed from BEV charger link flows and
+      adjusted for V2G participation.
+    - Hydrogen demand from fuel-cell land transport.
+    - Liquid fuels for aviation and navigation scaled by domestic fractions,
+      plus land-transport oil.
+
+    Raises
+    ------
+    ValueError
+        If BEV charging flow sign conventions are violated.
+    TypeError
+        If intermediate statistics return mixed result types.
     """
-    # sum over all transportation-relevant sectors - 2 different units involved.
-    res = (
-        n.statistics.energy_balance(
-            carrier=[
-                "land transport EV",
-                "land transport fuel cell",
-                "land transport oil",
-                "kerosene for aviation",
-                "shipping methanol",
-                "shipping oil",
-            ],
+
+    domestic_aviation_fraction = get_energy_totals_domestic_share(
+        energy_totals, "aviation"
+    )
+    domestic_navigation_fraction = get_energy_totals_domestic_share(
+        energy_totals, "navigation"
+    )
+
+    # get Final Energy [by Sector]|Transportation|Electricity
+    # get elec load losses from BEV-charger links
+    bev_charger_efficiencies = n.links.loc[
+        n.links.carrier == "BEV charger", "efficiency"
+    ].dropna()
+    if bev_charger_efficiencies.nunique() == 1:
+        eff = bev_charger_efficiencies.iloc[0]
+    else:
+        eff = bev_charger_efficiencies.mean()
+        print(
+            "WARNING: Network includes different efficiencies for BEV chargers. Using mean value for variable Final_Energy_by_Sector__Transportation"
+        )
+
+    elec = n.statistics.withdrawal(
+        bus_carrier="low voltage",
+        carrier="BEV charger",
+        components="Link",
+        aggregate_time=aggregate_per_year,
+        **kwargs,
+    ).mul(
+        1 / eff
+    )  # ( 1 + ((1/eff)-1))
+
+    # get Final Energy [by Sector]|Transportation|Hydrogen
+    h2 = n.statistics.withdrawal(
+        carrier="land transport fuel cell",
+        components="Load",
+        aggregate_time=aggregate_per_year,
+        **kwargs,
+    )
+
+    # get Final Energy [by Sector]|Transportation|Liquids
+    aviation_liquids = (
+        n.statistics.withdrawal(
+            carrier="kerosene for aviation",
             components="Load",
-            groupby=["carrier", "unit", "location"],
-            direction="withdrawal",  # for positive values
-            groupby_time=aggregate_per_year,
+            aggregate_time=aggregate_per_year,
+            **kwargs,
         )
-        .groupby(["location", "unit"])
-        .sum()
+        * domestic_aviation_fraction
     )
-    return res
+
+    navigation_liquids = (
+        n.statistics.withdrawal(
+            carrier=["shipping oil", "shipping methanol"],
+            components="Load",
+            aggregate_time=aggregate_per_year,
+            **kwargs,
+        )
+        * domestic_navigation_fraction
+    )
+
+    land_transport_liquids = n.statistics.withdrawal(
+        carrier="land transport oil",
+        components="Load",
+        aggregate_time=aggregate_per_year,
+        **kwargs,
+    )
+
+    series_list = [
+        elec,
+        h2,
+        aviation_liquids,
+        navigation_liquids,
+        land_transport_liquids,
+    ]
+    series_list = [series for series in series_list if not series.empty]
+
+    total = pd.concat(series_list)
+    return total
 
 
 def Final_Energy_by_Sector__Industry(

pypsa_validation_processing/utils.py

@@ -1,5 +1,9 @@
 """Static information and general utility functions for pypsa_validation_processing."""
 
+import pandas as pd
+from pathlib import Path
+
+
 EU27_COUNTRY_CODES: dict[str, str] = {
     "AL": "Albania",
     "AT": "Austria",
@@ -121,6 +125,50 @@
     "MWh_el": "MWh",
     "MWh_LHV": "MWh",
     "MWh_th": "MWh",
+    "land transport": "MWh",
     "t_co2": "t",
     "": "",
 }
+
+## standards for statistics-functions
+# standardize kwargs for pypsa-statistics statements
+statistics_kwargs = {
+    "groupby": ["location", "unit"],
+    "nice_names": False,
+}
+statistics_kwargs_for_filtering = {
+    "groupby": ["name", "bus", "carrier", "location", "unit"],
+    "nice_names": False,
+}
+# standardize MultiIndex
+statistics_grouping_index = ["location", "unit"]
+
+
+## UTILS FUNCTIONS
+
+
+def get_energy_totals_domestic_share(
+    energy_totals: Path,
+    kind: str,
+) -> pd.Series:
+    """
+    Return the domestic share of energy totals for a given kind.
+
+    Parameters
+    ----------
+    energy_totals
+        Path to the energy totals csv file.
+    kind: {'aviation', 'navigation'}
+        The kind of energy totals to calculate the factor for.
+
+    Returns
+    -------
+    :
+        The share of national aviation or navigation per country.
+    """
+    # TODO generalize energy totals for all countries
+    energy_totals = pd.read_csv(energy_totals, index_col="country").loc["AT"]
+    energy_totals = energy_totals[energy_totals.year == 2020]
+    domestic = energy_totals[f"total domestic {kind}"]
+    international = energy_totals[f"total international {kind}"]
+    return (domestic / (domestic + international)).values[0]

tests/conftest.py

@@ -12,7 +12,7 @@
 class MockStatisticsAccessor:
     """Mock PyPSA Statistics accessor for testing.
 
-    This mock provides the energy_balance method that returns realistic
+    This mock provides energy_balance/withdrawal methods that return realistic
     pandas Series with MultiIndex structure matching PyPSA output.
     """
 
@@ -29,6 +29,8 @@ def energy_balance(
         direction: str = "withdrawal",
         at_port: list[str] | None = None,
         groupby_time: bool = True,
+        nice_names: bool | None = None,
+        **_: object,
     ) -> pd.Series | pd.DataFrame:
         """Mock energy_balance method for PyPSA Network.statistics.
 
@@ -53,6 +55,8 @@ def energy_balance(
         groupby_time : bool
             If ``True`` (default) return an aggregated Series.
             If ``False`` return a DataFrame with 4 hourly timestamps as columns.
+        nice_names : bool | None
+            Ignored in mock, accepted for compatibility with production calls.
 
         Returns
         -------
@@ -83,14 +87,22 @@ def energy_balance(
                 for unit in ["MWh_el", "MWh_th"]:
                     # Create index tuple based on groupby keys
                     idx_dict = {
+                        "name": f"{c}_{location}",
+                        "bus": f"{location} bus",
                         "carrier": c,
                         "location": location,
                         "unit": unit,
                     }
-                    idx_tuple = tuple(idx_dict[key] for key in groupby)
+                    idx_tuple = tuple(idx_dict.get(key, f"mock_{key}") for key in groupby)
                     index_tuples.append(idx_tuple)
-                    # Mock value: roughly realistic energy value
-                    values.append(1000.0)
+                    # Provide deterministic signs for link ports used in
+                    # transportation charging-loss calculations.
+                    if components == "Link" and at_port == ["bus0"]:
+                        values.append(-1000.0)
+                    elif components == "Link" and at_port == ["bus1"]:
+                        values.append(900.0)
+                    else:
+                        values.append(1000.0)
 
         # Create MultiIndex
         index = pd.MultiIndex.from_tuples(index_tuples, names=groupby)
@@ -106,6 +118,33 @@ def energy_balance(
                 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,
+        **kwargs: object,
+    ) -> pd.Series | pd.DataFrame:
+        """Mock withdrawal method forwarding to energy_balance.
+
+        PyPSA's statistics.withdrawal uses ``aggregate_time`` while
+        ``energy_balance`` uses ``groupby_time``. This adapter keeps tests
+        compatible with either call style.
+        """
+        return self.energy_balance(
+            bus_carrier=bus_carrier,
+            carrier=carrier,
+            components=components,
+            groupby=groupby,
+            direction="withdrawal",
+            groupby_time=aggregate_time,
+            nice_names=nice_names,
+            **kwargs,
+        )
+
 
 class MockPyPSANetwork:
     """Minimal mock PyPSA Network for unit testing.
@@ -129,6 +168,13 @@ def __init__(self, name: str = "test_network", **kwargs):
             "wildcards": {"planning_horizons": 2020},
         }
         self.statistics = MockStatisticsAccessor()
+        self.links = pd.DataFrame(
+            {
+                "carrier": ["BEV charger", "BEV charger", "other link"],
+                "efficiency": [0.9, 0.9, 1.0],
+            },
+            index=["bev_charger_at1", "bev_charger_at2", "other_link_at1"],
+        )
         # Add carriers attribute with empty index by default
         self.carriers = pd.DataFrame(index=[])
 
@@ -212,3 +258,21 @@ def temp_config_file(tmp_path: Path) -> Path:
     config_file = tmp_path / "config.yaml"
     config_file.write_text(config_content)
     return config_file
+
+
+@pytest.fixture
+def energy_totals_csv(tmp_path: Path) -> Path:
+    """Fixture providing a minimal energy_totals.csv for domestic-share tests."""
+    energy_totals = pd.DataFrame(
+        {
+            "country": ["AT", "AT"],
+            "year": [2020, 2021],
+            "total domestic aviation": [30.0, 0.0],
+            "total international aviation": [70.0, 0.0],
+            "total domestic navigation": [20.0, 0.0],
+            "total international navigation": [80.0, 0.0],
+        }
+    )
+    csv_path = tmp_path / "energy_totals.csv"
+    energy_totals.to_csv(csv_path, index=False)
+    return csv_path