junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-08 11:17:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

updating EpcClean to use inside of the backend lambda and pushing property targets to database

Browse source
This commit is contained in:
Khalim Conn-Kowlessar 2023-08-01 14:26:49 +01:00
parent 6653ae9fbb
commit 75a358ff4c
6 changed files with 140 additions and 3678 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
backend/Property.py
View file

@ -44,6 +44,9 @@ class Property(BaseUtility):
self.solar_pv = None
self.solar_hot_water = None
self.wind_turbine = None
self.number_of_open_fireplaces = None
self.number_of_extensions = None
self.number_of_storeys = None
if epc_client:
self.epc_client = epc_client
@ -181,6 +184,30 @@ class Property(BaseUtility):
"wind_turbine": wind_turbine_count,
}
def set_property_counts(self):
"""
For EPC fields that are just counts, we'll set them here
These are fields that are integers but may contain additional values such as "" so we can't do a direct
conversion straight to an integer
:return:
"""
fields = {
"number_of_open_fireplaces": "number-open-fireplaces",
"number_of_extensions": "extension-count",
"number_of_storeys": "flat-storey-count",
}
for attribute, epc_field in fields.items():
value = self.data["extension-count"]
if value == "" or value in self.DATA_ANOMALY_MATCHES:
value = 0
else:
value = int(value)
setattr(self, attribute, value)
def get_components(self, cleaned):
"""
Given the cleaning that has been performed, we'll use this to identify the property
@ -200,11 +227,16 @@ class Property(BaseUtility):
self.set_solar_pv()
self.set_solar_hot_water()
self.set_wind_turbine()
self.set_property_counts()
for description, attribute in cleaned.items():
if self.data[description] in self.DATA_ANOMALY_MATCHES:
setattr(self, self.ATTRIBUTE_MAP[description], {"original_description": self.data[description]})
setattr(
self,
self.ATTRIBUTE_MAP[description],
{"original_description": self.data[description], "clean_description": self.data[description]}
)
continue
attributes = [

27
backend/app/db/functions/property_functions.py
View file

@ -3,7 +3,7 @@
###
import datetime
from sqlalchemy.orm import sessionmaker
from backend.app.db.models.portfolio import PropertyModel, PropertyCreationStatus, PortfolioStatus
from backend.app.db.models.portfolio import PropertyModel, PropertyCreationStatus, PortfolioStatus, PropertyTargetsModel
from backend.app.db.connection import db_engine
from sqlalchemy.orm.exc import NoResultFound
@ -57,3 +57,28 @@ def create_property(portfolio_id: int, address: str, postcode: str) -> (int, boo
session.commit()
return new_property.id, True
def create_property_targets(property_id: int, portfolio_id: int, epc_target=None, heat_demand_target=None):
"""
This function will create a record for the property targets in the database if it does not exist.
:param property_id: The ID of the property the targets belong to
:param portfolio_id: The ID of the portfolio the property belongs to
:param epc_target: Goal EPC value for the property
:param heat_demand_target: Heat demand target for the property in kwh/m^2/year
:return:
"""
Session = sessionmaker(bind=db_engine)
now = datetime.datetime.now()
with Session() as session:
new_target = PropertyTargetsModel(
property_id=property_id,
portfolio_id=portfolio_id,
created_at=now,
epc=epc_target,
heat_demand=heat_demand_target
)
session.add(new_target)
session.commit()
return True

2
backend/app/db/models/portfolio.py
View file

@ -143,7 +143,7 @@ class PropertyDetailsMeter(Base):
meter_reading_gas = Column(Float)
class PropertyTargets(Base):
class PropertyTargetsModel(Base):
__tablename__ = 'property_targets'
id = Column(Integer, primary_key=True, autoincrement=True)
property_id = Column(Integer, ForeignKey('property.id'), nullable=False)

84
backend/app/plan/router.py
View file

@ -10,12 +10,12 @@ from utils.logger import setup_logger
from recommendations.FloorRecommendations import FloorRecommendations
from recommendations.WallRecommendations import WallRecommendations
from utils.uvalue_estimates import classify_decile_newvalues
from model_data.EpcClean import EpcClean
# database interaction functions
from backend.app.db.functions.property_functions import create_property
from backend.app.db.functions.property_functions import create_property, create_property_targets
# TODO: This is placeholder until data is stored in DB
from backend.app.plan.temp_cleaned_data import cleaned
from backend.app.plan.uvalue_estimates_walls import uvalue_estimates_walls
from backend.app.plan.uvalue_estimates_floors import uvalue_estimates_floors
@ -94,14 +94,12 @@ async def trigger_plan(body: PlanTriggerRequest):
if not is_new:
continue
# TODO: push property targets
# TODO: Need to add heat demand target
property_targets = {
"property_id": property_id,
"portfolio_id": body.portfolio_id,
"created_at": datetime.datetime.now(),
"epc": body.goal_value,
}
create_property_targets(
property_id=property_id,
portfolio_id=body.portfolio_id,
epc_target=body.goal_value,
)
input_properties.append(
Property(
@ -130,6 +128,9 @@ async def trigger_plan(body: PlanTriggerRequest):
)
p.set_is_in_conservation_area(in_conservation_area)
cleaner = EpcClean(data=[x.data for x in input_properties])
cleaner.clean()
logger.info("Getting components and properties recommendations")
recommendations = []
for property_id, p in enumerate(input_properties):
@ -141,7 +142,7 @@ async def trigger_plan(body: PlanTriggerRequest):
)[0]
# Property recommendations
p.get_components(cleaned)
p.get_components(cleaner.cleaned)
# This is placeholder, until the full dataset is loaded into the database and we just make a read to the
# database
@ -228,37 +229,40 @@ async def trigger_plan(body: PlanTriggerRequest):
property_data = clean_upload_data(property_data, to_clean_values=p.DATA_ANOMALY_MATCHES)
rating_lookup = {
"Very Good": 5,
"Good": 4,
"Average": 3,
"Poor": 2,
"Very Poor": 1,
"N/A": None
}
def prepare_rating(field):
rating_lookup = {
"Very Good": 5,
"Good": 4,
"Average": 3,
"Poor": 2,
"Very Poor": 1,
"N/A": None,
}
return rating_lookup[field] if field not in p.DATA_ANOMALY_MATCHES else None
property_details_epc = {
"property_id": p.id,
"portfolio_id": body.portfolio_id,
"full_address": p.data["address"],
"total_floor_area": float(p.data["total-floor-area"]),
"walls": p.walls["cleaned_description"],
"walls_rating": rating_lookup[p.data["walls-energy-eff"]],
"roof": p.roof["cleaned_description"],
"roof_rating": rating_lookup[p.data["roof-energy-eff"]],
"floor": p.floor["cleaned_description"],
"floor_rating": rating_lookup[p.data["floor-energy-eff"]],
"windows": p.windows["cleaned_description"],
"windows_rating": rating_lookup[p.data["windows-energy-eff"]],
"heating": p.main_heating["cleaned_description"],
"heating_rating": rating_lookup[p.data["mainheat-energy-eff"]],
"heating_controls": p.main_heating_controls["cleaned_description"],
"heating_controls_rating": rating_lookup[p.data["mainheatc-energy-eff"]],
"hot_water": p.hotwater["cleaned_description"],
"hot_water_rating": rating_lookup[p.data["hot-water-energy-eff"]],
"lighting": p.lighting["cleaned_description"],
"lighting_rating": rating_lookup[p.data["lighting-energy-eff"]],
"mainfuel": p.main_fuel["cleaned_description"],
"walls": p.walls["clean_description"],
"walls_rating": prepare_rating(p.data["walls-energy-eff"]),
"roof": p.roof["clean_description"],
"roof_rating": prepare_rating(p.data["roof-energy-eff"]),
"floor": p.floor["clean_description"],
"floor_rating": prepare_rating(p.data["floor-energy-eff"]),
"windows": p.windows["clean_description"],
"windows_rating": prepare_rating(p.data["windows-energy-eff"]),
"heating": p.main_heating["clean_description"],
"heating_rating": prepare_rating(p.data["mainheat-energy-eff"]),
"heating_controls": p.main_heating_controls["clean_description"],
"heating_controls_rating": prepare_rating(p.data["mainheatc-energy-eff"]),
"hot_water": p.hotwater["clean_description"],
"hot_water_rating": prepare_rating(p.data["hot-water-energy-eff"]),
"lighting": p.lighting["clean_description"],
"lighting_rating": prepare_rating(p.data["lighting-energy-eff"]),
"mainfuel": p.main_fuel["clean_description"],
"ventilation": p.ventilation["ventilation"],
"solar_pv": p.solar_pv["solar_pv"],
"solar_hot_water": p.solar_hot_water["solar_hot_water"],
@ -266,13 +270,13 @@ async def trigger_plan(body: PlanTriggerRequest):
"floor_height": p.data["floor-height"],
"heat_loss_corridor": p.data["heat-loss-corridor"],
"unheated_corridor_length": p.data["unheated-corridor-length"],
"number_of_open_fireplaces": int(p.data["number-open-fireplaces"]),
"number_of_extensions": int(p.data["extension-count"]),
"number_of_storeys": int(p.data["flat-storey-count"]),
"number_of_open_fireplaces": p.number_of_open_fireplaces,
"number_of_extensions": p.number_of_extensions,
"number_of_storeys": p.number_of_storeys,
"mains_gas": p.data["mains-gas-flag"],
"energy_tarrif": p.data["energy-tariff"],
"primary_energy_consumption": p.energy["primary-energy-consumption"],
"co2_emissions": p.energy["co2-emissions"],
"primary_energy_consumption": p.energy["primary_energy_consumption"],
"co2_emissions": p.energy["co2_emissions"],
}
return {"recommendations": recommendations}

3615
backend/app/plan/temp_cleaned_data.py
View file

File diff suppressed because it is too large Load diff

56
model_data/EpcClean.py
View file

@ -1,7 +1,6 @@
from typing import List, Dict, Any
from collections import Counter
import pandas as pd
from collections import defaultdict
from model_data.utils import correct_spelling
from model_data.epc_attributes.FloorAttributes import FloorAttributes
@ -47,29 +46,43 @@ class EpcClean:
def _calculate_lighting_averages(self):
"""
This is a simple utility function that for few textual lighting descritpions, will calculate the average
This is a simple utility function that for few textual lighting descriptions, will calculate the average
low energy lighting proportion. This is only valid for a very tiny number of cases and so a very simple
methodology is applied
:return: Dataframe of avergages for the corresponding descriptions
This is done without pandas so we can utilise this inside of our lambdas
:return: list of avergages for the corresponding descriptions
"""
df = pd.DataFrame(self.data)
aggs = df[
df["lighting-description"].isin(
[
'Below average lighting efficiency',
'Good lighting efficiency',
'Excelent lighting efficiency'
]
)
].copy()
aggs["low-energy-lighting"] = aggs["low-energy-lighting"].astype(float)
data = self.data
averages = aggs.groupby("lighting-description")["low-energy-lighting"].mean().reset_index()
averages["lighting-description"] = averages["lighting-description"].str.lower()
# Filter rows with the specified lighting descriptions
filtered_data = [
row for row in data if row["lighting-description"] in [
'Below average lighting efficiency',
'Good lighting efficiency',
'Excelent lighting efficiency'
]
]
# Correct spelling mistakes in averages
averages["lighting-description"] = averages["lighting-description"].apply(correct_spelling)
# Convert low-energy-lighting to float
for row in filtered_data:
row["low-energy-lighting"] = float(row["low-energy-lighting"])
# Calculate averages
sums = defaultdict(float)
counts = defaultdict(int)
for row in filtered_data:
description = row["lighting-description"]
sums[description] += row["low-energy-lighting"]
counts[description] += 1
averages = [{
"lighting-description": correct_spelling(description.lower()),
"low-energy-lighting": total / counts[description]
} for description, total in sums.items()]
return averages
@ -103,9 +116,12 @@ class EpcClean:
def clean_wrapper(self, field, cleaning_cls, **kwargs):
for description in self.unique_vals[field].keys():
cln = cleaning_cls(description, **kwargs)
self.cleaned[field].append(
{
"original_description": description,
**cleaning_cls(description, **kwargs).process()
"clean_description": cln.description.capitalize(),
**cln.process()
}
)