s_ranjbar/system_assignation
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Added repo methods to fetch cities and buildings

Browse Source
This commit is contained in:
Peter Yefi 2022-11-08 21:03:48 -05:00
parent ffb4f8b6d8
commit 04c2c07c6d
4 changed files with 112 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
helpers/city_util.py
View File

@ -49,7 +49,6 @@ class CityUtil:
serialized_points.append(point)
return serialized_points
def extract_building_data(self, building: Building) -> Dict:
"""
Extracts various values from a building
@ -65,30 +64,33 @@ class CityUtil:
dict_building[key] = []
elif len(value) > 0:
if type(value[0]) is Surface:
surfaces = []
for surface in value:
surface_dict = vars(surface)
perimeter_polygon = self._class_object_to_dict(surface_dict['_perimeter_polygon'])
solid_polygon = self._class_object_to_dict(surface_dict['_solid_polygon'])
holes_polygon = self._class_object_to_dict(surface_dict['_holes_polygons'])
try:
surfaces = []
for surface in value:
surface_dict = vars(surface)
perimeter_polygon = self._class_object_to_dict(surface_dict['_perimeter_polygon'])
solid_polygon = self._class_object_to_dict(surface_dict['_solid_polygon'])
holes_polygon = self._class_object_to_dict(surface_dict['_holes_polygons'])
if perimeter_polygon is not None:
perimeter_polygon['_coordinates'] = self._ndarray_to_list('_coordinates', perimeter_polygon)
perimeter_polygon['_points'] = self._serialize_points(perimeter_polygon['_points'])
surface_dict['_perimeter_polygon'] = perimeter_polygon
if perimeter_polygon is not None:
perimeter_polygon['_coordinates'] = self._ndarray_to_list('_coordinates', perimeter_polygon)
perimeter_polygon['_points'] = self._serialize_points(perimeter_polygon['_points'])
surface_dict['_perimeter_polygon'] = perimeter_polygon
if holes_polygon is not None:
holes_polygon['_coordinates'] = self._ndarray_to_list('_coordinates', holes_polygon)
holes_polygon['_points'] = self._serialize_points(holes_polygon['_points'])
surface_dict['_holes_polygons'] = holes_polygon
if holes_polygon is not None:
holes_polygon['_coordinates'] = self._ndarray_to_list('_coordinates', holes_polygon)
holes_polygon['_points'] = self._serialize_points(holes_polygon['_points'])
surface_dict['_holes_polygons'] = holes_polygon
if solid_polygon is not None:
solid_polygon['_coordinates'] = self._ndarray_to_list('_coordinates', solid_polygon)
solid_polygon['_points'] = self._serialize_points(solid_polygon['_points'])
surface_dict['_solid_polygon'] = solid_polygon
if solid_polygon is not None:
solid_polygon['_coordinates'] = self._ndarray_to_list('_coordinates', solid_polygon)
solid_polygon['_points'] = self._serialize_points(solid_polygon['_points'])
surface_dict['_solid_polygon'] = solid_polygon
surfaces.append(surface_dict)
dict_building[key] = surfaces
surfaces.append(surface_dict)
dict_building[key] = surfaces
except KeyError as err:
print(f'Dictionary key error: {err}')
elif value is None:
dict_building[key] = None
elif type(value) in [str, int, dict, np.float64]:

3
persistence/models/City.py
View File

@ -24,10 +24,9 @@ class City(Base):
"""
__tablename__ = "city"
id = Column(Integer, Sequence('city_id_seq'), primary_key=True)
uid = Column(String, nullable=False, unique=True)
name = Column(String, nullable=False)
time_zone = Column(String, nullable=True)
country_code = Column(String, nullable=False)
latitude = Column(Float)
longitude = Column(Float)
building = relationship('Building', backref='city', lazy=True, cascade="all, delete-orphan")
buildings = relationship('Building', backref='city', lazy=True, cascade="all, delete-orphan")

65
persistence/repositories/BuildingRepo.py
View File

@ -5,12 +5,12 @@ Copyright © 2022 Concordia CERC group
Project Coder Peter Yefi peteryefi@gmail.com
"""
import json
from city_model_structure.building import Building as CityBuilding
from persistence.models import City
from sqlalchemy.exc import SQLAlchemyError
from persistence.models import Building
from persistence import BaseRepo
from helpers.city_util import CityUtil
from sqlalchemy import select
class BuildingRepo(BaseRepo):
@ -26,6 +26,7 @@ class BuildingRepo(BaseRepo):
:param city_id: the city the building belongs to
:return:
"""
model_building = Building()
model_building.name = building.name
model_building.function = building.function
@ -33,13 +34,59 @@ class BuildingRepo(BaseRepo):
model_building.floor_area = building.floor_area
model_building.city_id = city_id
model_building.data = self._city_util.extract_building_data(building)
self.session.add(model_building)
self.session.flush()
self.session.commit()
return model_building
try:
self.session.add(model_building)
self.session.flush()
self.session.commit()
return model_building
except SQLAlchemyError as err:
print(f'Error while adding building: {err}')
def get_by_id(self, building_id: int):
return self.session.query(Building).filter(Building.id == building_id)
def get_by_id(self, building_id: int) -> Building:
"""
Fetch a building based on the id
:param building_id: the building id
:return: a Building
"""
try:
return self.session.execute(select(Building).where(Building.id == building_id)).first()[0]
except SQLAlchemyError as err:
print(f'Error while fetching building: {err}')
def get_by_name(self, building_name: str) -> [Building]:
"""
Fetch a building based on the name
:param building_name: the name of the building
:return: [Building] with the provided name
"""
try:
result_set = self.session.execute(select(Building).where(Building.name == building_name))
return [building[0] for building in result_set]
except SQLAlchemyError as err:
print(f'Error while fetching buildings: {err}')
def get_by_construction_year(self, year: int):
"""
Fetch a building based on the year of construction
:param year: the construction year of the building
:return: [Building]
"""
try:
result_set = self.session.execute(select(Building).where(Building.construction_year == year))
return [building[0] for building in result_set]
except SQLAlchemyError as err:
print(f'Error while fetching buildings: {err}')
def get_by_city(self, city_id: int):
return self.session.query(Building).filter(Building.city_id == city_id)
"""
Get all the buildings in a city
:param city_id: the city id
:return: [Building]
"""
try:
result_set = self.session.execute(select(Building).where(Building.city_id == city_id))
return [building[0] for building in result_set]
except SQLAlchemyError as err:
print(f'Error while fetching buildings: {err}')

37
persistence/repositories/CityRepo.py
View File

@ -8,6 +8,8 @@ Project Coder Peter Yefi peteryefi@gmail.com
from city_model_structure.city import City
from persistence.models import City as ModelCity
from persistence import BaseRepo
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy import select
class CityRepo(BaseRepo):
@ -16,13 +18,38 @@ class CityRepo(BaseRepo):
def insert(self, city: City):
model_city = ModelCity()
model_city.uid = city.uid
model_city.name = city.name
model_city.longitude = city.longitude
model_city.latitude = city.latitude
model_city.country_code = city.country_code
model_city.time_zone = city.time_zone
self.session.add(model_city)
self.session.flush()
self.session.commit()
return model_city
try:
self.session.add(model_city)
self.session.flush()
self.session.commit()
return model_city
except SQLAlchemyError as err:
print(f'Error while adding city: {err}')
def get_by_id(self, city_id: int) -> ModelCity:
"""
Fetch a City based on the id
:param city_id: the city id
:return: a city
"""
try:
return self.session.execute(select(ModelCity).where(ModelCity.id == city_id)).first()[0]
except SQLAlchemyError as err:
print(f'Error while fetching city: {err}')
def get_by_name(self, city_name: str) -> [ModelCity]:
"""
Fetch city based on the name
:param city_name: the name of the building
:return: [ModelCity] with the provided name
"""
try:
result_set = self.session.execute(select(ModelCity).where(ModelCity.name == city_name))
return [building[0] for building in result_set]
except SQLAlchemyError as err:
print(f'Error while fetching city by name: {err}')