import pandas as pd
from buildingmodel.logger import duration_logging
from buildingmodel.utils import get_beta_distribution
from buildingmodel import data_path
from pathlib import Path
import numpy as np
import polars as pl
cp_water = 4186.0 # [J/kg.K]
rho_water = 1.0 # [kg/l]
daily_dhw_use_by_occupant = {
"min": 10.0, # in litres at 40°C
"max": 150.0,
"alpha": 2.5, # parameters of beta distribution
"beta": 4.5,
}
conventional_dhw_use_by_occupant = 57.0 # in litres
specific_annual_need_by_occupant = {
"house": {
"min": 900.0, # in kWh
"max": 3000.0,
"alpha": 2.5, # parameters of beta distribution
"beta": 4.5,
},
"apartment": {
"min": 700.0, # in kWh
"max": 2600.0,
"alpha": 2, # parameters of beta distribution
"beta": 4.5,
},
}
internal_gain_by_occupant = 50.0 # in W
specific_needs_peak_time_constant = 2000.0 # in hours
cooking_needs_peak_time_constant = 1000.0 # in hours
energy_use_path = Path(data_path["energy_use"])
specific_electricity = pd.read_hdf(energy_use_path / "specific_electricity_sample.hdf")
cooking_energy = pd.read_hdf(energy_use_path / "cooking_energy_sample.hdf")
[docs]
def daily_dhw_need(dhw_use, temperature):
"""
Args:
dhw_use:
temperature:
Returns:
"""
return (40 - temperature) * cp_water / 3600.0 * rho_water * dhw_use
[docs]
def water_temperature(climate):
"""
Args:
climate:
Returns:
"""
return climate.ground_temperature.min(), climate.ground_temperature.mean()
[docs]
@duration_logging
def conventional_needs(dwellings, mean_water_temperature):
"""
Conventional dhw need and occupant gains are calculated from the living area of the dwelling
Args:
dwellings:
Returns:
"""
dwellings = dwellings.with_columns(
conventional_occupant_gains=0.07
* pl.col("living_area")
* 365 # 70 Wh/m² by day
)
dwellings = dwellings.with_columns(
conventional_occupant_count=pl.when(pl.col("living_area") <= 10.0)
.then(pl.lit(1.0))
.when(pl.col("living_area").is_between(10.0, 50.0))
.then(0.81 + 0.01875 * pl.col("living_area"))
.when(pl.col("living_area") > 50.0)
.then(1.125 + 0.0105 * pl.col("living_area"))
)
dwellings = dwellings.with_columns(
conventional_dhw_use=pl.col("conventional_occupant_count")
* conventional_dhw_use_by_occupant
)
dwellings = dwellings.with_columns(
conventional_dhw_needs=daily_dhw_need(
pl.col("conventional_dhw_use"), mean_water_temperature
)
* 365.0
/ 1000.0
)
return dwellings
[docs]
@duration_logging
def actual_dhw_needs(
dwellings, min_water_temperature, mean_water_temperature, parameters
):
"""
The dhw needs are calculated by drawing the daily hot water consumption in a beta distribution and then using
ground temperature to calculate the equivalent energy needs
Args:
dwellings:
Returns:
"""
use = (
get_beta_distribution(daily_dhw_use_by_occupant, dwellings.shape[0])
* parameters.energy_use_factor
)
dwellings = dwellings.with_columns(
annual_dhw_use=pl.col("occupant_count") * use,
peak_dhw_use=occupant_count_factor(pl.col("occupant_count")) * use,
)
dwellings = dwellings.with_columns(
annual_dhw_use=pl.col("annual_dhw_use") * pl.col("occupation_factor"),
)
dwellings = dwellings.with_columns(
annual_dhw_needs=daily_dhw_need(
pl.col("annual_dhw_use"), mean_water_temperature
)
* 365.0
/ 1000.0,
peak_dhw_needs=daily_dhw_need(pl.col("peak_dhw_use"), min_water_temperature),
)
return dwellings
[docs]
@duration_logging
def actual_specific_needs(dwellings, parameters):
"""
Specific needs are obtained by drawing in a sample of specific energy consumption obtained from the dwelling survey
and conditioned by the number of occupants
Args:
dwellings:
Returns:
"""
annual_specific_needs = np.zeros(dwellings.shape[0])
for occ_count in dwellings["occupant_count"].unique():
if occ_count > 6:
occ_count = 6
if occ_count > 0:
dw_mask = dwellings["occupant_count"] == occ_count
smp_mask = specific_electricity["occupant_count"] == occ_count
sample = np.random.choice(
specific_electricity.loc[smp_mask, "specific_electricity"],
int(dw_mask.sum()),
)
annual_specific_needs[dw_mask] = sample
dwellings = dwellings.with_columns(
annual_specific_needs=annual_specific_needs * parameters.energy_use_factor,
)
dwellings = dwellings.with_columns(
peak_specific_needs=pl.col("annual_specific_needs")
/ specific_needs_peak_time_constant,
)
dwellings = dwellings.with_columns(
annual_specific_needs=pl.col("annual_specific_needs")
* pl.col("occupation_factor"),
)
return dwellings
[docs]
@duration_logging
def actual_cooking_needs(dwellings, parameters):
"""
Specific needs are obtained by drawing in a sample of specific energy consumption obtained from the dwelling survey
and conditioned by the number of occupants
Args:
dwellings:
Returns:
"""
annual_cooking_needs = np.zeros(dwellings.shape[0])
# np.random.seed()
for occ_count in dwellings["occupant_count"].unique():
if occ_count > 6:
occ_count = 6
if occ_count > 0:
dw_mask = dwellings["occupant_count"] == occ_count
smp_mask = cooking_energy["occupant_count"] == occ_count
weights = cooking_energy.loc[smp_mask, "cooking_energy"]
p = weights / weights.sum()
sample = np.random.choice(
cooking_energy.loc[smp_mask, "cooking_energy"],
size=int(dw_mask.sum()),
replace=True,
p=p,
)
annual_cooking_needs[dw_mask] = sample
dwellings = dwellings.with_columns(
annual_cooking_needs=annual_cooking_needs * parameters.energy_use_factor,
)
dwellings = dwellings.with_columns(
peak_cooking_needs=pl.col("annual_cooking_needs")
/ cooking_needs_peak_time_constant,
)
dwellings = dwellings.with_columns(
annual_cooking_needs=pl.col("annual_cooking_needs")
* pl.col("occupation_factor"),
)
return dwellings
@duration_logging
def occupant_gains(dwellings, parameters):
dwellings = dwellings.with_columns(
occupant_gains=pl.col("occupant_count")
* internal_gain_by_occupant
* 8760.0
/ 1000.0
+ 0.5 * (pl.col("annual_specific_needs") + pl.col("annual_cooking_needs"))
)
dwellings = dwellings.with_columns(
occupant_gains=pl.col("occupant_gains") * pl.col("occupation_factor")
)
return dwellings
[docs]
def occupant_count_factor(occupant_count):
"""
Specific and dhw use by occupant decreases when the number of occupants in a dwelling increases. This is represented
by the use of a factor that decreases iwth the number of occupants.
Args:
occupant_count:
Returns:
"""
return occupant_count ** (1 / 1.2)
def occupation_factor(dwellings, parameters):
return dwellings.with_columns(
occupation_factor=pl.col("occupancy_type")
.cast(pl.String)
.replace(parameters.occupation_factors)
.cast(pl.Float64)
)
[docs]
def run_models(dwellings, climate, parameters):
"""
Args:
dwellings: a Dataframe containing dwellings parameters and results
Returns:
"""
min_water_temperature, mean_water_temperature = water_temperature(climate)
dwellings = occupation_factor(dwellings, parameters)
dwellings = conventional_needs(dwellings, mean_water_temperature)
dwellings = actual_dhw_needs(
dwellings, min_water_temperature, mean_water_temperature, parameters
)
dwellings = actual_specific_needs(dwellings, parameters)
dwellings = actual_cooking_needs(dwellings, parameters)
dwellings = occupant_gains(dwellings, parameters)
return dwellings