New entrant dynamic properties

src/ispypsa/data_fetch/local_cache.py

@@ -10,6 +10,20 @@
     for table_name in [key + "_" + gen_type for gen_type in val]
 ]
 
+_NEW_ENTRANTS_COST_TABLES = [
+    "build_costs_scenario_mapping",
+    "build_costs_current_policies",
+    "build_costs_global_nze_by_2050",
+    "build_costs_global_nze_post_2050",
+    "build_costs_pumped_hydro",
+    "connection_costs_for_wind_and_solar",
+    "connection_costs_other",
+    "connection_cost_forecast_wind_and_solar_progressive_change",
+    "connection_cost_forecast_wind_and_solar_step_change&green_energy_exports",
+    "connection_cost_forecast_non_rez_progressive_change",
+    "connection_cost_forecast_non_rez_step_change&green_energy_exports",
+]
+
 _NETWORK_REQUIRED_TABLES = [
     "sub_regional_reference_nodes",
     "regional_topology_representation",
@@ -38,6 +52,7 @@
     _NETWORK_REQUIRED_TABLES
     + _GENERATORS_STORAGE_REQUIRED_SUMMARY_TABLES
     + _GENERATORS_REQUIRED_PROPERTY_TABLES
+    + _NEW_ENTRANTS_COST_TABLES
 )
 
 

src/ispypsa/templater/dynamic_generator_properties.py

@@ -2,8 +2,14 @@
 import re
 from pathlib import Path
 
+import numpy as np
 import pandas as pd
 
+from ispypsa.templater.helpers import (
+    _add_units_to_financial_year_columns,
+    _convert_financial_year_columns_to_float,
+)
+
 from .helpers import _snakecase_string
 from .lists import _ECAA_GENERATOR_TYPES
 
@@ -28,10 +34,23 @@ def template_generator_dynamic_properties(
     coal_prices = _template_coal_prices(parsed_workbook_path, scenario)
     gas_prices = _template_gas_prices(parsed_workbook_path, scenario)
     liquid_fuel_prices = _template_liquid_fuel_prices(parsed_workbook_path, scenario)
-    full_outage_forecasts = _template_full_outage_forecasts(parsed_workbook_path)
-    partial_outage_forecasts = _template_partial_outage_forecasts(parsed_workbook_path)
+    full_outage_forecasts = _template_existing_generators_full_outage_forecasts(
+        parsed_workbook_path
+    )
+    partial_outage_forecasts = _template_existing_generators_partial_outage_forecasts(
+        parsed_workbook_path
+    )
     seasonal_ratings = _template_seasonal_ratings(parsed_workbook_path)
     closure_years = _template_closure_years(parsed_workbook_path)
+    build_costs = _template_new_entrant_build_costs(parsed_workbook_path, scenario)
+    wind_and_solar_connection_costs = (
+        _template_new_entrant_wind_and_solar_connection_costs(
+            parsed_workbook_path, scenario
+        )
+    )
+    non_vre_connection_costs = _template_new_entrant_non_vre_connection_costs(
+        parsed_workbook_path
+    )
     return {
         "coal_prices": coal_prices,
         "gas_prices": gas_prices,
@@ -40,6 +59,10 @@ def template_generator_dynamic_properties(
         "partial_outage_forecasts": partial_outage_forecasts,
         "seasonal_ratings": seasonal_ratings,
         "closure_years": closure_years,
+        "build_costs": build_costs,
+        "new_entrant_build_costs": build_costs,
+        "new_entrant_wind_and_solar_connection_costs": wind_and_solar_connection_costs,
+        "new_entrant_non_vre_connection_costs": non_vre_connection_costs,
     }
 
 
@@ -63,7 +86,9 @@ def _template_coal_prices(
     coal_prices = pd.read_csv(
         Path(parsed_workbook_path, f"coal_prices_{snakecase_scenario}.csv")
     )
-    coal_prices.columns = _add_units_to_financial_year_columns(coal_prices.columns)
+    coal_prices.columns = _add_units_to_financial_year_columns(
+        coal_prices.columns, "$/GJ"
+    )
     coal_prices = coal_prices.drop(columns="coal_price_scenario")
     coal_prices = coal_prices.set_index("generator")
     coal_prices = _convert_financial_year_columns_to_float(coal_prices)
@@ -90,7 +115,7 @@ def _template_gas_prices(
     gas_prices = pd.read_csv(
         Path(parsed_workbook_path, f"gas_prices_{snakecase_scenario}.csv")
     )
-    cols = _add_units_to_financial_year_columns(gas_prices.columns)
+    cols = _add_units_to_financial_year_columns(gas_prices.columns, "$/GJ")
     cols[0] = "generator"
     gas_prices.columns = cols
     gas_prices = gas_prices.drop(columns="gas_price_scenario").set_index("generator")
@@ -115,46 +140,21 @@ def _template_liquid_fuel_prices(
         Path(parsed_workbook_path, "liquid_fuel_prices.csv")
     )
     liquid_fuel_prices.columns = _add_units_to_financial_year_columns(
-        liquid_fuel_prices.columns
-    )
-    liquid_fuel_prices = liquid_fuel_prices.drop(columns="liquid_fuel_price").set_index(
-        "liquid_fuel_price_scenario"
-    )
-    liquid_fuel_prices = _convert_financial_year_columns_to_float(liquid_fuel_prices)
-    liquid_fuel_prices_scenario = liquid_fuel_prices.loc[scenario, :]
-    liquid_fuel_prices_scenario.index.set_names("FY", inplace=True)
-    return liquid_fuel_prices_scenario
-
-
-def _template_liquid_fuel_prices(
-    parsed_workbook_path: Path | str, scenario: str
-) -> pd.Series:
-    """Creates a liquid fuel prices template
-
-    Args:
-        parsed_workbook_path: Path to directory with table CSVs that are the
-            outputs from the `isp-workbook-parser`.
-        scenario: Scenario obtained from the model configuration
-
-    Returns:
-        `pd.DataFrame`: ISPyPSA template for liquid fuel prices
-    """
-    liquid_fuel_prices = pd.read_csv(
-        Path(parsed_workbook_path, "liquid_fuel_prices.csv")
-    )
-    liquid_fuel_prices.columns = _add_units_to_financial_year_columns(
-        liquid_fuel_prices.columns
+        liquid_fuel_prices.columns, "$/GJ"
     )
     liquid_fuel_prices = liquid_fuel_prices.drop(columns="liquid_fuel_price").set_index(
         "liquid_fuel_price_scenario"
     )
     liquid_fuel_prices = _convert_financial_year_columns_to_float(liquid_fuel_prices)
     liquid_fuel_prices_scenario = liquid_fuel_prices.loc[scenario, :]
     liquid_fuel_prices_scenario.index.set_names("FY", inplace=True)
+    liquid_fuel_prices_scenario.name = "fuel_price"
     return liquid_fuel_prices_scenario
 
 
-def _template_full_outage_forecasts(parsed_workbook_path: Path | str) -> pd.DataFrame:
+def _template_existing_generators_full_outage_forecasts(
+    parsed_workbook_path: Path | str,
+) -> pd.DataFrame:
     """Creates a full outage forecast template for existing generators
 
     Args:
@@ -178,7 +178,7 @@ def _template_full_outage_forecasts(parsed_workbook_path: Path | str) -> pd.Data
     return full_outages_forecast
 
 
-def _template_partial_outage_forecasts(
+def _template_existing_generators_partial_outage_forecasts(
     parsed_workbook_path: Path | str,
 ) -> pd.DataFrame:
     """Creates a partial outage forecast template for existing generators
@@ -242,24 +242,157 @@ def _template_seasonal_ratings(
     return seasonal_rating
 
 
-def _add_units_to_financial_year_columns(columns: pd.Index) -> list[str]:
-    """Adds _$/GJ to the financial year columns"""
-    cols = [
-        _snakecase_string(col + "_$/GJ")
+def _template_new_entrant_build_costs(
+    parsed_workbook_path: Path | str, scenario: str
+) -> pd.DataFrame:
+    """Creates a new entrants build cost template
+
+    The function behaviour depends on the `scenario` specified in the model
+    configuration.
+
+    Args:
+        parsed_workbook_path: Path to directory with table CSVs that are the
+            outputs from the `isp-workbook-parser`.
+        scenario: Scenario obtained from the model configuration
+
+    Returns:
+        `pd.DataFrame`: ISPyPSA template for new entrant build costs
+    """
+    scenario_mapping = pd.read_csv(
+        Path(parsed_workbook_path, "build_costs_scenario_mapping.csv"), index_col=0
+    )
+    scenario_mapping = scenario_mapping.transpose().squeeze()
+    gencost_scenario_desc = re.match(
+        r"GenCost\s(.*)", scenario_mapping[scenario]
+    ).group(1)
+    build_costs_scenario = pd.read_csv(
+        Path(
+            parsed_workbook_path,
+            f"build_costs_{_snakecase_string(gencost_scenario_desc)}.csv",
+        )
+    )
+    build_costs_phes = pd.read_csv(
+        Path(
+            parsed_workbook_path,
+            f"build_costs_pumped_hydro.csv",
+        )
+    )
+    build_costs = pd.concat([build_costs_scenario, build_costs_phes], axis=0)
+    build_costs = _convert_financial_year_columns_to_float(build_costs)
+    build_costs = build_costs.drop(columns=["Source"])
+    # convert data in $/kW to $/MW
+    build_costs.columns = _add_units_to_financial_year_columns(
+        build_costs.columns, "$/MW"
+    )
+    build_costs = build_costs.set_index("technology")
+    build_costs *= 1000.0
+    return build_costs
+
+
+def _template_new_entrant_wind_and_solar_connection_costs(
+    parsed_workbook_path: Path | str, scenario: str
+) -> pd.DataFrame:
+    """Creates a new entrant wind and solar connection cost template
+
+    The function behaviour depends on the `scenario` specified in the model
+    configuration.
+
+    Args:
+        parsed_workbook_path: Path to directory with table CSVs that are the
+            outputs from the `isp-workbook-parser`.
+        scenario: Scenario obtained from the model configuration
+
+    Returns:
+        `pd.DataFrame`: ISPyPSA template for new entrant wind and solar connection costs
+    """
+    scenario = _snakecase_string(scenario)
+    if scenario == "step_change" or scenario == "green_energy_exports":
+        file_scenario = "step_change&green_energy_exports"
+    else:
+        file_scenario = scenario
+    # get rez cost forecasts and concatenate non-rez cost forecasts
+    wind_solar_connection_cost_forecasts = pd.read_csv(
+        Path(
+            parsed_workbook_path,
+            f"connection_cost_forecast_wind_and_solar_{file_scenario}.csv",
+        )
+    ).set_index("REZ names")
+    wind_solar_connection_cost_forecasts = wind_solar_connection_cost_forecasts.rename(
+        columns={"REZ network voltage (kV)": "Network voltage (kV)"}
+    )
+    non_rez_connection_cost_forecasts = pd.read_csv(
+        Path(
+            parsed_workbook_path,
+            f"connection_cost_forecast_non_rez_{file_scenario}.csv",
+        )
+    ).set_index("Non-REZ name")
+    wind_solar_connection_cost_forecasts = pd.concat(
+        [wind_solar_connection_cost_forecasts, non_rez_connection_cost_forecasts],
+        axis=0,
+    )
+    # get system strength connection cost from the initial connection cost table
+    initial_wind_solar_connection_costs = pd.read_csv(
+        Path(
+            parsed_workbook_path,
+            f"connection_costs_for_wind_and_solar.csv",
+        )
+    ).set_index("REZ names")
+    system_strength_cost = (
+        initial_wind_solar_connection_costs["System Strength connection cost ($/kW)"]
+        * 1000
+    ).rename("System strength connection cost ($/MW)")
+    wind_solar_connection_cost_forecasts = pd.concat(
+        [wind_solar_connection_cost_forecasts, system_strength_cost], axis=1
+    )
+    # remove notes
+    wind_solar_connection_cost_forecasts = wind_solar_connection_cost_forecasts.replace(
+        "Note 1", np.nan
+    )
+    # calculate $/MW by dividing total cost by connection capacity in MVA
+    wind_solar_connection_cost_forecasts = _convert_financial_year_columns_to_float(
+        wind_solar_connection_cost_forecasts
+    )
+    fy_cols = [
+        col
+        for col in wind_solar_connection_cost_forecasts.columns
         if re.match(r"[0-9]{4}-[0-9]{2}", col)
-        else _snakecase_string(col)
-        for col in columns
     ]
-    return cols
+    for col in fy_cols:
+        wind_solar_connection_cost_forecasts[col] /= (
+            wind_solar_connection_cost_forecasts["Connection capacity (MVA)"]
+        )
+    wind_solar_connection_cost_forecasts.columns = _add_units_to_financial_year_columns(
+        wind_solar_connection_cost_forecasts.columns, "$/MW"
+    )
+    return wind_solar_connection_cost_forecasts
 
 
-def _convert_financial_year_columns_to_float(df: pd.DataFrame) -> pd.DataFrame:
-    """Forcefully converts FY columns to float columns"""
-    cols = [
-        df[col].astype(float) if re.match(r"[0-9]{4}_[0-9]{2}", col) else df[col]
-        for col in df.columns
-    ]
-    return pd.concat(cols, axis=1)
+def _template_new_entrant_non_vre_connection_costs(
+    parsed_workbook_path: Path | str,
+) -> pd.DataFrame:
+    """Creates a new entrant non-VRE connection cost template
+
+    The function behaviour depends on the `scenario` specified in the model
+    configuration.
+
+    Args:
+        parsed_workbook_path: Path to directory with table CSVs that are the
+            outputs from the `isp-workbook-parser`.
+        scenario: Scenario obtained from the model configuration
+
+    Returns:
+        `pd.DataFrame`: ISPyPSA template for new entrant non-VRE connection costs
+    """
+    connection_costs = pd.read_csv(
+        Path(parsed_workbook_path, "connection_costs_other.csv")
+    ).set_index("Region")
+    # conveert to $/MW and add units to columns
+    col_rename_map = {}
+    for col in connection_costs.columns:
+        connection_costs[col] *= 1000
+        col_rename_map[col] = _snakecase_string(col) + "_$/mw"
+    connection_costs = connection_costs.rename(columns=col_rename_map)
+    return connection_costs
 
 
 def _convert_seasonal_columns_to_float(df: pd.DataFrame) -> pd.DataFrame:

src/ispypsa/templater/helpers.py

@@ -205,3 +205,25 @@ def _where_any_substring_appears(
         for i in range(2, len(wheres)):
             boolean = np.logical_or(boolean, wheres[i])
     return boolean
+
+
+def _add_units_to_financial_year_columns(
+    columns: pd.Index, units_str: str
+) -> list[str]:
+    """Adds '_{units_str}' to the financial year columns"""
+    cols = [
+        _snakecase_string(col + f"_{units_str}")
+        if re.match(r"[0-9]{4}-[0-9]{2}", col)
+        else _snakecase_string(col)
+        for col in columns
+    ]
+    return cols
+
+
+def _convert_financial_year_columns_to_float(df: pd.DataFrame) -> pd.DataFrame:
+    """Forcefully converts FY columns to float columns"""
+    cols = [
+        df[col].astype(float) if re.match(r"[0-9]{4}_[0-9]{2}", col) else df[col]
+        for col in df.columns
+    ]
+    return pd.concat(cols, axis=1)