108 lines
4.3 KiB
Python
108 lines
4.3 KiB
Python
"""
|
|
Heat pump simulation repository with database CRUD operations
|
|
SPDX - License - Identifier: LGPL - 3.0 - or -later
|
|
Copyright © 2022 Concordia CERC group
|
|
Project Coder Peter Yefi peteryefi@gmail.com
|
|
"""
|
|
|
|
from persistence import BaseRepo, CityRepo
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy import select
|
|
from persistence.models import HeatPumpSimulation
|
|
from typing import Union, Dict
|
|
|
|
|
|
class HeatPumpSimulationRepo(BaseRepo):
|
|
_instance = None
|
|
|
|
def __init__(self, db_name, dotenv_path, app_env):
|
|
super().__init__(db_name, dotenv_path, app_env)
|
|
self._city_repo = CityRepo(db_name, dotenv_path, app_env)
|
|
|
|
def __new__(cls, db_name, dotenv_path, app_env):
|
|
"""
|
|
Implemented for a singleton pattern
|
|
"""
|
|
if cls._instance is None:
|
|
cls._instance = super(HeatPumpSimulationRepo, cls).__new__(cls)
|
|
return cls._instance
|
|
|
|
def insert(self, hp_sim_data: Dict, city_id: int) -> Union[HeatPumpSimulation, Dict]:
|
|
"""
|
|
Inserts the results of heat pump simulation
|
|
:param hp_sim_data: dictionary with heatpump the simulation inputs and output
|
|
:param city_id: the city that was used in running the simulation
|
|
:return: HeatPumpSimulation
|
|
"""
|
|
|
|
city = self._city_repo.get_by_id(city_id)
|
|
if city is None:
|
|
return {'message': 'city not found in database'}
|
|
|
|
try:
|
|
hp_simulation = HeatPumpSimulation(city_id, hp_sim_data["HourlyElectricityDemand"],
|
|
hp_sim_data["DailyElectricityDemand"], hp_sim_data["MonthlyElectricityDemand"],
|
|
hp_sim_data["DailyFossilFuelConsumption"],
|
|
hp_sim_data["MonthlyFossilFuelConsumption"])
|
|
hp_simulation.city_id = city_id
|
|
hp_simulation.end_year = hp_sim_data["EndYear"]
|
|
hp_simulation.start_year = hp_sim_data["StartYear"]
|
|
hp_simulation.max_demand_storage_hour = hp_sim_data["HoursOfStorageAtMaxDemand"]
|
|
hp_simulation.max_hp_energy_input = hp_sim_data["MaximumHPEnergyInput"]
|
|
hp_simulation.building_supply_temp = hp_sim_data["BuildingSuppTemp"]
|
|
hp_simulation.temp_difference = hp_sim_data["TemperatureDifference"]
|
|
hp_simulation.fuel_lhv = hp_sim_data["FuelLHV"]
|
|
hp_simulation.fuel_price = hp_sim_data["FuelPrice"]
|
|
hp_simulation.fuel_efficiency = hp_sim_data["FuelEF"]
|
|
hp_simulation.fuel_density = hp_sim_data["FuelDensity"]
|
|
hp_simulation.hp_supply_temp = hp_sim_data["HPSupTemp"]
|
|
hp_simulation.simulation_type = hp_sim_data["SimulationType"]
|
|
hp_simulation.heat_pump_model = hp_sim_data["HeatPumpModel"]
|
|
hp_simulation.heat_pump_type = hp_sim_data["HeatPumpType"]
|
|
|
|
# Persist heat pump simulation data
|
|
self.session.add(hp_simulation)
|
|
self.session.flush()
|
|
self.session.commit()
|
|
return hp_simulation
|
|
except SQLAlchemyError as err:
|
|
print(f'Error while saving heat pump simulation data: {err}')
|
|
except KeyError as err:
|
|
print(f'A required field is missing in your heat pump simulation dictionary: {err}')
|
|
|
|
def get_by_id(self, hp_simulation_id: int) -> HeatPumpSimulation:
|
|
"""
|
|
Fetches heat pump simulation data
|
|
:param hp_simulation_id: the city id
|
|
:return: a HeatPumpSimulation
|
|
"""
|
|
try:
|
|
return self.session.execute(select(HeatPumpSimulation).where(HeatPumpSimulation.id == hp_simulation_id)).first()[
|
|
0]
|
|
except SQLAlchemyError as err:
|
|
print(f'Error while fetching city: {err}')
|
|
|
|
def get_by_city(self, city_id: int) -> [HeatPumpSimulation]:
|
|
"""
|
|
Fetch heat pump simulation results by city
|
|
:param city_id: the name of the building
|
|
:return: [HeatPumpSimulation] with the provided name
|
|
"""
|
|
try:
|
|
result_set = self.session.execute(select(HeatPumpSimulation).where(HeatPumpSimulation.city_id == city_id))
|
|
return [sim_data[0] for sim_data in result_set]
|
|
except SQLAlchemyError as err:
|
|
print(f'Error while fetching city by name: {err}')
|
|
|
|
def delete_hp_simulation(self, hp_simulation_id: int):
|
|
"""
|
|
Deletes a heat pump simulation results
|
|
:param hp_simulation_id: the heat pump simulation results id
|
|
:return:
|
|
"""
|
|
try:
|
|
self.session.query(HeatPumpSimulation).filter(HeatPumpSimulation.id == hp_simulation_id).delete()
|
|
self.session.commit()
|
|
except SQLAlchemyError as err:
|
|
print(f'Error while fetching city: {err}')
|