mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
102 lines
3.2 KiB
Python
102 lines
3.2 KiB
Python
from sqlalchemy.orm import sessionmaker
|
|
|
|
from backend.app.db.connection import db_engine
|
|
from backend.app.db.models.portfolio import PropertyModel, PropertyDetailsEpcModel
|
|
from backend.app.db.models.recommendations import Recommendation, Plan, PlanRecommendations, Scenario
|
|
|
|
|
|
class Outputs:
|
|
FORMATS = ["mds"]
|
|
|
|
def __init__(self, format, portfolio_id):
|
|
"""
|
|
This class handles the creation of standard outputs for the backend. For example, creation of
|
|
an excel output, to be used for the MDS data sheet, required by E.ON
|
|
|
|
:param format: The format of the output, e.g. mds
|
|
:param portfolio_id: The id of the portfolio for which the output is being created
|
|
"""
|
|
|
|
if format not in self.FORMATS:
|
|
raise ValueError("Invalid format, should be one of {}".format(self.FORMATS))
|
|
|
|
self.format = format
|
|
self.portfolio_id = portfolio_id
|
|
|
|
# Connect to the database
|
|
self.session = sessionmaker(bind=db_engine)()
|
|
|
|
def get_properties_from_db(self):
|
|
# Get properties and their details for a specific portfolio
|
|
self.session.begin()
|
|
properties_query = self.session.query(
|
|
PropertyModel,
|
|
PropertyDetailsEpcModel
|
|
).join(
|
|
PropertyDetailsEpcModel,
|
|
PropertyModel.id == PropertyDetailsEpcModel.property_id
|
|
).filter(
|
|
PropertyModel.portfolio_id == self.portfolio_id # Filter by portfolio ID
|
|
).all()
|
|
|
|
# Transform properties data to include all fields dynamically
|
|
properties_data = [
|
|
{**{col.name: getattr(prop.PropertyModel, col.name) for col in PropertyModel.__table__.columns},
|
|
**{col.name: getattr(prop.PropertyDetailsEpcModel, col.name) for col in
|
|
PropertyDetailsEpcModel.__table__.columns}}
|
|
for prop in properties_query
|
|
]
|
|
|
|
self.session.close()
|
|
return properties_data
|
|
|
|
def get_plans_from_db(self):
|
|
|
|
self.session.begin()
|
|
|
|
plans_query = self.session.query(Plan).all()
|
|
# Transform plans data to include all fields dynamically
|
|
plans_data = [
|
|
{col.name: getattr(plan, col.name) for col in Plan.__table__.columns}
|
|
for plan in plans_query
|
|
]
|
|
|
|
self.session.close()
|
|
return plans_data
|
|
|
|
def export_mds(self):
|
|
"""
|
|
This function will export the data in the MDS format
|
|
Core data required:
|
|
- Property address
|
|
- Property postcode
|
|
- uprn
|
|
- recommended measures
|
|
- pre-EPC
|
|
- pre-SAP
|
|
- pre Heat Demand
|
|
- Property Type
|
|
- Built form
|
|
- Wall type
|
|
- Tenure
|
|
- Fuel type
|
|
- Estimated bill
|
|
- Recommended measures
|
|
- Post EPC
|
|
- Post heat demand
|
|
- Bill savings
|
|
- Kwh savings
|
|
"""
|
|
|
|
properties_data = self.get_properties_from_db()
|
|
|
|
plans_data = self.get_plans_from_db()
|
|
|
|
plan_ids = [plan['id'] for plan in plans_data]
|
|
|
|
def export(self):
|
|
"""
|
|
This function will export the data in the required format
|
|
"""
|
|
if self.format == "mds":
|
|
self.export_mds()
|