diff --git a/etl/customers/vander_elliot/non_intrusives.py b/etl/customers/vander_elliot/non_intrusives.py index e11bc3f8..57e0522b 100644 --- a/etl/customers/vander_elliot/non_intrusives.py +++ b/etl/customers/vander_elliot/non_intrusives.py @@ -1,3 +1,4 @@ +from datetime import datetime from etl.non_intrusive_surveys.upload.UploadNonIntrusives import UploadNonIntrusives @@ -7,6 +8,60 @@ def app(): :return: """ + # In the future, we can just use the ordnance survey api + uprn_lookup = [ + {'House Number': 79, + 'Address Line 1': 'Clare Road', + 'Address Line 2': 'Liverpool', + 'Postcode': 'L20 9LZ', + 'uprn': 41018850}, + {'House Number': 'Flat 1', + 'Address Line 1': '2 Linacre Lane', + 'Address Line 2': 'Liverpool', + 'Postcode': 'L20 5AH', + 'uprn': 41052320}, + {'House Number': 'Flat 2', + 'Address Line 1': '2 Linacre Lane', + 'Address Line 2': 'Liverpool', + 'Postcode': 'L20 5AH', + 'uprn': 41052321}, + {'House Number': 'Flat 3', + 'Address Line 1': '2 Linacre Lane', + 'Address Line 2': 'Liverpool', + 'Postcode': 'L20 5AH', + 'uprn': 41052322}, + {'House Number': 'Flat 4', + 'Address Line 1': '2 Linacre Lane', + 'Address Line 2': 'Liverpool', + 'Postcode': 'L20 5AH', + 'uprn': 41222759}, + {'House Number': 'Flat 1', + 'Address Line 1': '4 Linacre Lane', + 'Address Line 2': 'Liverpool', + 'Postcode': 'L20 5AH', + 'uprn': 41222760}, + {'House Number': 'Flat 2 (NO ACCESS)', + 'Address Line 1': '4 Linacre Lane', + 'Address Line 2': 'Liverpool', + 'Postcode': 'L20 5AH', + 'uprn': 41222761}, + {'House Number': 'Flat 3', + 'Address Line 1': '4 Linacre Lane', + 'Address Line 2': 'Liverpool', + 'Postcode': 'L20 5AH', + 'uprn': 41212534}, + {'House Number': 'Flat 1 (NO ACCESS)', + 'Address Line 1': '29 Bedford Road', + 'Address Line 2': 'Liverpool', + 'Postcode': 'L4 5PS', + 'uprn': 38237316}, + {'House Number': 'Flat 2 (NO ACCESS)', + 'Address Line 1': '29 Bedford Road', + 'Address Line 2': 'Liverpool', + 'Postcode': 'L4 5PS', + 'uprn': 38237317} + ] + non_intrusive_s3_filename = ( "customers/Vander Elliot/Non-intrusive survey template V2 - Amazon Management Services.xlsx" ) @@ -14,4 +69,6 @@ def app(): non_intrusive = UploadNonIntrusives( s3_template_location=non_intrusive_s3_filename, s3_bucket="retrofit-datalake-dev", + uprn_lookup=uprn_lookup, + survey_date=datetime.strptime('2024-06-21', '%Y-%m-%d') ) diff --git a/etl/non_intrusive_surveys/upload/UploadNonIntrusives.py b/etl/non_intrusive_surveys/upload/UploadNonIntrusives.py index 00f707e9..824f41f9 100644 --- a/etl/non_intrusive_surveys/upload/UploadNonIntrusives.py +++ b/etl/non_intrusive_surveys/upload/UploadNonIntrusives.py @@ -1,4 +1,10 @@ from utils.s3 import read_excel_from_s3 +from utils.logger import setup_logger +from sqlalchemy.orm import sessionmaker +from backend.app.db.connection import db_engine +from backend.app.db.functions.non_intrusive_surveys import upload_non_intrusive_survey_notes + +logger = setup_logger() class UploadNonIntrusives: @@ -6,13 +12,139 @@ class UploadNonIntrusives: This class handles the upload of findings from the non-intrusive surveys, to the database """ - def __init__(self, s3_template_location, s3_bucket): + COLUMN_PREFIXES: dict = { + 'Surveyor First Name': 'Surveyor', + 'Surveyor Last Name': 'Surveyor', + 'House Number': 'Property Details', + 'Address Line 1': 'Property Details', + 'Address Line 2': 'Property Details', + 'Postcode': 'Property Details', + 'Property Year Built': 'Property Details', + 'Wall Construction': 'Walls', + 'Wall Construction Notes': 'Walls', + 'Existing insulation?': 'Walls', + 'Retro Drilled?': 'Walls', + 'Condition (cracks & damp)': 'Walls', + 'Condition Notes': 'Walls', + 'Alternative walls': 'Walls', + 'Alternative walls percentage': 'Walls', + 'Adequate Ventilation?': 'Walls', + 'Ventilation notes': 'Walls', + 'Party wall': 'Walls', + 'Floor Type': 'Floor', + 'Wall render': 'Wall Render', + 'Wall Render Condition': 'Wall Render', + 'Roof Type': 'Roof', + 'Roof insulation ': 'Roof', + 'Roof Condition': 'Roof', + 'Obvious Roof Shading': 'Roof', + 'Roof orientation - Primary': 'Roof', + 'Roof orientation - Secondary': 'Roof', + 'Obstructions on the roof': 'Roof', + 'Flue type': 'Heating', + 'Is there an extension?': 'Access', + 'Are there any out-buildings?': 'Access', + 'Is there a conservatory?': 'Access', + 'Is the property straight onto a footpath?': 'Access', + 'Is there a requirement for planning consent for works?': 'Access', + 'Is there space for an external unit?': 'Air Source Heat Pump', + 'Could a cylinder fit in the loft?': 'Air Source Heat Pump', + 'Are there obvious areas of heat loss from the walls?': 'Thermography', + 'Are there obvious areas of heat loss from the roof?': 'Thermography', + 'Does the existing insulation exhibit signs of inconsistent performance or underperformance?': 'Thermography', + 'Is there excessive levels of heat loss from windows?': 'Thermography', + 'Is there excessive levels of heat loss from doors?': 'Thermography', + 'Material inside the walls': 'Borescope Test', + 'Cavity depth (mm)': 'Borescope Test', + 'Is there rubble in the cavity?': 'Borescope Test', + 'Wall tie type': 'Borescope Test', + 'Wall tie integrity': 'Borescope Test', + 'Inner block work': 'Borescope Test', + 'Current glazing': 'Windows', + 'Windows Age (pre/post 2002)': 'Windows', + 'Glazing gap': 'Windows', + 'Are there obvious trickle vents in the windows?': 'Windows', + 'Is there sufficient space in the garden?': 'Ground Source Heat Pump', + 'Does the property need a CIGA check?': 'Funding', + 'Is the property eligible for GBIS?': 'Funding', + 'Is the property eligible for ECO4?': 'Funding', + 'Is the property eligible for the Local Authority Flex Scheme?': 'Funding', + 'Is the property eligible for HUG?': 'Funding', + 'Is the property eligible for LAD?': 'Funding', + 'Other funding recommendations': 'Funding' + } + + def __init__(self, s3_template_location, s3_bucket, uprn_lookup, survey_date): self.s3_template_location = s3_template_location self.s3_bucket = s3_bucket self.template = self.read_template() + self.uprn_lookup = uprn_lookup + self.survey_date = survey_date + def read_template(self): """ This method reads the template from S3 """ - return read_excel_from_s3(file_key=self.s3_template_location, bucket_name=self.s3_bucket, header_row=0) + return read_excel_from_s3(file_key=self.s3_template_location, bucket_name=self.s3_bucket, header_row=2) + + def upload(self): + """ + This method uploads the non-intrusive survey data to the database + """ + + if self.uprn_lookup is None: + raise Exception("Implement call to ordnance survey to get uprn lookup data") + + logger.info("Preparing non-intrusive notes") + non_intrusives = self.template.to_dict(orient="records") + + non_invasive_notes = [] + for survey in non_intrusives: + # Remove any NAN entries + survey_clean = {self.COLUMN_PREFIXES[k] + ": " + k: v for k, v in survey.items() if v == v} + + uprn_data = [ + x for x in self.uprn_lookup if ( + str(x['House Number']).strip() == str(survey_clean['Property Details: House Number']).strip() and + x['Address Line 1'] == survey_clean['Property Details: Address Line 1'].strip() and + x['Address Line 2'] == survey_clean['Property Details: Address Line 2'].strip() and + x['Postcode'] == survey_clean['Property Details: Postcode'].strip() + ) + ] + if len(uprn_data) != 1: + address = ( + str(survey_clean['Property Details: House Number']) + ' ' + + survey_clean['Property Details: Address Line 1'] + ' ' + + survey_clean['Property Details: Address Line 2'] + ' ' + + survey_clean['Property Details: Postcode'] + ) + raise Exception(f"Failed to find UPRN data for {address}") + + surveyor = ( + survey_clean.pop("Surveyor: Surveyor First Name") + " " + + survey_clean.pop("Surveyor: Surveyor Last Name") + ) + + # Include all of the information apart from data that includes the Property details prefix and the + # surveyor - we do however include Property Details: Property Year Built + notes_to_upload = { + k: v for k, v in survey_clean.items() if k == "Property Details: Property Year Built" or ( + not k.startswith("Property Details") and + not k.startswith("Surveyor") + ) + } + + non_invasive_notes.append({ + "uprn": uprn_data[0]['uprn'], + "surveyor": surveyor, + "survey_date": self.survey_date, + **notes_to_upload + }) + + # Implement call to upload notes_to_upload to the database + logger.info("Uploading non-intrusive notes to the database") + + session = sessionmaker(bind=db_engine)() + + upload_non_intrusive_survey_notes(session=session, non_invasive_notes=non_invasive_notes, batch_size=500)