diff --git a/etl/development.py b/etl/development.py index 06a9bb1..1e26f5f 100644 --- a/etl/development.py +++ b/etl/development.py @@ -15,7 +15,7 @@ pdfReader = pdfReaderToText(DATA_LOC_1) doc2 = pdfReader.get_reader() pdfReader2 = pdfReaderToText(DATA_LOC_2) doc1 = pdfReader2.get_reader() -print(doc1.survey_information) +print(doc2.property_description.waterHeating) # Transform @@ -32,7 +32,7 @@ if __name__ == "__main__": # Current todo list # - [x] Finish off scraping for section that I need to finish -# - [] Pydantic format for deemed report +# - [in progress] Pydantic format for deemed report # - [] Generate deemed report # - [] Docker compose to an sql database in docker compose (2 hours, then work on sql) # - [] Deploy via terraform to aws (1 day) \ No newline at end of file diff --git a/etl/pdfReader/sitenotes.py b/etl/pdfReader/sitenotes.py index edd30fc..a001d05 100644 --- a/etl/pdfReader/sitenotes.py +++ b/etl/pdfReader/sitenotes.py @@ -1,5 +1,10 @@ from etl.pdfReader.reportType import ReportType -from transform.types import CompanyInfo, SurverySummaryInfo, AssessorInfo, PropertyDescription, PropertyDetail, Dimension +from transform.types import ( + CompanyInfo, SurverySummaryInfo, AssessorInfo, + PropertyDescription, PropertyDetail, Dimension, + Walls, Roofs, Floors, Door, VentilationAndCooling, + Lighting, WaterHeating, HotWaterCylinder, +) from datetime import datetime class SiteNotesExtractor(): @@ -32,18 +37,17 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): """ self.transform_summary_information() self.transform_sections() - # self.get_section_7() - # self.get_section_8() - # self.get_section_9() - # self.get_section_10() + + # Saves windows till the end as that requires thought # self.get_section_11() - # self.get_section_12() - # self.get_section_13() + + # Heaters require thought so will complete later # self.get_section_14() # self.get_section_14_1() # self.get_section_14_2() - # self.get_section_15_0() - # self.get_section_15_1() + + + # These one are quick fixes can be done on train or one at a time # self.get_section_16() # self.get_section_17() # self.get_section_18() @@ -155,7 +159,7 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): # Section 2 - data = self.raw_data[self.raw_data.index("2.0 Number Of"):self.raw_data.index("3.0 Date Built")] + data = self.get_data_between("2.0 Number Of","3.0 Date Built") avoid = [ '2.0 Number Of', @@ -175,7 +179,7 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): # 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) @@ -193,6 +197,29 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): # Section 5 conservatory = self.is_there_a_conservatory() + # Section 7 + walls = self.get_walls() + + # Section 8 + roofs = self.get_roof() + + # Section 9 + floors = self.get_floors() + + # Section 10 + door = self.get_door() + + # Section 12 + ventilationAndCooling = self.get_ventilation_and_cooling() + + # Section 13 + lighting = self.get_lighting() + + # Section 15.0 Water Heating + waterHeating = self.get_water_heating() + + # Section 15.1 Hot Water Cylinder + hotWaterCylinder = self.get_hot_water_cylinder() self.property_description = PropertyDescription( built_form = get_value("Built Form"), @@ -209,22 +236,39 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): 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 [], + age_band=age_bands[0], + dimensions=dimensions["main"] if "main" in dimensions else [], + wall=walls[0], + roof=roofs[0], + floor=floors[0], )if no_of_main_property > 0 else None, ex1_property=PropertyDetail( age_band= age_bands[1], dimensions= dimensions["ex1"] if "ex1" in dimensions else [], + wall=walls[1], + roof=roofs[1], + floor=floors[1], )if no_of_extension_1 > 0 else None, ex2_property=PropertyDetail( age_band= age_bands[2], dimensions= dimensions["ex2"] if "ex2" in dimensions else [], + wall=walls[2], + roof=roofs[2], + floor=floors[2], )if no_of_extension_2 > 0 else None, ex3_property=PropertyDetail( age_band= age_bands[3], dimensions=dimensions["ex3"] if "ex3" in dimensions else [], + wall=walls[3], + roof=roofs[3], + floor=floors[3], )if no_of_extension_4 > 0 else None, conservatory=conservatory, + door=door, + ventilationAndCooling=ventilationAndCooling, + lighting=lighting, + waterHeating=waterHeating, + hotWaterCylinder=hotWaterCylinder, ) @@ -320,7 +364,7 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): def get_section_6(self): pass - def get_section_7(self): + def get_walls(self): data = self.get_data_between('7.0 Walls','8.0 Roofs') sub_titles = [ @@ -345,9 +389,27 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): "Extension 3", "Extension 4", ] - self.two_column_with_extension_processor(data, sub_titles, main_titles, 7) + lst = self.two_column_with_extension_processor(data, sub_titles, main_titles) + + walls = [] + for ext in lst: + walls.append(Walls( + construction=ext["construction"] if ext["construction"] is not None else "", + insulation=str(ext["insulation"]), + insulation_thickness_mm=str(ext["insulation_thickness(mm)"]), + wall_thickness_measured=True if ext["wall_thickness_measured"].upper() == "YES" else False, + wall_thickness_mm=int(ext.get("wall_thickness(mm)",-1)), + u_value_known=True if ext["u_value_known"].upper() == "YES" else False, + u_value_w_m2_k=float(ext.get("u_value_(w/m²k)", -1.0)), + dry_lining=True if ext.get("dry_lining", "NO").upper() == "YES" else False, + alternative_wall_present=True if ext.get("alternative_wall_present", "NO").upper() == "YES" else False, + )) + return walls - def get_section_8(self): + + + + def get_roof(self): data = self.raw_data[self.raw_data.index('8.0 Roofs'): self.raw_data.index('9.0 Floors')] sub_titles = [ "Construction", @@ -364,9 +426,19 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): "Extension 4", ] - self.two_column_with_extension_processor(data, sub_titles, titles, 8) + lst = self.two_column_with_extension_processor(data, sub_titles, titles) + roofs = [] + for roof in lst: + roofs.append(Roofs( + construction=roof["construction"] if roof["construction"] is not None else "", + insulation_type=str(roof["insulation_type"]), + insulation_thickness=str(roof["insulation_thickness"]), + u_value_known=True if roof["u_value_known"].upper() == "YES" else False, + )) + return roofs - def two_column_with_extension_processor(self, data, sub_titles, main_titles, section): + + def two_column_with_extension_processor(self, data, sub_titles, main_titles): def get_value(key): try: index = proc_data.index(key) @@ -374,24 +446,35 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): return None if value in sub_titles else value except (ValueError, IndexError): return None - title = None proc_data = data + return_dict = {} + group_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: + if return_dict: + group_data.append(return_dict) + return_dict = {} proc_data = data[data.index(main_titles[index]):] continue else: + if return_dict: + group_data.append(return_dict) break if title is None: continue if items in sub_titles: - setattr(self, f"section_{section}_{title}_{items.lower().replace(' ', '_').replace('-','_')}", get_value(items)) + return_dict.update({f"{items.lower().replace("?", "").replace(' ', '_').replace('-','_')}": get_value(items)}) - def get_section_9(self): + if return_dict: + group_data.append(return_dict) + + return group_data + + def get_floors(self): data = self.raw_data[self.raw_data.index('9.0 Floors'): self.raw_data.index('10.0 Doors')] sub_titles = [ "Floor Type", @@ -410,9 +493,19 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): ] - self.two_column_with_extension_processor(data, sub_titles, titles, 9) + lst = self.two_column_with_extension_processor(data, sub_titles, titles) + floors = [] + for floor in lst: + floors.append(Floors( + floor_type=floor.get("floor_type", ""), + ground_floor_construction=floor.get("get_floor_construction", ""), + ground_floor_insulation_type=floor.get("ground_floor_insulation_type", ""), + floor_insulation_thickness_mm=floor.get("floor_insulation_thickness_(mm)", -1) if floor.get("floor_insulation_thickness_(mm)", -1) is not None else -1, + u_value_known=True if floor.get("u_value_known").upper() == "YES" else False, + )) + return floors - def get_section_10(self): + def get_door(self): data = self.raw_data[self.raw_data.index("10.0 Doors"): self.raw_data.index("11.0 Windows")] avoid = [ "10.0 Doors", @@ -423,9 +516,15 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): "Number of Insulated Doors", "U-value (W/m²K)", ] - self.two_columns_processor(data, sub_titles, avoid, 10) + dict_ = self.two_columns_processor(data, sub_titles, avoid) - def two_columns_processor(self, data, sub_titles_to_gather, avoid, section, indexAdd = 1): + return Door( + no_of_doors=int(dict_.get("no_of_doors", -1)), + no_of_insulated_doors=int(dict_.get("number_of_insulated_doors", -1)), + u_value_w_m2_k=dict_.get("u_value_(w/m²k)", "") if dict_.get("u_value_(w/m²k), '')") is not None else "", + ) + + def two_columns_processor(self, data, sub_titles_to_gather, avoid, indexAdd = 1): def get_value(key): try: index = data.index(key) @@ -433,12 +532,13 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): return None if value in avoid else value except (ValueError, IndexError): return None - + dict_ = {} 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)) + dict_.update({f"{items.lower().replace('-', '_').replace(' ','_')}":get_value(items)}) + return dict_ def get_section_11(self): data = self.get_data_between("Window Location", "12.0 Ventilation & Cooling") @@ -502,7 +602,7 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): lst.append(dict_) until = index + 3 + i - def get_section_12(self): + def get_ventilation_and_cooling(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', @@ -514,9 +614,14 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): "Space Cooling System Present", ] - self.two_columns_processor(data, sub_titles, avoid, 12) + dict_ = self.two_columns_processor(data, sub_titles, avoid) + return VentilationAndCooling( + no_of_open_fireplaces=int(dict_.get("number_of_open_fireplaces", -1)), + ventilation_type=str(dict_.get("ventilation_type", "")), + space_cooling_system_present=True if dict_.get("space_cooling_system_present").upper() == "YES" else False + ) - def get_section_13(self): + def get_lighting(self): data = self.raw_data[self.raw_data.index('13.0 Lighting'): self.raw_data.index('14.0 Main Heating1')] avoid = [ "13.0 Lighting", @@ -526,7 +631,12 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): "Total number of light fittings", "Total number of L.E.L. fittings", ] - self.two_columns_processor(data, sub_titles, avoid = avoid, section=13) + dict_ = self.two_columns_processor(data, sub_titles, avoid = avoid) + print(dict_) + return Lighting( + total_no_of_light_fittings=int(dict_["total_number_of_light_fittings"]), + total_no_of_lel_fittings=int(dict_["total_number_of_l.e.l._fittings"]), + ) 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')] @@ -585,7 +695,7 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): ] self.two_columns_processor(data, sub_titles, avoid, 14.2) - def get_section_15_0(self): + def get_water_heating(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", @@ -595,9 +705,14 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): "15.0 Water Heating", "15.1 Hot Water Cylinder", ] - self.two_columns_processor(data, sub_titles, avoid, 15.0) + dict_ = self.two_columns_processor(data, sub_titles, avoid) - def get_section_15_1(self): + return WaterHeating( + heating_type=dict_.get("heating_type", ""), + fuel_type=dict_.get("fuel_type", ""), + ) + + def get_hot_water_cylinder(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", @@ -609,7 +724,13 @@ class QuidosSiteNotesExtractor(SiteNotesExtractor): "16.0 Solar Water Heating", "15.1 Hot Water Cylinder" ] - self.two_columns_processor(data, sub_tites, avoid, 15.1) + dict_ = self.two_columns_processor(data, sub_tites, avoid) + return HotWaterCylinder( + volume=dict_.get("volume", ""), + insulation_type=dict_.get("insulation_type", ""), + insulation_thickness=dict_.get("insulation_thickness", ""), + thermostat=True if dict_.get("thermostat", "NO").upper() == "YES" else False, + ) 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")] diff --git a/etl/transform/types.py b/etl/transform/types.py index 1058637..f83d9b4 100644 --- a/etl/transform/types.py +++ b/etl/transform/types.py @@ -41,6 +41,34 @@ class SurverySummaryInfo(BaseModel): current_annual_emission_including_0925_multiplayer: str current_annual_energy_costs: str +class Walls(BaseModel): + construction: str + insulation: str + insulation_thickness_mm: str + wall_thickness_measured: bool + wall_thickness_mm: Optional[int] + u_value_known: bool + u_value_w_m2_k: Optional[float] + dry_lining: bool + alternative_wall_present: bool + +class Roofs(BaseModel): + construction: str + insulation_type: str + insulation_thickness: str + u_value_known: bool + +class Floors(BaseModel): + floor_type: str + ground_floor_construction: str + ground_floor_insulation_type: str + floor_insulation_thickness_mm: Optional[float] = -1 + u_value_known: bool + +class Door(BaseModel): + no_of_doors: int + no_of_insulated_doors: int + u_value_w_m2_k: Optional[str] class AssessorInfo(BaseModel): accreditation_number: str @@ -48,9 +76,56 @@ class AssessorInfo(BaseModel): phone_number: Optional[str] = None email_address: Optional[EmailStr] = None +class Windows(BaseModel): + glazing_type: str + area_m2: float + roof_window: bool + orientation: str + u_value_w_m2_k: int + g_value: int + +class VentilationAndCooling(BaseModel): + no_of_open_fireplaces: int + ventilation_type: str + space_cooling_system_present: bool + +class Lighting(BaseModel): + total_no_of_light_fittings: int + total_no_of_lel_fittings: int + +class MainHeatingSystemControls(BaseModel): + control_type: str + flue_type: str + fan_assisted_flue: str + heat_emitter_type: str + electricity_meter_type: Optional[str] + mains_gas_available: Optional[bool] + +class MainHeating(BaseModel): + heating_source: str + efficiency_source: str + heating_fuel: str + brand_name: str + model_name: str + model_qualifer: str + controls: MainHeatingSystemControls + +class WaterHeating(BaseModel): + heating_type: str + fuel_type: str + +class HotWaterCylinder(BaseModel): + volume: str + insulation_type: str + insulation_thickness: str + thermostat: bool + class PropertyDetail(BaseModel): age_band: str dimensions: List[Dimension] = [] + wall: Optional[Walls] = None + roof: Optional[Roofs] = None + floor: Optional[Floors] = None class PropertyDescription(BaseModel): built_form: str @@ -71,4 +146,9 @@ class PropertyDescription(BaseModel): ex2_property: Optional[PropertyDetail] = None ex3_property: Optional[PropertyDetail] = None ex4_property: Optional[PropertyDetail] = None - conservatory: bool \ No newline at end of file + conservatory: bool + door: Optional[Door] + ventilationAndCooling: Optional[VentilationAndCooling] + lighting: Optional[Lighting] + waterHeating: Optional[WaterHeating] + hotWaterCylinder: Optional[HotWaterCylinder] \ No newline at end of file