mirror of
https://github.com/Hestia-Homes/survey-extraction.git
synced 2026-06-30 13:10:56 +00:00
save current progress but need to finsih section 4
This commit is contained in:
Jun-te Kim
parent
50600b5192
commit
01f3a0cec6
2 changed files with 158 additions and 32 deletions
|
|
@ -1,5 +1,5 @@
|
|||
# SQLModel mapping for ConditionReportModel using BaseModel
|
||||
from sqlmodel import SQLModel, Field, Relationship
|
||||
from sqlmodel import SQLModel, Field, Relationship, Column, JSON
|
||||
from typing import Optional, List
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
|
|
@ -234,3 +234,64 @@ class Rooms(BaseModel, table=True):
|
|||
|
||||
bedrooms: List[Bedroom] = Relationship(back_populates="rooms")
|
||||
bathrooms: List[Bathroom] = Relationship(back_populates="rooms")
|
||||
|
||||
class GeneralConditionHeatingSystem(BaseModel, table=True):
|
||||
is_the_heating_system_in_working_order: bool
|
||||
does_the_occupant_have_a_smart_meter: bool
|
||||
are_there_any_smart_monitoring_devices: bool
|
||||
are_the_gas_and_electricity_meters_accessible: bool
|
||||
dual_or_single_electric_meter: str
|
||||
|
||||
class MainHeatingOne(BaseModel, table=True):
|
||||
as_defined_by: str
|
||||
fuel: str
|
||||
type: str
|
||||
|
||||
class MainHeatingTwo(BaseModel, table=True):
|
||||
is_there_a_main_heating_two: bool
|
||||
|
||||
class SecondaryHeating(BaseModel, table=True):
|
||||
is_there_a_secondary_heating: bool
|
||||
fuel: str
|
||||
electric_heating_type: str
|
||||
gas_heating_type: str
|
||||
|
||||
class HeatingByRoom(BaseModel, table=True):
|
||||
rooms_heated_by_main_system_one: List[str] = Field(sa_column=Column(JSON))
|
||||
rooms_heated_by_main_system_two: List[str] = Field(sa_column=Column(JSON))
|
||||
rooms_heated_by_secondary_heating: List[str] = Field(sa_column=Column(JSON))
|
||||
are_there_any_partially_heated_rooms: bool
|
||||
partially_heated_rooms: Optional[List[str]] = Field(sa_column=Column(JSON))
|
||||
are_there_any_unheated_rooms: bool
|
||||
unheated_rooms: List[str] = Field(sa_column=Column(JSON))
|
||||
|
||||
|
||||
class Renewables(BaseModel, table=True):
|
||||
is_there_any_renewable_energy_system_in_place: bool
|
||||
suitable_roof_orientation_for_solar_pv_water: str
|
||||
is_there_a_water_tank: bool
|
||||
type: str
|
||||
size: str
|
||||
tank_location: str
|
||||
is_the_tank_insulated: bool
|
||||
type_of_insulation: str
|
||||
thickness_of_insulation_in_mm: int
|
||||
|
||||
class HeatingSystem(BaseModel, table=True):
|
||||
general_condition_id: uuid.UUID = Field(foreign_key="generalconditionheatingsystem.id")
|
||||
general_condition: GeneralConditionHeatingSystem
|
||||
|
||||
main_heating_one_id: uuid.UUID = Field(foreign_key="mainheatingone.id")
|
||||
main_heating_one: MainHeatingOne
|
||||
|
||||
main_heating_two_id: uuid.UUID = Field(foreign_key="maineheatingtwo.id")
|
||||
main_heating_two: MainHeatingTwo
|
||||
|
||||
secondary_heating_id: uuid.UUID = Field(foreign_key="secondaryheating.id")
|
||||
secondary_heating: SecondaryHeating
|
||||
|
||||
heating_by_room_id: uuid.UUID = Field(foreign_key="heatingbyroom.id")
|
||||
heating_by_room: HeatingByRoom#
|
||||
|
||||
renewables_id: uuid.UUID = Field(foreign_key="renewables.id")
|
||||
renewables: Renewables
|
||||
|
|
@ -41,7 +41,14 @@ from etl.models.conditionReport import (
|
|||
RoomInRoof,
|
||||
Bedroom,
|
||||
Bathroom,
|
||||
Rooms
|
||||
Rooms,
|
||||
GeneralConditionHeatingSystem,
|
||||
MainHeatingOne,
|
||||
MainHeatingTwo,
|
||||
SecondaryHeating,
|
||||
HeatingByRoom,
|
||||
Renewables,
|
||||
HeatingSystem
|
||||
)
|
||||
|
||||
from etl.models.topLevel import(
|
||||
|
|
@ -76,7 +83,7 @@ class surveyedDataProcessor():
|
|||
# My task to complete load
|
||||
# [x] general_information = self.get_section_1()
|
||||
# [x] access_and_elevations = self.get_section_2()
|
||||
# [] rooms = self.get_section_3()
|
||||
# [x] rooms = self.get_section_3()
|
||||
# [] heating_system = self.get_section_4()
|
||||
# [] occupant_assessment = self.get_section_5()
|
||||
# [] site_name, reference_code, address, postcode = self.get_section_0()
|
||||
|
|
@ -92,35 +99,87 @@ class surveyedDataProcessor():
|
|||
# heating_system=heating_system,
|
||||
# occupancy_assessment=occupant_assessment,
|
||||
# )
|
||||
#
|
||||
# hallway = self.get_hallway()
|
||||
# living_room = self.get_living_room()
|
||||
# dining_room = self.get_dining_room()
|
||||
# kitchen = self.get_kitchen()
|
||||
# utility = self.get_utility()
|
||||
# wc = self.get_wc()
|
||||
# landing = self.get_landing()
|
||||
# bedrooms = self.get_bedrooms()
|
||||
# bathrooms = self.get_bathroom()
|
||||
# loft_space = self.get_loft_space()
|
||||
# room_in_roof = self.get_room_in_roof()
|
||||
# class HeatingSystem(BaseModel):
|
||||
# heating_by_room: HeatingByRoom
|
||||
# renewables: Renewables
|
||||
|
||||
# return Rooms(
|
||||
# hallway=hallway,
|
||||
# living_room=living_room,
|
||||
# dining_room=dining_room,
|
||||
# kitchen=kitchen,
|
||||
# utility=utility,
|
||||
# wash_chamber=wc,
|
||||
# landing=landing,
|
||||
# bedrooms=bedrooms,
|
||||
# bathrooms=bathrooms,
|
||||
# loft_space=loft_space,
|
||||
# room_in_roof=room_in_roof,
|
||||
# )
|
||||
general_information = self.load_general_information_from_condition_report(db_session)
|
||||
access_and_elevations = self.load_access_and_elevations_from_condition_report(db_session)
|
||||
rooms = self.load_rooms_from_condition_report(db_session)
|
||||
heating_system = self.load_heating_system_from_condition_report(db_session)
|
||||
|
||||
|
||||
def load_heating_system_from_condition_report(self, db_session):
|
||||
general_condition = self.load_general_condition_of_heating_system_from_condition_report(db_session)
|
||||
main_heating_one = self.load_main_heating_one_from_condition_report(db_session)
|
||||
main_heating_two = self.load_main_heating_two_from_condition_report(db_session)
|
||||
secondary_heating = self.load_secondary_heating_from_condition_report(db_session)
|
||||
heating_by_room = self.load_heating_by_room_from_condition_report(db_session)
|
||||
renwables = self.load_renewables_from_condition_report(db_session)
|
||||
|
||||
return self.get_attribute_and_load(
|
||||
self.condition_report.master_obj,
|
||||
"heating_system",
|
||||
HeatingSystem,
|
||||
db_session,
|
||||
additional_fields={
|
||||
"general_condition_id": general_condition.id,
|
||||
"main_heating_one_id": main_heating_one.id,
|
||||
"main_heating_two_id": main_heating_two.id,
|
||||
"secondary_heating": secondary_heating.id,
|
||||
"heating_by_room_id": heating_by_room.id,
|
||||
"renewables_id": renwables.id,
|
||||
},
|
||||
)
|
||||
|
||||
def load_renewables_from_condition_report(self, db_session):
|
||||
return self.get_attribute_and_load(
|
||||
self.condition_report.master_obj.heating_system,
|
||||
"renewables",
|
||||
Renewables,
|
||||
db_session,
|
||||
)
|
||||
|
||||
def load_heating_by_room_from_condition_report(self, db_session):
|
||||
return self.get_attribute_and_load(
|
||||
self.condition_report.master_obj.heating_system,
|
||||
"heating_by_room",
|
||||
HeatingByRoom,
|
||||
db_session,
|
||||
exclude_list=False
|
||||
)
|
||||
|
||||
def load_secondary_heating_from_condition_report(self, db_session):
|
||||
return self.get_attribute_and_load(
|
||||
self.condition_report.master_obj.heating_system,
|
||||
"secondary_heating",
|
||||
SecondaryHeating,
|
||||
db_session,
|
||||
)
|
||||
|
||||
def load_main_heating_two_from_condition_report(self, db_session):
|
||||
return self.get_attribute_and_load(
|
||||
self.condition_report.master_obj.heating_system,
|
||||
"main_heating_two",
|
||||
MainHeatingTwo,
|
||||
db_session,
|
||||
)
|
||||
|
||||
def load_main_heating_one_from_condition_report(self, db_session):
|
||||
return self.get_attribute_and_load(
|
||||
self.condition_report.master_obj.heating_system,
|
||||
"main_heating_one",
|
||||
MainHeatingOne,
|
||||
db_session,
|
||||
)
|
||||
|
||||
def load_general_condition_of_heating_system_from_condition_report(self, db_session):
|
||||
return self.get_attribute_and_load(
|
||||
self.condition_report.master_obj.heating_system,
|
||||
"general_condition",
|
||||
GeneralConditionHeatingSystem,
|
||||
db_session,
|
||||
)
|
||||
|
||||
def load_rooms_from_condition_report(self, db_session):
|
||||
hallway = self.load_hallway_from_condition_report(db_session)
|
||||
|
|
@ -564,7 +623,7 @@ class surveyedDataProcessor():
|
|||
lookup_field="UPRN",
|
||||
)
|
||||
|
||||
def get_attribute_and_load(self, obj, attr_string, pydanticModel, db_session, lookup_field=None, additional_fields={}):
|
||||
def get_attribute_and_load(self, obj, attr_string, pydanticModel, db_session, lookup_field=None, additional_fields={}, exclude_list=True):
|
||||
found = getattr(obj, attr_string, None)
|
||||
if found:
|
||||
print(f"Uploading to data base {found}")
|
||||
|
|
@ -576,7 +635,8 @@ class surveyedDataProcessor():
|
|||
model_class=pydanticModel,
|
||||
data_dict=found.model_dump(),
|
||||
lookup_field=lookup_field,
|
||||
additional_fields=additional_fields
|
||||
additional_fields=additional_fields,
|
||||
exclude_list=exclude_list,
|
||||
)
|
||||
return db
|
||||
return None
|
||||
|
|
@ -905,10 +965,15 @@ class surveyedDataProcessor():
|
|||
data_dict,
|
||||
lookup_field=None,
|
||||
update_if_exists: bool = False,
|
||||
additional_fields: dict = None
|
||||
additional_fields: dict = None,
|
||||
exclude_list=True
|
||||
):
|
||||
def remove_nested_dicts_and_lists(data: dict):
|
||||
return {k: v for k, v in data.items() if not isinstance(v, (dict, list))}
|
||||
if exclude_list:
|
||||
return {k: v for k, v in data.items() if not isinstance(v, (dict, list))}
|
||||
else:
|
||||
return {k: v for k, v in data.items() if not isinstance(v, (dict))}
|
||||
|
||||
clean_data = remove_nested_dicts_and_lists(data_dict)
|
||||
|
||||
# Merge additional fields if provided
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue