mirror of
https://github.com/Hestia-Homes/survey-extraction.git
synced 2026-06-08 11:17:29 +00:00
code finsihed for flats
This commit is contained in:
parent
515c4601f1
commit
0f2579ec4d
2 changed files with 85 additions and 41 deletions
|
|
@ -154,6 +154,11 @@ def uprn_to_address():
|
|||
mapping = df.set_index('Address')['UPRN'].to_dict()
|
||||
return mapping
|
||||
|
||||
def stories_to_address():
|
||||
df = pd.read_excel("../../../../../home/Downloads/data.xlsx", sheet_name="All Energy Breakdown ")
|
||||
mapping = df.set_index('Address')['Storeys'].to_dict()
|
||||
return mapping
|
||||
|
||||
def parse_s3_uri(uri: str):
|
||||
"""
|
||||
Parse an S3 URI or HTTPS S3 URL into bucket and key.
|
||||
|
|
@ -268,6 +273,7 @@ def create_or_update_uploaded_file_entry(
|
|||
def handler(event, context):
|
||||
|
||||
uprn_mapping = uprn_to_address()
|
||||
flats_to_stories = stories_to_address()
|
||||
|
||||
# read data for houses only
|
||||
assets = process_complex("Houses Asset Data")
|
||||
|
|
@ -296,7 +302,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(
|
||||
|
|
@ -327,10 +333,13 @@ def handler(event, context):
|
|||
print(uprn_mapping[pseudo_name.upper()])
|
||||
|
||||
house.update({"UPRN": uprn_mapping[pseudo_name.upper()]})
|
||||
house["property_info"].update({"FLAT LEVEL": flats_to_stories[pseudo_name.upper()]})
|
||||
|
||||
|
||||
|
||||
for i,house in enumerate(flats):
|
||||
print(house["UPRN"])
|
||||
uprn = house["UPRN"]
|
||||
print(uprn)
|
||||
json_uri = upload_json_to_s3(house, generate_file_uri(house["UPRN"]))
|
||||
|
||||
# Save JSON locally
|
||||
|
|
@ -342,14 +351,30 @@ 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(
|
||||
db_session=session,
|
||||
uprn=uprn,
|
||||
doc_type=ReportType.DECENT_HOMES_SUMMARY,
|
||||
json_uri=json_uri_1,
|
||||
s3_file_uri=json_uri,
|
||||
)
|
||||
json_uri_1 = upload_json_to_s3(decent_home_meta, generate_file_uri(uprn), location="decent_homes/decent_homes_meta")
|
||||
with get_db_session() as session:
|
||||
create_or_update_uploaded_file_entry(
|
||||
db_session=session,
|
||||
uprn=uprn,
|
||||
doc_type=ReportType.DECENT_HOMES_PROPERTY_META,
|
||||
json_uri=json_uri_1,
|
||||
s3_file_uri=json_uri,
|
||||
)
|
||||
|
||||
# Keep track of saved file path
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# run a script that upload to s3 -> uprn -> jsonified -> walthamforest -> uri
|
||||
|
||||
# To Do:
|
||||
# [Jun-te] Spec of quesation that we have for waltham forest
|
||||
# [Khalim] A document that has our mapping and our understanding of our data
|
||||
|
|
@ -141,7 +141,12 @@ def decent_homes_calc(one_property):
|
|||
LABEL_NOISE = "Adequacy of Noise Insulation in Property"
|
||||
LABEL_COMMON_CIRC = "Circulation Space in Common Area" # flats only
|
||||
|
||||
STANDARD_HHSRS_MAPPING = {"pass": "TYPRISK", "fail": "MODRISK", "no_data": "TOBEASSESS"}
|
||||
|
||||
STANDARD_HHSRS_MAPPING = {
|
||||
"pass": ["TYPRISK"],
|
||||
"fail": ["MODRISK","SLIGHTRISK"],
|
||||
"no_data": ["TOBEASSESS"],
|
||||
}
|
||||
|
||||
# Criterion A - mapping of HHSRS variables to Waltham forest element codes
|
||||
HHSRS_MAPPING = {
|
||||
|
|
@ -351,8 +356,10 @@ def decent_homes_calc(one_property):
|
|||
if property_info["PROP TYPE"] in ["HOU"]:
|
||||
property_type = "house"
|
||||
elif property_info["PROP TYPE"] == "FLA":
|
||||
raise NotImplementedError("Implement distrinction between below and above 6 storeys")
|
||||
# property_type = "flat"
|
||||
if property_info["FLAT LEVEL"] < 6:
|
||||
property_type = "flat_below_6_storeys"
|
||||
else:
|
||||
property_type = "flat_above_6_storeys"
|
||||
else:
|
||||
raise NotImplementedError("Unknown property type")
|
||||
|
||||
|
|
@ -361,21 +368,20 @@ def decent_homes_calc(one_property):
|
|||
# If fail, why?
|
||||
for hhsrs_variable, mapping in HHSRS_MAPPING.items():
|
||||
element_code = list(mapping.keys())[0]
|
||||
|
||||
# Find the data in the JSON within data["elements"]
|
||||
check_pass = []
|
||||
for k, v in data["elements"].items():
|
||||
if v["ELEMENT CODE"] == element_code:
|
||||
# We check the attribute code
|
||||
# Check if pass
|
||||
if v["ATTRIBUTE CODE"] == mapping[element_code]["pass"]:
|
||||
if v["ATTRIBUTE CODE"] in mapping[element_code]["pass"]:
|
||||
result = "pass"
|
||||
elif v["ATTRIBUTE CODE"] == mapping[element_code]["fail"]:
|
||||
elif v["ATTRIBUTE CODE"] in mapping[element_code]["fail"]:
|
||||
result = "fail"
|
||||
elif v["ATTRIBUTE CODE"] == mapping[element_code]["no_data"]:
|
||||
elif v["ATTRIBUTE CODE"] in mapping[element_code]["no_data"]:
|
||||
result = "no_data"
|
||||
else:
|
||||
raise ValueError("Unknown attribute code")
|
||||
raise ValueError(f"Unknown attribute code: '{v[element_code]}")
|
||||
check_pass.append(result)
|
||||
append_result(
|
||||
decent_homes_meta,
|
||||
|
|
@ -409,15 +415,6 @@ def decent_homes_calc(one_property):
|
|||
|
||||
# Handle no-data or not-applicable
|
||||
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",
|
||||
# variable=component,
|
||||
# sub_variable=label,
|
||||
# result="pass",
|
||||
# install_date=None,
|
||||
# expiry_date=None,
|
||||
# )
|
||||
continue
|
||||
|
||||
# Special skip conditions for heating
|
||||
|
|
@ -444,8 +441,16 @@ 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 RuntimeError(f"no Install data label_data:{label_data}")
|
||||
|
||||
append_result(
|
||||
decent_homes_meta,
|
||||
criteria="B",
|
||||
variable=component,
|
||||
sub_variable=label,
|
||||
result="no_data",
|
||||
install_date=str(install_date),
|
||||
expiry_date=None,
|
||||
)
|
||||
continue
|
||||
component_lifetime = COMPONENT_LIFESPANS[component][property_type]
|
||||
is_old = years_between(today.to_pydatetime(), install_date.to_pydatetime()) > component_lifetime
|
||||
|
||||
|
|
@ -547,8 +552,10 @@ def decent_homes_calc(one_property):
|
|||
bth_attr_code = bath["ATTRIBUTE CODE"]
|
||||
if bth_attr_code in {"STDBTHADQ", "ADPBTHADQ"}:
|
||||
bathroom_location_result = "pass"
|
||||
elif bth_attr_code in {"STDBTHINAD"}:
|
||||
bathroom_location_result = "fail"
|
||||
else:
|
||||
raise NotImplementedError("No other observed codes yet")
|
||||
raise NotImplementedError(f"No other observed codes yet {bth_attr_code}")
|
||||
else:
|
||||
raise NotImplementedError("Bathroom data missing - pls check")
|
||||
|
||||
|
|
@ -580,14 +587,22 @@ def decent_homes_calc(one_property):
|
|||
|
||||
# 6) Adequate common entrance areas (flats only)
|
||||
if is_flat:
|
||||
raise Exception("Pls check this")
|
||||
common = get_element(data["elements"], LABEL_COMMON_CIRC)
|
||||
if common:
|
||||
circ_desc = common.get("ATTRIBUTE CODE DESCRIPTION", "")
|
||||
common_areas_result = adequacy_result_by_text(circ_desc)
|
||||
circ_desc = common["ATTRIBUTE CODE DESCRIPTION"]
|
||||
if circ_desc in {"Adequate Circulation Space in Common Area"}:
|
||||
common_areas_result = "pass"
|
||||
else:
|
||||
raise NotImplementedError(f"New description on common area {circ_desc}")
|
||||
else:
|
||||
common_areas_result = "no_data"
|
||||
append_result(decent_homes_meta, "adequate_common_entrance_areas", common_areas_result)
|
||||
append_result(
|
||||
decent_homes_meta=decent_homes_meta,
|
||||
criteria="C",
|
||||
variable="adequate_common_entrance_areas",
|
||||
sub_variable="adequate_common_entrance_areas",
|
||||
result=common_areas_result,
|
||||
)
|
||||
|
||||
# ---------------- Criterion D ----------------
|
||||
# heating system type
|
||||
|
|
@ -651,17 +666,21 @@ def decent_homes_calc(one_property):
|
|||
loft_result = "fail"
|
||||
elif loft_code.isnumeric():
|
||||
loft_result = "pass"
|
||||
elif loft_code == "UNKNOWN":
|
||||
loft_result = None
|
||||
else:
|
||||
raise NotImplementedError("Unknown loft insulation code - pls check")
|
||||
raise NotImplementedError(f"Unknown loft insulation code - pls check {loft_code}")
|
||||
else:
|
||||
raise NotImplementedError("Loft insulation data missing - pls check")
|
||||
append_result(
|
||||
decent_homes_meta,
|
||||
criteria="D",
|
||||
variable="loft_insulation_sufficient",
|
||||
sub_variable="loft_insulation_sufficient",
|
||||
result=loft_result
|
||||
)
|
||||
|
||||
if loft_result:
|
||||
append_result(
|
||||
decent_homes_meta,
|
||||
criteria="D",
|
||||
variable="loft_insulation_sufficient",
|
||||
sub_variable="loft_insulation_sufficient",
|
||||
result=loft_result
|
||||
)
|
||||
|
||||
# Wall insulation check
|
||||
if wall:
|
||||
|
|
@ -673,7 +692,7 @@ def decent_homes_calc(one_property):
|
|||
elif wall_code in {"SOLID"}:
|
||||
wall_result = "fail"
|
||||
else:
|
||||
raise NotImplementedError(f"No other observed codes yet")
|
||||
raise NotImplementedError(f"No other observed codes yet {wall_code}")
|
||||
else:
|
||||
raise NotImplementedError("Wall insulation data missing - pls check")
|
||||
append_result(
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue