survey-extraction/etl/pdfReader/sitenotes.py

706 lines
No EOL
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from etl.pdfReader.reportType import ReportType
from transform.types import CompanyInfo, SurverySummaryInfo, AssessorInfo, PropertyDescription, PropertyDetail, Dimension
from datetime import datetime
class SiteNotesExtractor():
def __init__(self, data_list):
self.raw_data = data_list
def get_x_occurance(self, lst, value, x=1):
try:
return [i for i, v in enumerate(lst) if v == value][x]
except IndexError:
return None # Return None if the value does not occur twice
def get_data_between(self, a, b):
return self.raw_data[self.raw_data.index(a):self.raw_data.index(b)]
class QuidosSiteNotesExtractor(SiteNotesExtractor):
def __init__(self, data_list):
super().__init__(data_list)
self.type = ReportType.QUIDOS_SITE_NOTE
self.company_information = None
self.survey_information = None
self.property_description = None
self.setup()
def setup(self):
"""
A function to read QUIDOS SITE REPORT and get all data
"""
self.transform_summary_information()
self.transform_sections()
# self.get_section_7()
# self.get_section_8()
# self.get_section_9()
# self.get_section_10()
# self.get_section_11()
# self.get_section_12()
# self.get_section_13()
# self.get_section_14()
# self.get_section_14_1()
# self.get_section_14_2()
# self.get_section_15_0()
# self.get_section_15_1()
# self.get_section_16()
# self.get_section_17()
# self.get_section_18()
# self.get_section_19()
# self.get_section_20()
# self.get_section_21()
# self.get_section_22()
def transform_summary_information(self):
# Summary Information
avoid = [
"Reference Number",
"EPC Language",
"UPRN",
"Postcode",
"Region",
"Address",
"Town",
"County",
"Property Tenure",
"Transaction Type",
"Inspection Date",
'Assessors accreditation number',
'Assessors name',
'Company name/trading name',
'Address',
'POST CODE',
'Phone number',
'Fax number',
'E-mail address',
'Related party disclosure',
'Current SAP rating',
'Potential SAP rating',
'Current EI rating',
'Current annual emissions',
'Current annual energy costs',
'Emission figures including 9.92 emission factor of 0.925',
]
get_value = lambda key: None if self.raw_data[self.raw_data.index(key) + 1] in avoid else self.raw_data[self.raw_data.index(key) + 1]
index = self.get_x_occurance(self.raw_data, "Current annual emissions")
if index:
including_9_92_emission_factor = self.raw_data[index + 1]
else:
including_9_92_emission_factor = None
self.survey_information = SurverySummaryInfo(
reference_number = get_value('Reference Number'),
epc_language = get_value('EPC Language'),
uprn = get_value('UPRN'),
postcode = get_value('Postcode'),
region = get_value('Region'),
address = get_value('Address'),
town = get_value('Town'),
county = get_value('County'),
property_tenure = get_value('Property Tenure'),
transaction_type = get_value('Transaction Type'),
inspection_date = datetime.strptime(get_value('Inspection Date'), '%d %B %Y'),
current_sap = get_value('Current SAP rating'),
potential_sap = get_value('Potential SAP rating'),
current_ei = get_value('Current EI rating'),
potential_ei = get_value('Potential EI rating'),
current_annual_emissions = get_value('Current annual emissions'),
current_annual_energy_costs = get_value('Current annual energy costs'),
current_annual_emission_including_0925_multiplayer=including_9_92_emission_factor,
)
self.company_information = CompanyInfo(
trading_name = get_value('Company name/trading name'),
post_code = get_value('POST CODE'),
fax_number = get_value('Fax number'),
related_party_disclosure = get_value("Related party disclosure")
)
self.assessor_information = AssessorInfo(
accreditation_number = get_value("Assessors accreditation number"),
name = get_value("Assessors name"),
phone_number = get_value("Phone number"),
email_address = get_value("E-mail address"),
)
index = self.get_x_occurance(self.raw_data, "Address")
if index:
assessor_address = self.raw_data[index + 1]
else:
assessor_address = None
self.assessor_information = AssessorInfo(
accreditation_number = get_value("Assessors accreditation number"),
name = get_value("Assessors name"),
phone_number = get_value("Phone number"),
email_address = get_value("E-mail address"),
address = assessor_address,
)
def transform_sections(self):
# Section 1
data = self.get_data_between("1.0 Property Type","2.0 Number Of")
avoid = [
"1.0 Property Type",
"Built Form",
"Detachment/Position",
"2.0 Number Of"
]
get_value = lambda key: None if self.raw_data[self.raw_data.index(key) + 1] in avoid else self.raw_data[self.raw_data.index(key) + 1]
# Section 2
data = self.raw_data[self.raw_data.index("2.0 Number Of"):self.raw_data.index("3.0 Date Built")]
avoid = [
'2.0 Number Of',
'Main Property',
'Extension 1',
'Extension 2',
'Extension 3',
'Extension 4',
'Number of Habitable Rooms',
'Number of Heated Habitable Rooms',
'Heated Basement',
'Conservatory Type',
'Terrain Type',
'Percentage of Draught Proofed(%)',
"3.0 Date Built",
]
# Section 3
age_bands = self.get_age_band()
# Section 4
no_of_main_property = int(get_value("Main Property"))
no_of_extension_1 = int(get_value('Extension 1') or 0)
no_of_extension_2 = int(get_value('Extension 2') or 0)
no_of_extension_3 = int(get_value('Extension 3') or 0)
no_of_extension_4 = int(get_value('Extension 4') or 0)
dimensions = self.get_dimensions(
no_of_main_property,
no_of_extension_1,
no_of_extension_2,
no_of_extension_3,
no_of_extension_4,
)
# Section 5
conservatory = self.is_there_a_conservatory()
self.property_description = PropertyDescription(
built_form = get_value("Built Form"),
detachment_or_position = get_value("Detachment/Position"),
no_of_main_property = no_of_main_property,
no_of_extension_1 = no_of_extension_1,
no_of_extension_2 = no_of_extension_2,
no_of_extension_3 = no_of_extension_3,
no_of_extension_4 = no_of_extension_4,
no_of_habitable_rooms = int(get_value('Number of Habitable Rooms')),
no_of_heated_rooms = int(get_value('Number of Heated Habitable Rooms')),
heated_basement = False if get_value('Heated Basement') == "NO" else True,
conservatory_type = get_value('Conservatory Type'),
terrain_type = get_value('Terrain Type'),
percentage_of_draught_proofed= int(get_value('Percentage of Draught Proofed(%)')),
main_property=PropertyDetail(
age_band= age_bands[0],
dimensions= dimensions["main"] if "main" in dimensions else [],
)if no_of_main_property > 0 else None,
ex1_property=PropertyDetail(
age_band= age_bands[1],
dimensions= dimensions["ex1"] if "ex1" in dimensions else [],
)if no_of_extension_1 > 0 else None,
ex2_property=PropertyDetail(
age_band= age_bands[2],
dimensions= dimensions["ex2"] if "ex2" in dimensions else [],
)if no_of_extension_2 > 0 else None,
ex3_property=PropertyDetail(
age_band= age_bands[3],
dimensions=dimensions["ex3"] if "ex3" in dimensions else [],
)if no_of_extension_4 > 0 else None,
conservatory=conservatory,
)
def get_age_band(self):
data = self.raw_data[self.raw_data.index('3.0 Date Built'):self.raw_data.index('4.0 Dimensions')]
avoid = [
'3.0 Date Built',
'Age Band',
'Main Property',
'Extension 1',
'Extension 2',
'Extension 3',
'Extension 4',
'4.0 Dimensions',
]
property_age = []
get_value = lambda x: None if data[data.index(x) + 1] in avoid else data[data.index(x) + 1]
age = (get_value("Main Property"))
if age:
property_age.append(age)
else:
property_age.append(None)
for i in range(1,4):
if f"Extension {i}" in data:
property_age.append(get_value(f"Extension {i}"))
else:
property_age.append(None)
return property_age
def get_dimensions(self, main, ext1=0, ext2=0, ext3=0, ext4=0):
data = self.get_data_between('4.0 Dimensions','5.0 Conservatory')
avoid = [
'4.0 Dimensions',
'5.0 Conservatory',
'Dimension Type internal',
'Part',
'Floor Area (m2)',
'Room Height (m)',
'Loss Perimeter (m)',
'Party Wall Length (m)',
'Main Property',
'Extension 1 Property',
'Extension 2 Property',
'Extension 3 Property',
'Extension 4 Property',
]
def create_dimensions_array(key, no_of_property):
index = data.index(key) + 1
allNumbers = []
for propertyNo in range(no_of_property):
numbers = []
for i in range(4):
numbers.append(data[i + propertyNo*4 + index])
details = Dimension(
floor_area_m2=float(numbers[0]),
room_height_m=float(numbers[1]),
loss_perimeter_m=float(numbers[2]),
party_wall_length_m=float(numbers[3]),
)
allNumbers.append(details)
return allNumbers
return_dict = {}
return_dict.update({"main": create_dimensions_array("Main Property", main)})
ext = [ext1, ext2, ext3, ext4]
for i in range(1,5):
if f"Extension {i} Property" in data:
return_dict.update({f"ex{i}" : create_dimensions_array(f"Extension {i} Property", ext[i-1])})
return return_dict
def is_there_a_conservatory(self):
data = self.raw_data[self.raw_data.index('5.0 Conservatory'):self.raw_data.index('7.0 Walls')]
avoid = [
'Is there a conservatory?',
'7.0 Walls'
]
get_value = lambda key: None if self.raw_data[self.raw_data.index(key) + 1] in avoid else self.raw_data[self.raw_data.index(key) + 1]
return True if get_value("Is there a conservatory?").upper() == "YES" else False
def get_section_6(self):
pass
def get_section_7(self):
data = self.get_data_between('7.0 Walls','8.0 Roofs')
sub_titles = [
"Construction",
"Insulation",
"Insulation Thickness(mm)",
"Wall Thickness Measured?",
"Wall Thickness Measured",
"Wall Thickness(mm)",
"U-value Known?",
"U-value Known",
"U-value (W/m²K)",
"Dry-lining?",
"Alternative Wall Present?",
"Alternative Wall Present",
]
main_titles = [
"Main Property",
"Extension 1",
"Extension 2",
"Extension 3",
"Extension 4",
]
self.two_column_with_extension_processor(data, sub_titles, main_titles, 7)
def get_section_8(self):
data = self.raw_data[self.raw_data.index('8.0 Roofs'): self.raw_data.index('9.0 Floors')]
sub_titles = [
"Construction",
"Insulation Type",
"Insulation Thickness",
"U-value Known",
]
titles = [
"Main Property",
"Extension 1",
"Extension 2",
"Extension 3",
"Extension 4",
]
self.two_column_with_extension_processor(data, sub_titles, titles, 8)
def two_column_with_extension_processor(self, data, sub_titles, main_titles, section):
def get_value(key):
try:
index = proc_data.index(key)
value = proc_data[index + 1]
return None if value in sub_titles else value
except (ValueError, IndexError):
return None
title = None
proc_data = data
for items in data:
if items in main_titles:
title = items.lower().replace(" ", "_").replace("-", "_")
index = main_titles.index(items)
if main_titles[index] in data:
proc_data = data[data.index(main_titles[index]):]
continue
else:
break
if title is None:
continue
if items in sub_titles:
setattr(self, f"section_{section}_{title}_{items.lower().replace(' ', '_').replace('-','_')}", get_value(items))
def get_section_9(self):
data = self.raw_data[self.raw_data.index('9.0 Floors'): self.raw_data.index('10.0 Doors')]
sub_titles = [
"Floor Type",
"Ground Floor Construction",
"Ground Floor Insulation Type",
"Floor Insulation Thickness (mm)",
"U-value Known",
]
titles = [
"Main Property",
"Extension 1",
"Extension 2",
"Extension 3",
"Extension 4",
]
self.two_column_with_extension_processor(data, sub_titles, titles, 9)
def get_section_10(self):
data = self.raw_data[self.raw_data.index("10.0 Doors"): self.raw_data.index("11.0 Windows")]
avoid = [
"10.0 Doors",
"11.0 Windows",
]
sub_titles = [
"Number of Doors",
"Number of Insulated Doors",
"U-value (W/m²K)",
]
self.two_columns_processor(data, sub_titles, avoid, 10)
def two_columns_processor(self, data, sub_titles_to_gather, avoid, section, indexAdd = 1):
def get_value(key):
try:
index = data.index(key)
value = data[index + indexAdd]
return None if value in avoid else value
except (ValueError, IndexError):
return None
for items in data:
if items in avoid:
continue
elif items in sub_titles_to_gather:
setattr(self, f"section_{section}_{items.lower().replace('-', '_').replace(' ','_')}", get_value(items))
def get_section_11(self):
data = self.get_data_between("Window Location", "12.0 Ventilation & Cooling")
headers = data[:8]
data_entries = data[8:]
num_attributes = 5
subtitles=[
"Main Property",
"Extension 1",
"Extension 2",
"Extension 3",
"Extension 4",
]
orientation = [
"north",
"east",
"west",
"south",
"n",
"w",
"s",
"e",
"nw",
"ne",
"sw",
"se",
"south west",
"south east",
"north west",
"north east",
]
def find_compose_index(lst, compose):
for i, item in enumerate(lst):
if item.lower() in compose:
return i
return None
title = None
until = 0
for i, items in enumerate(data_entries):
if data_entries[i] in subtitles:
title = data_entries[i].lower().replace(" ", "_").replace("-", "_")
setattr(self, f"section_11_{title}_window", [])
if title and until == i:
entry = data_entries[i:]
index = find_compose_index(entry,orientation)
new_entry = entry[index-3:index+3]
dict_ = {
"glazing type": new_entry[0],
"Area (m2)": new_entry[1],
"Roof Window": new_entry[2],
"Orientation": new_entry[3],
"U-value (W/m²K)": new_entry[4],
"g-value": new_entry[5],
}
lst = getattr(self, f"section_11_{title}_window")
lst.append(dict_)
until = index + 3 + i
def get_section_12(self):
data = self.raw_data[self.raw_data.index('12.0 Ventilation & Cooling'): self.raw_data.index('13.0 Lighting')]
avoid = [
'12.0 Ventilation & Cooling',
'13.0 Lighting'
]
sub_titles = [
"Number of Open Fireplaces",
"Ventilation Type",
"Space Cooling System Present",
]
self.two_columns_processor(data, sub_titles, avoid, 12)
def get_section_13(self):
data = self.raw_data[self.raw_data.index('13.0 Lighting'): self.raw_data.index('14.0 Main Heating1')]
avoid = [
"13.0 Lighting",
"14.0 Main Heating1"
]
sub_titles = [
"Total number of light fittings",
"Total number of L.E.L. fittings",
]
self.two_columns_processor(data, sub_titles, avoid = avoid, section=13)
def get_section_14(self):
data = self.raw_data[self.raw_data.index('14.0 Main Heating1'): self.raw_data.index('14.1 Main Heating2')]
main_titles = [
"Main Heating Type",
"Product Database"
"Main Heating System Controls",
]
sub_titles = [
"Heating Source",
"Efficiency Source",
"Heating Fuel",
"Brand Name",
"Model Name",
"Model Qualifier",
"Control Type",
"Flue Type",
"Fan Assisted Flue",
"Heat Emitter Type",
"Electricity Meter Type",
"Mains Gas Available",
]
self.two_column_with_extension_processor(data, sub_titles, main_titles, 14.0)
def get_section_14_1(self):
data = self.raw_data[self.raw_data.index("14.1 Main Heating2"):self.raw_data.index("14.2 Secondary Heating Type")]
main_titles = [
"Second Main Heating Type",
"Main Heating System Controls",
]
sub_titles = [
"Percentage of Heated Floor Area Served (%)",
"Heating Source",
"Efficiency Source",
"Heating Fuel",
"SAP 2009 Table 4a/4b",
"Heating Type",
"Heating Description",
"Control Type",
"Flue Type",
"Fan Assisted Flue",
"Heat Emitter Type",
]
self.two_column_with_extension_processor(data, sub_titles, main_titles, 14.1)
def get_section_14_2(self):
data = self.raw_data[self.raw_data.index("14.2 Secondary Heating Type"):self.raw_data.index("15.0 Water Heating")]
avoid = [
"14.2 Secondary Heating Type",
"15.0 Water Heating",
]
sub_titles = [
"Heating Type",
"Fuel Type",
]
self.two_columns_processor(data, sub_titles, avoid, 14.2)
def get_section_15_0(self):
data = self.raw_data[self.raw_data.index("15.0 Water Heating"):self.raw_data.index("15.1 Hot Water Cylinder")]
sub_titles = [
"Heating Type",
"Fuel Type",
]
avoid = [
"15.0 Water Heating",
"15.1 Hot Water Cylinder",
]
self.two_columns_processor(data, sub_titles, avoid, 15.0)
def get_section_15_1(self):
data = self.raw_data[self.raw_data.index("15.1 Hot Water Cylinder"):self.raw_data.index("16.0 Solar Water Heating")]
sub_tites = [
"Volume",
"Insulation Type",
"Insulation Thickness",
"Thermostat",
]
avoid = [
"16.0 Solar Water Heating",
"15.1 Hot Water Cylinder"
]
self.two_columns_processor(data, sub_tites, avoid, 15.1)
def get_section_16(self):
data = self.raw_data[self.raw_data.index("16.0 Solar Water Heating"):self.raw_data.index("17.0 Waste Water Heat Recovery System")]
avoid = [
"16.0 Solar Water Heating",
"17.0 Waste Water Heat Recovery System",
]
sub_titles = [
"Solar Water Heating Details Known?"
]
self.two_columns_processor(data, sub_titles, avoid, 16.0)
def get_section_17(self):
pass
def get_section_18(self):
data = self.get_data_between("18.0 Showers And Baths", "19.0 Flue Gas Heat Recovery System")
sub_titles = [
"Number of Rooms with Bath and/or Shower",
]
avoid = [
"18.0 Showers And Baths",
"19.0 Flue Gas Heat Recovery System",
]
self.two_columns_processor(data, sub_titles, avoid, 18.0)
avoid = [
"18.0 Showers And Baths",
"19.0 Flue Gas Heat Recovery System",
]
sub_titles = [
"Number of Rooms with Mixer Shower and no", # Number of Rooms with Mixer Shower and no Bath
"Number of Rooms with Mixer Shower and", # Number of Rooms with Mixer Shower and Bath
]
self.two_columns_processor(data, sub_titles, avoid, 18.0, 2)
def get_section_19(self):
data = self.get_data_between("19.0 Flue Gas Heat Recovery System","20.0 Photovoltaic Panel")
sub_titles = [
"FGHRS Present",
]
avoid = [
"19.0 Flue Gas Heat Recovery System",
"20.0 Photovoltaic Panel",
]
self.two_columns_processor(data, sub_titles, avoid, 19)
def get_section_20(self):
data = self.get_data_between("20.0 Photovoltaic Panel","21.0 Wind Turbine")
sub_titles = [
"Percentage of External Roof Area with PVs"
]
avoid = [
"20.0 Photovoltaic Panel",
"21.0 Wind Turbine",
]
self.two_columns_processor(data, sub_titles, avoid, 20)
sub_titles = [
"PVs are connected to dwelling electricity" # PVs are connected to dwelling electricity meter
]
self.two_columns_processor(data, sub_titles, avoid, 20, 2)
def get_section_21(self):
data = self.get_data_between("21.0 Wind Turbine","22.0 Other Details")
sub_titles = [
"Wind Turbine",
]
avoid = [
"21.0 Wind Turbine",
"22.0 Other Details",
]
self.two_columns_processor(data, sub_titles, avoid, 21)
def get_section_22(self):
data = self.get_data_between("22.0 Other Details","Recommendations (Carbon Saving Figures Are For Guidance Only)")
sub_titles = [
"Electricity Meter Type",
"Mains Gas Available",
]
avoid = [
"22.0 Other Details",
"Recommendations (Carbon Saving Figures Are For Guidance Only)",
]
self.two_columns_processor(data, sub_titles, avoid, 22)
# Section and 11
# Extract
# Transform ( wiht validation pydantnic)
# Load