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

decent home stuff save

Browse source
This commit is contained in:
Jun-te Kim 2025-09-24 11:01:52 +00:00
parent 37a1e60522
commit 515c4601f1
2 changed files with 42 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -296,6 +296,7 @@ def handler(event, context):
json.dump(house, f, indent=2, ensure_ascii=False, default=_json_default)
property_decent_home, decent_home_meta = decent_homes_calc(filepath)
json_uri_1 = upload_json_to_s3(property_decent_home, generate_file_uri(uprn), location="decent_homes/property_decent_home")
with get_db_session() as session:
create_or_update_uploaded_file_entry(

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

@ -408,7 +408,7 @@ def decent_homes_calc(one_property):
label_data = get_element(data["elements"], label)
# Handle no-data or not-applicable
if label_data["ATTRIBUTE CODE"] in ["UNKNOWN", "NONE", "UNKNOWNG", "UNKNOWNS"]:
if label_data["ATTRIBUTE CODE"] in ["UNKNOWN", "NONE", "UNKNOWNG", "UNKNOWNS", "UNKNOWNMAT"] and pd.isnull(label_data["INSTALL DATE"]):
# append_result(
# decent_homes_meta,
# criteria="B",
@ -444,25 +444,28 @@ 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):
install_date = None
raise RuntimeError(f"no Install data label_data:{label_data}")
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"]):
append_result(
decent_homes_meta,
criteria="B",
variable=component,
sub_variable=label,
result="no_data",
install_date=str(install_date),
expiry_date=None,
)
continue
has_failed = label_data["REMAINING LIFE"] < 0
expiry_date = install_date + pd.DateOffset(years=component_lifetime)
expiry_date = today.to_pydatetime() + pd.DateOffset(years=label_data["REMAINING LIFE"])
if has_failed:
component_result = "fail" if is_old and has_failed else "pass"
else:
component_result = "no_data"
else:
component_result = "no_data"
component_result = "fail" if is_old and has_failed else "pass"
# Push into decent_homes_meta
append_result(
@ -488,7 +491,7 @@ def decent_homes_calc(one_property):
kit_age_years = years_between(today.to_pydatetime(), kit_install.to_pydatetime())
kitchen_age_result = "pass" if kit_age_years <= CRITERION_C_AGE_LIMITS["kitchen_years_max"] else "fail"
# For transparency, store next renewal as install + 20 years (criterion C perspective)
kit_next_due = kit_install + pd.DateOffset(years=CRITERION_C_AGE_LIMITS["kitchen_years_max"])
kit_next_due = today.to_pydatetime() + pd.DateOffset(years=kitchen["REMAINING LIFE"])
else:
raise NotImplementedError("Kitchen data missing - pls check")
append_result(
@ -526,7 +529,7 @@ def decent_homes_calc(one_property):
bth_install = pd.to_datetime(bth_install_raw)
bth_age_years = years_between(today.to_pydatetime(), bth_install.to_pydatetime())
bathroom_age_result = "pass" if bth_age_years <= CRITERION_C_AGE_LIMITS["bathroom_years_max"] else "fail"
bth_next_due = bth_install + pd.DateOffset(years=CRITERION_C_AGE_LIMITS["bathroom_years_max"])
bth_next_due = today.to_pydatetime() + pd.DateOffset(years=bath["REMAINING LIFE"])
else:
raise NotImplementedError("Bathroom data missing - pls check")
append_result(
@ -663,13 +666,13 @@ def decent_homes_calc(one_property):
# Wall insulation check
if wall:
wall_code = wall["ATTRIBUTE CODE"]
if wall_code in {"NONE", "SOLID"}: # Means no insulation improvement required
if wall_code in {"NONE"}: # Means no insulation improvement required
wall_result = "pass"
elif wall_code in {"UNKNOWN"}:
wall_result = "no_data"
elif wall_code in {"SOLID"}:
wall_result = "fail"
else:
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")
@ -759,6 +762,22 @@ def decent_homes_calc(one_property):
criterion_d_result = "no_data"
# ---------------- Append to property_decent_homes ----------------
check_pass = [
criterion_a_result,
criterion_b_result,
criterion_c_result,
criterion_d_result
]
decent_homes_result = "no_data"
if all(v == "pass" for v in check_pass):
decent_homes_result = "pass"
elif any(v == "fail" for v in check_pass):
decent_homes_result = "fail"
elif any(v=="no_data" for v in check_pass):
decent_homes_result = "no_data"
property_decent_homes.append({
"uprn": data.get("UPRN"), # TODO: Need UPRN
"creation_date": datetime.now().date().isoformat(),
@ -766,11 +785,7 @@ def decent_homes_calc(one_property):
"criterion_b": criterion_b_result,
"criterion_c": criterion_c_result,
"criterion_d": criterion_d_result,
"decent_homes": (
criterion_a_result == "pass"
and criterion_c_result == "pass"
and criterion_d_result == "pass"
)
"decent_homes": decent_homes_result,
})
return property_decent_homes[0], decent_homes_meta,