Source code for buildingmodel.models.energy_consumption

import polars as pl

dhw_efficiencies = {
    "electricity": 0.7,
    "fossil_gas": 0.6,
    "liquified_petroleum_gas": 0.6,
    "oil": 0.6,
    "biomass": 0.6,
    "district_network": 0.6,
}


def cooking_consumption(buildings, dwellings, parameters):
    dw_res_to_agg = [
        "annual_cooking_needs",
        "peak_cooking_needs",
    ]

    agg_cols = ["building_id", "main_cooking_energy", "secondary_cooking_energy"]
    dwelling_agg = dwellings.group_by(agg_cols).agg(
        **{res: pl.col(res).sum() for res in dw_res_to_agg}
    )
    cols_names = []
    expr_dict = {"main": {}, "secondary": {}}
    cooking_energy = [
        e for e in parameters.energy_list if e not in ["oil", "district_network"]
    ]
    for value_type in ["annual", "peak"]:
        for energy in cooking_energy:
            col_name = f"{value_type}_{energy}_cooking"
            cols_names.append(col_name)
            for priority in ["main", "secondary"]:
                expr_dict[priority][col_name] = (
                    pl.when(pl.col(f"{priority}_cooking_energy") == energy)
                    .then(pl.col(col_name) + pl.col(f"{value_type}_cooking_needs"))
                    .otherwise(pl.col(col_name))
                )

    dwelling_agg = (
        dwelling_agg.with_columns(**{c: pl.lit(0.0) for c in cols_names})
        .with_columns(**expr_dict["main"])
        .with_columns(**expr_dict["secondary"])
    )
    buildings = buildings.join(
        dwelling_agg.group_by("building_id").agg(
            **{c: pl.col(c).sum() for c in cols_names}
        ),
        on="building_id",
    )
    return buildings


def specific_consumption(buildings, dwellings):
    dw_res_to_agg = {
        "annual_specific_needs": "annual_electricity_specific",
        "peak_specific_needs": "peak_electricity_specific",
    }

    dwelling_agg = dwellings.group_by("building_id").agg(
        **{name: pl.col(res).sum() for res, name in dw_res_to_agg.items()}
    )
    buildings = buildings.join(dwelling_agg, on="building_id")

    return buildings


def heating_dhw_consumption(buildings, dwellings, parameters):
    dw_res_to_agg = [
        "annual_dhw_needs",
        "peak_dhw_needs",
        "conventional_dhw_needs",
    ]

    dwelling_agg = dwellings.group_by("building_id").agg(
        *[pl.col(res).sum() for res in dw_res_to_agg]
    )
    buildings = (
        buildings.join(dwelling_agg, on="building_id")
        .with_columns(pl.col("backup_heating_share").fill_null(0.0).fill_nan(0.0))
        .with_columns(pl.col("backup_heating_system_efficiency").fill_null(0.6).fill_nan(0.6))
        .with_columns(pl.col("dhw_efficiency").fill_null(0.6).fill_nan(0.6))
        .with_columns(main_heating_share=1 - pl.col("backup_heating_share"))
    )

    expr_dict = {"main": {}, "backup": {}, "dhw": {}}
    cols_names = []
    for value_type in ["annual", "peak", "conventional"]:
        for energy in parameters.energy_list:
            col_name = f"{value_type}_{energy}_heating"
            cols_names.append(col_name)
            for priority in ["main", "backup"]:
                expr_dict[priority][col_name] = (
                    pl.when(pl.col(f"{priority}_heating_energy") == energy)
                    .then(
                        pl.col(col_name)
                        + pl.col(f"{value_type}_heating_needs")
                        * pl.col(f"{priority}_heating_share")
                        / pl.col(f"{priority}_heating_system_efficiency")
                    )
                    .otherwise(pl.col(col_name))
                )

            col_name = f"{value_type}_{energy}_dhw"
            expr_dict["dhw"][col_name] = (
                pl.when(pl.col("dhw_energy") == energy)
                .then(pl.col(f"{value_type}_dhw_needs") / pl.col(f"dhw_efficiency"))
                .otherwise(pl.lit(0.0))
            )

    buildings = (
        buildings.with_columns(**{c: pl.lit(0.0) for c in cols_names})
        .with_columns(**expr_dict["main"])
        .with_columns(**expr_dict["backup"])
        .with_columns(**expr_dict["dhw"])
    )

    return buildings


def agg_use_by_energy(buildings, parameters):
    expr_dict = {}
    for value_type in ["annual", "peak", "conventional"]:
        for energy in parameters.energy_list:
            cols_to_sum = [
                c for c in buildings.columns if (value_type in c) and (energy in c)
            ]
            expr_dict[f"{value_type}_{energy}_consumption"] = pl.sum_horizontal(
                *cols_to_sum
            )

    buildings = buildings.with_columns(**expr_dict)
    return buildings




def run_models(buildings, dwellings, parameters):
    """
    runs the energy consumption model

    Args:
        buildings:

    Returns:

    """
    buildings = heating_dhw_consumption(buildings, dwellings, parameters)
    buildings = specific_consumption(buildings, dwellings)
    buildings = cooking_consumption(buildings, dwellings, parameters)
    buildings = agg_use_by_energy(buildings, parameters)

    return buildings