Skip to content

Add electricity

Misc collection of functions supporting network prep still to be cleaned up

add_missing_carriers(n, carriers)

Function to add missing carriers to the network without raising errors.

Parameters:

Name Type Description Default
n Network

the pypsa network object

 required
carriers list | set

a list of carriers that should be included

 required
Source code in workflow/scripts/add_electricity.py 
175
176
177
178
179
180
181
182
183
184
def add_missing_carriers(n: pypsa.Network, carriers: list | set) -> None:
    """Function to add missing carriers to the network without raising errors.

    Args:
        n (pypsa.Network): the pypsa network object
        carriers (list | set): a list of carriers that should be included
    """
    missing_carriers = set(carriers) - set(n.carriers.index)
    if len(missing_carriers) > 0:
        n.add("Carrier", missing_carriers)

calculate_annuity(lifetime, discount_rate)

Calculate the annuity factor for an asset with lifetime n years and discount rate of r, e.g. annuity(20, 0.05) * 20 = 1.6

Parameters:

Name Type Description Default
lifetime int

ecomic asset lifetime for discounting/NPV calc

 required
discount_rate float

the WACC

 required

Returns:

Name Type Description
float float

the annuity factor
Source code in workflow/scripts/add_electricity.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def calculate_annuity(lifetime: int, discount_rate: float) -> float:
    """Calculate the annuity factor for an asset with lifetime n years and
    discount rate of r, e.g. annuity(20, 0.05) * 20 = 1.6

    Args:
        lifetime (int): ecomic asset lifetime for discounting/NPV calc
        discount_rate (float): the WACC

    Returns:
        float: the annuity factor
    """
    r = discount_rate
    n = lifetime

    if isinstance(r, pd.Series):
        if r.any() < 0:
            raise ValueError("Discount rate must be positive")
        if r.any() < 0:
            raise ValueError("Discount rate must be positive")
        return pd.Series(1 / n, index=r.index).where(r == 0, r / (1.0 - 1.0 / (1.0 + r) ** n))
    elif r < 0:
        raise ValueError("Discount rate must be positive")
    elif r < 0:
        raise ValueError("Discount rate must be positive")
    elif r > 0:
        return r / (1.0 - 1.0 / (1.0 + r) ** n)
    else:
        return 1 / n

load_costs(tech_costs, cost_config, elec_config, cost_year, n_years, econ_lifetime=40)

Calculate the anualised capex costs and OM costs for the technologies based on the input data

Parameters:

Name Type Description Default
tech_costs PathLike

the csv containing the costs

 required
cost_config dict

the snakemake pypsa-china cost config

 required
elec_config dict

the snakemake pypsa-china electricity config

 required
cost_year int

the year for which the costs are retrived

 required
n_years int

the # of years represented by the snapshots/investment period

 required
econ_lifetime int

the max lifetime over which to discount. Defaults to 40.

 40

Returns:

Type Description
DataFrame

pd.DataFrame: costs dataframe in [CURRENCY] per MW_ ... or per MWh_ ...
Source code in workflow/scripts/add_electricity.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def load_costs(
    tech_costs: PathLike,
    cost_config: dict,
    elec_config: dict,
    cost_year: int,
    n_years: int,
    econ_lifetime=40,
) -> pd.DataFrame:
    """Calculate the anualised capex costs and OM costs for the technologies based on the input data

    Args:
        tech_costs (PathLike): the csv containing the costs
        cost_config (dict): the snakemake pypsa-china cost config
        elec_config (dict): the snakemake pypsa-china electricity config
        cost_year (int): the year for which the costs are retrived
        n_years (int): the # of years represented by the snapshots/investment period
        econ_lifetime (int, optional): the max lifetime over which to discount. Defaults to 40.

    Returns:
        pd.DataFrame: costs dataframe in [CURRENCY] per MW_ ... or per MWh_ ...
    """

    # set all asset costs and other parameters
    costs = pd.read_csv(tech_costs, index_col=list(range(3))).sort_index()
    costs.fillna(" ", inplace=True)
    # correct units to MW and EUR
    costs.loc[costs.unit.str.contains("/kW"), "value"] *= 1e3
    costs.loc[costs.unit.str.contains("USD"), "value"] *= cost_config["USD2013_to_EUR2013"]
    costs.loc[costs.unit.str.contains("USD"), "value"] *= cost_config["USD2013_to_EUR2013"]

    cost_year = float(cost_year)
    costs = (
        costs.loc[idx[:, cost_year, :], "value"]
        .unstack(level=2)
        .groupby("technology")
        .sum(min_count=1)
    )

    # TODO set default lifetime as option
    if "discount rate" not in costs.columns:
        costs.loc[:, "discount rate"] = cost_config["discountrate"]
    costs = costs.fillna(
        {
            "CO2 intensity": 0,
            "FOM": 0,
            "VOM": 0,
            "discount rate": cost_config["discountrate"],
            "efficiency": 1,
            "fuel": 0,
            "investment": 0,
            "lifetime": 25,
        }
    )

    discount_period = costs["lifetime"].apply(lambda x: min(x, econ_lifetime))
    costs["capital_cost"] = (
        (calculate_annuity(discount_period, costs["discount rate"]) + costs["FOM"] / 100.0)
        * costs["investment"]
        * n_years
    )

    costs.at["OCGT", "fuel"] = costs.at["gas", "fuel"]
    costs.at["CCGT", "fuel"] = costs.at["gas", "fuel"]
    costs.at["CCGT-CCS", "fuel"] = costs.at["gas", "fuel"]

    costs["marginal_cost"] = costs["VOM"] + costs["fuel"] / costs["efficiency"]

    costs = costs.rename(columns={"CO2 intensity": "co2_emissions"})

    costs.at["OCGT", "co2_emissions"] = costs.at["gas", "co2_emissions"]
    costs.at["CCGT", "co2_emissions"] = costs.at["gas", "co2_emissions"]

    def costs_for_storage(store, link1, link2=None, max_hours=1.0):
        capital_cost = link1["capital_cost"] + max_hours * store["capital_cost"]
        if link2 is not None:
            capital_cost += link2["capital_cost"]
        return pd.Series(dict(capital_cost=capital_cost, marginal_cost=0.0, co2_emissions=0.0))

    max_hours = elec_config["max_hours"]
    costs.loc["battery"] = costs_for_storage(
        costs.loc["battery storage"], costs.loc["battery inverter"], max_hours=max_hours["battery"]
    )

    for attr in ("marginal_cost", "capital_cost"):
        overwrites = cost_config.get(attr)
        overwrites = cost_config.get(attr)
        if overwrites is not None:
            overwrites = pd.Series(overwrites)
            costs.loc[overwrites.index, attr] = overwrites

    return costs

sanitize_carriers(n, config)

Sanitize the carrier information in a PyPSA Network object.

The function ensures that all unique carrier names are present in the network's carriers attribute, and adds nice names and colors for each carrier according to the provided configuration dictionary.

Parameters:

Name Type Description Default
n Network

PyPSA Network object representing the electrical power system.

 required
config dict

A dictionary containing configuration information, specifically the "plotting" key with "nice_names" and "tech_colors" keys for carriers.

 required
Source code in workflow/scripts/add_electricity.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def sanitize_carriers(n: pypsa.Network, config: dict) -> None:
    """Sanitize the carrier information in a PyPSA Network object.

    The function ensures that all unique carrier names are present in the network's
    carriers attribute, and adds nice names and colors for each carrier according
    to the provided configuration dictionary.

    Args:
        n (pypsa.Network): PyPSA Network object representing the electrical power system.
        config (dict): A dictionary containing configuration information, specifically the
               "plotting" key with "nice_names" and "tech_colors" keys for carriers.
    """
    # update default nice names w user settings
    nice_names = NICE_NAMES_DEFAULT.update(config["plotting"].get("nice_names", {}))
    for c in n.iterate_components():
        if "carrier" in c.df:
            add_missing_carriers(n, c.df.carrier)

    # sort the nice names to match carriers and fill missing with "ugly" names
    carrier_i = n.carriers.index
    nice_names = pd.Series(nice_names).reindex(carrier_i).fillna(carrier_i.to_series())
    # replace empty nice names with nice names
    n.carriers.nice_name.where(n.carriers.nice_name != "", nice_names, inplace=True)

    # TODO make less messy, avoid using map
    tech_colors = config["plotting"]["tech_colors"]
    colors = pd.Series(tech_colors).reindex(carrier_i)
    # try to fill missing colors with tech_colors after renaming
    missing_colors_i = colors[colors.isna()].index
    colors[missing_colors_i] = missing_colors_i.map(lambda x: rename_techs(x, nice_names)).map(
        tech_colors
    )
    if colors.isna().any():
        missing_i = list(colors.index[colors.isna()])
        logger.warning(f"tech_colors for carriers {missing_i} not defined in config.")
    n.carriers["color"] = n.carriers.color.where(n.carriers.color != "", colors)