Source code for buildingmodel.models.thermal_needs

from ..inference.dwelling import living_area_share_limit
import numpy as np
import pandas as pd
import polars as pl
from pathlib import Path

from buildingmodel.utils import get_beta_distribution
from buildingmodel import data_path
from ..logger import duration_logging

building_stat_path = Path(data_path["building_statistics"])
regulation_factors = pd.read_csv(building_stat_path / "regulation_factors.csv")




@duration_logging
def calculate_thermal_needs(buildings, boundaries, dwellings, parameters):
    """
    Calculates the annual and peak thermal needs by combining the boundary losses, ventilation losses, solar gains,
    occupant internal gains and intermittency factor

    Args:
        buildings (GeoDataframe): a GeoDataframe containing the building geometries and parameters
        boundaries (GeoDataframe): a GeoDataframe containing the boundary geometries and parameters

    Returns:

    """

    buildings = buildings.with_columns(
        heated_area_share=pl.col("living_area") / pl.col("floor_area")
    )
    heating_season_ratio = buildings["heating_season_duration"][0] / 8760.0

    dwelling_agg = dwellings.group_by(["building_id"]).agg(
        annual_occupant_gains=pl.col("occupant_gains").sum() * heating_season_ratio,
        conventional_occupant_gains=pl.col("conventional_occupant_gains").sum()
        * heating_season_ratio,
    )

    boundary_agg = boundaries.group_by(["building_id"]).agg(
        annual_boundary_losses=pl.col("annual_thermal_losses").sum(),
        conventional_boundary_losses=pl.col("conventional_thermal_losses").sum(),
        annual_solar_gains=pl.col("transmitted_solar_gain").sum(),
        peak_boundary_losses=pl.col("peak_thermal_losses").sum(),
    )
    buildings = buildings.join(dwelling_agg, on="building_id").join(
        boundary_agg, on="building_id"
    )
    gain_share = parameters.maximal_occupant_gain_share
    solar_share = parameters.maximal_solar_gain_share

    buildings = (
        buildings.with_columns(
            annual_thermal_losses=pl.col("annual_boundary_losses")
            + pl.col("annual_ventilation_losses"),
        )
        .with_columns(
            annual_occupant_gains=pl.col("annual_occupant_gains").clip(
                0.0, gain_share * pl.col("annual_thermal_losses")
            ),
            annual_solar_gains=pl.col("annual_solar_gains").clip(
                0.0, solar_share * pl.col("annual_thermal_losses")
            ),
        )
        .with_columns(
            annual_heating_needs=(
                pl.col("annual_thermal_losses")
                - pl.col("annual_solar_gains")
                - pl.col("annual_occupant_gains")
            )
            * pl.col("heated_area_share")
            * pl.col("intermittency_factor")
            * pl.col("occupation_factor")
        )
        .with_columns(
            peak_heating_needs=(
                pl.col("peak_boundary_losses") + pl.col("peak_ventilation_losses")
            ),
        )
        .with_columns(
            conventional_thermal_losses=pl.col("conventional_boundary_losses")
            + pl.col("conventional_ventilation_losses"),
        )
        .with_columns(
            conventional_occupant_gains=pl.col("conventional_occupant_gains").clip(
                0.0, gain_share * pl.col("conventional_thermal_losses")
            ),
        )
        .with_columns(
            conventional_heating_needs=(
                pl.col("conventional_thermal_losses")
                - pl.col("annual_solar_gains")
                - pl.col("conventional_occupant_gains")
            )
            * pl.col("heated_area_share")
            * pl.col("intermittency_factor")
        )
    )

    return buildings



def aggregate_occupation_factor(buildings, dwellings):
    mean_occupation_factor = dwellings.group_by("building_id").agg(
        pl.col("occupation_factor").mean()
    )
    buildings = buildings.join(mean_occupation_factor, on="building_id")
    return buildings




def run_models(buildings, boundaries, dwellings, parameters):
    """

    Args:
        buildings (GeoDataframe): a GeoDataframe containing the building geometries and parameters
        boundaries (GeoDataframe): a GeoDataframe containing the boundary geometries and parameters

    Returns:

    """
    buildings = aggregate_occupation_factor(buildings, dwellings)
    buildings = calculate_thermal_needs(buildings, boundaries, dwellings, parameters)

    return buildings