junte/survey-extraction
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/survey-extraction.git synced 2026-06-30 13:10:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

flats requre some thought

Browse source
This commit is contained in:
Jun-te Kim 2025-09-23 14:53:57 +00:00
parent 6c214c9f89
commit 37a1e60522
2 changed files with 41 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
deployment/lambda/walthamforest_etl/docker/app.py
View file

@ -115,7 +115,10 @@ def combine_records_for_flats(assets: dict, simple: list) -> dict:
for record in assets:
# Make sure record is a dict
record.update({"BLOCK_INFO": block_info})
# record.update({"BLOCK_INFO": block_info})
for ele_desc in block_info["elements"]:
if ele_desc not in record["elements"]:
record["elements"].update({ele_desc:block_info["elements"][ele_desc]})
return assets
@ -237,7 +240,7 @@ def create_or_update_uploaded_file_entry(
"""
existing = (
db_session.query(uploaded_files)
.filter(uploaded_files.uprn == uprn, uploaded_files.doc_type == doc_type)
.filter(uploaded_files.uprn == str(uprn), uploaded_files.doc_type == doc_type)
.one_or_none()
)
@ -254,7 +257,7 @@ def create_or_update_uploaded_file_entry(
s3_json_uri=json_uri,
s3_json_upload_timestamp=datetime.now(timezone.utc),
s3_file_uri=s3_file_uri,
uprn=uprn,
uprn=str(uprn),
)
db_session.add(obj)
@ -278,9 +281,8 @@ def handler(event, context):
#upload to s3
saved_paths = []
for house in houses:
for i,house in enumerate(houses):
uprn = house["UPRN"]
print(uprn)
json_uri = upload_json_to_s3(house, generate_file_uri(house["UPRN"]), location="decent_homes/raw_data")
@ -308,12 +310,10 @@ def handler(event, context):
create_or_update_uploaded_file_entry(
db_session=session,
uprn=uprn,
doc_type=ReportType.DECENT_HOMES_SUMMARY,
doc_type=ReportType.DECENT_HOMES_PROPERTY_META,
json_uri=json_uri_1,
s3_file_uri=json_uri,
)
# Keep track of saved file path
saved_paths.append(filepath)
# read data for flats
assets = process_complex("Chingford Rd 236-256 Properties")
@ -328,9 +328,10 @@ def handler(event, context):
house.update({"UPRN": uprn_mapping[pseudo_name.upper()]})
for house in flats:
for i,house in enumerate(flats):
print(house["UPRN"])
json_uri = upload_json_to_s3(house, generate_file_uri(house["UPRN"]))
# Save JSON locally
filename = f"{house['UPRN']}.json"
filepath = os.path.join("output", filename) # saves inside an "output" folder
@ -339,8 +340,9 @@ def handler(event, context):
with open(filepath, "w", encoding="utf-8") as f:
json.dump(house, f, indent=2, ensure_ascii=False, default=_json_default)
property_decent_home, decent_home_meta = decent_homes_calc(filepath)
# Keep track of saved file path
saved_paths.append(filepath)

43
deployment/lambda/walthamforest_etl/docker/decent_homes_pilot.py
View file

@ -444,17 +444,25 @@ def decent_homes_calc(one_property):
# Normal case: evaluate install date + lifetime + remaining life
install_date = pd.to_datetime(label_data["INSTALL DATE"])
if pd.isnull(install_date):
raise ValueError(f"Missing install date for {component}/{label}")
install_date = None
if install_date:
component_lifetime = COMPONENT_LIFESPANS[component][property_type]
is_old = years_between(today.to_pydatetime(), install_date.to_pydatetime()) > component_lifetime
component_lifetime = COMPONENT_LIFESPANS[component][property_type]
is_old = years_between(today.to_pydatetime(), install_date.to_pydatetime()) > component_lifetime
if pd.isnull(label_data["REMAINING LIFE"]):
has_failed = None
else:
has_failed = label_data["REMAINING LIFE"] < 0
if pd.isnull(label_data["REMAINING LIFE"]):
raise ValueError(f"Missing remaining life for {component}/{label}")
has_failed = label_data["REMAINING LIFE"] < 0
expiry_date = install_date + pd.DateOffset(years=component_lifetime)
expiry_date = install_date + pd.DateOffset(years=component_lifetime)
component_result = "fail" if is_old and has_failed else "pass"
if has_failed:
component_result = "fail" if is_old and has_failed else "pass"
else:
component_result = "no_data"
else:
component_result = "no_data"
# Push into decent_homes_meta
append_result(
@ -608,7 +616,12 @@ def decent_homes_calc(one_property):
# For the observed case, there was no heating and wet heating needed to be installed in full so the value
# was unknown
heating_dist_result = "no_data"
elif dist_code in {"RADIATORS", "ELECWARMAR"}:
# Found one with heating distribution - check with Khalim if this is pass
heating_dist_result = "pass"
else:
print(f"heating_dist {heating_dist}")
print(f"dist-code {dist_code}")
raise NotImplementedError("No other observed codes yet")
else:
raise NotImplementedError("Heating distribution element missing in dataset")
@ -650,10 +663,14 @@ def decent_homes_calc(one_property):
# Wall insulation check
if wall:
wall_code = wall["ATTRIBUTE CODE"]
if wall_code in {"NONE"}: # Means no insulation improvement required
if wall_code in {"NONE", "SOLID"}: # Means no insulation improvement required
wall_result = "pass"
elif wall_code in {"UNKNOWN"}:
wall_result = "no_data"
else:
raise NotImplementedError("No other observed codes yet")
print(f"wall {wall}")
print(f"wall_code {wall_code}")
raise NotImplementedError(f"No other observed codes yet")
else:
raise NotImplementedError("Wall insulation data missing - pls check")
append_result(
@ -688,10 +705,8 @@ def decent_homes_calc(one_property):
comp_result = "no_data"
elif any(r == "fail" for r in comp_sub_results):
comp_result = "fail"
elif all(r == "pass" for r in comp_sub_results if r != "no_data"):
elif all(r == "pass" for r in comp_sub_results):
comp_result = "pass"
elif all(r == "no_data" for r in comp_sub_results):
comp_result = "no_data"
else:
comp_result = "no_data"
@ -704,7 +719,7 @@ def decent_homes_calc(one_property):
criterion_b_result = "fail"
elif len(other_fails) >= 2:
criterion_b_result = "fail"
elif all(r == "no_data" for r in component_results.values()):
elif any(r == "no_data" for r in component_results.values()):
criterion_b_result = "no_data"
else:
criterion_b_result = "pass"