refactor: refactor clean_projects to reduce redundancy in outputs

Author: tgilon

scripts/cba/clean_projects.py

@@ -71,7 +71,7 @@
 }
 
 
-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.
 
@@ -392,12 +392,29 @@ 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:
+    """
+    Define the method to apply on a project. The method can be PINT (default) or TOOT and can verify 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
+        List of the method to apply on projects.
+
+    """
     guidelines = guidelines.rename(
         columns={
             "ID": "project_id",
@@ -441,14 +458,14 @@ 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")
+    return assigned
 
 
 if __name__ == "__main__":

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,38 @@
 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 the CBA method name.
+
+    If the method is not recognized as either "pint" or "toot", a ValueError is raised.
+    """
+    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 +60,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 +176,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}")