mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
170 lines
7.8 KiB
Python
170 lines
7.8 KiB
Python
from fastapi import APIRouter, Depends
|
|
from backend.app.dependencies import validate_token
|
|
from backend.app.plan.schemas import PlanTriggerRequest
|
|
from backend.app.utils import read_csv_from_s3
|
|
from backend.app.config import get_settings
|
|
from model_data.Property import Property
|
|
from epc_api.client import EpcClient
|
|
from utils.logger import setup_logger
|
|
from recommendations.FloorRecommendations import FloorRecommendations
|
|
from recommendations.WallRecommendations import WallRecommendations
|
|
from utils.uvalue_estimates import classify_decile_newvalues
|
|
|
|
# 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
|
|
|
|
logger = setup_logger()
|
|
|
|
router = APIRouter(
|
|
prefix="/plan",
|
|
tags=["plan"],
|
|
dependencies=[Depends(validate_token)],
|
|
responses={404: {"description": "Not found"}}
|
|
)
|
|
|
|
# TODO: Load this data from db
|
|
open_uprn_data = [
|
|
{'UPRN': 6032920, 'X_COORDINATE': 535110.0, 'Y_COORDINATE': 181819.0, 'LATITUDE': 51.5191407,
|
|
'LONGITUDE': -0.0540506},
|
|
{'UPRN': 6038625, 'X_COORDINATE': 535374.0, 'Y_COORDINATE': 182784.0, 'LATITUDE': 51.5277492,
|
|
'LONGITUDE': -0.0498772},
|
|
{'UPRN': 34153991, 'X_COORDINATE': 523238.74, 'Y_COORDINATE': 178003.02, 'LATITUDE': 51.4875579,
|
|
'LONGITUDE': -0.226392},
|
|
{'UPRN': 10008299676, 'X_COORDINATE': 533285.0, 'Y_COORDINATE': 184711.0, 'LATITUDE': 51.5455629,
|
|
'LONGITUDE': -0.0792445},
|
|
{'UPRN': 10008299677, 'X_COORDINATE': 533285.0, 'Y_COORDINATE': 184711.0, 'LATITUDE': 51.5455629,
|
|
'LONGITUDE': -0.0792445},
|
|
{'UPRN': 100021039066, 'X_COORDINATE': 535506.0, 'Y_COORDINATE': 185624.0, 'LATITUDE': 51.5532385,
|
|
'LONGITUDE': -0.0468833},
|
|
{'UPRN': 100021226060, 'X_COORDINATE': 529247.0, 'Y_COORDINATE': 187959.0, 'LATITUDE': 51.5756908,
|
|
'LONGITUDE': -0.1362513},
|
|
{'UPRN': 200003489276, 'X_COORDINATE': 533210.0, 'Y_COORDINATE': 179442.0, 'LATITUDE': 51.4982309,
|
|
'LONGITUDE': -0.0823165}
|
|
]
|
|
|
|
in_conservation_area_data = [
|
|
{'uprn': 6032920, 'is_in_conservation_area': 'not_in_conservation_area'},
|
|
{'uprn': 6038625, 'is_in_conservation_area': 'not_in_conservation_area'},
|
|
{'uprn': 34153991, 'is_in_conservation_area': 'unknown'},
|
|
{'uprn': 10008299676, 'is_in_conservation_area': 'in_conservation_area'},
|
|
{'uprn': 10008299677, 'is_in_conservation_area': 'in_conservation_area'},
|
|
{'uprn': 100021039066, 'is_in_conservation_area': 'not_in_conservation_area'},
|
|
{'uprn': 100021226060, 'is_in_conservation_area': 'in_conservation_area'},
|
|
{'uprn': 200003489276, 'is_in_conservation_area': 'in_conservation_area'}
|
|
]
|
|
|
|
# TODO: db
|
|
floors_decile_data = {
|
|
'decile_labels': ['Decile 1', 'Decile 2', 'Decile 3', 'Decile 4', 'Decile 5', 'Decile 6', 'Decile 7', 'Decile 8',
|
|
'Decile 9', 'Decile 10'], 'decile_boundaries': [6., 50., 56., 69., 77.6, 87., 98., 112.,
|
|
127., 150., 2279.]}
|
|
|
|
walls_decile_data = {
|
|
'decile_labels': ['Decile 1', 'Decile 2', 'Decile 3', 'Decile 4', 'Decile 5', 'Decile 6', 'Decile 7', 'Decile 8',
|
|
'Decile 9', 'Decile 10'], 'decile_boundaries': [6., 49., 51., 55., 64., 71., 76., 83., 96.,
|
|
120., 2279.]}
|
|
|
|
|
|
@router.post("/trigger")
|
|
async def trigger_plan(body: PlanTriggerRequest):
|
|
logger.info("Getting the inputs")
|
|
# Read in the trigger file from s3
|
|
bucket_name = get_settings().PLAN_TRIGGER_BUCKET
|
|
plan_input = read_csv_from_s3(bucket_name=bucket_name, filepath=body.trigger_file_path)
|
|
|
|
# TODO: Add validation to the file
|
|
|
|
logger.info("Getting EPC data")
|
|
epc_client = EpcClient(auth_token=get_settings().EPC_AUTH_TOKEN)
|
|
input_properties = [
|
|
Property(postcode=config['postcode'], address1=config['address'], epc_client=epc_client)
|
|
for config in plan_input
|
|
]
|
|
|
|
for p in input_properties:
|
|
p.search_address_epc()
|
|
p.set_year_built()
|
|
|
|
logger.info("Getting coordinates")
|
|
# This is placeholder, until the full dataset is loaded into the database
|
|
for p in input_properties:
|
|
coordinate_data = [x for x in open_uprn_data if x['UPRN'] == int(p.data['uprn'])][0]
|
|
p.set_coordinates(coordinate_data)
|
|
|
|
logger.info("Check if property is in conservation area")
|
|
for p in input_properties:
|
|
in_conservation_area = [x for x in in_conservation_area_data if x['uprn'] == int(p.data['uprn'])][0].get(
|
|
"is_in_conservation_area"
|
|
)
|
|
p.set_is_in_conservation_area(in_conservation_area)
|
|
|
|
logger.info("Getting components and properties recommendations")
|
|
recommendations = []
|
|
for property_id, p in enumerate(input_properties):
|
|
# For each property, classiy floor area decide
|
|
total_floor_area_group_decile = classify_decile_newvalues(
|
|
decile_boundaries=floors_decile_data["decile_boundaries"],
|
|
decile_labels=floors_decile_data["decile_labels"],
|
|
new_values=[float(p.data["total-floor-area"])],
|
|
)[0]
|
|
|
|
# Property recommendations
|
|
p.get_components(cleaned)
|
|
|
|
# This is placeholder, until the full dataset is loaded into the database and we just make a read to the
|
|
# database
|
|
floors_u_value_estimate = [
|
|
x for x in uvalue_estimates_floors
|
|
if (x['local-authority'] == p.data["local-authority"]) &
|
|
(x['property-type'] == p.data["property-type"]) &
|
|
(x['built-form'] == p.data["built-form"]) &
|
|
(x['floor-energy-eff'] == p.data["floor-energy-eff"] if p.data["floor-energy-eff"] != 'N/A' else True) &
|
|
(x['floor-env-eff'] == p.data["floor-env-eff"] if p.data["floor-env-eff"] != 'N/A' else True)
|
|
]
|
|
|
|
# Floor recommendations
|
|
floor_recommender = FloorRecommendations(
|
|
property_instance=p, uvalue_estimates=floors_u_value_estimate,
|
|
total_floor_area_group_decile=total_floor_area_group_decile
|
|
)
|
|
floor_recommender.recommend()
|
|
# insert property id
|
|
for rec in floor_recommender.recommendations:
|
|
rec["property_id"] = property_id
|
|
|
|
recommendations.extend(floor_recommender.recommendations)
|
|
|
|
# Wall recommendations
|
|
# We would make this u-value query directly to the database
|
|
total_floor_area_group_decile = classify_decile_newvalues(
|
|
decile_boundaries=walls_decile_data["decile_boundaries"],
|
|
decile_labels=walls_decile_data["decile_labels"],
|
|
new_values=[float(p.data["total-floor-area"])],
|
|
)[0]
|
|
|
|
# This is placeholder, until the full dataset is loaded into the database and we just make a read to the
|
|
# database
|
|
walls_u_value_estimate = [
|
|
x for x in uvalue_estimates_walls
|
|
if (x['local-authority'] == p.data["local-authority"]) &
|
|
(x['property-type'] == p.data["property-type"]) &
|
|
(x['built-form'] == p.data["built-form"]) &
|
|
(x['walls-energy-eff'] == p.data["walls-energy-eff"] if p.data["walls-energy-eff"] != 'N/A' else True) &
|
|
(x['walls-env-eff'] == p.data["walls-env-eff"] if p.data["walls-env-eff"] != 'N/A' else True)
|
|
]
|
|
|
|
wall_recomendations = WallRecommendations(
|
|
property_instance=p,
|
|
uvalue_estimates=walls_u_value_estimate,
|
|
total_floor_area_group_decile=total_floor_area_group_decile
|
|
)
|
|
wall_recomendations.recommend()
|
|
# insert property id
|
|
for rec in wall_recomendations.recommendations:
|
|
rec["property_id"] = property_id
|
|
|
|
recommendations.extend(wall_recomendations.recommendations)
|
|
|
|
return {"recommendations": recommendations}
|