refactor: refactor clean_projects to reduce redundancy in outputs

doc/release_notes.md

@@ -80,6 +80,8 @@
 
 * Use `combine_indicators` instead of `collect_indicators` in the function `summary_benchmark_indicators` ([#797](https://github.com/open-energy-transition/open-tyndp/pull/797)).
 
+* Refactor `clean_projects` script to reduce the redundancy in its outputs ([#807](https://github.com/open-energy-transition/open-tyndp/issues/807)).
+
 
 ## Upcoming PyPSA-Eur Release
 

scripts/cba/clean_projects.py

@@ -16,17 +16,17 @@
 the user can modify existing projects and add new ones.
 
 - Using an existing combination (`project_id`, `bus0`, `bus1`), the user can overwrite any
-field of an existing project with a custom value. Not all fields need to be specified; leaving
-a field empty keeps its existing value.
+  field of an existing project with a custom value. Not all fields need to be specified; leaving
+  a field empty keeps its existing value.
 
-- New projects are added as PINT projects. A new `project_id` must be used, and custom projects are added as links.
+- New projects must use a new `project_id` and are added as PINT links.
 
 **Inputs**
 
 - `data/tyndp_2024_bundle/cba_projects/20250312_export_transmission.xlsx`: Excel file containing CBA transmission projects
 - `data/tyndp_2024_bundle/cba_projects/20250312_export_storage.xlsx`: Excel file containing CBA storage projects (not yet processed)
 - `rules.retrieve_tyndp.output.nodes`: List of nodes defined in the workflow
-- `rules.retrieve_cba_guidelines_reference_projects.output.file`: Overview of the projects included in the reference grid, as defined in the Implementation Guidelines.
+- `rules.retrieve_cba_guidelines_reference_projects.output.file`: Overview of the projects included in the reference grid, as defined in the Implementation Guidelines
 - `data/custom_cba_transmission_projects.csv`: File used to configure custom transmission projects. With it, the user can modify existing projects and add new ones.
 
 **Outputs**
@@ -45,6 +45,8 @@
 
 - `resources/cba/storage_projects.csv`: Empty CSV with columns project_id and project_name (stub implementation)
 
+- `resources/cba/cba_project_methods.csv`: List of the method to apply on projects
+
 """
 
 import logging
@@ -71,21 +73,23 @@
 }
 
 
-def read_tyndp_electricity_buses(buses_fn: str):
+def read_tyndp_electricity_buses(buses_fn: str) -> pd.Index:
     """
     Read node list for electricity from tyndp data input.
 
     Parameters
     ----------
-        - buses_fn (str): Path to "LIST OF NODES.xlsx" from tyndp bundle
+    buses_fn : str
+        Path to "LIST OF NODES.xlsx" from tyndp bundle.
 
     Returns
     -------
-        - buses: Index of electricity buses as used in Open-TYNDP
+    pd.Index
+        Index of electricity buses as used in Open-TYNDP.
 
     See Also
     --------
-        build_tyndp_network.py : build_buses
+    build_tyndp_network.py : build_buses
     """
     buses = pd.Index(
         pd.read_excel(buses_fn)
@@ -148,7 +152,7 @@ def remove_no_capacity(projects: pd.DataFrame) -> pd.DataFrame:
     Returns
     -------
     pd.DataFrame
-        Curated list of projects with a defined capacity
+        Curated list of projects with a defined capacity.
     """
     if projects.empty:
         return projects
@@ -227,7 +231,7 @@ def extract_transmission_projects(
         }
     )
 
-    # Clean the project list by removing unclear border and projects with no capacity
+    # Clean the project list by removing projects with unclear borders or no capacity
     projects = remove_unclear_border(projects, existing_buses)
     projects = remove_no_capacity(projects)
 
@@ -358,7 +362,7 @@ def overwrite_projects(
     new_projects = custom_projects.index.difference(projects.index)
     existing_projects = custom_projects.index.intersection(projects.index)
 
-    # Fill missing values with existing projects
+    # Fill missing values from existing projects
     custom_projects = custom_projects.reindex(columns=projects.columns).fillna(projects)
     custom_projects["is_crossborder"] = (
         custom_projects["is_crossborder"].fillna(True).astype(bool)
@@ -395,12 +399,28 @@ def normalize_yes_no(value: str) -> str:
 
 
 def compute_method(flag: str) -> str:
-    return "TOOT" if flag == "yes" else "PINT"
+    return "toot" if flag == "yes" else "pint"
 
 
 def build_method_assignments(
     guidelines: pd.DataFrame, projects: pd.DataFrame
 ) -> pd.DataFrame:
+    """
+    Determine the CBA assessment method for each project. The method is PINT (default) or TOOT and
+    depends on the planning horizon (2030 or 2040).
+
+    Parameters
+    ----------
+    guidelines : pd.DataFrame
+        Overview of the projects included in the reference grid, as defined in the Implementation Guidelines.
+    projects : pd.DataFrame
+        List of projects.
+
+    Returns
+    -------
+    pd.DataFrame
+        DataFrame of projects with the corresponding method to apply.
+    """
     guidelines = guidelines.rename(
         columns={
             "ID": "project_id",
@@ -444,14 +464,16 @@ def build_method_assignments(
                     "in_ref_grid_2030": "no",
                     "in_ref_grid_2040": "no",
                     "planning_horizon": horizon,
-                    "method": "PINT",
+                    "method": "pint",
                 }
             )
             rows = pd.concat([rows, missing_rows], ignore_index=True)
         assigned.append(rows)
 
-    assigned = pd.concat(assigned, ignore_index=True)
-    return projects.merge(assigned, on="project_id", how="left")
+    assigned = pd.concat(assigned, ignore_index=True).query(
+        "project_id in @projects.project_id"
+    )
+    return assigned
 
 
 if __name__ == "__main__":
@@ -492,7 +514,7 @@ def build_method_assignments(
 
     # Storage projects
     storage_projects = extract_storage_projects(storage_path, existing_buses)
-    # TODO Add overwrite_projects for storages
+    # TODO Add overwrite_projects for storage projects
     storage_projects.to_csv(snakemake.output.storage_projects, index=False)
 
     # Method definition (PINT / TOOT)

scripts/cba/make_indicators.py

@@ -35,6 +35,7 @@
 import pypsa
 
 from scripts._helpers import configure_logging, set_scenario_config
+from scripts.cba.prepare_project import load_method
 from scripts.prepare_sector_network import get
 
 logger = logging.getLogger(__name__)
@@ -130,18 +131,6 @@ def calculate_total_system_cost(
     }
 
 
-def check_method(method: str) -> str:
-    """
-    Normalize and validate the CBA method name.
-
-    If the method is not recognized as either "pint" or "toot", a ValueError is raised.
-    """
-    method = method.lower()
-    if method not in ["pint", "toot"]:
-        raise ValueError(f"Method must be 'pint' or 'toot', got: {method}")
-    return method
-
-
 def difference_by_method(reference: float, project: float, method: str) -> float:
     """
     Calculate differences for indicator values by method (TOOT / PINT).
@@ -999,16 +988,7 @@ def load_benchmark_rows(
     # Detect method from assignments (toot or pint)
     cba_project = snakemake.wildcards.cba_project
     project_id = int(cba_project[1:])
-    methods = pd.read_csv(snakemake.input.methods)
-    method_row = methods[
-        (methods["project_id"] == project_id)
-        & (methods["planning_horizon"] == planning_horizon)
-    ]
-    if method_row.empty:
-        raise ValueError(
-            f"Missing CBA method for project {project_id} and horizon {planning_horizon}"
-        )
-    method = check_method(method_row["method"].iloc[0])
+    method = load_method(snakemake.input.methods, project_id, planning_horizon)
 
     # Calculate indicators
     indicators = {}

scripts/cba/prepare_project.py

@@ -20,7 +20,40 @@
 logger = logging.getLogger(__name__)
 
 
-def load_method(methods: pd.DataFrame, project_id: int, planning_horizon: int) -> str:
+def check_method(method: str) -> str:
+    """
+    Normalize and validate a given CBA method name.
+
+    Raises
+    ------
+    ValueError
+        If the normalized value is neither "pint" nor "toot".
+    """
+    method = method.lower().strip()
+    if method not in ["pint", "toot"]:
+        raise ValueError(f"Method must be 'pint' or 'toot', got: {method}")
+    return method
+
+
+def load_method(methods_fn: str, project_id: int, planning_horizon: int) -> str:
+    """
+    Load the method for a specific project and planning horizon.
+
+    Parameters
+    ----------
+    methods_fn : str
+        Path to the file defining the methods.
+    project_id : int
+        Project reference ID.
+    planning_horizon : int
+        Planning horizon.
+
+    Returns
+    -------
+    str
+        Method to be used to assess a project at a planning horizon.
+    """
+    methods = pd.read_csv(methods_fn)
     row = methods[
         (methods["project_id"] == project_id)
         & (methods["planning_horizon"] == planning_horizon)
@@ -29,7 +62,7 @@ def load_method(methods: pd.DataFrame, project_id: int, planning_horizon: int) -
         raise ValueError(
             f"Missing CBA method for project {project_id} and horizon {planning_horizon}"
         )
-    return str(row["method"].iloc[0]).strip().upper()
+    return check_method(row["method"].iloc[0])
 
 
 def apply_toot(
@@ -145,38 +178,37 @@ def apply_pint(
     configure_logging(snakemake)
     set_scenario_config(snakemake)
 
-    n = pypsa.Network(snakemake.input.network)
+    cba_project = snakemake.wildcards.cba_project
+    planning_horizon = int(snakemake.wildcards.planning_horizons)
+    methods_fn = snakemake.input.methods
     transmission_projects = pd.read_csv(snakemake.input.transmission_projects)
-    methods = pd.read_csv(snakemake.input.methods)
+    n = pypsa.Network(snakemake.input.network)
+    hurdle_costs = snakemake.params.hurdle_costs
+    costs = pd.read_csv(snakemake.input.costs, index_col=0)
 
-    cba_project = snakemake.wildcards.cba_project
     project_id = int(cba_project[1:])
-    planning_horizon = int(snakemake.wildcards.planning_horizons)
     if planning_horizon not in [2030, 2040]:
         logger.warning(
             "CBA methods are only available for 2030 or 2040. Using 2040 for planning horizon %s.",
             snakemake.wildcards.planning_horizons,
         )
         planning_horizon = 2040
 
-    method = load_method(methods, project_id, planning_horizon)
-    hurdle_costs = snakemake.params.hurdle_costs
+    method = load_method(methods_fn, project_id, planning_horizon)
     negative_toot_capacity = snakemake.config["cba"].get(
         "negative_toot_capacity", "zero"
     )
 
-    costs = pd.read_csv(snakemake.input.costs, index_col=0)
-
     transmission_project = transmission_projects[
         transmission_projects["project_id"] == project_id
     ]
     assert not transmission_project.empty, (
         f"Transmission project {project_id} not found."
     )
 
-    if method == "TOOT":
+    if method == "toot":
         apply_toot(n, transmission_project, negative_toot_capacity)
-    elif method == "PINT":
+    elif method == "pint":
         apply_pint(n, transmission_project, hurdle_costs, costs)
     else:
         raise ValueError(f"Unknown method {method} for project {project_id}")