Modules

npa_howtopay.model

YearContext dataclass

Context for all values needed in a given year

Source code in src/npa_howtopay/model.py

@dataclass
class YearContext:
    """Context for all values needed in a given year"""

    year: int
    gas_ratebase: float
    electric_ratebase: float
    gas_depreciation_expense: float
    electric_depreciation_expense: float
    gas_maintenance_cost: float
    electric_maintenance_cost: float
    gas_npa_opex: float
    electric_npa_opex: float
    gas_performance_incentive: float

calculate_avg_bill_per_user(inflation_adjusted_revenue, num_users)

Calculate the average bill per user by dividing total revenue by number of users.

Parameters:

Name	Type	Description	Default
inflation_adjusted_revenue	float	Total revenue adjusted for inflation	required
num_users	int	Total number of users to divide revenue among	required

Returns:

Name	Type	Description
float	float	Average bill amount per user

Source code in src/npa_howtopay/model.py

def calculate_avg_bill_per_user(inflation_adjusted_revenue: float, num_users: int) -> float:
    """Calculate the average bill per user by dividing total revenue by number of users.

    Args:
        inflation_adjusted_revenue: Total revenue adjusted for inflation
        num_users: Total number of users to divide revenue among

    Returns:
        float: Average bill amount per user
    """
    return inflation_adjusted_revenue / num_users

calculate_converts_electric_bill_per_user(electric_fixed_charge, electric_variable_tariff, per_user_electric_need, per_user_heating_need, per_user_water_heating_need, hp_efficiency, water_heater_efficiency)

Calculate electric bill per user for converts (includes heating).

Parameters:

Name	Type	Description	Default
electric_fixed_charge	float	Fixed charge per user	required
electric_variable_tariff	float	Variable tariff per kWh	required
per_user_electric_need	float	Electric need per user	required
per_user_heating_need	float	Heating need per user (therms)	required
per_user_water_heating_need	float	Water heating need per user (therms)	required
hp_efficiency	float	Heat pump efficiency	required
water_heater_efficiency	float	Water heater efficiency	required

Returns:

Name	Type	Description
float	float	Electric bill per user for converts

Source code in src/npa_howtopay/model.py

def calculate_converts_electric_bill_per_user(
    electric_fixed_charge: float,
    electric_variable_tariff: float,
    per_user_electric_need: float,
    per_user_heating_need: float,
    per_user_water_heating_need: float,
    hp_efficiency: float,
    water_heater_efficiency: float,
) -> float:
    """Calculate electric bill per user for converts (includes heating).

    Args:
        electric_fixed_charge: Fixed charge per user
        electric_variable_tariff: Variable tariff per kWh
        per_user_electric_need: Electric need per user
        per_user_heating_need: Heating need per user (therms)
        per_user_water_heating_need: Water heating need per user (therms)
        hp_efficiency: Heat pump efficiency
        water_heater_efficiency: Water heater efficiency

    Returns:
        float: Electric bill per user for converts
    """
    add_on_usage = (
        per_user_heating_need * KWH_PER_THERM / hp_efficiency
        + per_user_water_heating_need * KWH_PER_THERM / water_heater_efficiency
    )
    return electric_fixed_charge + electric_variable_tariff * (per_user_electric_need + add_on_usage)

calculate_converts_total_bill_per_user(converts_gas_bill, converts_electric_bill)

Calculate total bill per user for converts (gas + electric).

Parameters:

Name	Type	Description	Default
converts_gas_bill	float	Gas bill per user for converts	required
converts_electric_bill	float	Electric bill per user for converts	required

Returns:

Name	Type	Description
float	float	Total bill per user for converts

Source code in src/npa_howtopay/model.py

def calculate_converts_total_bill_per_user(converts_gas_bill: float, converts_electric_bill: float) -> float:
    """Calculate total bill per user for converts (gas + electric).

    Args:
        converts_gas_bill: Gas bill per user for converts
        converts_electric_bill: Electric bill per user for converts

    Returns:
        float: Total bill per user for converts
    """
    return converts_gas_bill + converts_electric_bill

calculate_electric_fixed_charge_per_user(fixed_charge)

Return electric fixed charge per user. Currently a user defined constant

Source code in src/npa_howtopay/model.py

def calculate_electric_fixed_charge_per_user(fixed_charge: float) -> float:
    """Return electric fixed charge per user. Currently a user defined constant"""
    return fixed_charge

calculate_electric_variable_tariff_per_kwh(electric_infl_adj_revenue, total_electric_usage_kwh, fixed_charge, num_users)

Calculate electric variable cost per kWh.

Parameters:

Name	Type	Description	Default
electric_infl_adj_revenue	float	Total electric revenue adjusted for inflation	required
total_electric_usage_kwh	float	Total electric usage in kWh	required
fixed_charge	float	Fixed charge per user	required
num_users	int	Number of users	required

Returns:

Name	Type	Description
float	float	Variable cost per kWh calculated by subtracting total fixed charges from revenue and dividing by total usage

Source code in src/npa_howtopay/model.py

def calculate_electric_variable_tariff_per_kwh(
    electric_infl_adj_revenue: float, total_electric_usage_kwh: float, fixed_charge: float, num_users: int
) -> float:
    """Calculate electric variable cost per kWh.

    Args:
        electric_infl_adj_revenue: Total electric revenue adjusted for inflation
        total_electric_usage_kwh: Total electric usage in kWh
        fixed_charge: Fixed charge per user
        num_users: Number of users

    Returns:
        float: Variable cost per kWh calculated by subtracting total fixed charges
              from revenue and dividing by total usage
    """
    return (electric_infl_adj_revenue - num_users * fixed_charge) / total_electric_usage_kwh

calculate_gas_fixed_charge_per_user(fixed_charge)

Return gas fixed cost per user. Currently a user defined constant

Source code in src/npa_howtopay/model.py

def calculate_gas_fixed_charge_per_user(fixed_charge: float) -> float:
    """Return gas fixed cost per user. Currently a user defined constant"""
    return fixed_charge

calculate_gas_variable_tariff_per_therm(gas_infl_adj_revenue, total_gas_usage_therms, fixed_charge, num_users)

Calculate gas variable cost per therm.

Parameters:

Name	Type	Description	Default
gas_infl_adj_revenue	float	Total gas revenue adjusted for inflation	required
total_gas_usage_therms	float	Total gas usage in therms	required
fixed_charge	float	Fixed charge per user	required
num_users	int	Number of users	required

Returns:

Name	Type	Description
float	float	Variable cost per therm calculated by subtracting total fixed charges from revenue and dividing by total usage

Source code in src/npa_howtopay/model.py

def calculate_gas_variable_tariff_per_therm(
    gas_infl_adj_revenue: float,
    total_gas_usage_therms: float,
    fixed_charge: float,
    num_users: int,
) -> float:
    """Calculate gas variable cost per therm.

    Args:
        gas_infl_adj_revenue: Total gas revenue adjusted for inflation
        total_gas_usage_therms: Total gas usage in therms
        fixed_charge: Fixed charge per user
        num_users: Number of users

    Returns:
        float: Variable cost per therm calculated by subtracting total fixed charges
              from revenue and dividing by total usage
    """
    return (gas_infl_adj_revenue - num_users * fixed_charge) / total_gas_usage_therms

calculate_nonconverts_electric_bill_per_user(electric_fixed_charge, electric_variable_tariff, per_user_electric_need)

Calculate electric bill per user for nonconverts (no heating).

Parameters:

Name	Type	Description	Default
electric_fixed_charge	float	Fixed charge per user	required
electric_variable_tariff	float	Variable tariff per kWh	required
per_user_electric_need	float	Electric need per user	required

Returns:

Name	Type	Description
float	float	Electric bill per user for nonconverts

Source code in src/npa_howtopay/model.py

def calculate_nonconverts_electric_bill_per_user(
    electric_fixed_charge: float, electric_variable_tariff: float, per_user_electric_need: float
) -> float:
    """Calculate electric bill per user for nonconverts (no heating).

    Args:
        electric_fixed_charge: Fixed charge per user
        electric_variable_tariff: Variable tariff per kWh
        per_user_electric_need: Electric need per user

    Returns:
        float: Electric bill per user for nonconverts
    """
    return electric_fixed_charge + electric_variable_tariff * per_user_electric_need

calculate_nonconverts_gas_bill_per_user(gas_fixed_charge, gas_variable_tariff, per_user_heating_need)

Calculate gas bill per user for nonconverts.

Parameters:

Name	Type	Description	Default
gas_fixed_charge	float	Fixed charge per user	required
gas_variable_tariff	float	Variable tariff per therm	required
per_user_heating_need	float	Heating need per user	required

Returns:

Name	Type	Description
float	float	Gas bill per user for nonconverts

Source code in src/npa_howtopay/model.py

def calculate_nonconverts_gas_bill_per_user(
    gas_fixed_charge: float, gas_variable_tariff: float, per_user_heating_need: float
) -> float:
    """Calculate gas bill per user for nonconverts.

    Args:
        gas_fixed_charge: Fixed charge per user
        gas_variable_tariff: Variable tariff per therm
        per_user_heating_need: Heating need per user

    Returns:
        float: Gas bill per user for nonconverts
    """
    return gas_fixed_charge + gas_variable_tariff * per_user_heating_need

calculate_nonconverts_total_bill_per_user(nonconverts_gas_bill, nonconverts_electric_bill)

Calculate total bill per user for nonconverts (gas + electric).

Parameters:

Name	Type	Description	Default
nonconverts_gas_bill	float	Gas bill per user for nonconverts	required
nonconverts_electric_bill	float	Electric bill per user for nonconverts	required

Returns:

Name	Type	Description
float	float	Total bill per user for nonconverts

Source code in src/npa_howtopay/model.py

def calculate_nonconverts_total_bill_per_user(nonconverts_gas_bill: float, nonconverts_electric_bill: float) -> float:
    """Calculate total bill per user for nonconverts (gas + electric).

    Args:
        nonconverts_gas_bill: Gas bill per user for nonconverts
        nonconverts_electric_bill: Electric bill per user for nonconverts

    Returns:
        float: Total bill per user for nonconverts
    """
    return nonconverts_gas_bill + nonconverts_electric_bill

compute_bill_costs(df, input_params)

Compute bill costs and tariffs for gas and electric utilities.

Takes a DataFrame with revenue requirements and usage data and computes: - Inflation adjusted revenue requirements for gas and electric - Fixed charges per user and Variable tariffs per therm (gas) and kWh (electric) - Utility bills per user for converts and nonconverts - Total bills per user for converts and nonconverts

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame containing revenue requirements and usage data	required
input_params	InputParams	Input parameters containing utility rates and user counts	required

Returns:

Type	Description
DataFrame	DataFrame with added columns for adjusted revenue requirements and tariffs

Source code in src/npa_howtopay/model.py

def compute_bill_costs(
    df: pl.DataFrame,
    input_params: InputParams,
) -> pl.DataFrame:
    """Compute bill costs and tariffs for gas and electric utilities.

    Takes a DataFrame with revenue requirements and usage data and computes:
    - Inflation adjusted revenue requirements for gas and electric
    - Fixed charges per user and Variable tariffs per therm (gas) and kWh (electric)
    - Utility bills per user for converts and nonconverts
    - Total bills per user for converts and nonconverts

    Args:
        df: DataFrame containing revenue requirements and usage data
        input_params: Input parameters containing utility rates and user counts

    Returns:
        DataFrame with added columns for adjusted revenue requirements and tariffs
    """
    start_year = df.select(pl.col("year")).min().item()

    # Create inflation-adjusted revenue requirement columns
    df = df.with_columns([
        pl.struct(["gas_revenue_requirement", "year"])
        .map_elements(
            lambda x: inflation_adjust_revenue_requirement(
                x["gas_revenue_requirement"], x["year"], start_year, input_params.shared.real_dollar_discount_rate
            ),
            return_dtype=pl.Float64,
        )
        .alias("gas_inflation_adjusted_revenue_requirement"),
        pl.struct(["electric_revenue_requirement", "year"])
        .map_elements(
            lambda x: inflation_adjust_revenue_requirement(
                x["electric_revenue_requirement"], x["year"], start_year, input_params.shared.real_dollar_discount_rate
            ),
            return_dtype=pl.Float64,
        )
        .alias("electric_inflation_adjusted_revenue_requirement"),
        (pl.col("gas_revenue_requirement") + pl.col("electric_revenue_requirement")).alias("total_revenue_requirement"),
    ])

    # Create gas and electric tariffs columns (and total inflation adjusted revenue requirement)
    df = df.with_columns([
        (
            pl.col("gas_inflation_adjusted_revenue_requirement")
            + pl.col("electric_inflation_adjusted_revenue_requirement")
        ).alias("total_inflation_adjusted_revenue_requirement"),
        pl.struct(["gas_inflation_adjusted_revenue_requirement", "total_gas_usage_therms"])
        .map_elements(
            lambda x: calculate_gas_variable_tariff_per_therm(
                x["gas_inflation_adjusted_revenue_requirement"],
                x["total_gas_usage_therms"],
                input_params.gas.user_bill_fixed_charge,
                input_params.gas.num_users_init,
            ),
            return_dtype=pl.Float64,
        )
        .alias("gas_variable_tariff_per_therm"),
        pl.struct(["electric_inflation_adjusted_revenue_requirement", "total_electric_usage_kwh"])
        .map_elements(
            lambda x: calculate_electric_variable_tariff_per_kwh(
                x["electric_inflation_adjusted_revenue_requirement"],
                x["total_electric_usage_kwh"],
                input_params.electric.user_bill_fixed_charge,
                input_params.electric.num_users_init,
            ),
            return_dtype=pl.Float64,
        )
        .alias("electric_variable_tariff_per_kwh"),
        pl.lit(
            calculate_electric_fixed_charge_per_user(
                input_params.electric.user_bill_fixed_charge,
            )
        ).alias("electric_fixed_charge_per_user"),
        pl.lit(
            calculate_gas_fixed_charge_per_user(
                input_params.gas.user_bill_fixed_charge,
            )
        ).alias("gas_fixed_charge_per_user"),
    ])

    # Create per-user gas bill columns and total inflation adjusted revenue requirement
    df = df.with_columns([
        pl.struct(["gas_inflation_adjusted_revenue_requirement", "gas_num_users"])
        .map_elements(
            lambda x: calculate_avg_bill_per_user(x["gas_inflation_adjusted_revenue_requirement"], x["gas_num_users"]),
            return_dtype=pl.Float64,
        )
        .alias("gas_avg_bill_per_user"),
        pl.struct(["gas_fixed_charge_per_user", "gas_variable_tariff_per_therm"])
        .map_elements(
            lambda x: calculate_nonconverts_gas_bill_per_user(
                x["gas_fixed_charge_per_user"],
                x["gas_variable_tariff_per_therm"],
                input_params.gas.per_user_heating_need_therms,
            ),
            return_dtype=pl.Float64,
        )
        .alias("gas_nonconverts_bill_per_user"),
        pl.lit(0.0).alias("gas_converts_bill_per_user"),
    ])

    # Create converts and nonconverts electric bills
    df = df.with_columns([
        pl.struct(["electric_inflation_adjusted_revenue_requirement", "electric_num_users"])
        .map_elements(
            lambda x: calculate_avg_bill_per_user(
                x["electric_inflation_adjusted_revenue_requirement"], x["electric_num_users"]
            ),
            return_dtype=pl.Float64,
        )
        .alias("electric_avg_bill_per_user"),
        pl.struct(["electric_fixed_charge_per_user", "electric_variable_tariff_per_kwh"])
        .map_elements(
            lambda x: calculate_converts_electric_bill_per_user(
                x["electric_fixed_charge_per_user"],
                x["electric_variable_tariff_per_kwh"],
                input_params.electric.per_user_electric_need_kwh,
                input_params.gas.per_user_heating_need_therms,
                input_params.gas.per_user_water_heating_need_therms,
                input_params.electric.hp_efficiency,
                input_params.electric.water_heater_efficiency,
            ),
            return_dtype=pl.Float64,
        )
        .alias("electric_converts_bill_per_user"),
        pl.struct(["electric_fixed_charge_per_user", "electric_variable_tariff_per_kwh"])
        .map_elements(
            lambda x: calculate_nonconverts_electric_bill_per_user(
                x["electric_fixed_charge_per_user"],
                x["electric_variable_tariff_per_kwh"],
                input_params.electric.per_user_electric_need_kwh,
            ),
            return_dtype=pl.Float64,
        )
        .alias("electric_nonconverts_bill_per_user"),
    ])

    # Create total bill calculations for converts and nonconverts
    df = df.with_columns([
        pl.struct(["gas_converts_bill_per_user", "electric_converts_bill_per_user"])
        .map_elements(
            lambda x: calculate_converts_total_bill_per_user(
                x["gas_converts_bill_per_user"], x["electric_converts_bill_per_user"]
            ),
            return_dtype=pl.Float64,
        )
        .alias("converts_total_bill_per_user"),
        pl.struct(["gas_nonconverts_bill_per_user", "electric_nonconverts_bill_per_user"])
        .map_elements(
            lambda x: calculate_nonconverts_total_bill_per_user(
                x["gas_nonconverts_bill_per_user"], x["electric_nonconverts_bill_per_user"]
            ),
            return_dtype=pl.Float64,
        )
        .alias("nonconverts_total_bill_per_user"),
    ])

    return df

compute_intermediate_cols_electric(context, input_params, ts_params)

Compute intermediate columns for electric utility calculations.

Calculates electric utility metrics for a given year including: - Number of users and cumulative heat pump converts - Total electric usage in kWh (base usage + heating from converts) - Fixed and volumetric costs - Operating expenses and revenue requirements

Parameters:

Name	Type	Description	Default
context	YearContext	Year context containing ratebase, depreciation and maintenance costs	required
input_params	InputParams	Input parameters with utility rates and user counts	required
ts_params	TimeSeriesParams	Time series parameters with NPA projects and overhead costs	required

Returns:

Type	Description
DataFrame	DataFrame with calculated electric utility metrics for the given year

Source code in src/npa_howtopay/model.py

def compute_intermediate_cols_electric(
    context: YearContext, input_params: InputParams, ts_params: TimeSeriesParams
) -> pl.DataFrame:
    """Compute intermediate columns for electric utility calculations.

    Calculates electric utility metrics for a given year including:
    - Number of users and cumulative heat pump converts
    - Total electric usage in kWh (base usage + heating from converts)
    - Fixed and volumetric costs
    - Operating expenses and revenue requirements

    Args:
        context: Year context containing ratebase, depreciation and maintenance costs
        input_params: Input parameters with utility rates and user counts
        ts_params: Time series parameters with NPA projects and overhead costs

    Returns:
        DataFrame with calculated electric utility metrics for the given year
    """
    electric_fixed_overhead_costs = (
        ts_params.electric_fixed_overhead_costs.filter(pl.col("year") == context.year)
        .select(pl.col("cost"))
        .sum()
        .item()
    )
    total_converts_cumul = npa.compute_hp_converts_from_df(
        context.year, ts_params.npa_projects, cumulative=True, npa_only=False
    )
    electric_num_users = input_params.electric.num_users_init
    added_usage = (
        total_converts_cumul
        * input_params.gas.per_user_heating_need_therms
        * KWH_PER_THERM
        / input_params.electric.hp_efficiency
        + total_converts_cumul
        * input_params.gas.per_user_water_heating_need_therms
        * KWH_PER_THERM
        / input_params.electric.water_heater_efficiency
    )
    total_usage = input_params.electric.num_users_init * input_params.electric.per_user_electric_need_kwh + added_usage
    costs_volumetric = total_usage * input_params.electric.electricity_generation_cost_per_kwh(context.year)
    costs_fixed = electric_fixed_overhead_costs + context.electric_maintenance_cost + context.electric_npa_opex
    opex_costs = costs_fixed + costs_volumetric
    revenue_requirement = (
        context.electric_ratebase * input_params.electric.ror + opex_costs + context.electric_depreciation_expense
    )
    return_on_ratebase_pct = (
        context.electric_ratebase * input_params.electric.ror
    ) / revenue_requirement  # Return on Rate Base as % of Revenue Requirement

    return pl.DataFrame({
        "year": [context.year],
        "electric_num_users": [electric_num_users],
        "total_converts_cumul": [total_converts_cumul],
        "electric_added_usage_kwh": [added_usage],
        "total_electric_usage_kwh": [total_usage],
        "electric_costs_volumetric": [costs_volumetric],
        "electric_costs_fixed": [costs_fixed],
        "electric_opex_costs": [opex_costs],
        "electric_revenue_requirement": [revenue_requirement],
        "electric_return_on_ratebase_pct": [return_on_ratebase_pct],
    })

compute_intermediate_cols_gas(context, input_params, ts_params)

Compute intermediate columns for gas utility calculations.

Calculates gas utility metrics for a given year including: - Number of users (accounting for heat pump converts) - Total gas usage in therms - Fixed and volumetric costs - Operating expenses and revenue requirements

Parameters:

Name	Type	Description	Default
context	YearContext	Year context containing ratebase, depreciation and maintenance costs	required
input_params	InputParams	Input parameters with utility rates and user counts	required
ts_params	TimeSeriesParams	Time series parameters with NPA projects and overhead costs	required

Returns:

Type	Description
DataFrame	DataFrame with calculated gas utility metrics for the given year

Source code in src/npa_howtopay/model.py

def compute_intermediate_cols_gas(
    context: YearContext, input_params: InputParams, ts_params: TimeSeriesParams
) -> pl.DataFrame:
    """Compute intermediate columns for gas utility calculations.

    Calculates gas utility metrics for a given year including:
    - Number of users (accounting for heat pump converts)
    - Total gas usage in therms
    - Fixed and volumetric costs
    - Operating expenses and revenue requirements

    Args:
        context: Year context containing ratebase, depreciation and maintenance costs
        input_params: Input parameters with utility rates and user counts
        ts_params: Time series parameters with NPA projects and overhead costs

    Returns:
        DataFrame with calculated gas utility metrics for the given year
    """
    gas_fixed_overhead_costs = (
        ts_params.gas_fixed_overhead_costs.filter(pl.col("year") == context.year).select(pl.col("cost")).sum().item()
    )
    gas_num_users = input_params.gas.num_users_init - npa.compute_hp_converts_from_df(
        context.year, ts_params.npa_projects, cumulative=True, npa_only=False
    )
    total_usage = gas_num_users * input_params.gas.per_user_heating_need_therms
    costs_volumetric = total_usage * input_params.gas.gas_generation_cost_per_therm(context.year)
    costs_fixed = gas_fixed_overhead_costs + context.gas_maintenance_cost + context.gas_npa_opex
    opex_costs = costs_fixed + costs_volumetric
    revenue_requirement = (
        context.gas_ratebase * input_params.gas.ror
        + opex_costs
        + context.gas_depreciation_expense
        + context.gas_performance_incentive
    )
    return_on_ratebase_pct = (
        context.gas_ratebase * input_params.gas.ror
    ) / revenue_requirement  # Return on Rate Base as % of Revenue Requirement

    return pl.DataFrame({
        "year": [context.year],
        "gas_num_users": [gas_num_users],
        "total_gas_usage_therms": [total_usage],
        "gas_costs_volumetric": [costs_volumetric],
        "gas_costs_fixed": [costs_fixed],
        "gas_opex_costs": [opex_costs],
        "gas_revenue_requirement": [revenue_requirement],
        "gas_return_on_ratebase_pct": [return_on_ratebase_pct],
    })

create_scenario_runs(start_year, end_year, gas_electric, capex_opex)

Create a dictionary of scenario parameters for different model runs.

Creates scenarios for business-as-usual (BAU), taxpayer-funded, and combinations of gas/electric and capex/opex interventions. Each scenario specifies the time period and configuration parameters.

Parameters:

Name	Type	Description	Default
start_year	int	First year of the scenario analysis	required
end_year	int	Last year of the scenario analysis	required
gas_electric	list[Literal['gas', 'electric']]	List specifying which utilities to analyze ("gas" and/or "electric")	required
capex_opex	list[Literal['capex', 'opex']]	List specifying which cost types to analyze ("capex" and/or "opex")	required

Returns:

Type	Description
dict[str, ScenarioParams]	Dictionary mapping scenario names to ScenarioParams objects containing the
dict[str, ScenarioParams]	configuration for that scenario. Includes "bau" and "taxpayer" base scenarios
dict[str, ScenarioParams]	plus combinations of gas/electric and capex/opex parameters.

Source code in src/npa_howtopay/model.py

def create_scenario_runs(
    start_year: int,
    end_year: int,
    gas_electric: list[Literal["gas", "electric"]],
    capex_opex: list[Literal["capex", "opex"]],
) -> dict[str, ScenarioParams]:
    """Create a dictionary of scenario parameters for different model runs.

    Creates scenarios for business-as-usual (BAU), taxpayer-funded, and combinations of
    gas/electric and capex/opex interventions. Each scenario specifies the time period
    and configuration parameters.

    Args:
        start_year: First year of the scenario analysis
        end_year: Last year of the scenario analysis
        gas_electric: List specifying which utilities to analyze ("gas" and/or "electric")
        capex_opex: List specifying which cost types to analyze ("capex" and/or "opex")

    Returns:
        Dictionary mapping scenario names to ScenarioParams objects containing the
        configuration for that scenario. Includes "bau" and "taxpayer" base scenarios
        plus combinations of gas/electric and capex/opex parameters.
    """
    scenarios = {
        "bau": ScenarioParams(start_year=start_year, end_year=end_year, bau=True),
        "taxpayer": ScenarioParams(start_year=start_year, end_year=end_year, taxpayer=True),
        "performance_incentive": ScenarioParams(
            start_year=start_year, end_year=end_year, performance_incentive=True, gas_electric="gas", capex_opex="opex"
        ),
    }

    # Add the regular scenarios
    for ge in gas_electric:
        for co in capex_opex:
            scenarios[f"{ge}_{co}"] = ScenarioParams(
                start_year=start_year, end_year=end_year, gas_electric=ge, capex_opex=co
            )

    return scenarios

inflation_adjust_revenue_requirement(revenue_req, year, start_year, real_dollar_discount_rate)

Adjust revenue requirement for inflation using discount rate.

Source code in src/npa_howtopay/model.py

def inflation_adjust_revenue_requirement(
    revenue_req: float, year: int, start_year: int, real_dollar_discount_rate: float
) -> float:
    """Adjust revenue requirement for inflation using discount rate."""
    if year < start_year:
        raise ValueError(f"Year {year} cannot be before start year {start_year}")
    return revenue_req / ((1 + real_dollar_discount_rate) ** (year - start_year))

npa_howtopay.params

TimeSeriesParams

Source code in src/npa_howtopay/params.py

@define
class TimeSeriesParams:
    npa_projects: pl.DataFrame
    scattershot_electrification: pl.DataFrame
    gas_fixed_overhead_costs: pl.DataFrame
    electric_fixed_overhead_costs: pl.DataFrame
    gas_bau_lpp_costs_per_year: pl.DataFrame

    def __attrs_post_init__(self) -> None:
        """Automatically append scattershot electrification to npa projects. In the BAU scenario, this will only return the scattershot electrification dataframe."""

        self.npa_projects = append_scattershot_electrification_df(self.npa_projects, self.scattershot_electrification)

__attrs_post_init__()

Automatically append scattershot electrification to npa projects. In the BAU scenario, this will only return the scattershot electrification dataframe.

Source code in src/npa_howtopay/params.py

def __attrs_post_init__(self) -> None:
    """Automatically append scattershot electrification to npa projects. In the BAU scenario, this will only return the scattershot electrification dataframe."""

    self.npa_projects = append_scattershot_electrification_df(self.npa_projects, self.scattershot_electrification)

get_available_runs(data_dir='data')

Get list of available run_names from YAML files

Source code in src/npa_howtopay/params.py

def get_available_runs(data_dir: str = "data") -> list[str]:
    """Get list of available run_names from YAML files"""
    import os
    import glob

    yaml_files = glob.glob(f"{data_dir}/*.yaml")
    return [os.path.splitext(os.path.basename(f))[0] for f in yaml_files]

load_scenario_from_yaml(run_name, data_dir='data')

Load default parameters for a specific run_name from its YAML file

Source code in src/npa_howtopay/params.py

def load_scenario_from_yaml(run_name: str, data_dir: str = "data") -> InputParams:
    """Load default parameters for a specific run_name from its YAML file"""
    from pathlib import Path

    # Get the package directory
    package_dir = Path(__file__).parent
    yaml_path = package_dir / data_dir / f"{run_name}.yaml"

    return _load_params_from_yaml(str(yaml_path))

load_time_series_params_from_web_params(web_params, start_year, end_year, cost_inflation_rate=0.0)

Load time series parameters from web parameters (scalar values)

Source code in src/npa_howtopay/params.py

def load_time_series_params_from_web_params(
    web_params: dict, start_year: int, end_year: int, cost_inflation_rate: float = 0.0
) -> TimeSeriesParams:
    """Load time series parameters from web parameters (scalar values)"""

    web_params_obj = WebParams(**web_params)
    generated_data = create_time_series_from_web_params(web_params_obj, start_year, end_year, cost_inflation_rate)

    return TimeSeriesParams(
        npa_projects=generated_data["npa_projects"],
        scattershot_electrification=generated_data["scattershot_electrification_users_per_year"],
        gas_fixed_overhead_costs=generated_data["gas_fixed_overhead_costs"],
        electric_fixed_overhead_costs=generated_data["electric_fixed_overhead_costs"],
        gas_bau_lpp_costs_per_year=generated_data["gas_bau_lpp_costs_per_year"],
    )

load_time_series_params_from_yaml(run_name, data_dir='data')

Load time series parameters from YAML file

Source code in src/npa_howtopay/params.py

def load_time_series_params_from_yaml(run_name: str, data_dir: str = "data") -> TimeSeriesParams:
    """Load time series parameters from YAML file"""
    from pathlib import Path

    # Get the package directory
    package_dir = Path(__file__).parent
    yaml_path = package_dir / data_dir / f"{run_name}.yaml"

    return _load_time_series_params_from_yaml(str(yaml_path))

npa_howtopay.capex_project

NpvSavingsProject

Represents NPV savings from NPA projects that generate performance incentives.

Source code in src/npa_howtopay/capex_project.py

@define
class NpvSavingsProject:
    """Represents NPV savings from NPA projects that generate performance incentives."""

    project_year: int = field()
    savings_amount: float = field(validator=validators.ge(0.0))
    payback_period: int = field(validator=validators.ge(1))

    def to_df(self) -> pl.DataFrame:
        return pl.DataFrame({
            "project_year": [self.project_year],
            "savings_amount": [self.savings_amount],
            "payback_period": pl.Series([self.payback_period], dtype=pl.Int64),
            "end_year": [self.project_year + self.payback_period],
        })

compute_depreciation_expense_from_capex_projects(year, df)

Compute annual depreciation expense for capital projects.

For each project, depreciation expense is the original cost divided evenly over the depreciation lifetime. Depreciation starts the year after the project year and continues for the depreciation lifetime.

Parameters:

Name	Type	Description	Default
year	int	The year to compute depreciation expense for	required
df	DataFrame	DataFrame containing capital projects with columns: - project_year: int - Year project was initiated - original_cost: float - Original cost of the project - depreciation_lifetime: int - Number of years to depreciate over	required

Returns:

Name	Type	Description
float	float	Total depreciation expense for the year across all projects

Source code in src/npa_howtopay/capex_project.py

def compute_depreciation_expense_from_capex_projects(year: int, df: pl.DataFrame) -> float:
    """Compute annual depreciation expense for capital projects.

    For each project, depreciation expense is the original cost divided evenly over the depreciation lifetime.
    Depreciation starts the year after the project year and continues for the depreciation lifetime.

    Args:
        year: The year to compute depreciation expense for
        df: DataFrame containing capital projects with columns:
            - project_year: int - Year project was initiated
            - original_cost: float - Original cost of the project
            - depreciation_lifetime: int - Number of years to depreciate over

    Returns:
        float: Total depreciation expense for the year across all projects
    """
    return float(
        df.select(
            pl.when(
                (pl.lit(year) > pl.col("project_year"))
                & (pl.lit(year) <= pl.col("project_year") + pl.col("depreciation_lifetime"))
            )
            .then(pl.col("original_cost") / pl.col("depreciation_lifetime"))
            .otherwise(pl.lit(0))
        )
        .sum()
        .item()
    )

compute_maintanence_costs(year, df, maintenance_cost_pct)

Compute annual maintenance costs for capital projects.

Parameters:

Name	Type	Description	Default
year	int	The year to compute maintenance costs for	required
df	DataFrame	DataFrame containing capital projects with columns: - project_type: str - Type of project (npa or other) - original_cost: float - Original cost of the project	required
maintenance_cost_pct	float	float - Annual maintenance cost as percentage of original cost	required

Returns:

Name	Type	Description
float	float	Total maintenance costs for the year, excluding NPA projects

Source code in src/npa_howtopay/capex_project.py

def compute_maintanence_costs(year: int, df: pl.DataFrame, maintenance_cost_pct: float) -> float:
    """Compute annual maintenance costs for capital projects.

    Args:
        year: The year to compute maintenance costs for
        df: DataFrame containing capital projects with columns:
            - project_type: str - Type of project (npa or other)
            - original_cost: float - Original cost of the project
        maintenance_cost_pct: float - Annual maintenance cost as percentage of original cost

    Returns:
        float: Total maintenance costs for the year, excluding NPA projects
    """
    df = df.filter(pl.col("project_type") != "npa", pl.col("project_year") <= year, pl.col("retirement_year") >= year)
    return float(df.select(pl.col("original_cost")).sum().item() * maintenance_cost_pct)

compute_npv_of_capex_investment(initial_cost, lifetime, ror, real_dollar_discount_rate, year)

Compute NPV of a capex investment. Year one incurrs a cost, then subsequent years earn a return equal to the ror on the annually depreciated value of the investment.

Parameters:

Name	Type	Description	Default
initial_cost	float	Initial investment cost	required
lifetime	int	Investment lifetime in years	required
ror	float	Rate of return on ratebase	required
real_dollar_discount_rate	float	Discount rate for NPV	required
year	int	Year of investment	required

Returns:

Name	Type	Description
float	float	NPV of the investment

Source code in src/npa_howtopay/capex_project.py

def compute_npv_of_capex_investment(
    initial_cost: float, lifetime: int, ror: float, real_dollar_discount_rate: float, year: int
) -> float:
    """Compute NPV of a capex investment. Year one incurrs a cost, then subsequent years earn a return equal to the ror on the annually depreciated value of the investment.

    Args:
        initial_cost: Initial investment cost
        lifetime: Investment lifetime in years
        ror: Rate of return on ratebase
        real_dollar_discount_rate: Discount rate for NPV
        year: Year of investment

    Returns:
        float: NPV of the investment
    """
    if initial_cost == 0:
        return 0.0

    # Initial cost (negative cash flow)
    npv = -initial_cost

    # Annual returns over lifetime
    for t in range(1, lifetime + 1):
        # Annual return = remaining_ratebase_value * ror
        # Remaining value declines linearly from initial_cost to 0
        remaining_value_fraction = max(0, 1 - (t - 1) / lifetime)
        # Return on ratebase
        return_on_ratebase = initial_cost * remaining_value_fraction * ror
        # Depreciation recovery
        depreciation_recovery = initial_cost / lifetime

        # Total annual cash flow
        annual_cash_flow = return_on_ratebase + depreciation_recovery

        # Discount to present value
        discount_factor = 1 / ((1 + real_dollar_discount_rate) ** t)
        npv += annual_cash_flow * discount_factor

    return npv

compute_npv_savings_from_npa_projects(year, npa_projects, npa_install_cost, npa_lifetime, pipeline_depreciation_lifetime, gas_ror, npv_discount_rate, performance_incentive_pct, incentive_payback_period)

Generate NPV savings projects for NPA installations.

This function calculates the NPV difference between NPA investment and avoided LPP spending. The savings are tracked as projects that generate performance incentives over a payback period.

Parameters:

Name	Type	Description	Default
year	int	The year to generate NPV savings projects for	required
npa_projects	DataFrame	DataFrame containing NPA project details	required
npa_install_cost	float	Cost per household of installing an NPA	required
npa_lifetime	int	Expected lifetime in years of an NPA installation	required
pipeline_depreciation_lifetime	int	Depreciation lifetime for avoided pipe projects	required
gas_ror	float	Rate of return on gas utility investments	required
npv_discount_rate	float	Discount rate for NPV calculations	required
performance_incentive_pct	float	Percentage of savings on which gas utility receives a performance incentive	required

Returns: pl.DataFrame with columns: - project_year: Year the NPV savings project was initiated - savings_amount: Total NPV savings amount - payback_period: Number of years to pay incentives - end_year: Year the incentive payments end

Source code in src/npa_howtopay/capex_project.py

def compute_npv_savings_from_npa_projects(
    year: int,
    npa_projects: pl.DataFrame,
    npa_install_cost: float,
    npa_lifetime: int,
    pipeline_depreciation_lifetime: int,
    gas_ror: float,
    npv_discount_rate: float,
    performance_incentive_pct: float,
    incentive_payback_period: int,
) -> pl.DataFrame:
    """Generate NPV savings projects for NPA installations.

    This function calculates the NPV difference between NPA investment and avoided LPP spending.
    The savings are tracked as projects that generate performance incentives over a payback period.

    Args:
        year: The year to generate NPV savings projects for
        npa_projects: DataFrame containing NPA project details
        npa_install_cost: Cost per household of installing an NPA
        npa_lifetime: Expected lifetime in years of an NPA installation
        pipeline_depreciation_lifetime: Depreciation lifetime for avoided pipe projects
        gas_ror: Rate of return on gas utility investments
        npv_discount_rate: Discount rate for NPV calculations
        performance_incentive_pct: Percentage of savings on which gas utility receives a performance incentive
    Returns:
        pl.DataFrame with columns:
            - project_year: Year the NPV savings project was initiated
            - savings_amount: Total NPV savings amount
            - payback_period: Number of years to pay incentives
            - end_year: Year the incentive payments end
    """
    npas_this_year = npa_projects.filter(pl.col("project_year") == year)

    if npas_this_year.height == 0:
        return return_empty_npv_savings_df()

    # Calculate costs
    num_converts = compute_hp_converts_from_df(year, npas_this_year, cumulative=False, npa_only=True)
    npa_investment_cost = npa_install_cost * num_converts
    avoided_lpp_cost = compute_npa_pipe_cost_avoided_from_df(year, npas_this_year)

    # Calculate NPVs
    npa_npv = npa_investment_cost  # npa investment is opex so costs are recouped in the same year with no ror

    avoided_lpp_npv = compute_npv_of_capex_investment(
        initial_cost=avoided_lpp_cost,
        lifetime=pipeline_depreciation_lifetime,
        ror=gas_ror,
        real_dollar_discount_rate=npv_discount_rate,
        year=year,
    )

    savings_amount = (avoided_lpp_npv - npa_npv) * performance_incentive_pct

    if savings_amount > 0:
        return NpvSavingsProject(
            project_year=year,
            savings_amount=savings_amount,
            payback_period=incentive_payback_period,  # 10-year incentive period
        ).to_df()
    else:
        return return_empty_npv_savings_df()

compute_performance_incentive_this_year(year, df)

Compute the ratebase value for a given year from NPV savings projects.

For each savings project, the ratebase value is the savings amount divided by payback period. Projects that haven't started yet (year < project_year) have zero ratebase value. Projects that are fully paid back have zero ratebase value.

Parameters:

Name	Type	Description	Default
year	int	The year to compute ratebase for	required
df	DataFrame	DataFrame containing NPV savings projects with columns: - project_year: int - Year project was initiated - savings_amount: float - Total NPV savings amount - payback_period: int - Number of years to pay incentives	required

Returns:

Name	Type	Description
float	float	Total ratebase value for the year across all savings projects

Source code in src/npa_howtopay/capex_project.py

def compute_performance_incentive_this_year(year: int, df: pl.DataFrame) -> float:
    """Compute the ratebase value for a given year from NPV savings projects.

    For each savings project, the ratebase value is the savings amount divided by payback period.
    Projects that haven't started yet (year < project_year) have zero ratebase value.
    Projects that are fully paid back have zero ratebase value.

    Args:
        year: The year to compute ratebase for
        df: DataFrame containing NPV savings projects with columns:
            - project_year: int - Year project was initiated
            - savings_amount: float - Total NPV savings amount
            - payback_period: int - Number of years to pay incentives

    Returns:
        float: Total ratebase value for the year across all savings projects
    """
    if df.height == 0:
        return 0.0
    df = df.with_columns(
        pl.when((pl.lit(year) >= pl.col("project_year")) & (pl.lit(year) < pl.col("end_year")))
        .then(pl.col("savings_amount") / pl.col("payback_period"))
        .otherwise(pl.lit(0))
        .alias("annual_ratebase_contribution")
    )
    return float(df.select(pl.col("annual_ratebase_contribution")).sum().item())

compute_ratebase_from_capex_projects(year, df)

Compute the ratebase value for a given year from capital projects.

For each project, the ratebase value declines linearly from the original cost to zero over the depreciation lifetime. Projects that haven't started yet (year < project_year) have zero ratebase value. Projects that are fully depreciated have zero ratebase value. Projects that are in the year of the project have the full original cost added to the ratebase.

Parameters:

Name	Type	Description	Default
year	int	The year to compute ratebase for	required
df	DataFrame	DataFrame containing capital projects with columns: - project_year: int - Year project was initiated - original_cost: float - Original cost of the project - depreciation_lifetime: int - Number of years to depreciate over	required

Returns:

Name	Type	Description
float	float	Total ratebase value for the year across all projects

Source code in src/npa_howtopay/capex_project.py

def compute_ratebase_from_capex_projects(year: int, df: pl.DataFrame) -> float:
    """Compute the ratebase value for a given year from capital projects.

    For each project, the ratebase value declines linearly from the original cost to zero over the depreciation lifetime.
    Projects that haven't started yet (year < project_year) have zero ratebase value.
    Projects that are fully depreciated have zero ratebase value.
    Projects that are in the year of the project have the full original cost added to the ratebase.

    Args:
        year: The year to compute ratebase for
        df: DataFrame containing capital projects with columns:
            - project_year: int - Year project was initiated
            - original_cost: float - Original cost of the project
            - depreciation_lifetime: int - Number of years to depreciate over

    Returns:
        float: Total ratebase value for the year across all projects
    """
    df = df.with_columns(
        pl.when(pl.lit(year) < pl.col("project_year"))
        .then(pl.lit(0))
        .otherwise((1 - (pl.lit(year) - pl.col("project_year")) / pl.col("depreciation_lifetime")).clip(lower_bound=0))
        .alias("depreciation_fraction")
    )
    return float(df.select(pl.col("depreciation_fraction") * pl.col("original_cost")).sum().item())

get_grid_upgrade_capex_projects(year, npa_projects, peak_hp_kw, peak_aircon_kw, distribution_cost_per_peak_kw_increase, grid_upgrade_depreciation_lifetime)

Generate capex projects for grid upgrades needed to support NPA installations.

This function calculates the required grid upgrades based on the peak power increase from heat pumps and air conditioners installed as part of NPA projects. The cost scales linearly with the total peak power increase.

Parameters:

Name	Type	Description	Default
year	int	The year to generate projects for	required
npa_projects	DataFrame	DataFrame containing NPA project details	required
peak_hp_kw	float	Peak power draw in kW for a heat pump	required
peak_aircon_kw	float	Peak power draw in kW for an air conditioner	required
distribution_cost_per_peak_kw_increase	float	Cost per kW of increasing grid capacity in year of project	required
grid_upgrade_depreciation_lifetime	int	Depreciation lifetime in years for grid upgrades	required

Returns:

Type	Description
DataFrame	pl.DataFrame with columns: - project_year: Year the project was initiated - project_type: "grid_upgrade" for grid capacity upgrades - original_cost: Total cost of grid upgrades - depreciation_lifetime: Depreciation lifetime in years

Source code in src/npa_howtopay/capex_project.py

def get_grid_upgrade_capex_projects(
    year: int,
    npa_projects: pl.DataFrame,
    peak_hp_kw: float,
    peak_aircon_kw: float,
    distribution_cost_per_peak_kw_increase: float,
    grid_upgrade_depreciation_lifetime: int,
) -> pl.DataFrame:
    """
    Generate capex projects for grid upgrades needed to support NPA installations.

    This function calculates the required grid upgrades based on the peak power increase
    from heat pumps and air conditioners installed as part of NPA projects. The cost
    scales linearly with the total peak power increase.

    Args:
        year: The year to generate projects for
        npa_projects: DataFrame containing NPA project details
        peak_hp_kw: Peak power draw in kW for a heat pump
        peak_aircon_kw: Peak power draw in kW for an air conditioner
        distribution_cost_per_peak_kw_increase: Cost per kW of increasing grid capacity in year of project
        grid_upgrade_depreciation_lifetime: Depreciation lifetime in years for grid upgrades

    Returns:
        pl.DataFrame with columns:
            - project_year: Year the project was initiated
            - project_type: "grid_upgrade" for grid capacity upgrades
            - original_cost: Total cost of grid upgrades
            - depreciation_lifetime: Depreciation lifetime in years
    """
    npas_this_year = npa_projects.filter(pl.col("project_year") == year)
    peak_kw_increase = compute_peak_kw_increase_from_df(year, npas_this_year, peak_hp_kw, peak_aircon_kw)
    if peak_kw_increase > 0:
        return CapexProject(
            project_year=year,
            project_type="grid_upgrade",
            original_cost=peak_kw_increase * distribution_cost_per_peak_kw_increase,
            depreciation_lifetime=grid_upgrade_depreciation_lifetime,
        ).to_df()
    else:
        return return_empty_capex_df()

get_lpp_gas_capex_projects(year, gas_bau_lpp_costs_per_year, npa_projects, depreciation_lifetime)

Generate capex projects for leak-prone pipe (LPP) replacement in the gas system.

This function calculates the remaining pipe replacement costs after accounting for pipe replacements avoided by NPA projects. If NPAs avoid all planned pipe replacements in a given year, returns an empty dataframe.

Parameters:

Name	Type	Description	Default
year	int	The year to generate projects for	required
gas_bau_lpp_costs_per_year	DataFrame	DataFrame containing business-as-usual pipe replacement costs with columns: - year: Year of planned replacement - cost: Cost of planned replacement Note: Multiple entries may exist per year	required
npa_projects	DataFrame	DataFrame containing NPA project details, used to calculate avoided pipe costs	required
depreciation_lifetime	int	Depreciation lifetime in years for pipe replacement projects	required

Returns:

Type	Description
DataFrame	pl.DataFrame with columns: - project_year: Year the project was initiated - project_type: "pipeline" for pipe replacement projects - original_cost: Cost of the project after subtracting avoided costs - depreciation_lifetime: Depreciation lifetime in years - retirement_year: Year the project is fully depreciated

Source code in src/npa_howtopay/capex_project.py

def get_lpp_gas_capex_projects(
    year: int,
    gas_bau_lpp_costs_per_year: pl.DataFrame,
    npa_projects: pl.DataFrame,
    depreciation_lifetime: int,
) -> pl.DataFrame:
    """
    Generate capex projects for leak-prone pipe (LPP) replacement in the gas system.

    This function calculates the remaining pipe replacement costs after accounting for pipe
    replacements avoided by NPA projects. If NPAs avoid all planned pipe replacements in a given
    year, returns an empty dataframe.

    Args:
        year: The year to generate projects for
        gas_bau_lpp_costs_per_year: DataFrame containing business-as-usual pipe replacement costs
            with columns:
            - year: Year of planned replacement
            - cost: Cost of planned replacement
            Note: Multiple entries may exist per year
        npa_projects: DataFrame containing NPA project details, used to calculate avoided pipe costs
        depreciation_lifetime: Depreciation lifetime in years for pipe replacement projects

    Returns:
        pl.DataFrame with columns:
            - project_year: Year the project was initiated
            - project_type: "pipeline" for pipe replacement projects
            - original_cost: Cost of the project after subtracting avoided costs
            - depreciation_lifetime: Depreciation lifetime in years
            - retirement_year: Year the project is fully depreciated
    """
    npas_this_year = npa_projects.filter(pl.col("project_year") == year)
    npa_pipe_costs_avoided = compute_npa_pipe_cost_avoided_from_df(year, npas_this_year)
    bau_pipe_replacement_costs = (
        gas_bau_lpp_costs_per_year.filter(pl.col("year") == year).select(pl.col("cost")).sum().item()
    )
    remaining_pipe_replacement_cost = np.maximum(0, bau_pipe_replacement_costs - npa_pipe_costs_avoided)
    if remaining_pipe_replacement_cost > 0:
        return CapexProject(
            project_year=year,
            project_type="pipeline",
            original_cost=remaining_pipe_replacement_cost,
            depreciation_lifetime=depreciation_lifetime,
        ).to_df()
    else:
        return return_empty_capex_df()

get_non_lpp_gas_capex_projects(year, current_ratebase, baseline_non_lpp_gas_ratebase_growth, depreciation_lifetime, construction_inflation_rate)

Generate capex projects for non-LPP (non-leak prone pipe) gas infrastructure.

These represent routine gas infrastructure investments not related to pipe replacement or npas, such as meter replacements, regulator stations, etc. The cost is calculated as a percentage of current ratebase, adjusted for construction cost inflation.

Parameters:

Name	Type	Description	Default
year	int	The year to generate projects for	required
current_ratebase	float	Current value of the gas utility's ratebase	required
baseline_non_lpp_gas_ratebase_growth	float	Annual growth rate for non-LPP capex as fraction of ratebase	required
depreciation_lifetime	int	Blended depreciation lifetime in years for these projects	required
construction_inflation_rate	float	Annual inflation rate for construction costs	required

Returns:

Type	Description
DataFrame	pl.DataFrame with columns: - project_year: Year the project was initiated - project_type: "misc" for miscellaneous gas infrastructure - original_cost: Cost of the project - depreciation_lifetime: Depreciation lifetime in years - retirement_year: Year the project is fully depreciated

Source code in src/npa_howtopay/capex_project.py

def get_non_lpp_gas_capex_projects(
    year: int,
    current_ratebase: float,
    baseline_non_lpp_gas_ratebase_growth: float,
    depreciation_lifetime: int,
    construction_inflation_rate: float,
) -> pl.DataFrame:
    """
    Generate capex projects for non-LPP (non-leak prone pipe) gas infrastructure.

    These represent routine gas infrastructure investments not related to pipe replacement or npas,
    such as meter replacements, regulator stations, etc. The cost is calculated as a
    percentage of current ratebase, adjusted for construction cost inflation.

    Args:
        year: The year to generate projects for
        current_ratebase: Current value of the gas utility's ratebase
        baseline_non_lpp_gas_ratebase_growth: Annual growth rate for non-LPP capex as fraction of ratebase
        depreciation_lifetime: Blended depreciation lifetime in years for these projects
        construction_inflation_rate: Annual inflation rate for construction costs

    Returns:
        pl.DataFrame with columns:
            - project_year: Year the project was initiated
            - project_type: "misc" for miscellaneous gas infrastructure
            - original_cost: Cost of the project
            - depreciation_lifetime: Depreciation lifetime in years
            - retirement_year: Year the project is fully depreciated
    """
    return CapexProject(
        project_year=year,
        project_type="misc",
        original_cost=current_ratebase * baseline_non_lpp_gas_ratebase_growth * (1 + construction_inflation_rate),
        depreciation_lifetime=depreciation_lifetime,
    ).to_df()

get_non_npa_electric_capex_projects(year, current_ratebase, baseline_electric_ratebase_growth, depreciation_lifetime, construction_inflation_rate)

Generate capex projects for non-NPA non-grid upgrade electric system upgrades.

This function calculates the baseline capital expenditures for the electric system, excluding NPA-related projects. The expenditure grows with both the baseline growth rate and construction inflation.

Parameters:

Name	Type	Description	Default
year	int	The year to generate projects for	required
current_ratebase	float	Current value of the electric utility's ratebase	required
baseline_electric_ratebase_growth	float	Annual growth rate of non-NPA electric capex as fraction of ratebase	required
depreciation_lifetime	int	Blended depreciation lifetime in years for electric system projects	required
construction_inflation_rate	float	Annual inflation rate for construction costs	required

Returns:

Type	Description
DataFrame	pl.DataFrame with columns: - project_year: Year the project was initiated - project_type: "misc" for miscellaneous electric system upgrades - original_cost: Cost of the project including construction inflation - depreciation_lifetime: Depreciation lifetime in years

Source code in src/npa_howtopay/capex_project.py

def get_non_npa_electric_capex_projects(
    year: int,
    current_ratebase: float,
    baseline_electric_ratebase_growth: float,
    depreciation_lifetime: int,
    construction_inflation_rate: float,
) -> pl.DataFrame:
    """
    Generate capex projects for non-NPA non-grid upgrade electric system upgrades.

    This function calculates the baseline capital expenditures for the electric system,
    excluding NPA-related projects. The expenditure grows with both the baseline growth rate
    and construction inflation.

    Args:
        year: The year to generate projects for
        current_ratebase: Current value of the electric utility's ratebase
        baseline_electric_ratebase_growth: Annual growth rate of non-NPA electric capex as fraction of ratebase
        depreciation_lifetime: Blended depreciation lifetime in years for electric system projects
        construction_inflation_rate: Annual inflation rate for construction costs

    Returns:
        pl.DataFrame with columns:
            - project_year: Year the project was initiated
            - project_type: "misc" for miscellaneous electric system upgrades
            - original_cost: Cost of the project including construction inflation
            - depreciation_lifetime: Depreciation lifetime in years
    """
    return CapexProject(
        project_year=year,
        project_type="misc",
        original_cost=current_ratebase * baseline_electric_ratebase_growth * (1 + construction_inflation_rate),
        depreciation_lifetime=depreciation_lifetime,
    ).to_df()

get_npa_capex_projects(year, npa_projects, npa_install_cost, npa_lifetime)

Generate capex projects for NPA (non-pipe alternative) installations.

This function calculates the capital costs associated with installing NPAs in a given year. The total cost is based on the number of heat pump conversions and the per-unit installation cost.

Parameters:

Name	Type	Description	Default
year	int	The year to generate projects for	required
npa_projects	DataFrame	DataFrame containing NPA project details	required
npa_install_cost	float	Cost per household of installing an NPA	required
npa_lifetime	int	Expected lifetime in years of an NPA installation	required

Returns:

Type	Description
DataFrame	pl.DataFrame with columns: - project_year: Year the project was initiated - project_type: "npa" for NPA installations - original_cost: Total cost of NPA installations - depreciation_lifetime: Depreciation lifetime in years

Source code in src/npa_howtopay/capex_project.py

def get_npa_capex_projects(
    year: int, npa_projects: pl.DataFrame, npa_install_cost: float, npa_lifetime: int
) -> pl.DataFrame:
    """
    Generate capex projects for NPA (non-pipe alternative) installations.

    This function calculates the capital costs associated with installing NPAs in a given year.
    The total cost is based on the number of heat pump conversions and the per-unit installation cost.

    Args:
        year: The year to generate projects for
        npa_projects: DataFrame containing NPA project details
        npa_install_cost: Cost per household of installing an NPA
        npa_lifetime: Expected lifetime in years of an NPA installation

    Returns:
        pl.DataFrame with columns:
            - project_year: Year the project was initiated
            - project_type: "npa" for NPA installations
            - original_cost: Total cost of NPA installations
            - depreciation_lifetime: Depreciation lifetime in years
    """
    npas_this_year = npa_projects.filter(pl.col("project_year") == year)
    npa_total_cost = npa_install_cost * compute_hp_converts_from_df(
        year, npas_this_year, cumulative=False, npa_only=True
    )
    if npa_total_cost > 0:
        return CapexProject(
            project_year=year, project_type="npa", original_cost=npa_total_cost, depreciation_lifetime=npa_lifetime
        ).to_df()
    else:
        return return_empty_capex_df()

get_synthetic_initial_capex_projects(start_year, initial_ratebase, depreciation_lifetime)

Generate synthetic capex projects to represent the projects that make up the initial ratebase.

Creates a series of historical capex projects that would result in the given initial ratebase value, assuming straight-line depreciation. Uses the triangular number formula to create a uniform distribution of projects over the depreciation lifetime, where each project has the same original cost. Projects are distributed evenly over depreciation_lifetime years leading up to start_year.

Parameters:

Name	Type	Description	Default
start_year	int	The first year of the model	required
initial_ratebase	float	The target ratebase value at start_year	required
depreciation_lifetime	int	The blended depreciation lifetime for the synthetic projects	required

Returns:

Type	Description
DataFrame	pl.DataFrame with columns: - project_year: Year the project was initiated - project_type: "synthetic_initial" - original_cost: Cost of the project - depreciation_lifetime: Depreciation lifetime in years - retirement_year: Year the project is fully depreciated

Source code in src/npa_howtopay/capex_project.py

def get_synthetic_initial_capex_projects(
    start_year: int, initial_ratebase: float, depreciation_lifetime: int
) -> pl.DataFrame:
    """
    Generate synthetic capex projects to represent the projects that make up the initial ratebase.

    Creates a series of historical capex projects that would result in the given initial ratebase value, assuming straight-line depreciation. Uses the triangular number formula to create a uniform distribution of projects over the depreciation lifetime, where each project has the same original cost. Projects are distributed evenly over depreciation_lifetime years leading up to start_year.

    Args:
        start_year: The first year of the model
        initial_ratebase: The target ratebase value at start_year
        depreciation_lifetime: The blended depreciation lifetime for the synthetic projects

    Returns:
        pl.DataFrame with columns:
            - project_year: Year the project was initiated
            - project_type: "synthetic_initial"
            - original_cost: Cost of the project
            - depreciation_lifetime: Depreciation lifetime in years
            - retirement_year: Year the project is fully depreciated
    """
    total_weight = (depreciation_lifetime * (depreciation_lifetime + 1) / 2) / depreciation_lifetime
    est_original_cost_per_year = initial_ratebase / total_weight
    project_years = range(start_year - depreciation_lifetime + 1, start_year + 1)
    return pl.DataFrame({
        "project_year": project_years,
        "project_type": ["synthetic_initial"] * depreciation_lifetime,
        "original_cost": est_original_cost_per_year,
        "depreciation_lifetime": pl.Series([depreciation_lifetime] * depreciation_lifetime, dtype=pl.Int64),
        "retirement_year": pl.Series([year + depreciation_lifetime for year in project_years], dtype=pl.Int64),
    })

return_empty_npv_savings_df()

Return empty DataFrame for NPV savings projects with proper schema.

Source code in src/npa_howtopay/capex_project.py

def return_empty_npv_savings_df() -> pl.DataFrame:
    """Return empty DataFrame for NPV savings projects with proper schema."""
    return pl.DataFrame({
        "project_year": pl.Series([], dtype=pl.Int64),
        "savings_amount": pl.Series([], dtype=pl.Float64),
        "payback_period": pl.Series([], dtype=pl.Int64),
        "end_year": pl.Series([], dtype=pl.Int64),
    })

npa_howtopay.npa_project

append_scattershot_electrification_df(npa_projects_df, scattershot_electrification_df)

Append a dataframe of scattershot electrification projects to the npa projects df. Scattershot electrification projects match the schema for npa projects, but will only affect the number of users and total electric usage, not anything related to pipe value or grid upgrades. It is meant to capture customers leaving the gas network independent of NPA projects.

Source code in src/npa_howtopay/npa_project.py

def append_scattershot_electrification_df(
    npa_projects_df: pl.DataFrame,
    scattershot_electrification_df: pl.DataFrame,
) -> pl.DataFrame:
    """
    Append a dataframe of scattershot electrification projects to the npa projects df.
    Scattershot electrification projects match the schema for npa projects, but will only affect the number of users and total electric usage, not anything related to pipe value or grid upgrades. It is meant to capture customers leaving the gas network independent of NPA projects.
    """
    scattershot_with_npa_cols = scattershot_electrification_df.with_columns(
        pl.lit(0.0).alias("pipe_value_per_user"),
        pl.lit(0.0).alias("pipe_decomm_cost_per_user"),
        pl.lit(np.inf).alias("peak_kw_winter_headroom"),
        pl.lit(np.inf).alias("peak_kw_summer_headroom"),
        pl.lit(0.0).alias("aircon_percent_adoption_pre_npa"),
        pl.lit(True).alias("is_scattershot"),
    )
    return pl.concat([npa_projects_df, scattershot_with_npa_cols])

npa_howtopay.web_params

create_scattershot_electrification_df(web_params, start_year, end_year)

Generate a dataframe of scattershot electrification projects, one per year. The projects are distributed evenly across the years. These match the schema for npa projects, but will only affect the number of users, not anything related to pipe value or grid upgrades

Source code in src/npa_howtopay/web_params.py

def create_scattershot_electrification_df(
    web_params: WebParams,
    start_year: int,
    end_year: int,
) -> pl.DataFrame:
    """
    Generate a dataframe of scattershot electrification projects, one per year.
    The projects are distributed evenly across the years. These match the schema for npa projects, but will only
    affect the number of users, not anything related to pipe value or grid upgrades
    """
    years = range(start_year, end_year + 1)
    num_years = len(years)

    return pl.DataFrame({
        "project_year": list(years),
        "num_converts": [web_params.scattershot_electrification_users_per_year] * num_years,
    })

create_time_series_from_web_params(web_params, start_year, end_year, cost_inflation_rate=0.0)

Create all time series DataFrames from web parameters

Source code in src/npa_howtopay/web_params.py

def create_time_series_from_web_params(
    web_params: WebParams, start_year: int, end_year: int, cost_inflation_rate: float = 0.0
) -> dict[str, pl.DataFrame]:
    """Create all time series DataFrames from web parameters"""
    npa_year_end = web_params.npa_year_end if web_params.npa_year_end is not None else end_year
    npa_year_start = web_params.npa_year_start if web_params.npa_year_start is not None else start_year

    if npa_year_start < start_year:
        raise ValueError("npa_year_start must be greater than or equal to SharedParams.start_year")
    if npa_year_end > end_year:
        raise ValueError("npa_year_end must be less than or equal to npa_end_year")

    return {
        "npa_projects": create_npa_projects(web_params, npa_year_start, npa_year_end),
        "scattershot_electrification_users_per_year": create_scattershot_electrification_df(
            web_params, start_year, end_year
        ),
        "gas_fixed_overhead_costs": create_gas_fixed_overhead_costs(
            web_params, start_year, end_year, cost_inflation_rate
        ),
        "electric_fixed_overhead_costs": create_electric_fixed_overhead_costs(
            web_params, start_year, end_year, cost_inflation_rate
        ),
        "gas_bau_lpp_costs_per_year": create_gas_bau_lpp_costs_per_year(
            web_params, start_year, end_year, cost_inflation_rate
        ),
    }