CERC/cerc_persistence
19
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 7 Wiki Activity

Compare commits

base: CERC:main
Branches Tags
CERC:main
CERC:1.0.0.2
CERC:1.0.0.1
CERC:1.0.0.0
CERC:0.1.0.4
CERC:v0.1.0.3
CERC:v0.1.0.2
CERC:test
CERC:test_jenkins
CERC:jenkins_test
CERC:v0.1.0.1
..
compare: CERC:v0.1.0.1
Branches Tags
CERC:main
CERC:1.0.0.2
CERC:1.0.0.1
CERC:1.0.0.0
CERC:0.1.0.4
CERC:v0.1.0.3
CERC:v0.1.0.2
CERC:test
CERC:test_jenkins
CERC:jenkins_test
CERC:v0.1.0.1

No commits in common. "main" and "v0.1.0.1" have entirely different histories.
main ... v0.1.0.1

27 changed files with 559 additions and 583 deletions
Show Stats Expand all files Collapse all files

1
.gitignore vendored
View File

@ -10,4 +10,3 @@
**/__pycache__/
**/.idea/
./dist
./cerc_persistence.egg-info

0
unittests/tests_outputs/.gitignore → cerc_persistence.egg-info/.gitignore vendored
View File

3
cerc_persistence/__init__.py
View File

@ -1,3 +0,0 @@
"""
Repositories Package
"""

5
cerc_persistence/configuration.py
View File

@ -20,6 +20,11 @@ class Configuration:
"""
def __init__(self, db_name: str, dotenv_path: str, app_env='TEST'):
"""
:param db_name: database name
:param app_env: application environment, test or production
:param dotenv_path: the absolute path to dotenv file
"""
try:
# load environmental variables
if not Path(dotenv_path).exists():

40
cerc_persistence/db_control.py
View File

@ -4,6 +4,7 @@ SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project CoderPeter Yefi peteryefi@gmail.com
"""
import json
from typing import Dict
from cerc_persistence.repositories.application import Application
@ -73,8 +74,8 @@ class DBControl:
:
"""
cities = self._city.get_by_user_id_application_id_and_scenario(user_id, application_id, scenario)
city = [city[0].id for city in cities]
result = self._city_object.building_in_cities_info(name, city)
for city in cities:
result = self.building_info(name, city[0].id)
if result is not None:
return result
return None
@ -88,14 +89,17 @@ class DBControl:
"""
return self._city_object.get_by_name_or_alias_and_city(name, city_id)
def building_info_in_cities(self, name, cities) -> CityObject:
def buildings_info(self, request_values, city_id) -> [CityObject]:
"""
Retrieve the building info from the database
:param name: Building name
:param cities: [City ID]
:return: CityObject
Retrieve the buildings info from the database
:param request_values: Building names
:param city_id: City ID
:return: [CityObject]
"""
return self._city_object.get_by_name_or_alias_in_cities(name, cities)
buildings = []
for name in request_values['names']:
buildings.append(self.building_info(name, city_id))
return buildings
def results(self, user_id, application_id, request_values, result_names=None) -> Dict:
"""
@ -117,19 +121,22 @@ class DBControl:
)
if result_sets is None:
continue
for result_set in result_sets:
city_id = result_set[0].id
results[scenario_name] = []
city_ids = [r[0].id for r in result_sets]
for building_name in scenario[scenario_name]:
_building = self._city_object.get_by_name_or_alias_in_cities(building_name, city_ids)
_building = self._city_object.get_by_name_or_alias_and_city(building_name, city_id)
if _building is None:
continue
city_object_id = _building.id
_ = self._simulation_results.get_simulation_results_by_city_object_id_and_names(
_ = self._simulation_results.get_simulation_results_by_city_id_city_object_id_and_names(
city_id,
city_object_id,
result_names)
for value in _:
values = value.values
values = json.loads(value.values)
values["building"] = building_name
results[scenario_name].append(values)
return results
@ -236,13 +243,6 @@ class DBControl:
def delete_application(self, application_uuid):
"""
Deletes a single application from the database
:param application_uuid: the id of the application to delete
:param application_uuid: the id of the application to get
"""
self._application.delete(application_uuid)
def get_city(self, city_id):
"""
Get a single city by id
:param city_id: the id of the city to get
"""
return self._city.get_by_id(city_id)

16
cerc_persistence/db_setup.py
View File

@ -23,6 +23,14 @@ class DBSetup:
"""
def __init__(self, db_name, app_env, dotenv_path, admin_password, application_uuid):
"""
Creates database tables a default admin user and a default admin app with the given password and uuid
:param db_name: database name
:param app_env: application environment type [TEST|PROD]
:param dotenv_path: .env file path
:param admin_password: administrator password for the application uuid
:application_uuid: application uuid
"""
repository = Repository(db_name=db_name, app_env=app_env, dotenv_path=dotenv_path)
# Create the tables using the models
@ -42,14 +50,14 @@ class DBSetup:
name = 'AdminTool'
description = 'Admin tool to control city persistence and to test the API v1.4'
logging.info('Creating default admin tool application...')
application = application_repo.insert(name, description, application_uuid)
application_id = application_repo.insert(name, description, application_uuid)
if isinstance(application, dict):
logging.info(application)
if isinstance(application_id, dict):
logging.info(application_id)
else:
msg = f'Created Admin tool with application_uuid: {application_uuid}'
logging.info(msg)
return application.id
return application_id
@staticmethod
def _create_admin_user(user_repo, admin_password, application_id):

2
cerc_persistence/models/application.py
View File

@ -16,7 +16,7 @@ from cerc_persistence.configuration import Models
class Application(Models):
"""
Application(Models) class
A model representation of an application
"""
__tablename__ = 'application'
id = Column(Integer, Sequence('application_id_seq'), primary_key=True)

3
cerc_persistence/models/city.py
View File

@ -14,8 +14,7 @@ from cerc_persistence.configuration import Models
class City(Models):
"""
City(Models) class
"""A model representation of a city
"""
__tablename__ = 'city'
id = Column(Integer, Sequence('city_id_seq'), primary_key=True)

5
cerc_persistence/models/city_object.py
View File

@ -17,7 +17,7 @@ from cerc_persistence.configuration import Models
class CityObject(Models):
"""
CityObject(Models) class
A model representation of an application
"""
__tablename__ = 'city_object'
id = Column(Integer, Sequence('city_object_id_seq'), primary_key=True)
@ -51,8 +51,7 @@ class CityObject(Models):
self.area = building.floor_area
self.roof_area = sum(roof.solid_polygon.area for roof in building.roofs)
self.total_pv_area = sum(
roof.solid_polygon.area * roof.solar_collectors_area_reduction_factor for roof in building.roofs
)
roof.solid_polygon.area * roof.solar_collectors_area_reduction_factor for roof in building.roofs)
storeys = building.storeys_above_ground
wall_area = 0
window_ratio = 0

2
cerc_persistence/models/simulation_results.py
View File

@ -15,7 +15,7 @@ from cerc_persistence.configuration import Models
class SimulationResults(Models):
"""
SimulationResults(Models) class
A model representation of an application
"""
__tablename__ = 'simulation_results'
id = Column(Integer, Sequence('simulation_results_id_seq'), primary_key=True)

2
cerc_persistence/models/user.py
View File

@ -24,7 +24,7 @@ class UserRoles(enum.Enum):
class User(Models):
"""
User(Models) class
A model representation of a city
"""
__tablename__ = 'user'
id = Column(Integer, Sequence('user_id_seq'), primary_key=True)

7
cerc_persistence/repositories/__init__.py
View File

@ -1,3 +1,10 @@
"""
Repositories Package
"""
import datetime
import logging
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from cerc_persistence.repository import Repository

15
cerc_persistence/repositories/application.py
View File

@ -4,21 +4,13 @@ SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Guille Gutierrez Guillermo.GutierrezMorote@concordia.ca
"""
import datetime
import logging
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm.session import Session
from cerc_persistence.repository import Repository
from cerc_persistence.models import Application as Model
from . import datetime, logging, select, SQLAlchemyError, Session, Repository
class Application(Repository):
"""
Application(Repository) class
Application repository
"""
_instance = None
@ -26,6 +18,9 @@ class Application(Repository):
super().__init__(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(Application, cls).__new__(cls)
return cls._instance

22
cerc_persistence/repositories/city.py
View File

@ -4,23 +4,16 @@ SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Peter Yefi peteryefi@gmail.com
"""
import datetime
import logging
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from hub.version import __version__
from hub.city_model_structure.city import City as CityHub
from cerc_persistence.repository import Repository
from hub.version import __version__
from cerc_persistence.models import City as Model
from cerc_persistence.models import CityObject
from . import datetime, logging, select, SQLAlchemyError, Session, Repository
class City(Repository):
"""
City(Repository) class
City repository
"""
_instance = None
@ -28,6 +21,9 @@ class City(Repository):
super().__init__(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(City, cls).__new__(cls)
return cls._instance
@ -37,7 +33,7 @@ class City(Repository):
Inserts a city
:param city: The complete city instance
:param pickle_path: Path to the pickle
param scenario: Simulation scenario name
:param scenario: Simulation scenario name
:param application_id: Application id owning the instance
:param user_id: User id owning the instance
:return: Identity id
@ -133,7 +129,3 @@ class City(Repository):
except SQLAlchemyError as err:
logging.error('Error while fetching city by name %s', err)
raise SQLAlchemyError from err
def get_by_id(self, city_id) -> Model:
with Session(self.engine) as session:
return session.execute(select(Model).where(Model.id == city_id)).first()[0]

87
cerc_persistence/repositories/city_object.py
View File

@ -4,22 +4,14 @@ SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Guille Gutierrez Guillermo.GutierrezMorote@concordia.ca
"""
import datetime
import logging
from typing import Union
from hub.city_model_structure.building import Building
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from cerc_persistence.models import CityObject as Model
from cerc_persistence.repository import Repository
from . import datetime, logging, select, SQLAlchemyError, Session, Repository
class CityObject(Repository):
"""
CityObject(Repository) class
City object repository
"""
_instance = None
@ -27,6 +19,9 @@ class CityObject(Repository):
super().__init__(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(CityObject, cls).__new__(cls)
return cls._instance
@ -70,7 +65,7 @@ class CityObject(Repository):
with Session(self.engine) as session:
session.query(Model).filter(Model.name == building.name, Model.city_id == city_id).update(
{'name': building.name,
'aliases': building.aliases,
'alias': building.alias,
'object_type': building.type,
'year_of_construction': building.year_of_construction,
'function': building.function,
@ -95,44 +90,10 @@ class CityObject(Repository):
session.query(Model).filter(Model.city_id == city_id, Model.name == name).delete()
session.commit()
except SQLAlchemyError as err:
logging.error('Error while deleting city object %s', err)
logging.error('Error while deleting city_object %s', err)
raise SQLAlchemyError from err
def building_in_cities_info(self, name, cities):
"""
Fetch a city object based on name and city id
:param name: city object name
:param cities: city identifiers
:return: [CityObject] with the provided name or alias belonging to the city with id city_id
"""
try:
# search by name first
with Session(self.engine) as session:
city_object = session.execute(select(Model).where(
Model.name == name, Model.city_id.in_(cities))
).first()
if city_object is not None:
return city_object[0]
# name not found, so search by alias instead
city_objects = session.execute(
select(Model).where(Model.aliases.contains(name), Model.city_id.in_(cities))
).all()
for city_object in city_objects:
aliases = city_object[0].aliases.replace('{', '').replace('}', '').split(',')
for alias in aliases:
if alias == name:
# force the name as the alias
city_object[0].name = name
return city_object[0]
return None
except SQLAlchemyError as err:
logging.error('Error while fetching city object by name and city: %s', err)
raise SQLAlchemyError from err
except IndexError as err:
logging.error('Error while fetching city object by name and city, empty result %s', err)
raise IndexError from err
def get_by_name_or_alias_and_city(self, name, city_id) -> Union[Model, None]:
def get_by_name_or_alias_and_city(self, name, city_id) -> Model:
"""
Fetch a city object based on name and city id
:param name: city object name
@ -163,35 +124,3 @@ class CityObject(Repository):
except IndexError as err:
logging.error('Error while fetching city object by name and city, empty result %s', err)
raise IndexError from err
def get_by_name_or_alias_in_cities(self, name, city_ids) -> Model:
"""
Fetch a city object based on name and city ids
:param name: city object name
:param city_ids: a list of city identifiers
:return: [CityObject] with the provided name or alias belonging to the city with id city_id
"""
try:
# search by name first
with Session(self.engine) as session:
city_object = session.execute(select(Model).where(Model.name == name, Model.city_id.in_(tuple(city_ids)))).first()
if city_object is not None:
return city_object[0]
# name not found, so search by alias instead
city_objects = session.execute(
select(Model).where(Model.aliases.contains(name), Model.city_id.in_(tuple(city_ids)))
).all()
for city_object in city_objects:
aliases = city_object[0].aliases.replace('{', '').replace('}', '').split(',')
for alias in aliases:
if alias == name:
# force the name as the alias
city_object[0].name = name
return city_object[0]
return None
except SQLAlchemyError as err:
logging.error('Error while fetching city object by name and city: %s', err)
raise SQLAlchemyError from err
except IndexError as err:
logging.error('Error while fetching city object by name and city, empty result %s', err)
raise IndexError from err

43
cerc_persistence/repositories/simulation_results.py
View File

@ -4,23 +4,16 @@ SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Guille Gutierrez Guillermo.GutierrezMorote@concordia.ca
"""
import datetime
import logging
from sqlalchemy import or_
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from cerc_persistence.repository import Repository
from cerc_persistence.models import City
from cerc_persistence.models import CityObject
from cerc_persistence.models import SimulationResults as Model
from . import datetime, logging, select, SQLAlchemyError, Session, Repository
class SimulationResults(Repository):
"""
SimulationResults(Repository) class
Simulation results repository
"""
_instance = None
@ -28,6 +21,9 @@ class SimulationResults(Repository):
super().__init__(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(SimulationResults, cls).__new__(cls)
return cls._instance
@ -88,7 +84,7 @@ class SimulationResults(Repository):
else:
raise NotImplementedError('Missing either city_id or city_object_id')
except SQLAlchemyError as err:
logging.error('Error while updating simulation results for %s', err)
logging.error('Error while updating simulation results %s', err)
raise SQLAlchemyError from err
def delete(self, name: str, city_id=None, city_object_id=None):
@ -139,7 +135,8 @@ class SimulationResults(Repository):
logging.error('Error while fetching city by city_id: %s', err)
raise SQLAlchemyError from err
def get_simulation_results_by_city_id_city_object_id_and_names(self, city_id, city_object_id, result_names=None) -> [Model]:
def get_simulation_results_by_city_id_city_object_id_and_names(self, city_id, city_object_id,
result_names=None) -> [Model]:
"""
Fetch the simulation results based in the city_id or city_object_id with the given names or all
:param city_id: the city id
@ -164,27 +161,3 @@ class SimulationResults(Repository):
except SQLAlchemyError as err:
logging.error('Error while fetching city by city_id: %s', err)
raise SQLAlchemyError from err
def get_simulation_results_by_city_object_id_and_names(self, city_object_id, result_names=None) -> [Model]:
"""
Fetch the simulation results based in the city_object_id with the given names or all
:param city_object_id: the city object id
:param result_names: if given filter the results
:return: [SimulationResult]
"""
try:
with Session(self.engine) as session:
result_set = session.execute(select(Model).where(
Model.city_object_id == city_object_id
))
results = [r[0] for r in result_set]
if not result_names:
return results
filtered_results = []
for result in results:
if result.name in result_names:
filtered_results.append(result)
return filtered_results
except SQLAlchemyError as err:
logging.error('Error while fetching city by city_id: %s', err)
raise SQLAlchemyError from err

11
cerc_persistence/repositories/user.py
View File

@ -4,21 +4,17 @@ SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Peter Yefi peteryefi@gmail.com
"""
import datetime
import logging
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session
from hub.helpers.auth import Auth
from cerc_persistence.repository import Repository
from cerc_persistence.models import User as Model, Application as ApplicationModel, UserRoles
from . import datetime, logging, select, SQLAlchemyError, Session, Repository
class User(Repository):
"""
User(Repository) class
User class
"""
_instance = None
@ -26,6 +22,9 @@ class User(Repository):
super().__init__(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(User, cls).__new__(cls)
return cls._instance

2
cerc_persistence/version.py
View File

@ -1,4 +1,4 @@
"""
CERC Persistence version number
"""
__version__ = '1.0.0.2'
__version__ = '0.1.0.1'

2
requirements.txt
View File

@ -2,5 +2,3 @@ pathlib
python-dotenv
SQLAlchemy
cerc-hub
psycopg2-binary
sqlalchemy_utils

9
setup.py
View File

@ -20,13 +20,8 @@ with open(version) as f:
setup(
name='cerc-persistence',
version=main_ns['__version__'],
description="CERC Persistence consist of a set of classes to store and retrieve Cerc Hub cities and results",
long_description="CERC Persistence consist of a set of classes to store and retrieve Cerc Hub cities and results.\n\n"
"Developed at Concordia university in Canada as part of the research group from the Next Generation "
"Cities Institute, our aim among others is to provide a comprehensive set of tools to help "
"researchers and urban developers to make decisions to improve the livability and efficiency of our "
"cities, cerc persistence will store the simulation results and city information to make those "
"available for other researchers.",
description="",
long_description="",
classifiers=[
"License :: OSI Approved :: GNU Library or Lesser General Public License (LGPL)",
"Programming Language :: Python",

309
tests/test_db_factory.py Normal file
View File

@ -0,0 +1,309 @@
"""
Test db factory
SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Peter Yefi peteryefi@gmail.com
"""
import distutils.spawn
import glob
import json
import logging
import os
import subprocess
import unittest
from pathlib import Path
from unittest import TestCase
import sqlalchemy.exc
from sqlalchemy import create_engine
from sqlalchemy.exc import ProgrammingError
import hub.helpers.constants as cte
from hub.exports.energy_building_exports_factory import EnergyBuildingsExportsFactory
from hub.exports.exports_factory import ExportsFactory
from hub.helpers.data.montreal_function_to_hub_function import MontrealFunctionToHubFunction
from hub.imports.construction_factory import ConstructionFactory
from hub.imports.energy_systems_factory import EnergySystemsFactory
from hub.imports.geometry_factory import GeometryFactory
from hub.imports.results_factory import ResultFactory
from hub.imports.usage_factory import UsageFactory
from hub.imports.weather_factory import WeatherFactory
from cerc_persistence.db_control import DBControl
from cerc_persistence.models import City, Application, CityObject, SimulationResults
from cerc_persistence.models import User, UserRoles
from cerc_persistence.repository import Repository
class Control:
_skip_test = False
_skip_reason = 'PostgreSQL not properly installed in host machine'
def __init__(self):
"""
Test
setup
:return: None
"""
self._skip_test = False
# Create test database
dotenv_path = Path("{}/.local/etc/hub/.env".format(os.path.expanduser('~'))).resolve()
if not dotenv_path.exists():
self._skip_test = True
self._skip_reason = f'.env file missing at {dotenv_path}'
return
dotenv_path = str(dotenv_path)
repository = Repository(db_name='test_db', app_env='TEST', dotenv_path=dotenv_path)
engine = create_engine(repository.configuration.connection_string)
try:
# delete test database if it exists
connection = engine.connect()
connection.close()
except ProgrammingError:
logging.info('Database does not exist. Nothing to delete')
except sqlalchemy.exc.OperationalError as operational_error:
self._skip_test = True
self._skip_reason = f'{operational_error}'
return
Application.__table__.create(bind=repository.engine, checkfirst=True)
User.__table__.create(bind=repository.engine, checkfirst=True)
City.__table__.create(bind=repository.engine, checkfirst=True)
CityObject.__table__.create(bind=repository.engine, checkfirst=True)
SimulationResults.__table__.create(bind=repository.engine, checkfirst=True)
city_file = Path('tests_data/test.geojson').resolve()
output_path = Path('tests_outputs/').resolve()
self._city = GeometryFactory('geojson',
city_file,
height_field='citygml_me',
year_of_construction_field='ANNEE_CONS',
aliases_field=['ID_UEV', 'CIVIQUE_DE', 'NOM_RUE'],
function_field='CODE_UTILI',
function_to_hub=MontrealFunctionToHubFunction().dictionary).city
ConstructionFactory('nrcan', self._city).enrich()
UsageFactory('nrcan', self._city).enrich()
WeatherFactory('epw', self._city).enrich()
ExportsFactory('sra', self._city, output_path).export()
sra_file = str((output_path / f'{self._city.name}_sra.xml').resolve())
subprocess.run([self.sra, sra_file], stdout=subprocess.DEVNULL)
ResultFactory('sra', self._city, output_path).enrich()
for building in self._city.buildings:
building.energy_systems_archetype_name = 'system 1 gas pv'
EnergySystemsFactory('montreal_custom', self._city).enrich()
EnergyBuildingsExportsFactory('insel_monthly_energy_balance', self._city, output_path).export()
_insel_files = glob.glob(f'{output_path}/*.insel')
for insel_file in _insel_files:
subprocess.run([self.insel, str(insel_file)], stdout=subprocess.DEVNULL)
ResultFactory('insel_monthly_energy_balance', self._city, output_path).enrich()
self._database = DBControl(
db_name=repository.configuration.db_name,
app_env='TEST',
dotenv_path=dotenv_path)
self._application_uuid = 'b9e0ce80-1218-410c-8a64-9d9b7026aad8'
self._application_id = 1
self._user_id = 1
try:
self._application_id = self._database.persist_application(
'test',
'test',
self.application_uuid
)
except sqlalchemy.exc.SQLAlchemyError:
self._application_id = self._database.application_info(self.application_uuid).id
try:
self._user_id = self._database.create_user('test', self._application_id, 'test', UserRoles.Admin)
except sqlalchemy.exc.SQLAlchemyError:
self._user_id = self._database.user_info(name='test', password='test', application_id=self._application_id).id
self._pickle_path = Path('tests_data/pickle_path.bz2').resolve()
@property
def database(self):
return self._database
@property
def application_uuid(self):
return self._application_uuid
@property
def application_id(self):
return self._application_id
@property
def user_id(self):
return self._user_id
@property
def skip_test(self):
return self._skip_test
@property
def insel(self):
return distutils.spawn.find_executable('insel')
@property
def sra(self):
return distutils.spawn.find_executable('sra')
@property
def skip_insel_test(self):
return self.insel is None
@property
def skip_reason(self):
return self._skip_reason
@property
def message(self):
return self._skip_reason
@property
def city(self):
return self._city
@property
def pickle_path(self):
return self._pickle_path
control = Control()
class TestDBFactory(TestCase):
"""
TestDBFactory
"""
@unittest.skipIf(control.skip_test, control.skip_reason)
def test_save_city(self):
control.city.name = "Montreal"
city_id = control.database.persist_city(
control.city,
control.pickle_path,
control.city.name,
control.application_id,
control.user_id)
control.database.delete_city(city_id)
@unittest.skipIf(control.skip_test, control.skip_reason)
def test_get_update_city(self):
city_id = control.database.persist_city(control.city,
control.pickle_path,
control.city.name,
control.application_id,
control.user_id)
control.city.name = "Ottawa"
control.database.update_city(city_id, control.city)
cities = control.database.cities_by_user_and_application(
control.user_id,
control.application_id)
for updated_city in cities:
if updated_city.id == city_id:
self.assertEqual(updated_city.name, control.city.name)
break
control.database.delete_city(city_id)
@unittest.skipIf(control.skip_test, control.skip_reason)
@unittest.skipIf(control.skip_insel_test, 'insel is not installed')
def test_save_results(self):
city_id = control.database.persist_city(control.city,
control.pickle_path,
'current status',
control.application_id,
control.user_id)
city_objects_id = []
for building in control.city.buildings:
_building = control.database.building_info(building.name, city_id)
if cte.MONTH not in building.cooling_demand:
print(f'building {building.name} not calculated')
continue
monthly_cooling_peak_load = building.cooling_peak_load[cte.MONTH]
yearly_cooling_peak_load = building.cooling_peak_load[cte.YEAR]
monthly_heating_peak_load = building.heating_peak_load[cte.MONTH]
yearly_heating_peak_load = building.heating_peak_load[cte.YEAR]
monthly_lighting_peak_load = building.lighting_peak_load[cte.MONTH]
yearly_lighting_peak_load = building.lighting_peak_load[cte.YEAR]
monthly_appliances_peak_load = building.appliances_peak_load[cte.MONTH]
yearly_appliances_peak_load = building.appliances_peak_load[cte.YEAR]
monthly_cooling_demand = building.cooling_demand[cte.MONTH]
yearly_cooling_demand = building.cooling_demand[cte.YEAR]
monthly_heating_demand = building.heating_demand[cte.MONTH]
yearly_heating_demand = building.heating_demand[cte.YEAR]
monthly_lighting_electrical_demand = building.lighting_electrical_demand[cte.MONTH]
yearly_lighting_electrical_demand = building.lighting_electrical_demand[cte.YEAR]
monthly_appliances_electrical_demand = building.appliances_electrical_demand[cte.MONTH]
yearly_appliances_electrical_demand = building.appliances_electrical_demand[cte.YEAR]
monthly_domestic_hot_water_heat_demand = building.domestic_hot_water_heat_demand[cte.MONTH]
yearly_domestic_hot_water_heat_demand = building.domestic_hot_water_heat_demand[cte.YEAR]
monthly_heating_consumption = building.heating_consumption[cte.MONTH]
yearly_heating_consumption = building.heating_consumption[cte.YEAR]
monthly_cooling_consumption = building.cooling_consumption[cte.MONTH]
yearly_cooling_consumption = building.cooling_consumption[cte.YEAR]
monthly_domestic_hot_water_consumption = building.domestic_hot_water_consumption[cte.MONTH]
yearly_domestic_hot_water_consumption = building._domestic_hot_water_consumption[cte.YEAR]
monthly_distribution_systems_electrical_consumption = building.distribution_systems_electrical_consumption[
cte.MONTH]
yearly_distribution_systems_electrical_consumption = building.distribution_systems_electrical_consumption[
cte.YEAR]
monthly_on_site_electrical_production = [x * cte.WATTS_HOUR_TO_JULES
for x in building.onsite_electrical_production[cte.MONTH]]
yearly_on_site_electrical_production = [x * cte.WATTS_HOUR_TO_JULES
for x in building.onsite_electrical_production[cte.YEAR]]
results = json.dumps({cte.INSEL_MEB: [
{'monthly_cooling_peak_load': monthly_cooling_peak_load},
{'yearly_cooling_peak_load': yearly_cooling_peak_load},
{'monthly_heating_peak_load': monthly_heating_peak_load},
{'yearly_heating_peak_load': yearly_heating_peak_load},
{'monthly_lighting_peak_load': monthly_lighting_peak_load},
{'yearly_lighting_peak_load': yearly_lighting_peak_load},
{'monthly_appliances_peak_load': monthly_appliances_peak_load},
{'yearly_appliances_peak_load': yearly_appliances_peak_load},
{'monthly_cooling_demand': monthly_cooling_demand},
{'yearly_cooling_demand': yearly_cooling_demand},
{'monthly_heating_demand': monthly_heating_demand},
{'yearly_heating_demand': yearly_heating_demand},
{'monthly_lighting_electrical_demand': monthly_lighting_electrical_demand},
{'yearly_lighting_electrical_demand': yearly_lighting_electrical_demand},
{'monthly_appliances_electrical_demand': monthly_appliances_electrical_demand},
{'yearly_appliances_electrical_demand': yearly_appliances_electrical_demand},
{'monthly_domestic_hot_water_heat_demand': monthly_domestic_hot_water_heat_demand},
{'yearly_domestic_hot_water_heat_demand': yearly_domestic_hot_water_heat_demand},
{'monthly_heating_consumption': monthly_heating_consumption},
{'yearly_heating_consumption': yearly_heating_consumption},
{'monthly_cooling_consumption': monthly_cooling_consumption},
{'yearly_cooling_consumption': yearly_cooling_consumption},
{'monthly_domestic_hot_water_consumption': monthly_domestic_hot_water_consumption},
{'yearly_domestic_hot_water_consumption': yearly_domestic_hot_water_consumption},
{
'monthly_distribution_systems_electrical_consumption': monthly_distribution_systems_electrical_consumption},
{
'yearly_distribution_systems_electrical_consumption': yearly_distribution_systems_electrical_consumption},
{'monthly_on_site_electrical_production': monthly_on_site_electrical_production},
{'yearly_on_site_electrical_production': yearly_on_site_electrical_production}
]})
db_building_id = _building.id
city_objects_id.append(db_building_id)
control.database.add_simulation_results(
cte.INSEL_MEB,
results, city_object_id=db_building_id)
self.assertEqual(17, len(city_objects_id), 'wrong number of results')
self.assertIsNotNone(city_objects_id[0], 'city_object_id is None')
for _id in city_objects_id:
control.database.delete_results_by_name('insel meb', city_object_id=_id)
control.database.delete_city(city_id)
@classmethod
@unittest.skipIf(control.skip_test, control.skip_reason)
def tearDownClass(cls):
control.database.delete_application(control.application_uuid)
control.database.delete_user(control.user_id)
os.unlink(control.pickle_path)
output_files = glob.glob('./tests_outputs/*')
for output_file in output_files:
os.unlink(output_file)

135
tests/test_db_retrieve.py Normal file
View File

@ -0,0 +1,135 @@
"""
Test db retrieve
SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Ruben Sanchez ruben.sanchez@mail.concordia.ca
"""
import distutils.spawn
import logging
import os
import unittest
from pathlib import Path
from unittest import TestCase
import sqlalchemy.exc
from sqlalchemy import create_engine
from sqlalchemy.exc import ProgrammingError
from cerc_persistence.db_control import DBControl
from cerc_persistence.repository import Repository
class Control:
_skip_test = False
_skip_reason = 'PostgreSQL not properly installed in host machine'
def __init__(self):
"""
Test
setup
:return: None
"""
self._skip_test = False
# Create test database
dotenv_path = Path("{}/.local/etc/hub/.env".format(os.path.expanduser('~'))).resolve()
if not dotenv_path.exists():
self._skip_test = True
self._skip_reason = f'.env file missing at {dotenv_path}'
return
dotenv_path = str(dotenv_path)
repository = Repository(db_name='montreal_retrofit_test', app_env='TEST', dotenv_path=dotenv_path)
engine = create_engine(repository.configuration.connection_string)
try:
# delete test database if it exists
connection = engine.connect()
connection.close()
except ProgrammingError:
logging.info('Database does not exist. Nothing to delete')
except sqlalchemy.exc.OperationalError as operational_error:
self._skip_test = True
self._skip_reason = f'{operational_error}'
return
self._database = DBControl(
db_name=repository.configuration.db_name,
app_env='TEST',
dotenv_path=dotenv_path)
self._application_uuid = '60b7fc1b-f389-4254-9ffd-22a4cf32c7a3'
self._application_id = 1
self._user_id = 1
self._pickle_path = 'tests_data/pickle_path.bz2'
@property
def database(self):
return self._database
@property
def application_uuid(self):
return self._application_uuid
@property
def application_id(self):
return self._application_id
@property
def user_id(self):
return self._user_id
@property
def skip_test(self):
return self._skip_test
@property
def insel(self):
return distutils.spawn.find_executable('insel')
@property
def sra(self):
return distutils.spawn.find_executable('sra')
@property
def skip_insel_test(self):
return self.insel is None
@property
def skip_reason(self):
return self._skip_reason
@property
def message(self):
return self._skip_reason
@property
def pickle_path(self):
return self._pickle_path
control = Control()
class TestDBFactory(TestCase):
"""
TestDBFactory
"""
@unittest.skipIf(control.skip_test, control.skip_reason)
def test_retrieve_results(self):
request_values = {
"scenarios": [
{
"current status": ["01002777", "01002773", "01036804"]
},
{
"skin retrofit": ["01002777", "01002773", "01036804"]
},
{
"system retrofit and pv": ["01002777", "01002773", "01036804"]
},
{
"skin and system retrofit with pv": ["01002777", "01002773", "01036804"]
}
]
}
results = control.database.results(control.user_id, control.application_id, request_values)
print(results)

0
unittests/tests_data/test.geojson → tests/tests_data/test.geojson
View File

3
tests/tests_outputs/.gitignore vendored Normal file
View File

@ -0,0 +1,3 @@
# Except this file
*
!.gitignore

171
unittests/control.py
View File

@ -1,171 +0,0 @@
"""
Test control
SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Guille Gutierrez <guillermo.gutierrezmorote@concordia.ca
"""
import distutils.spawn
import glob
import os
import subprocess
from pathlib import Path
import sqlalchemy
import sqlalchemy.exc
from hub.exports.energy_building_exports_factory import EnergyBuildingsExportsFactory
from hub.exports.exports_factory import ExportsFactory
from hub.helpers.data.montreal_function_to_hub_function import MontrealFunctionToHubFunction
from hub.imports.construction_factory import ConstructionFactory
from hub.imports.energy_systems_factory import EnergySystemsFactory
from hub.imports.geometry_factory import GeometryFactory
from hub.imports.results_factory import ResultFactory
from hub.imports.usage_factory import UsageFactory
from hub.imports.weather_factory import WeatherFactory
from sqlalchemy import create_engine
from sqlalchemy_utils import database_exists, create_database, drop_database
from cerc_persistence.db_control import DBControl
from cerc_persistence.models import City, Application, CityObject, SimulationResults, UserRoles
from cerc_persistence.models import User
from cerc_persistence.repository import Repository
class Control:
_skip_test = False
_skip_reason = 'PostgreSQL not properly installed in host machine'
def __init__(self):
"""
Test
setup
:return: None
"""
self._skip_test = False
# Create test database
dotenv_path = Path("{}/.local/etc/hub/.env".format(os.path.expanduser('~'))).resolve()
if not dotenv_path.exists():
self._skip_test = True
self._skip_reason = f'.env file missing at {dotenv_path}'
return
dotenv_path = str(dotenv_path)
repository = Repository(db_name='persistence_test_db', app_env='TEST', dotenv_path=dotenv_path)
engine = create_engine(repository.configuration.connection_string)
if database_exists(engine.url):
drop_database(engine.url)
create_database(engine.url)
Application.__table__.create(bind=engine, checkfirst=True)
User.__table__.create(bind=engine, checkfirst=True)
City.__table__.create(bind=engine, checkfirst=True)
CityObject.__table__.create(bind=engine, checkfirst=True)
SimulationResults.__table__.create(bind=engine, checkfirst=True)
self._database = DBControl(
db_name=repository.configuration.db_name,
app_env='TEST',
dotenv_path=dotenv_path)
example_path = (Path(__file__).parent / 'tests_data').resolve()
city_file = (example_path / 'test.geojson').resolve()
output_path = (Path(__file__).parent / 'tests_outputs').resolve()
self._application_uuid = '60b7fc1b-f389-4254-9ffd-22a4cf32c7a3'
self._application_id = 1
self._user_id = 1
self._pickle_path = 'tests_data/pickle_path.bz2'
self._city = GeometryFactory('geojson',
city_file,
height_field='citygml_me',
year_of_construction_field='ANNEE_CONS',
aliases_field=['ID_UEV', 'CIVIQUE_DE', 'NOM_RUE'],
function_field='CODE_UTILI',
function_to_hub=MontrealFunctionToHubFunction().dictionary).city
ConstructionFactory('nrcan', self._city).enrich()
UsageFactory('nrcan', self._city).enrich()
WeatherFactory('epw', self._city).enrich()
ExportsFactory('sra', self._city, output_path).export()
sra_file = str((output_path / f'{self._city.name}_sra.xml').resolve())
subprocess.run([self.sra, sra_file], stdout=subprocess.DEVNULL)
ResultFactory('sra', self._city, output_path).enrich()
for building in self._city.buildings:
building.energy_systems_archetype_name = 'system 1 gas pv'
EnergySystemsFactory('montreal_custom', self._city).enrich()
EnergyBuildingsExportsFactory('insel_monthly_energy_balance', self._city, output_path).export()
_insel_files = glob.glob(f'{output_path}/*.insel')
for insel_file in _insel_files:
subprocess.run([self.insel, str(insel_file)], stdout=subprocess.DEVNULL)
ResultFactory('insel_monthly_energy_balance', self._city, output_path).enrich()
self._database = DBControl(
db_name=repository.configuration.db_name,
app_env='TEST',
dotenv_path=dotenv_path)
self._application_uuid = 'b9e0ce80-1218-410c-8a64-9d9b7026aad8'
self._application_id = 1
self._user_id = 1
try:
self._application_id = self._database.persist_application(
'test',
'test',
self.application_uuid
)
except sqlalchemy.exc.SQLAlchemyError:
self._application_id = self._database.application_info(self.application_uuid).id
try:
self._user_id = self._database.create_user('test', self._application_id, 'test', UserRoles.Admin)
except sqlalchemy.exc.SQLAlchemyError:
self._user_id = self._database.user_info(name='test', password='test', application_id=self._application_id).id
self._pickle_path = (example_path / 'pickle_path.bz2').resolve()
self._city
@property
def database(self):
return self._database
@property
def application_uuid(self):
return self._application_uuid
@property
def application_id(self):
return self._application_id
@property
def user_id(self):
return self._user_id
@property
def skip_test(self):
return self._skip_test
@property
def insel(self):
return distutils.spawn.find_executable('insel')
@property
def sra(self):
return distutils.spawn.find_executable('sra')
@property
def skip_insel_test(self):
return self.insel is None
@property
def skip_reason(self):
return self._skip_reason
@property
def message(self):
return self._skip_reason
@property
def pickle_path(self):
return self._pickle_path
@property
def city(self):
return self._city

159
unittests/test_db_factory.py
View File

@ -1,159 +0,0 @@
"""
Test db factory
SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Peter Yefi peteryefi@gmail.com
"""
import glob
import os
import unittest
from unittest import TestCase
import hub.helpers.constants as cte
from unittests.control import Control
control = Control()
class TestDBFactory(TestCase):
"""
TestDBFactory
"""
@unittest.skipIf(control.skip_test, control.skip_reason)
def test_save_city(self):
control.city.name = "Montreal"
city_id = control.database.persist_city(
control.city,
control.pickle_path,
control.city.name,
control.application_id,
control.user_id)
control.database.delete_city(city_id)
@unittest.skipIf(control.skip_test, control.skip_reason)
def test_get_update_city(self):
city_id = control.database.persist_city(control.city,
control.pickle_path,
control.city.name,
control.application_id,
control.user_id)
control.city.name = "Ottawa"
control.database.update_city(city_id, control.city)
cities = control.database.cities_by_user_and_application(
control.user_id,
control.application_id)
for updated_city in cities:
if updated_city.id == city_id:
self.assertEqual(updated_city.name, control.city.name)
break
control.database.delete_city(city_id)
@unittest.skipIf(control.skip_test, control.skip_reason)
@unittest.skipIf(control.skip_insel_test, 'insel is not installed')
def test_save_results(self):
city_id = control.database.persist_city(control.city,
control.pickle_path,
'current status',
control.application_id,
control.user_id)
city_objects_id = []
request_values = {
'scenarios': [{'current status': ['1']}]
}
for building in control.city.buildings:
_building = control.database.building_info(building.name, city_id)
if cte.MONTH not in building.cooling_demand:
print(f'building {building.name} not calculated')
continue
monthly_cooling_peak_load = building.cooling_peak_load[cte.MONTH]
yearly_cooling_peak_load = building.cooling_peak_load[cte.YEAR]
monthly_heating_peak_load = building.heating_peak_load[cte.MONTH]
yearly_heating_peak_load = building.heating_peak_load[cte.YEAR]
monthly_lighting_peak_load = building.lighting_peak_load[cte.MONTH]
yearly_lighting_peak_load = building.lighting_peak_load[cte.YEAR]
monthly_appliances_peak_load = building.appliances_peak_load[cte.MONTH]
yearly_appliances_peak_load = building.appliances_peak_load[cte.YEAR]
monthly_cooling_demand = building.cooling_demand[cte.MONTH]
yearly_cooling_demand = building.cooling_demand[cte.YEAR]
monthly_heating_demand = building.heating_demand[cte.MONTH]
yearly_heating_demand = building.heating_demand[cte.YEAR]
monthly_lighting_electrical_demand = building.lighting_electrical_demand[cte.MONTH]
yearly_lighting_electrical_demand = building.lighting_electrical_demand[cte.YEAR]
monthly_appliances_electrical_demand = building.appliances_electrical_demand[cte.MONTH]
yearly_appliances_electrical_demand = building.appliances_electrical_demand[cte.YEAR]
monthly_domestic_hot_water_heat_demand = building.domestic_hot_water_heat_demand[cte.MONTH]
yearly_domestic_hot_water_heat_demand = building.domestic_hot_water_heat_demand[cte.YEAR]
monthly_heating_consumption = building.heating_consumption[cte.MONTH]
yearly_heating_consumption = building.heating_consumption[cte.YEAR]
monthly_cooling_consumption = building.cooling_consumption[cte.MONTH]
yearly_cooling_consumption = building.cooling_consumption[cte.YEAR]
monthly_domestic_hot_water_consumption = building.domestic_hot_water_consumption[cte.MONTH]
yearly_domestic_hot_water_consumption = building._domestic_hot_water_consumption[cte.YEAR]
monthly_distribution_systems_electrical_consumption = building.distribution_systems_electrical_consumption[
cte.MONTH]
yearly_distribution_systems_electrical_consumption = building.distribution_systems_electrical_consumption[
cte.YEAR]
monthly_on_site_electrical_production = [x * cte.WATTS_HOUR_TO_JULES
for x in building.onsite_electrical_production[cte.MONTH]]
yearly_on_site_electrical_production = [x * cte.WATTS_HOUR_TO_JULES
for x in building.onsite_electrical_production[cte.YEAR]]
results = {cte.INSEL_MEB: {
'monthly_cooling_peak_load': monthly_cooling_peak_load,
'yearly_cooling_peak_load': yearly_cooling_peak_load,
'monthly_heating_peak_load': monthly_heating_peak_load,
'yearly_heating_peak_load': yearly_heating_peak_load,
'monthly_lighting_peak_load': monthly_lighting_peak_load,
'yearly_lighting_peak_load': yearly_lighting_peak_load,
'monthly_appliances_peak_load': monthly_appliances_peak_load,
'yearly_appliances_peak_load': yearly_appliances_peak_load,
'monthly_cooling_demand': monthly_cooling_demand,
'yearly_cooling_demand': yearly_cooling_demand,
'monthly_heating_demand': monthly_heating_demand,
'yearly_heating_demand': yearly_heating_demand,
'monthly_lighting_electrical_demand': monthly_lighting_electrical_demand,
'yearly_lighting_electrical_demand': yearly_lighting_electrical_demand,
'monthly_appliances_electrical_demand': monthly_appliances_electrical_demand,
'yearly_appliances_electrical_demand': yearly_appliances_electrical_demand,
'monthly_domestic_hot_water_heat_demand': monthly_domestic_hot_water_heat_demand,
'yearly_domestic_hot_water_heat_demand': yearly_domestic_hot_water_heat_demand,
'monthly_heating_consumption': monthly_heating_consumption,
'yearly_heating_consumption': yearly_heating_consumption,
'monthly_cooling_consumption': monthly_cooling_consumption,
'yearly_cooling_consumption': yearly_cooling_consumption,
'monthly_domestic_hot_water_consumption': monthly_domestic_hot_water_consumption,
'yearly_domestic_hot_water_consumption': yearly_domestic_hot_water_consumption,
'monthly_distribution_systems_electrical_consumption': monthly_distribution_systems_electrical_consumption,
'yearly_distribution_systems_electrical_consumption': yearly_distribution_systems_electrical_consumption,
'monthly_on_site_electrical_production': monthly_on_site_electrical_production,
'yearly_on_site_electrical_production': yearly_on_site_electrical_production
}}
db_building_id = _building.id
city_objects_id.append(db_building_id)
control.database.add_simulation_results(
cte.INSEL_MEB,
results, city_object_id=db_building_id)
self.assertEqual(17, len(city_objects_id), 'wrong number of results')
self.assertIsNotNone(city_objects_id[0], 'city_object_id is None')
results = control.database.results(control.user_id, control.application_id, request_values)
self.assertEqual(
28,
len(results['current status'][0]['insel meb'].keys()),
'wrong number of results after retrieve'
)
for _id in city_objects_id:
control.database.delete_results_by_name('insel meb', city_object_id=_id)
control.database.delete_city(city_id)
@classmethod
@unittest.skipIf(control.skip_test, control.skip_reason)
def tearDownClass(cls):
control.database.delete_application(control.application_uuid)
control.database.delete_user(control.user_id)
os.unlink(control.pickle_path)
output_files = glob.glob('./tests_outputs/*')
for output_file in output_files:
os.unlink(output_file)

36
unittests/test_db_retrieve.py
View File

@ -1,36 +0,0 @@
"""
Test db retrieve
SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Ruben Sanchez ruben.sanchez@mail.concordia.ca
"""
import datetime
import unittest
from unittest import TestCase
from unittests.control import Control
control = Control()
class TestDBFactory(TestCase):
"""
TestDBFactory
"""
@unittest.skipIf(control.skip_test, control.skip_reason)
def test_retrieve_results(self):
datetime.datetime.now()
request_values = {
"scenarios": [
{"current status": ["01002777", "01002773", "01036804"]},
{"skin retrofit": ["01002777", "01002773", "01036804"]},
{"system retrofit and pv": ["01002777", "01002773", "01036804"]},
{"skin and system retrofit with pv": ["01002777", "01002773", "01036804"]}
]
}
results = control.database.results(control.user_id, control.application_id, request_values)
scenarios = ['current status', 'skin retrofit', 'system retrofit and pv', 'skin and system retrofit with pv']
for key, value in results.items():
self.assertTrue(key in scenarios, 'Wrong key value')
self.assertEqual(len(value), 0, 'wrong number of results')