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

Changed 4 spaces to 2 and added pylintrc

Browse Source
This commit is contained in:
Ruben Sanchez 2023-11-13 13:53:52 -05:00
parent 731afca116
commit 22417e49d2
19 changed files with 1893 additions and 1356 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@ -2,7 +2,7 @@
The persistence package includes classes to store different class objects in a Postgres database.
## Models
## Models
This defines models for all class objects that we want to persist. It is used for Object Relation Mapping (ORM)
of the class objects to database table columns
@ -27,7 +27,7 @@ All database operations are conducted with the production database (*PROD*) name
This Python file is a configuration class that contains variables that map to configuration parameters in a .env file.
It also contains a method ``def conn_string()`` which returns the connection string to a Postgres database.
## Base
## Base
This class has a constructor that establishes a database connection and returns a reference for database-related CRUD
operations.

88
cerc_persistence/configuration.py
View File

@ -15,53 +15,53 @@ Models = declarative_base()
class Configuration:
"""
Configuration class to hold common persistence configuration
"""
def __init__(self, db_name: str, dotenv_path: str, app_env='TEST'):
"""
Configuration class to hold common persistence configuration
: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():
error_message = f'dotenv file doesn\'t exists at {dotenv_path}'
logging.error(error_message)
raise FileNotFoundError(error_message)
load_dotenv(dotenv_path=dotenv_path)
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():
error_message = f'dotenv file doesn\'t exists at {dotenv_path}'
logging.error(error_message)
raise FileNotFoundError(error_message)
load_dotenv(dotenv_path=dotenv_path)
self._db_name = db_name
self._db_host = os.getenv(f'{app_env}_DB_HOST')
self._db_user = os.getenv(f'{app_env}_DB_USER')
self._db_pass = os.getenv(f'{app_env}_DB_PASSWORD')
self._db_port = os.getenv(f'{app_env}_DB_PORT')
self.hub_token = os.getenv('HUB_TOKEN')
except KeyError as err:
logging.error('Error with credentials: %s', err)
self._db_name = db_name
self._db_host = os.getenv(f'{app_env}_DB_HOST')
self._db_user = os.getenv(f'{app_env}_DB_USER')
self._db_pass = os.getenv(f'{app_env}_DB_PASSWORD')
self._db_port = os.getenv(f'{app_env}_DB_PORT')
self.hub_token = os.getenv('HUB_TOKEN')
except KeyError as err:
logging.error('Error with credentials: %s', err)
@property
def connection_string(self):
"""
Returns a connection string postgresql
:return: connection string
"""
if self._db_pass:
return f'postgresql://{self._db_user}:{self._db_pass}@{self._db_host}:{self._db_port}/{self._db_name}'
return f'postgresql://{self._db_user}@{self._db_host}:{self._db_port}/{self._db_name}'
@property
def connection_string(self):
"""
Returns a connection string postgresql
:return: connection string
"""
if self._db_pass:
return f'postgresql://{self._db_user}:{self._db_pass}@{self._db_host}:{self._db_port}/{self._db_name}'
return f'postgresql://{self._db_user}@{self._db_host}:{self._db_port}/{self._db_name}'
@property
def db_user(self):
"""
retrieve the configured username
"""
return self._db_user
@property
def db_user(self):
"""
retrieve the configured username
"""
return self._db_user
@property
def db_name(self):
"""
retrieve the configured database name
"""
return self._db_name
@property
def db_name(self):
"""
retrieve the configured database name
"""
return self._db_name

414
cerc_persistence/db_control.py
View File

@ -16,233 +16,233 @@ from cerc_persistence.repositories.user import UserRoles
class DBControl:
"""
DBFactory class
"""
def __init__(self, db_name, app_env, dotenv_path):
self._city = City(db_name=db_name, dotenv_path=dotenv_path, app_env=app_env)
self._application = Application(db_name=db_name, app_env=app_env, dotenv_path=dotenv_path)
self._user = User(db_name=db_name, app_env=app_env, dotenv_path=dotenv_path)
self._city_object = CityObject(db_name=db_name, app_env=app_env, dotenv_path=dotenv_path)
self._simulation_results = SimulationResults(db_name=db_name, dotenv_path=dotenv_path, app_env=app_env)
def application_info(self, application_uuid) -> Application:
"""
DBFactory class
Retrieve the application info for the given uuid from the database
:param application_uuid: the uuid for the application
:return: Application
"""
return self._application.get_by_uuid(application_uuid)
def __init__(self, db_name, app_env, dotenv_path):
self._city = City(db_name=db_name, dotenv_path=dotenv_path, app_env=app_env)
self._application = Application(db_name=db_name, app_env=app_env, dotenv_path=dotenv_path)
self._user = User(db_name=db_name, app_env=app_env, dotenv_path=dotenv_path)
self._city_object = CityObject(db_name=db_name, app_env=app_env, dotenv_path=dotenv_path)
self._simulation_results = SimulationResults(db_name=db_name, dotenv_path=dotenv_path, app_env=app_env)
def user_info(self, name, password, application_id) -> User:
"""
Retrieve the user info for the given name and password and application_id from the database
:param name: the username
:param password: the user password
:param application_id: the application id
:return: User
"""
return self._user.get_by_name_application_id_and_password(name, password, application_id)
def application_info(self, application_uuid) -> Application:
"""
Retrieve the application info for the given uuid from the database
:param application_uuid: the uuid for the application
:return: Application
"""
return self._application.get_by_uuid(application_uuid)
def user_login(self, name, password, application_uuid) -> User:
"""
Retrieve the user info from the database
:param name: the username
:param password: the user password
:param application_uuid: the application uuid
:return: User
"""
return self._user.get_by_name_application_uuid_and_password(name, password, application_uuid)
def user_info(self, name, password, application_id) -> User:
"""
Retrieve the user info for the given name and password and application_id from the database
:param name: the username
:param password: the user password
:param application_id: the application id
:return: User
"""
return self._user.get_by_name_application_id_and_password(name, password, application_id)
def cities_by_user_and_application(self, user_id, application_id) -> [City]:
"""
Retrieve the cities belonging to the user and the application from the database
:param user_id: User id
:param application_id: Application id
:return: [City]
"""
return self._city.get_by_user_id_and_application_id(user_id, application_id)
def user_login(self, name, password, application_uuid) -> User:
"""
Retrieve the user info from the database
:param name: the username
:param password: the user password
:param application_uuid: the application uuid
:return: User
"""
return self._user.get_by_name_application_uuid_and_password(name, password, application_uuid)
def building(self, name, user_id, application_id, scenario) -> CityObject:
"""
Retrieve the building from the database
:param name: Building name
:param user_id: User id
:param application_id: Application id
:param scenario: Scenario
:
"""
cities = self._city.get_by_user_id_application_id_and_scenario(user_id, application_id, scenario)
for city in cities:
result = self.building_info(name, city[0].id)
if result is not None:
return result
return None
def cities_by_user_and_application(self, user_id, application_id) -> [City]:
"""
Retrieve the cities belonging to the user and the application from the database
:param user_id: User id
:param application_id: Application id
:return: [City]
"""
return self._city.get_by_user_id_and_application_id(user_id, application_id)
def building_info(self, name, city_id) -> CityObject:
"""
Retrieve the building info from the database
:param name: Building name
:param city_id: City ID
:return: CityObject
"""
return self._city_object.get_by_name_or_alias_and_city(name, city_id)
def building(self, name, user_id, application_id, scenario) -> CityObject:
"""
Retrieve the building from the database
:param name: Building name
:param user_id: User id
:param application_id: Application id
:param scenario: Scenario
:
"""
cities = self._city.get_by_user_id_application_id_and_scenario(user_id, application_id, scenario)
for city in cities:
result = self.building_info(name, city[0].id)
if result is not None:
return result
return None
def buildings_info(self, request_values, city_id) -> [CityObject]:
"""
Retrieve the buildings info from the database
:param request_values: Building names
:param city_id: City ID
:return: [CityObject]
"""
buildings = []
for name in request_values['names']:
buildings.append(self.building_info(name, city_id))
return buildings
def building_info(self, name, city_id) -> CityObject:
"""
Retrieve the building info from the database
:param name: Building name
:param city_id: City ID
:return: CityObject
"""
return self._city_object.get_by_name_or_alias_and_city(name, city_id)
def results(self, user_id, application_id, request_values, result_names=None) -> Dict:
"""
Retrieve the simulation results for the given cities from the database
:param user_id: the user id owning the results
:param application_id: the application id owning the results
:param request_values: dictionary containing the scenario and building names to grab the results
:param result_names: if given, filter the results to the selected names
"""
if result_names is None:
result_names = []
results = {}
for scenario in request_values['scenarios']:
for scenario_name in scenario.keys():
result_sets = self._city.get_by_user_id_application_id_and_scenario(
user_id,
application_id,
scenario_name
)
if result_sets is None:
continue
for result_set in result_sets:
city_id = result_set[0].id
def buildings_info(self, request_values, city_id) -> [CityObject]:
"""
Retrieve the buildings info from the database
:param request_values: Building names
:param city_id: City ID
:return: [CityObject]
"""
buildings = []
for name in request_values['names']:
buildings.append(self.building_info(name, city_id))
return buildings
results[scenario_name] = []
for building_name in scenario[scenario_name]:
_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_id_city_object_id_and_names(
city_id,
city_object_id,
result_names)
def results(self, user_id, application_id, request_values, result_names=None) -> Dict:
"""
Retrieve the simulation results for the given cities from the database
:param user_id: the user id owning the results
:param application_id: the application id owning the results
:param request_values: dictionary containing the scenario and building names to grab the results
:param result_names: if given, filter the results to the selected names
"""
if result_names is None:
result_names = []
results = {}
for scenario in request_values['scenarios']:
for scenario_name in scenario.keys():
result_sets = self._city.get_by_user_id_application_id_and_scenario(
user_id,
application_id,
scenario_name
)
if result_sets is None:
continue
for result_set in result_sets:
city_id = result_set[0].id
for value in _:
values = json.loads(value.values)
values["building"] = building_name
results[scenario_name].append(values)
return results
results[scenario_name] = []
for building_name in scenario[scenario_name]:
_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_id_city_object_id_and_names(
city_id,
city_object_id,
result_names)
def persist_city(self, city: City, pickle_path, scenario, application_id: int, user_id: int):
"""
Creates a city into the database
:param city: City to be stored
:param pickle_path: Path to save the pickle file
:param scenario: Simulation scenario name
:param application_id: Application id owning this city
:param user_id: User who create the city
return identity_id
"""
return self._city.insert(city, pickle_path, scenario, application_id, user_id)
for value in _:
values = json.loads(value.values)
values["building"] = building_name
results[scenario_name].append(values)
return results
def update_city(self, city_id, city):
"""
Update an existing city in the database
:param city_id: the id of the city to update
:param city: the updated city object
"""
return self._city.update(city_id, city)
def persist_city(self, city: City, pickle_path, scenario, application_id: int, user_id: int):
"""
Creates a city into the database
:param city: City to be stored
:param pickle_path: Path to save the pickle file
:param scenario: Simulation scenario name
:param application_id: Application id owning this city
:param user_id: User who create the city
return identity_id
"""
return self._city.insert(city, pickle_path, scenario, application_id, user_id)
def persist_application(self, name: str, description: str, application_uuid: str):
"""
Creates information for an application in the database
:param name: name of application
:param description: the description of the application
:param application_uuid: the uuid of the application to be created
"""
return self._application.insert(name, description, application_uuid)
def update_city(self, city_id, city):
"""
Update an existing city in the database
:param city_id: the id of the city to update
:param city: the updated city object
"""
return self._city.update(city_id, city)
def update_application(self, name: str, description: str, application_uuid: str):
"""
Update the application information stored in the database
:param name: name of application
:param description: the description of the application
:param application_uuid: the uuid of the application to be created
"""
return self._application.update(application_uuid, name, description)
def persist_application(self, name: str, description: str, application_uuid: str):
"""
Creates information for an application in the database
:param name: name of application
:param description: the description of the application
:param application_uuid: the uuid of the application to be created
"""
return self._application.insert(name, description, application_uuid)
def add_simulation_results(self, name, values, city_id=None, city_object_id=None):
"""
Add simulation results to the city or to the city_object to the database
:param name: simulation and simulation engine name
:param values: simulation values in json format
:param city_id: city id or None
:param city_object_id: city object id or None
"""
return self._simulation_results.insert(name, values, city_id, city_object_id)
def update_application(self, name: str, description: str, application_uuid: str):
"""
Update the application information stored in the database
:param name: name of application
:param description: the description of the application
:param application_uuid: the uuid of the application to be created
"""
return self._application.update(application_uuid, name, description)
def create_user(self, name: str, application_id: int, password: str, role: UserRoles):
"""
Creates a new user in the database
:param name: the name of the user
:param application_id: the application id of the user
:param password: the password of the user
:param role: the role of the user
"""
return self._user.insert(name, password, role, application_id)
def add_simulation_results(self, name, values, city_id=None, city_object_id=None):
"""
Add simulation results to the city or to the city_object to the database
:param name: simulation and simulation engine name
:param values: simulation values in json format
:param city_id: city id or None
:param city_object_id: city object id or None
"""
return self._simulation_results.insert(name, values, city_id, city_object_id)
def update_user(self, user_id: int, name: str, password: str, role: UserRoles):
"""
Updates a user in the database
:param user_id: the id of the user
:param name: the name of the user
:param password: the password of the user
:param role: the role of the user
"""
return self._user.update(user_id, name, password, role)
def create_user(self, name: str, application_id: int, password: str, role: UserRoles):
"""
Creates a new user in the database
:param name: the name of the user
:param application_id: the application id of the user
:param password: the password of the user
:param role: the role of the user
"""
return self._user.insert(name, password, role, application_id)
def get_by_name_and_application(self, name: str, application: int):
"""
Retrieve a single user from the database
:param name: username
:param application: application accessing hub
"""
return self._user.get_by_name_and_application(name, application)
def update_user(self, user_id: int, name: str, password: str, role: UserRoles):
"""
Updates a user in the database
:param user_id: the id of the user
:param name: the name of the user
:param password: the password of the user
:param role: the role of the user
"""
return self._user.update(user_id, name, password, role)
def delete_user(self, user_id):
"""
Delete a single user from the database
:param user_id: the id of the user to delete
"""
self._user.delete(user_id)
def get_by_name_and_application(self, name: str, application: int):
"""
Retrieve a single user from the database
:param name: username
:param application: application accessing hub
"""
return self._user.get_by_name_and_application(name, application)
def delete_city(self, city_id):
"""
Deletes a single city from the database
:param city_id: the id of the city to get
"""
self._city.delete(city_id)
def delete_user(self, user_id):
"""
Delete a single user from the database
:param user_id: the id of the user to delete
"""
self._user.delete(user_id)
def delete_results_by_name(self, name, city_id=None, city_object_id=None):
"""
Deletes city object simulation results from the database
:param name: simulation name
:param city_id: if given, delete delete the results for the city with id city_id
:param city_object_id: if given, delete delete the results for the city object with id city_object_id
"""
self._simulation_results.delete(name, city_id=city_id, city_object_id=city_object_id)
def delete_city(self, city_id):
"""
Deletes a single city from the database
:param city_id: the id of the city to get
"""
self._city.delete(city_id)
def delete_results_by_name(self, name, city_id=None, city_object_id=None):
"""
Deletes city object simulation results from the database
:param name: simulation name
:param city_id: if given, delete delete the results for the city with id city_id
:param city_object_id: if given, delete delete the results for the city object with id city_object_id
"""
self._simulation_results.delete(name, city_id=city_id, city_object_id=city_object_id)
def delete_application(self, application_uuid):
"""
Deletes a single application from the database
:param application_uuid: the id of the application to get
"""
self._application.delete(application_uuid)
def delete_application(self, application_uuid):
"""
Deletes a single application from the database
:param application_uuid: the id of the application to get
"""
self._application.delete(application_uuid)

86
cerc_persistence/db_setup.py
View File

@ -18,53 +18,53 @@ from cerc_persistence.repositories.application import Application as Application
class DBSetup:
"""
Creates a Persistence database structure
"""
def __init__(self, db_name, app_env, dotenv_path, admin_password, application_uuid):
"""
Creates a Persistence database structure
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)
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
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)
# Create the tables using the models
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)
self._user_repo = UserRepository(db_name=db_name, app_env=app_env, dotenv_path=dotenv_path)
self._application_repo = ApplicationRepository(db_name=db_name, app_env=app_env, dotenv_path=dotenv_path)
application_id = self._create_admin_app(self._application_repo, application_uuid)
self._create_admin_user(self._user_repo, admin_password, application_id)
self._user_repo = UserRepository(db_name=db_name, app_env=app_env, dotenv_path=dotenv_path)
self._application_repo = ApplicationRepository(db_name=db_name, app_env=app_env, dotenv_path=dotenv_path)
application_id = self._create_admin_app(self._application_repo, application_uuid)
self._create_admin_user(self._user_repo, admin_password, application_id)
@staticmethod
def _create_admin_app(application_repo, application_uuid):
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)
@staticmethod
def _create_admin_app(application_repo, application_uuid):
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)
if isinstance(application, dict):
logging.info(application)
else:
msg = f'Created Admin tool with application_uuid: {application_uuid}'
logging.info(msg)
return application.id
if isinstance(application, dict):
logging.info(application)
else:
msg = f'Created Admin tool with application_uuid: {application_uuid}'
logging.info(msg)
return application.id
@staticmethod
def _create_admin_user(user_repo, admin_password, application_id):
password = admin_password
logging.info('Creating default admin user...')
user = user_repo.insert('Administrator', password, UserRoles.Admin, application_id)
if isinstance(user, dict):
logging.info(user)
else:
logging.info('Created Admin user')
@staticmethod
def _create_admin_user(user_repo, admin_password, application_id):
password = admin_password
logging.info('Creating default admin user...')
user = user_repo.insert('Administrator', password, UserRoles.Admin, application_id)
if isinstance(user, dict):
logging.info(user)
else:
logging.info('Created Admin user')

28
cerc_persistence/models/application.py
View File

@ -15,18 +15,18 @@ from cerc_persistence.configuration import Models
class Application(Models):
"""
A model representation of an application
"""
__tablename__ = 'application'
id = Column(Integer, Sequence('application_id_seq'), primary_key=True)
name = Column(String, nullable=False)
description = Column(String, nullable=False)
application_uuid = Column(UUID(as_uuid=True), nullable=False)
created = Column(DateTime, default=datetime.datetime.utcnow)
updated = Column(DateTime, default=datetime.datetime.utcnow)
"""
A model representation of an application
"""
__tablename__ = 'application'
id = Column(Integer, Sequence('application_id_seq'), primary_key=True)
name = Column(String, nullable=False)
description = Column(String, nullable=False)
application_uuid = Column(UUID(as_uuid=True), nullable=False)
created = Column(DateTime, default=datetime.datetime.utcnow)
updated = Column(DateTime, default=datetime.datetime.utcnow)
def __init__(self, name, description, application_uuid):
self.name = name
self.description = description
self.application_uuid = application_uuid
def __init__(self, name, description, application_uuid):
self.name = name
self.description = description
self.application_uuid = application_uuid

38
cerc_persistence/models/city.py
View File

@ -14,23 +14,23 @@ from cerc_persistence.configuration import Models
class City(Models):
"""A model representation of a city
"""
__tablename__ = 'city'
id = Column(Integer, Sequence('city_id_seq'), primary_key=True)
pickle_path = Column(String, nullable=False)
name = Column(String, nullable=False)
scenario = Column(String, nullable=False)
application_id = Column(Integer, ForeignKey('application.id', ondelete='CASCADE'), nullable=False)
user_id = Column(Integer, ForeignKey('user.id', ondelete='CASCADE'), nullable=True)
hub_release = Column(String, nullable=False)
created = Column(DateTime, default=datetime.datetime.utcnow)
updated = Column(DateTime, default=datetime.datetime.utcnow)
"""A model representation of a city
"""
__tablename__ = 'city'
id = Column(Integer, Sequence('city_id_seq'), primary_key=True)
pickle_path = Column(String, nullable=False)
name = Column(String, nullable=False)
scenario = Column(String, nullable=False)
application_id = Column(Integer, ForeignKey('application.id', ondelete='CASCADE'), nullable=False)
user_id = Column(Integer, ForeignKey('user.id', ondelete='CASCADE'), nullable=True)
hub_release = Column(String, nullable=False)
created = Column(DateTime, default=datetime.datetime.utcnow)
updated = Column(DateTime, default=datetime.datetime.utcnow)
def __init__(self, pickle_path, name, scenario, application_id, user_id, hub_release):
self.pickle_path = str(pickle_path)
self.name = name
self.scenario = scenario
self.application_id = application_id
self.user_id = user_id
self.hub_release = hub_release
def __init__(self, pickle_path, name, scenario, application_id, user_id, hub_release):
self.pickle_path = str(pickle_path)
self.name = name
self.scenario = scenario
self.application_id = application_id
self.user_id = user_id
self.hub_release = hub_release

122
cerc_persistence/models/city_object.py
View File

@ -16,66 +16,66 @@ from cerc_persistence.configuration import Models
class CityObject(Models):
"""
A model representation of an application
"""
__tablename__ = 'city_object'
id = Column(Integer, Sequence('city_object_id_seq'), primary_key=True)
city_id = Column(Integer, ForeignKey('city.id', ondelete='CASCADE'), nullable=False)
name = Column(String, nullable=False)
aliases = Column(String, nullable=True)
type = Column(String, nullable=False)
year_of_construction = Column(Integer, nullable=True)
function = Column(String, nullable=True)
usage = Column(String, nullable=True)
volume = Column(Float, nullable=False)
area = Column(Float, nullable=False)
total_heating_area = Column(Float, nullable=False)
wall_area = Column(Float, nullable=False)
windows_area = Column(Float, nullable=False)
roof_area = Column(Float, nullable=False)
total_pv_area = Column(Float, nullable=False)
system_name = Column(String, nullable=False)
created = Column(DateTime, default=datetime.datetime.utcnow)
updated = Column(DateTime, default=datetime.datetime.utcnow)
"""
A model representation of an application
"""
__tablename__ = 'city_object'
id = Column(Integer, Sequence('city_object_id_seq'), primary_key=True)
city_id = Column(Integer, ForeignKey('city.id', ondelete='CASCADE'), nullable=False)
name = Column(String, nullable=False)
aliases = Column(String, nullable=True)
type = Column(String, nullable=False)
year_of_construction = Column(Integer, nullable=True)
function = Column(String, nullable=True)
usage = Column(String, nullable=True)
volume = Column(Float, nullable=False)
area = Column(Float, nullable=False)
total_heating_area = Column(Float, nullable=False)
wall_area = Column(Float, nullable=False)
windows_area = Column(Float, nullable=False)
roof_area = Column(Float, nullable=False)
total_pv_area = Column(Float, nullable=False)
system_name = Column(String, nullable=False)
created = Column(DateTime, default=datetime.datetime.utcnow)
updated = Column(DateTime, default=datetime.datetime.utcnow)
def __init__(self, city_id, building: Building):
self.city_id = city_id
self.name = building.name
self.aliases = building.aliases
self.type = building.type
self.year_of_construction = building.year_of_construction
self.function = building.function
self.usage = building.usages_percentage
self.volume = building.volume
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)
storeys = building.storeys_above_ground
wall_area = 0
window_ratio = 0
try:
if storeys is None:
storeys = building.max_height / building.average_storey_height
for internal_zone in building.internal_zones:
for thermal_zone in internal_zone.thermal_zones_from_internal_zones:
for thermal_boundary in thermal_zone.thermal_boundaries:
window_ratio = thermal_boundary.window_ratio
break
except TypeError:
storeys = 0
logging.warning(
'building %s has no storey height so heating area, storeys and window ratio cannot be calculated',
self.name
)
self.total_heating_area = building.floor_area * storeys
def __init__(self, city_id, building: Building):
self.city_id = city_id
self.name = building.name
self.aliases = building.aliases
self.type = building.type
self.year_of_construction = building.year_of_construction
self.function = building.function
self.usage = building.usages_percentage
self.volume = building.volume
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)
storeys = building.storeys_above_ground
wall_area = 0
window_ratio = 0
try:
if storeys is None:
storeys = building.max_height / building.average_storey_height
for internal_zone in building.internal_zones:
for thermal_zone in internal_zone.thermal_zones_from_internal_zones:
for thermal_boundary in thermal_zone.thermal_boundaries:
window_ratio = thermal_boundary.window_ratio
break
except TypeError:
storeys = 0
logging.warning(
'building %s has no storey height so heating area, storeys and window ratio cannot be calculated',
self.name
)
self.total_heating_area = building.floor_area * storeys
for wall in building.walls:
wall_area += wall.solid_polygon.area
self.wall_area = wall_area
self.windows_area = wall_area * window_ratio
system_name = building.energy_systems_archetype_name
if system_name is None:
system_name = ''
self.system_name = system_name
for wall in building.walls:
wall_area += wall.solid_polygon.area
self.wall_area = wall_area
self.windows_area = wall_area * window_ratio
system_name = building.energy_systems_archetype_name
if system_name is None:
system_name = ''
self.system_name = system_name

32
cerc_persistence/models/simulation_results.py
View File

@ -14,20 +14,20 @@ from cerc_persistence.configuration import Models
class SimulationResults(Models):
"""
A model representation of an application
"""
__tablename__ = 'simulation_results'
id = Column(Integer, Sequence('simulation_results_id_seq'), primary_key=True)
city_id = Column(Integer, ForeignKey('city.id', ondelete='CASCADE'), nullable=True)
city_object_id = Column(Integer, ForeignKey('city_object.id', ondelete='CASCADE'), nullable=True)
name = Column(String, nullable=False)
values = Column(JSONB, nullable=False)
created = Column(DateTime, default=datetime.datetime.utcnow)
updated = Column(DateTime, default=datetime.datetime.utcnow)
"""
A model representation of an application
"""
__tablename__ = 'simulation_results'
id = Column(Integer, Sequence('simulation_results_id_seq'), primary_key=True)
city_id = Column(Integer, ForeignKey('city.id', ondelete='CASCADE'), nullable=True)
city_object_id = Column(Integer, ForeignKey('city_object.id', ondelete='CASCADE'), nullable=True)
name = Column(String, nullable=False)
values = Column(JSONB, nullable=False)
created = Column(DateTime, default=datetime.datetime.utcnow)
updated = Column(DateTime, default=datetime.datetime.utcnow)
def __init__(self, name, values, city_id=None, city_object_id=None):
self.name = name
self.values = values
self.city_id = city_id
self.city_object_id = city_object_id
def __init__(self, name, values, city_id=None, city_object_id=None):
self.name = name
self.values = values
self.city_id = city_id
self.city_object_id = city_object_id

42
cerc_persistence/models/user.py
View File

@ -15,28 +15,28 @@ from cerc_persistence.configuration import Models
class UserRoles(enum.Enum):
"""
User roles enum
"""
Admin = 'Admin'
Hub_Reader = 'Hub_Reader'
"""
User roles enum
"""
Admin = 'Admin'
Hub_Reader = 'Hub_Reader'
class User(Models):
"""
A model representation of a city
"""
__tablename__ = 'user'
id = Column(Integer, Sequence('user_id_seq'), primary_key=True)
name = Column(String, nullable=False)
password = Column(String, nullable=False)
role = Column(Enum(UserRoles), nullable=False, default=UserRoles.Hub_Reader)
application_id = Column(Integer, nullable=False)
created = Column(DateTime, default=datetime.datetime.utcnow)
updated = Column(DateTime, default=datetime.datetime.utcnow)
"""
A model representation of a city
"""
__tablename__ = 'user'
id = Column(Integer, Sequence('user_id_seq'), primary_key=True)
name = Column(String, nullable=False)
password = Column(String, nullable=False)
role = Column(Enum(UserRoles), nullable=False, default=UserRoles.Hub_Reader)
application_id = Column(Integer, nullable=False)
created = Column(DateTime, default=datetime.datetime.utcnow)
updated = Column(DateTime, default=datetime.datetime.utcnow)
def __init__(self, name, password, role, application_id):
self.name = name
self.password = password
self.role = role
self.application_id = application_id
def __init__(self, name, password, role, application_id):
self.name = name
self.password = password
self.role = role
self.application_id = application_id

172
cerc_persistence/repositories/application.py
View File

@ -17,95 +17,95 @@ from cerc_persistence.models import Application as Model
class Application(Repository):
"""
Application repository
"""
_instance = None
def __init__(self, db_name: str, dotenv_path: str, app_env: str):
super().__init__(db_name, dotenv_path, app_env)
def __new__(cls, db_name, dotenv_path, app_env):
"""
Application repository
Implemented for a singleton pattern
"""
_instance = None
if cls._instance is None:
cls._instance = super(Application, cls).__new__(cls)
return cls._instance
def __init__(self, db_name: str, dotenv_path: str, app_env: str):
super().__init__(db_name, dotenv_path, app_env)
def insert(self, name: str, description: str, application_uuid: str):
"""
Inserts a new application
:param name: Application name
:param description: Application description
:param application_uuid: Unique identifier for the application
:return: Identity id
"""
try:
application = self.get_by_uuid(application_uuid)
if application is not None:
raise SQLAlchemyError('application already exists')
except TypeError:
pass
try:
application = Model(name=name, description=description, application_uuid=application_uuid)
with Session(self.engine) as session:
session.add(application)
session.commit()
session.refresh(application)
return application.id
except SQLAlchemyError as err:
logging.error('An error occurred while creating application %s', err)
raise SQLAlchemyError from err
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
def update(self, application_uuid: str, name: str, description: str):
"""
Updates an application
:param application_uuid: the application uuid of the application to be updated
:param name: the application name
:param description: the application description
:return: None
"""
try:
with Session(self.engine) as session:
session.query(Model).filter(
Model.application_uuid == application_uuid
).update({'name': name, 'description': description, 'updated': datetime.datetime.utcnow()})
session.commit()
except SQLAlchemyError as err:
logging.error('Error while updating application %s', err)
raise SQLAlchemyError from err
def insert(self, name: str, description: str, application_uuid: str):
"""
Inserts a new application
:param name: Application name
:param description: Application description
:param application_uuid: Unique identifier for the application
:return: Identity id
"""
try:
application = self.get_by_uuid(application_uuid)
if application is not None:
raise SQLAlchemyError('application already exists')
except TypeError:
pass
try:
application = Model(name=name, description=description, application_uuid=application_uuid)
with Session(self.engine) as session:
session.add(application)
session.commit()
session.refresh(application)
return application.id
except SQLAlchemyError as err:
logging.error('An error occurred while creating application %s', err)
raise SQLAlchemyError from err
def delete(self, application_uuid: str):
"""
Deletes an application with the application_uuid
:param application_uuid: The application uuid
:return: None
"""
try:
with Session(self.engine) as session:
session.query(Model).filter(Model.application_uuid == application_uuid).delete()
session.flush()
session.commit()
except SQLAlchemyError as err:
logging.error('Error while deleting application %s', err)
raise SQLAlchemyError from err
def update(self, application_uuid: str, name: str, description: str):
"""
Updates an application
:param application_uuid: the application uuid of the application to be updated
:param name: the application name
:param description: the application description
:return: None
"""
try:
with Session(self.engine) as session:
session.query(Model).filter(
Model.application_uuid == application_uuid
).update({'name': name, 'description': description, 'updated': datetime.datetime.utcnow()})
session.commit()
except SQLAlchemyError as err:
logging.error('Error while updating application %s', err)
raise SQLAlchemyError from err
def delete(self, application_uuid: str):
"""
Deletes an application with the application_uuid
:param application_uuid: The application uuid
:return: None
"""
try:
with Session(self.engine) as session:
session.query(Model).filter(Model.application_uuid == application_uuid).delete()
session.flush()
session.commit()
except SQLAlchemyError as err:
logging.error('Error while deleting application %s', err)
raise SQLAlchemyError from err
def get_by_uuid(self, application_uuid: str) -> Model:
"""
Fetch Application based on the application uuid
:param application_uuid: the application uuid
:return: Application with the provided application_uuid
"""
try:
with Session(self.engine) as session:
result_set = session.execute(select(Model).where(
Model.application_uuid == application_uuid)
).first()
return result_set[0]
except SQLAlchemyError as err:
logging.error('Error while fetching application by application_uuid %s', err)
raise SQLAlchemyError from err
except TypeError as err:
logging.error('Error while fetching application, empty result %s', err)
raise TypeError from err
def get_by_uuid(self, application_uuid: str) -> Model:
"""
Fetch Application based on the application uuid
:param application_uuid: the application uuid
:return: Application with the provided application_uuid
"""
try:
with Session(self.engine) as session:
result_set = session.execute(select(Model).where(
Model.application_uuid == application_uuid)
).first()
return result_set[0]
except SQLAlchemyError as err:
logging.error('Error while fetching application by application_uuid %s', err)
raise SQLAlchemyError from err
except TypeError as err:
logging.error('Error while fetching application, empty result %s', err)
raise TypeError from err

220
cerc_persistence/repositories/city.py
View File

@ -19,120 +19,120 @@ from hub.version import __version__
class City(Repository):
"""
City repository
"""
_instance = None
def __init__(self, db_name: str, dotenv_path: str, app_env: str):
super().__init__(db_name, dotenv_path, app_env)
def __new__(cls, db_name, dotenv_path, app_env):
"""
City repository
Implemented for a singleton pattern
"""
_instance = None
if cls._instance is None:
cls._instance = super(City, cls).__new__(cls)
return cls._instance
def __init__(self, db_name: str, dotenv_path: str, app_env: str):
super().__init__(db_name, dotenv_path, app_env)
def insert(self, city: CityHub, pickle_path, scenario, application_id, user_id: int):
"""
Inserts a city
:param city: The complete city instance
:param pickle_path: Path to the pickle
:param scenario: Simulation scenario name
:param application_id: Application id owning the instance
:param user_id: User id owning the instance
:return: Identity id
"""
city.save_compressed(pickle_path)
try:
db_city = Model(
pickle_path,
city.name,
scenario,
application_id,
user_id,
__version__)
with Session(self.engine) as session:
session.add(db_city)
session.flush()
session.commit()
for building in city.buildings:
db_city_object = CityObject(db_city.id,
building)
session.add(db_city_object)
session.flush()
session.commit()
session.refresh(db_city)
return db_city.id
except SQLAlchemyError as err:
logging.error('An error occurred while creating a city %s', err)
raise SQLAlchemyError from err
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
def update(self, city_id: int, city: CityHub):
"""
Updates a city name (other updates makes no sense)
:param city_id: the id of the city to be updated
:param city: the city object
:return: None
"""
try:
now = datetime.datetime.utcnow()
with Session(self.engine) as session:
session.query(Model).filter(Model.id == city_id).update({'name': city.name, 'updated': now})
session.commit()
except SQLAlchemyError as err:
logging.error('Error while updating city %s', err)
raise SQLAlchemyError from err
def insert(self, city: CityHub, pickle_path, scenario, application_id, user_id: int):
"""
Inserts a city
:param city: The complete city instance
:param pickle_path: Path to the pickle
:param scenario: Simulation scenario name
:param application_id: Application id owning the instance
:param user_id: User id owning the instance
:return: Identity id
"""
city.save_compressed(pickle_path)
try:
db_city = Model(
pickle_path,
city.name,
scenario,
application_id,
user_id,
__version__)
with Session(self.engine) as session:
session.add(db_city)
session.flush()
session.commit()
for building in city.buildings:
db_city_object = CityObject(db_city.id,
building)
session.add(db_city_object)
session.flush()
session.commit()
session.refresh(db_city)
return db_city.id
except SQLAlchemyError as err:
logging.error('An error occurred while creating a city %s', err)
raise SQLAlchemyError from err
def delete(self, city_id: int):
"""
Deletes a City with the id
:param city_id: the city id
:return: None
"""
try:
with Session(self.engine) as session:
session.query(CityObject).filter(CityObject.city_id == city_id).delete()
session.query(Model).filter(Model.id == city_id).delete()
session.commit()
except SQLAlchemyError as err:
logging.error('Error while fetching city %s', err)
raise SQLAlchemyError from err
def update(self, city_id: int, city: CityHub):
"""
Updates a city name (other updates makes no sense)
:param city_id: the id of the city to be updated
:param city: the city object
:return: None
"""
try:
now = datetime.datetime.utcnow()
with Session(self.engine) as session:
session.query(Model).filter(Model.id == city_id).update({'name': city.name, 'updated': now})
session.commit()
except SQLAlchemyError as err:
logging.error('Error while updating city %s', err)
raise SQLAlchemyError from err
def get_by_user_id_application_id_and_scenario(self, user_id, application_id, scenario) -> [Model]:
"""
Fetch city based on the user who created it
:param user_id: the user id
:param application_id: the application id
:param scenario: simulation scenario name
:return: [ModelCity]
"""
try:
with Session(self.engine) as session:
result_set = session.execute(select(Model).where(Model.user_id == user_id,
Model.application_id == application_id,
Model.scenario == scenario
)).all()
return result_set
except SQLAlchemyError as err:
logging.error('Error while fetching city by name %s', err)
raise SQLAlchemyError from err
def delete(self, city_id: int):
"""
Deletes a City with the id
:param city_id: the city id
:return: None
"""
try:
with Session(self.engine) as session:
session.query(CityObject).filter(CityObject.city_id == city_id).delete()
session.query(Model).filter(Model.id == city_id).delete()
session.commit()
except SQLAlchemyError as err:
logging.error('Error while fetching city %s', err)
raise SQLAlchemyError from err
def get_by_user_id_application_id_and_scenario(self, user_id, application_id, scenario) -> [Model]:
"""
Fetch city based on the user who created it
:param user_id: the user id
:param application_id: the application id
:param scenario: simulation scenario name
:return: [ModelCity]
"""
try:
with Session(self.engine) as session:
result_set = session.execute(select(Model).where(Model.user_id == user_id,
Model.application_id == application_id,
Model.scenario == scenario
)).all()
return result_set
except SQLAlchemyError as err:
logging.error('Error while fetching city by name %s', err)
raise SQLAlchemyError from err
def get_by_user_id_and_application_id(self, user_id, application_id) -> [Model]:
"""
Fetch city based on the user who created it
:param user_id: the user id
:param application_id: the application id
:return: ModelCity
"""
try:
with Session(self.engine) as session:
result_set = session.execute(
select(Model).where(Model.user_id == user_id, Model.application_id == application_id)
)
return [r[0] for r in result_set]
except SQLAlchemyError as err:
logging.error('Error while fetching city by name %s', err)
raise SQLAlchemyError from err
def get_by_user_id_and_application_id(self, user_id, application_id) -> [Model]:
"""
Fetch city based on the user who created it
:param user_id: the user id
:param application_id: the application id
:return: ModelCity
"""
try:
with Session(self.engine) as session:
result_set = session.execute(
select(Model).where(Model.user_id == user_id, Model.application_id == application_id)
)
return [r[0] for r in result_set]
except SQLAlchemyError as err:
logging.error('Error while fetching city by name %s', err)
raise SQLAlchemyError from err

216
cerc_persistence/repositories/city_object.py
View File

@ -17,117 +17,117 @@ from cerc_persistence.models import CityObject as Model
class CityObject(Repository):
"""
City object repository
"""
_instance = None
def __init__(self, db_name: str, dotenv_path: str, app_env: str):
super().__init__(db_name, dotenv_path, app_env)
def __new__(cls, db_name, dotenv_path, app_env):
"""
City object repository
Implemented for a singleton pattern
"""
_instance = None
if cls._instance is None:
cls._instance = super(CityObject, cls).__new__(cls)
return cls._instance
def __init__(self, db_name: str, dotenv_path: str, app_env: str):
super().__init__(db_name, dotenv_path, app_env)
def insert(self, city_id: int, building: Building):
"""
Inserts a new city object
:param city_id: city id for the city owning this city object
:param building: the city object (only building for now) to be inserted
return Identity id
"""
city_object = self.get_by_name_or_alias_and_city(building.name, city_id)
if city_object is not None:
raise SQLAlchemyError(f'A city_object named {building.name} already exists in that city')
try:
city_object = Model(city_id=city_id,
building=building)
with Session(self.engine) as session:
session.add(city_object)
session.flush()
session.commit()
session.refresh(city_object)
return city_object.id
except SQLAlchemyError as err:
logging.error('An error occurred while creating city_object %s', err)
raise SQLAlchemyError from err
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
def update(self, city_id: int, building: Building):
"""
Updates an application
:param city_id: the city id of the city owning the city object
:param building: the city object
:return: None
"""
try:
object_usage = ''
for internal_zone in building.internal_zones:
for usage in internal_zone.usages:
object_usage = f'{object_usage}{usage.name}_{usage.percentage} '
object_usage = object_usage.rstrip()
with Session(self.engine) as session:
session.query(Model).filter(Model.name == building.name, Model.city_id == city_id).update(
{'name': building.name,
'alias': building.alias,
'object_type': building.type,
'year_of_construction': building.year_of_construction,
'function': building.function,
'usage': object_usage,
'volume': building.volume,
'area': building.floor_area,
'updated': datetime.datetime.utcnow()})
session.commit()
except SQLAlchemyError as err:
logging.error('Error while updating city object %s', err)
raise SQLAlchemyError from err
def insert(self, city_id: int, building: Building):
"""
Inserts a new city object
:param city_id: city id for the city owning this city object
:param building: the city object (only building for now) to be inserted
return Identity id
"""
city_object = self.get_by_name_or_alias_and_city(building.name, city_id)
def delete(self, city_id: int, name: str):
"""
Deletes an application with the application_uuid
:param city_id: The id for the city owning the city object
:param name: The city object name
:return: None
"""
try:
with Session(self.engine) as session:
session.query(Model).filter(Model.city_id == city_id, Model.name == name).delete()
session.commit()
except SQLAlchemyError as err:
logging.error('Error while deleting application %s', err)
raise SQLAlchemyError from err
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
:param city_id: a city identifier
: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 == city_id)).first()
if city_object is not None:
raise SQLAlchemyError(f'A city_object named {building.name} already exists in that city')
try:
city_object = Model(city_id=city_id,
building=building)
with Session(self.engine) as session:
session.add(city_object)
session.flush()
session.commit()
session.refresh(city_object)
return city_object.id
except SQLAlchemyError as err:
logging.error('An error occurred while creating city_object %s', err)
raise SQLAlchemyError from err
def update(self, city_id: int, building: Building):
"""
Updates an application
:param city_id: the city id of the city owning the city object
:param building: the city object
:return: None
"""
try:
object_usage = ''
for internal_zone in building.internal_zones:
for usage in internal_zone.usages:
object_usage = f'{object_usage}{usage.name}_{usage.percentage} '
object_usage = object_usage.rstrip()
with Session(self.engine) as session:
session.query(Model).filter(Model.name == building.name, Model.city_id == city_id).update(
{'name': building.name,
'alias': building.alias,
'object_type': building.type,
'year_of_construction': building.year_of_construction,
'function': building.function,
'usage': object_usage,
'volume': building.volume,
'area': building.floor_area,
'updated': datetime.datetime.utcnow()})
session.commit()
except SQLAlchemyError as err:
logging.error('Error while updating city object %s', err)
raise SQLAlchemyError from err
def delete(self, city_id: int, name: str):
"""
Deletes an application with the application_uuid
:param city_id: The id for the city owning the city object
:param name: The city object name
:return: None
"""
try:
with Session(self.engine) as session:
session.query(Model).filter(Model.city_id == city_id, Model.name == name).delete()
session.commit()
except SQLAlchemyError as err:
logging.error('Error while deleting application %s', err)
raise SQLAlchemyError from err
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
:param city_id: a city identifier
: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 == city_id)).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 == city_id)
).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
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 == city_id)
).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

278
cerc_persistence/repositories/simulation_results.py
View File

@ -19,152 +19,152 @@ from cerc_persistence.models import SimulationResults as Model
class SimulationResults(Repository):
"""
Simulation results repository
"""
_instance = None
def __init__(self, db_name: str, dotenv_path: str, app_env: str):
super().__init__(db_name, dotenv_path, app_env)
def __new__(cls, db_name, dotenv_path, app_env):
"""
Simulation results repository
Implemented for a singleton pattern
"""
_instance = None
if cls._instance is None:
cls._instance = super(SimulationResults, cls).__new__(cls)
return cls._instance
def __init__(self, db_name: str, dotenv_path: str, app_env: str):
super().__init__(db_name, dotenv_path, app_env)
def insert(self, name: str, values: str, city_id=None, city_object_id=None):
"""
Inserts simulations results linked either with a city as a whole or with a city object
:param name: results name
:param values: the simulation results in json format
:param city_id: optional city id
:param city_object_id: optional city object id
:return: Identity id
"""
if city_id is not None:
_ = self._get_city(city_id)
else:
_ = self._get_city_object(city_object_id)
try:
simulation_result = Model(name=name,
values=values,
city_id=city_id,
city_object_id=city_object_id)
with Session(self.engine) as session:
session.add(simulation_result)
session.flush()
session.commit()
session.refresh(simulation_result)
return simulation_result.id
except SQLAlchemyError as err:
logging.error('An error occurred while creating city_object %s', err)
raise SQLAlchemyError from err
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
def insert(self, name: str, values: str, city_id=None, city_object_id=None):
"""
Inserts simulations results linked either with a city as a whole or with a city object
:param name: results name
:param values: the simulation results in json format
:param city_id: optional city id
:param city_object_id: optional city object id
:return: Identity id
"""
def update(self, name: str, values: str, city_id=None, city_object_id=None):
"""
Updates simulation results for a city or a city object
:param name: The simulation results tool and workflow name
:param values: the simulation results in json format
:param city_id: optional city id
:param city_object_id: optional city object id
:return: None
"""
try:
with Session(self.engine) as session:
if city_id is not None:
_ = self._get_city(city_id)
session.query(Model).filter(Model.name == name, Model.city_id == city_id).update(
{
'values': values,
'updated': datetime.datetime.utcnow()
})
session.commit()
elif city_object_id is not None:
session.query(Model).filter(Model.name == name, Model.city_object_id == city_object_id).update(
{
'values': values,
'updated': datetime.datetime.utcnow()
})
session.commit()
else:
_ = self._get_city_object(city_object_id)
try:
simulation_result = Model(name=name,
values=values,
city_id=city_id,
city_object_id=city_object_id)
with Session(self.engine) as session:
session.add(simulation_result)
session.flush()
session.commit()
session.refresh(simulation_result)
return simulation_result.id
except SQLAlchemyError as err:
logging.error('An error occurred while creating city_object %s', err)
raise SQLAlchemyError from err
raise NotImplementedError('Missing either city_id or city_object_id')
except SQLAlchemyError as err:
logging.error('Error while updating city object %s', err)
raise SQLAlchemyError from err
def update(self, name: str, values: str, city_id=None, city_object_id=None):
"""
Updates simulation results for a city or a city object
:param name: The simulation results tool and workflow name
:param values: the simulation results in json format
:param city_id: optional city id
:param city_object_id: optional city object id
:return: None
"""
try:
with Session(self.engine) as session:
if city_id is not None:
session.query(Model).filter(Model.name == name, Model.city_id == city_id).update(
{
'values': values,
'updated': datetime.datetime.utcnow()
})
session.commit()
elif city_object_id is not None:
session.query(Model).filter(Model.name == name, Model.city_object_id == city_object_id).update(
{
'values': values,
'updated': datetime.datetime.utcnow()
})
session.commit()
else:
raise NotImplementedError('Missing either city_id or city_object_id')
except SQLAlchemyError as err:
logging.error('Error while updating city object %s', err)
raise SQLAlchemyError from err
def delete(self, name: str, city_id=None, city_object_id=None):
"""
Deletes an application with the application_uuid
:param name: The simulation results tool and workflow name
:param city_id: The id for the city owning the simulation results
:param city_object_id: the id for the city_object owning these simulation results
:return: None
"""
try:
with Session(self.engine) as session:
if city_id is not None:
session.query(Model).filter(Model.name == name, Model.city_id == city_id).delete()
session.commit()
elif city_object_id is not None:
session.query(Model).filter(Model.name == name, Model.city_object_id == city_object_id).delete()
session.commit()
else:
raise NotImplementedError('Missing either city_id or city_object_id')
except SQLAlchemyError as err:
logging.error('Error while deleting application: %s', err)
raise SQLAlchemyError from err
def delete(self, name: str, city_id=None, city_object_id=None):
"""
Deletes an application with the application_uuid
:param name: The simulation results tool and workflow name
:param city_id: The id for the city owning the simulation results
:param city_object_id: the id for the city_object owning these simulation results
:return: None
"""
try:
with Session(self.engine) as session:
if city_id is not None:
session.query(Model).filter(Model.name == name, Model.city_id == city_id).delete()
session.commit()
elif city_object_id is not None:
session.query(Model).filter(Model.name == name, Model.city_object_id == city_object_id).delete()
session.commit()
else:
raise NotImplementedError('Missing either city_id or city_object_id')
except SQLAlchemyError as err:
logging.error('Error while deleting application: %s', err)
raise SQLAlchemyError from err
def _get_city(self, city_id) -> City:
"""
Fetch a city object based city id
:param city_id: a city identifier
:return: [City] with the provided city_id
"""
try:
with Session(self.engine) as session:
return session.execute(select(City).where(City.id == city_id)).first()
except SQLAlchemyError as err:
logging.error('Error while fetching city by city_id: %s', err)
raise SQLAlchemyError from err
def _get_city(self, city_id) -> City:
"""
Fetch a city object based city id
:param city_id: a city identifier
:return: [City] with the provided city_id
"""
try:
with Session(self.engine) as session:
return session.execute(select(City).where(City.id == city_id)).first()
except SQLAlchemyError as err:
logging.error('Error while fetching city by city_id: %s', err)
raise SQLAlchemyError from err
def _get_city_object(self, city_object_id) -> [CityObject]:
"""
Fetch a city object based city id
:param city_object_id: a city object identifier
:return: [CityObject] with the provided city_object_id
"""
try:
with Session(self.engine) as session:
return session.execute(select(CityObject).where(CityObject.id == city_object_id)).first()
except SQLAlchemyError as err:
logging.error('Error while fetching city by city_id: %s', err)
raise SQLAlchemyError from err
def _get_city_object(self, city_object_id) -> [CityObject]:
"""
Fetch a city object based city id
:param city_object_id: a city object identifier
:return: [CityObject] with the provided city_object_id
"""
try:
with Session(self.engine) as session:
return session.execute(select(CityObject).where(CityObject.id == city_object_id)).first()
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_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
: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(or_(
Model.city_id == city_id,
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
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
: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(or_(
Model.city_id == city_id,
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

264
cerc_persistence/repositories/user.py
View File

@ -17,143 +17,143 @@ from cerc_persistence.models import User as Model, Application as ApplicationMod
class User(Repository):
"""
User class
"""
_instance = None
def __init__(self, db_name: str, dotenv_path: str, app_env: str):
super().__init__(db_name, dotenv_path, app_env)
def __new__(cls, db_name, dotenv_path, app_env):
"""
User class
Implemented for a singleton pattern
"""
_instance = None
if cls._instance is None:
cls._instance = super(User, cls).__new__(cls)
return cls._instance
def __init__(self, db_name: str, dotenv_path: str, app_env: str):
super().__init__(db_name, dotenv_path, app_env)
def insert(self, name: str, password: str, role: UserRoles, application_id: int):
"""
Inserts a new user
:param name: username
:param password: user password
:param role: user rol [Admin or Hub_Reader]
:param application_id: user application id
:return: Identity id
"""
try:
user = self.get_by_name_and_application(name, application_id)
if user is not None:
raise SQLAlchemyError(f'A user named {user.name} already exists for that application')
except TypeError:
pass
try:
user = Model(name=name, password=Auth.hash_password(password), role=role, application_id=application_id)
with Session(self.engine) as session:
session.add(user)
session.flush()
session.commit()
session.refresh(user)
return user.id
except SQLAlchemyError as err:
logging.error('An error occurred while creating user %s', err)
raise SQLAlchemyError from err
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
def update(self, user_id: int, name: str, password: str, role: UserRoles):
"""
Updates a user
:param user_id: the id of the user to be updated
:param name: the name of the user
:param password: the password of the user
:param role: the role of the user
:return: None
"""
try:
with Session(self.engine) as session:
session.query(Model).filter(Model.id == user_id).update({
'name': name,
'password': Auth.hash_password(password),
'role': role,
'updated': datetime.datetime.utcnow()
})
session.commit()
except SQLAlchemyError as err:
logging.error('Error while updating user: %s', err)
raise SQLAlchemyError from err
def insert(self, name: str, password: str, role: UserRoles, application_id: int):
"""
Inserts a new user
:param name: username
:param password: user password
:param role: user rol [Admin or Hub_Reader]
:param application_id: user application id
:return: Identity id
"""
try:
user = self.get_by_name_and_application(name, application_id)
if user is not None:
raise SQLAlchemyError(f'A user named {user.name} already exists for that application')
except TypeError:
pass
try:
user = Model(name=name, password=Auth.hash_password(password), role=role, application_id=application_id)
with Session(self.engine) as session:
session.add(user)
session.flush()
session.commit()
session.refresh(user)
return user.id
except SQLAlchemyError as err:
logging.error('An error occurred while creating user %s', err)
raise SQLAlchemyError from err
def delete(self, user_id: int):
"""
Deletes a user with the id
:param user_id: the user id
:return: None
"""
try:
with Session(self.engine) as session:
session.query(Model).filter(Model.id == user_id).delete()
session.commit()
except SQLAlchemyError as err:
logging.error('Error while fetching user: %s', err)
raise SQLAlchemyError from err
def update(self, user_id: int, name: str, password: str, role: UserRoles):
"""
Updates a user
:param user_id: the id of the user to be updated
:param name: the name of the user
:param password: the password of the user
:param role: the role of the user
:return: None
"""
try:
with Session(self.engine) as session:
session.query(Model).filter(Model.id == user_id).update({
'name': name,
'password': Auth.hash_password(password),
'role': role,
'updated': datetime.datetime.utcnow()
})
session.commit()
except SQLAlchemyError as err:
logging.error('Error while updating user: %s', err)
raise SQLAlchemyError from err
def get_by_name_and_application(self, name: str, application_id: int) -> Model:
"""
Fetch user based on the email address
:param name: Username
:param application_id: User application name
:return: User matching the search criteria or None
"""
try:
with Session(self.engine) as session:
user = session.execute(
select(Model).where(Model.name == name, Model.application_id == application_id)
).first()
session.commit()
return user[0]
except SQLAlchemyError as err:
logging.error('Error while fetching user by name and application: %s', err)
raise SQLAlchemyError from err
except TypeError as err:
logging.error('Error while fetching user, empty result %s', err)
raise TypeError from err
def delete(self, user_id: int):
"""
Deletes a user with the id
:param user_id: the user id
:return: None
"""
try:
with Session(self.engine) as session:
session.query(Model).filter(Model.id == user_id).delete()
session.commit()
except SQLAlchemyError as err:
logging.error('Error while fetching user: %s', err)
raise SQLAlchemyError from err
def get_by_name_application_id_and_password(self, name: str, password: str, application_id: int) -> Model:
"""
Fetch user based on the name, password and application id
:param name: Username
:param password: User password
:param application_id: Application id
:return: User
"""
try:
with Session(self.engine) as session:
user = session.execute(
select(Model).where(Model.name == name, Model.application_id == application_id)
).first()
if user:
if Auth.check_password(password, user[0].password):
return user[0]
except SQLAlchemyError as err:
logging.error('Error while fetching user by name: %s', err)
raise SQLAlchemyError from err
raise ValueError('Unauthorized')
def get_by_name_and_application(self, name: str, application_id: int) -> Model:
"""
Fetch user based on the email address
:param name: Username
:param application_id: User application name
:return: User matching the search criteria or None
"""
try:
with Session(self.engine) as session:
user = session.execute(
select(Model).where(Model.name == name, Model.application_id == application_id)
).first()
session.commit()
return user[0]
except SQLAlchemyError as err:
logging.error('Error while fetching user by name and application: %s', err)
raise SQLAlchemyError from err
except TypeError as err:
logging.error('Error while fetching user, empty result %s', err)
raise TypeError from err
def get_by_name_application_id_and_password(self, name: str, password: str, application_id: int) -> Model:
"""
Fetch user based on the name, password and application id
:param name: Username
:param password: User password
:param application_id: Application id
:return: User
"""
try:
with Session(self.engine) as session:
user = session.execute(
select(Model).where(Model.name == name, Model.application_id == application_id)
).first()
if user:
if Auth.check_password(password, user[0].password):
return user[0]
except SQLAlchemyError as err:
logging.error('Error while fetching user by name: %s', err)
raise SQLAlchemyError from err
raise ValueError('Unauthorized')
def get_by_name_application_uuid_and_password(self, name: str, password: str, application_uuid: str) -> Model:
"""
Fetch user based on the email and password
:param name: Username
:param password: User password
:param application_uuid: Application uuid
:return: User
"""
try:
with Session(self.engine) as session:
application = session.execute(
select(ApplicationModel).where(ApplicationModel.application_uuid == application_uuid)
).first()
return self.get_by_name_application_id_and_password(name, password, application[0].id)
except SQLAlchemyError as err:
logging.error('Error while fetching user by name: %s', err)
raise SQLAlchemyError from err
except ValueError as err:
raise ValueError from err
def get_by_name_application_uuid_and_password(self, name: str, password: str, application_uuid: str) -> Model:
"""
Fetch user based on the email and password
:param name: Username
:param password: User password
:param application_uuid: Application uuid
:return: User
"""
try:
with Session(self.engine) as session:
application = session.execute(
select(ApplicationModel).where(ApplicationModel.application_uuid == application_uuid)
).first()
return self.get_by_name_application_id_and_password(name, password, application[0].id)
except SQLAlchemyError as err:
logging.error('Error while fetching user by name: %s', err)
raise SQLAlchemyError from err
except ValueError as err:
raise ValueError from err

18
cerc_persistence/repository.py
View File

@ -10,13 +10,13 @@ from cerc_persistence.configuration import Configuration
class Repository:
"""
Base repository class to establish db connection
"""
"""
Base repository class to establish db connection
"""
def __init__(self, db_name, dotenv_path: str, app_env='TEST'):
try:
self.configuration = Configuration(db_name, dotenv_path, app_env)
self.engine = create_engine(self.configuration.connection_string)
except ValueError as err:
logging.error('Missing value for credentials: %s', err)
def __init__(self, db_name, dotenv_path: str, app_env='TEST'):
try:
self.configuration = Configuration(db_name, dotenv_path, app_env)
self.engine = create_engine(self.configuration.connection_string)
except ValueError as err:
logging.error('Missing value for credentials: %s', err)

537
pylintrc Normal file
View File

@ -0,0 +1,537 @@
[MASTER]
# A comma-separated list of package or module names from where C extensions may
# be loaded. Extensions are loading into the active Python interpreter and may
# run arbitrary code.
extension-pkg-whitelist=
# Specify a score threshold to be exceeded before program exits with error.
fail-under=10
# Add files or directories to the blacklist. They should be base names, not
# paths.
ignore=CVS, conf
# Add files or directories matching the regex patterns to the blacklist. The
# regex matches against base names, not paths.
ignore-patterns=docs.source.conf
# Python code to execute, usually for sys.path manipulation such as
# pygtk.require().
init-hook='sys.path = list(); sys.path.append("./helpers/"); sys.path.append("../")'
# Use multiple processes to speed up Pylint. Specifying 0 will auto-detect the
# number of processors available to use.
jobs=1
# Control the amount of potential inferred values when inferring a single
# object. This can help the performance when dealing with large functions or
# complex, nested conditions.
limit-inference-results=100
# List of plugins (as comma separated values of python module names) to load,
# usually to register additional checkers.
load-plugins=
# Pickle collected data for later comparisons.
persistent=yes
# When enabled, pylint would attempt to guess common misconfiguration and emit
# user-friendly hints instead of false-positive error messages.
suggestion-mode=yes
# Allow loading of arbitrary C extensions. Extensions are imported into the
# active Python interpreter and may run arbitrary code.
unsafe-load-any-extension=no
[MESSAGES CONTROL]
# Only show warnings with the listed confidence levels. Leave empty to show
# all. Valid levels: HIGH, INFERENCE, INFERENCE_FAILURE, UNDEFINED.
confidence=
# Disable the message, report, category or checker with the given id(s). You
# can either give multiple identifiers separated by comma (,) or put this
# option multiple times (only on the command line, not in the configuration
# file where it should appear only once). You can also use "--disable=all" to
# disable everything first and then reenable specific checks. For example, if
# you want to run only the similarities checker, you can use "--disable=all
# --enable=similarities". If you want to run only the classes checker, but have
# no Warning level messages displayed, use "--disable=all --enable=classes
# --disable=W".
disable=raw-checker-failed,
bad-inline-option,
locally-disabled,
file-ignored,
suppressed-message,
useless-suppression,
deprecated-pragma,
use-symbolic-message-instead,
import-error,
parse-error,
syntax-error,
no-name-in-module
# Enable the message, report, category or checker with the given id(s). You can
# either give multiple identifier separated by comma (,) or put this option
# multiple time (only on the command line, not in the configuration file where
# it should appear only once). See also the "--disable" option for examples.
enable=c-extension-no-member
[REPORTS]
# Python expression which should return a score less than or equal to 10. You
# have access to the variables 'error', 'warning', 'refactor', and 'convention'
# which contain the number of messages in each category, as well as 'statement'
# which is the total number of statements analyzed. This score is used by the
# global evaluation report (RP0004).
evaluation=10.0 - ((float(5 * error + warning + refactor + convention) / statement) * 10)
# Template used to display messages. This is a python new-style format string
# used to format the message information. See doc for all details.
#msg-template=
# Set the output format. Available formats are text, parseable, colorized, json
# and msvs (visual studio). You can also give a reporter class, e.g.
# mypackage.mymodule.MyReporterClass.
output-format=text
# Tells whether to display a full report or only the messages.
reports=no
# Activate the evaluation score.
score=yes
[REFACTORING]
# Maximum number of nested blocks for function / method body
max-nested-blocks=5
# Complete name of functions that never returns. When checking for
# inconsistent-return-statements if a never returning function is called then
# it will be considered as an explicit return statement and no message will be
# printed.
never-returning-functions=sys.exit
[TYPECHECK]
# List of decorators that produce context managers, such as
# contextlib.contextmanager. Add to this list to register other decorators that
# produce valid context managers.
contextmanager-decorators=contextlib.contextmanager
# List of members which are set dynamically and missed by pylint inference
# system, and so shouldn't trigger E1101 when accessed. Python regular
# expressions are accepted.
generated-members=
# Tells whether missing members accessed in mixin class should be ignored. A
# mixin class is detected if its name ends with "mixin" (case insensitive).
ignore-mixin-members=yes
# Tells whether to warn about missing members when the owner of the attribute
# is inferred to be None.
ignore-none=yes
# This flag controls whether pylint should warn about no-member and similar
# checks whenever an opaque object is returned when inferring. The inference
# can return multiple potential results while evaluating a Python object, but
# some branches might not be evaluated, which results in partial inference. In
# that case, it might be useful to still emit no-member and other checks for
# the rest of the inferred objects.
ignore-on-opaque-inference=yes
# List of class names for which member attributes should not be checked (useful
# for classes with dynamically set attributes). This supports the use of
# qualified names.
ignored-classes=optparse.Values,thread._local,_thread._local
# List of module names for which member attributes should not be checked
# (useful for modules/projects where namespaces are manipulated during runtime
# and thus existing member attributes cannot be deduced by static analysis). It
# supports qualified module names, as well as Unix pattern matching.
ignored-modules=pyproj, reverse_geocoder, matplotlib, shapely, numpy
# Show a hint with possible names when a member name was not found. The aspect
# of finding the hint is based on edit distance.
missing-member-hint=yes
# The minimum edit distance a name should have in order to be considered a
# similar match for a missing member name.
missing-member-hint-distance=1
# The total number of similar names that should be taken in consideration when
# showing a hint for a missing member.
missing-member-max-choices=1
# List of decorators that change the signature of a decorated function.
signature-mutators=
[LOGGING]
# The type of string formatting that logging methods do. `old` means using %
# formatting, `new` is for `{}` formatting.
logging-format-style=old
# Logging modules to check that the string format arguments are in logging
# function parameter format.
logging-modules=logging
[BASIC]
# Naming style matching correct argument names.
argument-naming-style=snake_case
# Regular expression matching correct argument names. Overrides argument-
# naming-style.
#argument-rgx=
# Naming style matching correct attribute names.
attr-naming-style=snake_case
# Regular expression matching correct attribute names. Overrides attr-naming-
# style.
#attr-rgx=
# Bad variable names which should always be refused, separated by a comma.
bad-names=foo,
bar,
baz,
toto,
tutu,
tata
# Bad variable names regexes, separated by a comma. If names match any regex,
# they will always be refused
bad-names-rgxs=
# Naming style matching correct class attribute names.
class-attribute-naming-style=any
# Regular expression matching correct class attribute names. Overrides class-
# attribute-naming-style.
#class-attribute-rgx=
# Naming style matching correct class names.
class-naming-style=PascalCase
# Regular expression matching correct class names. Overrides class-naming-
# style.
#class-rgx=
# Naming style matching correct constant names.
const-naming-style=UPPER_CASE
# Regular expression matching correct constant names. Overrides const-naming-
# style.
#const-rgx=
# Minimum line length for functions/classes that require docstrings, shorter
# ones are exempt.
docstring-min-length=-1
# Naming style matching correct function names.
function-naming-style=snake_case
# Regular expression matching correct function names. Overrides function-
# naming-style.
#function-rgx=
# Good variable names which should always be accepted, separated by a comma.
good-names=i,
j,
k,
ex,
Run,
x,
y,
z,
s,
df,
id,
he,
hi,
setUp,
_
# Good variable names regexes, separated by a comma. If names match any regex,
# they will always be accepted
good-names-rgxs=
# Include a hint for the correct naming format with invalid-name.
include-naming-hint=no
# Naming style matching correct inline iteration names.
inlinevar-naming-style=any
# Regular expression matching correct inline iteration names. Overrides
# inlinevar-naming-style.
#inlinevar-rgx=
# Naming style matching correct method names.
method-naming-style=snake_case
# Regular expression matching correct method names. Overrides method-naming-
# style.
#method-rgx=
# Naming style matching correct module names.
module-naming-style=snake_case
# Regular expression matching correct module names. Overrides module-naming-
# style.
#module-rgx=
# Colon-delimited sets of names that determine each other's naming style when
# the name regexes allow several styles.
name-group=
# Regular expression which should only match function or class names that do
# not require a docstring.
no-docstring-rgx=^_
# List of decorators that produce properties, such as abc.abstractproperty. Add
# to this list to register other decorators that produce valid properties.
# These decorators are taken in consideration only for invalid-name.
property-classes=abc.abstractproperty
# Naming style matching correct variable names.
variable-naming-style=snake_case
# Regular expression matching correct variable names. Overrides variable-
# naming-style.
#variable-rgx=
[STRING]
# This flag controls whether inconsistent-quotes generates a warning when the
# character used as a quote delimiter is used inconsistently within a module.
check-quote-consistency=no
# This flag controls whether the implicit-str-concat should generate a warning
# on implicit string concatenation in sequences defined over several lines.
check-str-concat-over-line-jumps=no
[SIMILARITIES]
# Ignore comments when computing similarities.
ignore-comments=yes
# Ignore docstrings when computing similarities.
ignore-docstrings=yes
# Ignore imports when computing similarities.
ignore-imports=no
# Minimum lines number of a similarity.
min-similarity-lines=4
[VARIABLES]
# List of additional names supposed to be defined in builtins. Remember that
# you should avoid defining new builtins when possible.
additional-builtins=
# Tells whether unused global variables should be treated as a violation.
allow-global-unused-variables=yes
# List of strings which can identify a callback function by name. A callback
# name must start or end with one of those strings.
callbacks=cb_,
_cb
# A regular expression matching the name of dummy variables (i.e. expected to
# not be used).
dummy-variables-rgx=_+$|(_[a-zA-Z0-9_]*[a-zA-Z0-9]+?$)|dummy|^ignored_|^unused_
# Argument names that match this expression will be ignored. Default to name
# with leading underscore.
ignored-argument-names=_.*|^ignored_|^unused_
# Tells whether we should check for unused import in __init__ files.
init-import=no
# List of qualified module names which can have objects that can redefine
# builtins.
redefining-builtins-modules=six.moves,past.builtins,future.builtins,builtins,io
[SPELLING]
# Limits count of emitted suggestions for spelling mistakes.
max-spelling-suggestions=4
# Spelling dictionary name. Available dictionaries: none. To make it work,
# install the python-enchant package.
spelling-dict=
# List of comma separated words that should not be checked.
spelling-ignore-words=
# A path to a file that contains the private dictionary; one word per line.
spelling-private-dict-file=
# Tells whether to store unknown words to the private dictionary (see the
# --spelling-private-dict-file option) instead of raising a message.
spelling-store-unknown-words=no
[FORMAT]
# Expected format of line ending, e.g. empty (any line ending), LF or CRLF.
expected-line-ending-format=
# Regexp for a line that is allowed to be longer than the limit.
ignore-long-lines=^\s*(# )?<?https?://\S+>?$
# Number of spaces of indent required inside a hanging or continued line.
indent-after-paren=4
# String used as indentation unit. This is usually " " (4 spaces) or "\t" (1
# tab).
indent-string=' '
# Maximum number of characters on a single line.
max-line-length=120
# Maximum number of lines in a module.
max-module-lines=1000
# List of optional constructs for which whitespace checking is disabled. `dict-
# separator` is used to allow tabulation in dicts, etc.: {1 : 1,\n222: 2}.
# `trailing-comma` allows a space between comma and closing bracket: (a, ).
# `empty-line` allows space-only lines.
no-space-check=trailing-comma,
dict-separator
# Allow the body of a class to be on the same line as the declaration if body
# contains single statement.
single-line-class-stmt=no
# Allow the body of an if to be on the same line as the test if there is no
# else.
single-line-if-stmt=no
[MISCELLANEOUS]
# List of note tags to take in consideration, separated by a comma.
notes=FIXME,
XXX,
TODO
# Regular expression of note tags to take in consideration.
#notes-rgx=
[IMPORTS]
# List of modules that can be imported at any level, not just the top level
# one.
allow-any-import-level=
# Allow wildcard imports from modules that define __all__.
allow-wildcard-with-all=no
# Analyse import fallback blocks. This can be used to support both Python 2 and
# 3 compatible code, which means that the block might have code that exists
# only in one or another interpreter, leading to false positives when analysed.
analyse-fallback-blocks=no
# Deprecated modules which should not be used, separated by a comma.
deprecated-modules=optparse,tkinter.tix
# Create a graph of external dependencies in the given file (report RP0402 must
# not be disabled).
ext-import-graph=
# Create a graph of every (i.e. internal and external) dependencies in the
# given file (report RP0402 must not be disabled).
import-graph=
# Create a graph of internal dependencies in the given file (report RP0402 must
# not be disabled).
int-import-graph=
# Force import order to recognize a module as part of the standard
# compatibility libraries.
known-standard-library=
# Force import order to recognize a module as part of a third party library.
known-third-party=enchant
# Couples of modules and preferred modules, separated by a comma.
preferred-modules=
[CLASSES]
# List of method names used to declare (i.e. assign) instance attributes.
defining-attr-methods=__init__,
__new__,
setUp,
__post_init__
# List of member names, which should be excluded from the protected access
# warning.
exclude-protected=_asdict,
_fields,
_replace,
_source,
_make
# List of valid names for the first argument in a class method.
valid-classmethod-first-arg=cls
# List of valid names for the first argument in a metaclass class method.
valid-metaclass-classmethod-first-arg=cls
[DESIGN]
# Maximum number of arguments for function / method.
max-args=5
# Maximum number of attributes for a class (see R0902).
max-attributes=7
# Maximum number of boolean expressions in an if statement (see R0916).
max-bool-expr=5
# Maximum number of branch for function / method body.
max-branches=12
# Maximum number of locals for function / method body.
max-locals=15
# Maximum number of parents for a class (see R0901).
max-parents=7
# Maximum number of public methods for a class (see R0904).
max-public-methods=20
# Maximum number of return / yield for function / method body.
max-returns=6
# Maximum number of statements in function / method body.
max-statements=50
# Minimum number of public methods for a class (see R0903).
min-public-methods=2
[EXCEPTIONS]
# Exceptions that will emit a warning when being caught. Defaults to
# "BaseException, Exception".
overgeneral-exceptions=BaseException,
Exception

48
setup.py
View File

@ -5,35 +5,35 @@ from distutils.util import convert_path
from setuptools import setup
with pathlib.Path('requirements.txt').open() as r:
install_requires = [
str(requirement).replace('\n', '')
for requirement
in r.readlines()
]
install_requires = [
str(requirement).replace('\n', '')
for requirement
in r.readlines()
]
install_requires.append('setuptools')
main_ns = {}
version = convert_path('cerc_persistence/version.py')
with open(version) as f:
exec(f.read(), main_ns)
exec(f.read(), main_ns)
setup(
name='cerc-persistence',
version=main_ns['__version__'],
description="",
long_description="",
classifiers=[
"License :: OSI Approved :: GNU Library or Lesser General Public License (LGPL)",
"Programming Language :: Python",
"Programming Language :: Python :: 3",
],
include_package_data=True,
packages=[
'cerc_persistence',
'cerc_persistence.models',
'cerc_persistence.repositories'
],
setup_requires=install_requires,
install_requires=install_requires,
data_files=[],
name='cerc-persistence',
version=main_ns['__version__'],
description="",
long_description="",
classifiers=[
"License :: OSI Approved :: GNU Library or Lesser General Public License (LGPL)",
"Programming Language :: Python",
"Programming Language :: Python :: 3",
],
include_package_data=True,
packages=[
'cerc_persistence',
'cerc_persistence.models',
'cerc_persistence.repositories'
],
setup_requires=install_requires,
install_requires=install_requires,
data_files=[],
)

460
tests/test_db_factory.py
View File

@ -35,266 +35,266 @@ from cerc_persistence.repository import Repository
class Control:
_skip_test = False
_skip_reason = 'PostgreSQL not properly installed in host machine'
_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
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)
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
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()
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()
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._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
self._application_uuid = 'b9e0ce80-1218-410c-8a64-9d9b7026aad8'
self._application_id = 1
self._user_id = 1
self._application_id = self._database.persist_application(
'test',
'test',
self.application_uuid
)
self._user_id = self._database.create_user('test', self._application_id, 'test', UserRoles.Admin)
self._application_id = self._database.persist_application(
'test',
'test',
self.application_uuid
)
self._user_id = self._database.create_user('test', self._application_id, 'test', UserRoles.Admin)
self._pickle_path = Path('tests_data/pickle_path.bz2').resolve()
self._pickle_path = Path('tests_data/pickle_path.bz2').resolve()
@property
def database(self):
return self._database
@property
def database(self):
return self._database
@property
def application_uuid(self):
return self._application_uuid
@property
def application_uuid(self):
return self._application_uuid
@property
def application_id(self):
return self._application_id
@property
def application_id(self):
return self._application_id
@property
def user_id(self):
return self._user_id
@property
def user_id(self):
return self._user_id
@property
def skip_test(self):
return self._skip_test
@property
def skip_test(self):
return self._skip_test
@property
def insel(self):
return distutils.spawn.find_executable('insel')
@property
def insel(self):
return distutils.spawn.find_executable('insel')
@property
def sra(self):
return distutils.spawn.find_executable('sra')
@property
def sra(self):
return distutils.spawn.find_executable('sra')
@property
def skip_insel_test(self):
return self.insel is None
@property
def skip_insel_test(self):
return self.insel is None
@property
def skip_reason(self):
return self._skip_reason
@property
def skip_reason(self):
return self._skip_reason
@property
def message(self):
return self._skip_reason
@property
def message(self):
return self._skip_reason
@property
def city(self):
return self._city
@property
def city(self):
return self._city
@property
def pickle_path(self):
return self._pickle_path
@property
def pickle_path(self):
return self._pickle_path
control = Control()
class TestDBFactory(TestCase):
"""
TestDBFactory
"""
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_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)
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}
]})
@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)
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)
@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)

182
tests/test_db_retrieve.py
View File

@ -19,117 +19,117 @@ from cerc_persistence.repository import Repository
class Control:
_skip_test = False
_skip_reason = 'PostgreSQL not properly installed in host machine'
_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
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._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'
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 database(self):
return self._database
@property
def application_uuid(self):
return self._application_uuid
@property
def application_uuid(self):
return self._application_uuid
@property
def application_id(self):
return self._application_id
@property
def application_id(self):
return self._application_id
@property
def user_id(self):
return self._user_id
@property
def user_id(self):
return self._user_id
@property
def skip_test(self):
return self._skip_test
@property
def skip_test(self):
return self._skip_test
@property
def insel(self):
return distutils.spawn.find_executable('insel')
@property
def insel(self):
return distutils.spawn.find_executable('insel')
@property
def sra(self):
return distutils.spawn.find_executable('sra')
@property
def sra(self):
return distutils.spawn.find_executable('sra')
@property
def skip_insel_test(self):
return self.insel is None
@property
def skip_insel_test(self):
return self.insel is None
@property
def skip_reason(self):
return self._skip_reason
@property
def skip_reason(self):
return self._skip_reason
@property
def message(self):
return self._skip_reason
@property
def message(self):
return self._skip_reason
@property
def pickle_path(self):
return self._pickle_path
@property
def pickle_path(self):
return self._pickle_path
control = Control()
class TestDBFactory(TestCase):
"""
TestDBFactory
"""
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"]
}
]
@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)
]
}
results = control.database.results(control.user_id, control.application_id, request_values)
print(results)