Functions supporting myopic pathway network creation

add_build_year_to_new_assets(n, baseyear)

add a build year to new assets

Parameters:
  • n (Network) –

    the network

  • baseyear (int) –

    year in which optimized assets are built

Source code in workflow/scripts/add_existing_baseyear.py
def add_build_year_to_new_assets(n: pypsa.Network, baseyear: int):
    """add a build year to new assets

    Args:
        n (pypsa.Network): the network
        baseyear (int): year in which optimized assets are built
    """

    # Give assets with lifetimes and no build year the build year baseyear
    for c in n.iterate_components(["Link", "Generator", "Store"]):
        attr = "e" if c.name == "Store" else "p"

        assets = c.df.index[(c.df.lifetime != np.inf) & (c.df[attr + "_nom_extendable"] is True)]

        # add -baseyear to name
        rename = pd.Series(c.df.index, c.df.index)
        rename[assets] += "-" + str(baseyear)
        c.df.rename(index=rename, inplace=True)

        assets = c.df.index[
            (c.df.lifetime != np.inf)
            & (c.df[attr + "_nom_extendable"] is True)
            & (c.df.build_year == 0)
        ]
        c.df.loc[assets, "build_year"] = baseyear

        # rename time-dependent
        selection = n.component_attrs[c.name].type.str.contains("series") & n.component_attrs[
            c.name
        ].status.str.contains("Input")
        for attr in n.component_attrs[c.name].index[selection]:
            c.pnl[attr].rename(columns=rename, inplace=True)

add_power_capacities_installed_before_baseyear(n, grouping_years, costs, baseyear, config)

Parameters

n : pypsa.Network grouping_years : intervals to group existing capacities costs : to read lifetime to estimate YearDecomissioning baseyear : int

Source code in workflow/scripts/add_existing_baseyear.py
def add_power_capacities_installed_before_baseyear(
    n: pypsa.Network, grouping_years, costs, baseyear, config
):
    """
    Parameters
    ----------
    n : pypsa.Network
    grouping_years :
        intervals to group existing capacities
    costs :
        to read lifetime to estimate YearDecomissioning
    baseyear : int
    """
    logger.info("adding power capacities installed before baseyear")

    df_agg = pd.DataFrame()

    # include renewables in df_agg
    add_existing_capacities(df_agg)

    df_agg["grouping_year"] = np.take(
        grouping_years, np.digitize(df_agg.DateIn, grouping_years, right=True)
    )

    df = df_agg.pivot_table(
        index=["grouping_year", "Tech"], columns="cluster_bus", values="Capacity", aggfunc="sum"
    )

    df.fillna(0)

    carrier = {
        "coal": "coal power plant",
        "CHP coal": "CHP coal",
        "CHP gas": "CHP gas",
        "OCGT": "OCGT gas",
        "solar": "solar",
        "solar thermal": "solar thermal",
        "onwind": "onwind",
        "offwind": "offwind",
        "coal boiler": "coal boiler",
        "ground heat pump": "heat pump",
        "nuclear": "nuclear",
    }

    for grouping_year, generator in df.index:

        # capacity is the capacity in MW at each node for this
        capacity = df.loc[grouping_year, generator]
        capacity = capacity[~capacity.isna()]
        capacity = capacity[capacity > config["existing_capacities"]["threshold_capacity"]]

        if generator in ["solar", "onwind", "offwind"]:
            p_max_pu = n.generators_t.p_max_pu[capacity.index + " " + generator]
            n.add(
                "Generator",
                capacity.index,
                suffix=" " + generator + "-" + str(grouping_year),
                bus=capacity.index,
                carrier=carrier[generator],
                p_nom=capacity,
                p_nom_min=capacity,
                p_nom_extendable=False,
                marginal_cost=costs.at[generator, "marginal_cost"],
                capital_cost=costs.at[generator, "capital_cost"],
                efficiency=costs.at[generator, "efficiency"],
                p_max_pu=p_max_pu.rename(columns=n.generators.bus),
                build_year=grouping_year,
                lifetime=costs.at[generator, "lifetime"],
            )

        if generator == "coal":
            n.add(
                "Generator",
                capacity.index,
                suffix=" " + generator + "-" + str(grouping_year),
                bus=capacity.index,
                carrier=carrier[generator],
                p_nom=capacity,
                p_nom_extendable=False,
                marginal_cost=costs.at[generator, "marginal_cost"],
                capital_cost=costs.at[generator, "capital_cost"],
                efficiency=costs.at[generator, "efficiency"],
                build_year=grouping_year,
                lifetime=costs.at[generator, "lifetime"],
            )

        if generator == "nuclear":
            n.add(
                "Generator",
                capacity.index,
                suffix=" " + generator + "-" + str(grouping_year),
                bus=capacity.index,
                carrier=carrier[generator],
                p_nom=capacity,
                p_nom_min=capacity,
                p_nom_extendable=False,
                p_min_pu=0.7,
                marginal_cost=costs.at[generator, "marginal_cost"],
                capital_cost=costs.at[generator, "capital_cost"],
                efficiency=costs.at[generator, "efficiency"],
                build_year=grouping_year,
                lifetime=costs.at[generator, "lifetime"],
            )

        if generator == "solar thermal":
            p_max_pu = n.generators_t.p_max_pu[capacity.index + " central " + generator]
            p_max_pu.columns = capacity.index
            n.add(
                "Generator",
                capacity.index,
                suffix=" central " + generator + "-" + str(grouping_year),
                bus=capacity.index + " central heat",
                carrier=carrier[generator],
                p_nom=capacity,
                p_nom_min=capacity,
                p_nom_extendable=False,
                marginal_cost=costs.at["central " + generator, "marginal_cost"],
                capital_cost=costs.at["central " + generator, "capital_cost"],
                p_max_pu=p_max_pu,
                build_year=grouping_year,
                lifetime=costs.at["central " + generator, "lifetime"],
            )

        if generator == "CHP coal":
            bus0 = capacity.index + " coal"
            n.add(
                "Link",
                capacity.index,
                suffix=" " + generator + " generator" + "-" + str(grouping_year),
                bus0=bus0,
                bus1=capacity.index,
                carrier=carrier[generator],
                marginal_cost=0.37 * costs.at["central coal CHP", "VOM"],  # NB: VOM is per MWel
                capital_cost=0.37
                * costs.at["central coal CHP", "capital_cost"],  # NB: fixed cost is per MWel,
                p_nom=capacity / 0.37,
                p_nom_min=capacity / 0.37,
                p_nom_extendable=False,
                efficiency=0.37,
                p_nom_ratio=1.0,
                c_b=0.75,
                build_year=grouping_year,
                lifetime=costs.at["central coal CHP", "lifetime"],
            )
            n.add(
                "Link",
                capacity.index,
                suffix=" " + generator + " boiler" + "-" + str(grouping_year),
                bus0=bus0,
                bus1=capacity.index + " central heat",
                carrier=carrier[generator],
                marginal_cost=0.37 * costs.at["central coal CHP", "VOM"],  # NB: VOM is per MWel
                p_nom=capacity / 0.37 * 0.15,
                p_nom_min=capacity / 0.37 * 0.15,
                p_nom_extendable=False,
                efficiency=0.37 / 0.15,
                build_year=grouping_year,
                lifetime=costs.at["central coal CHP", "lifetime"],
            )

        if generator == "CHP gas":
            bus0 = capacity.index + " gas"
            n.add(
                "Link",
                capacity.index,
                suffix=" " + generator + " generator" + "-" + str(grouping_year),
                bus0=bus0,
                bus1=capacity.index,
                carrier=carrier[generator],
                marginal_cost=costs.at["central gas CHP", "efficiency"]
                * costs.at["central gas CHP", "VOM"],  # NB: VOM is per MWel
                capital_cost=costs.at["central gas CHP", "efficiency"]
                * costs.at["central gas CHP", "capital_cost"],  # NB: fixed cost is per MWel,
                p_nom=capacity / costs.at["central gas CHP", "efficiency"],
                p_nom_min=capacity / costs.at["central gas CHP", "efficiency"],
                p_nom_extendable=False,
                efficiency=costs.at["central gas CHP", "efficiency"],
                p_nom_ratio=1.0,
                c_b=costs.at["central gas CHP", "c_b"],
                build_year=grouping_year,
                lifetime=costs.at["central gas CHP", "lifetime"],
            )
            n.add(
                "Link",
                capacity.index,
                suffix=" " + generator + " boiler" + "-" + str(grouping_year),
                bus0=bus0,
                bus1=capacity.index + " central heat",
                carrier=carrier[generator],
                marginal_cost=costs.at["central gas CHP", "efficiency"]
                * costs.at["central gas CHP", "VOM"],  # NB: VOM is per MWel
                p_nom=capacity
                / costs.at["central gas CHP", "efficiency"]
                * costs.at["central gas CHP", "c_v"],
                p_nom_min=capacity
                / costs.at["central gas CHP", "efficiency"]
                * costs.at["central gas CHP", "c_v"],
                p_nom_extendable=False,
                efficiency=costs.at["central gas CHP", "efficiency"]
                / costs.at["central gas CHP", "c_v"],
                build_year=grouping_year,
                lifetime=costs.at["central gas CHP", "lifetime"],
            )

        if generator == "OCGT":
            bus0 = capacity.index + " gas"
            n.add(
                "Link",
                capacity.index,
                suffix=" " + generator + "-" + str(grouping_year),
                bus0=bus0,
                bus1=capacity.index,
                carrier=carrier[generator],
                marginal_cost=costs.at[generator, "efficiency"]
                * costs.at[generator, "VOM"],  # NB: VOM is per MWel
                capital_cost=costs.at[generator, "efficiency"]
                * costs.at[generator, "capital_cost"],
                # NB: fixed cost is per MWel
                p_nom=capacity / costs.at[generator, "efficiency"],
                p_nom_min=capacity / costs.at[generator, "efficiency"],
                p_nom_extendable=False,
                efficiency=costs.at[generator, "efficiency"],
                build_year=grouping_year,
                lifetime=costs.at[generator, "lifetime"],
            )

        if generator == "coal boiler":
            bus0 = capacity.index + " coal"
            for cat in [" central "]:
                n.add(
                    "Link",
                    capacity.index,
                    suffix="" + cat + generator + "-" + str(grouping_year),
                    bus0=bus0,
                    bus1=capacity.index + cat + "heat",
                    carrier=carrier[generator],
                    marginal_cost=costs.at[cat.lstrip() + generator, "efficiency"]
                    * costs.at[cat.lstrip() + generator, "VOM"],
                    capital_cost=costs.at[cat.lstrip() + generator, "efficiency"]
                    * costs.at[cat.lstrip() + generator, "capital_cost"],
                    p_nom=capacity / costs.at[cat.lstrip() + generator, "efficiency"],
                    p_nom_min=capacity / costs.at[cat.lstrip() + generator, "efficiency"],
                    p_nom_extendable=False,
                    efficiency=costs.at[cat.lstrip() + generator, "efficiency"],
                    build_year=grouping_year,
                    lifetime=costs.at[cat.lstrip() + generator, "lifetime"],
                )
        # TODO fix centralise
        if generator == "ground heat pump":
            date_range = pd.date_range(
                "2025-01-01 00:00",
                "2025-12-31 23:00",
                freq=config["snapshots"]["freq"],
                tz="Asia/shanghai",
            )
            date_range = date_range.map(lambda t: t.replace(year=2020))

            with pd.HDFStore(snakemake.input.cop_name, mode="r") as store:
                gshp_cop = store["gshp_cop_profiles"]
                gshp_cop.index = gshp_cop.index.tz_localize("Asia/shanghai")
                gshp_cop = gshp_cop.loc[date_range].set_index(n.snapshots)

            n.add(
                "Link",
                capacity.index,
                suffix=" " + generator + "-" + str(grouping_year),
                bus0=capacity.index,
                bus1=capacity.index + " central heat",
                carrier="heat pump",
                efficiency=(
                    gshp_cop[capacity.index]
                    if config["time_dep_hp_cop"]
                    else costs.at["decentral ground-sourced heat pump", "efficiency"]
                ),
                capital_cost=costs.at["decentral ground-sourced heat pump", "efficiency"]
                * costs.at["decentral ground-sourced heat pump", "capital_cost"],
                marginal_cost=costs.at["decentral ground-sourced heat pump", "efficiency"]
                * costs.at["decentral ground-sourced heat pump", "marginal_cost"],
                p_nom=capacity / costs.at["decentral ground-sourced heat pump", "efficiency"],
                p_nom_min=capacity / costs.at["decentral ground-sourced heat pump", "efficiency"],
                p_nom_extendable=False,
                build_year=grouping_year,
                lifetime=costs.at["decentral ground-sourced heat pump", "lifetime"],
            )