mirror of
https://github.com/Hestia-Homes/survey-extraction.git
synced 2026-06-30 13:10:56 +00:00
Merge pull request #89 from Hestia-Homes/feautre/walthamforest_etl
Feautre/walthamforest etl
This commit is contained in:
Jun-te Kim
GitHub
commit
a5499fb784
12 changed files with 1640 additions and 600 deletions
1
.github/workflows/months_end.yml
vendored
1
.github/workflows/months_end.yml
vendored
|
|
@ -4,6 +4,7 @@ on:
|
|||
schedule:
|
||||
- cron: '0 7 * * 1' # Every Monday at 07:00 UTC
|
||||
- cron: '0 7 29 * *' # On the 29th of every month at 07:00 UTC
|
||||
- cron: '0 7 26 * *' # On the 26th of every month at 07:00 UTC
|
||||
workflow_dispatch:
|
||||
|
||||
jobs:
|
||||
|
|
|
|||
60
alembic/versions/4c67501b7451_added_more_enums.py
Normal file
60
alembic/versions/4c67501b7451_added_more_enums.py
Normal file
|
|
@ -0,0 +1,60 @@
|
|||
"""added more enums
|
||||
|
||||
Revision ID: 4c67501b7451
|
||||
Revises: ac8dba8cef50
|
||||
Create Date: 2025-09-23 10:22:20.648664
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = "4c67501b7451"
|
||||
down_revision: Union[str, None] = "ac8dba8cef50"
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
ENUM_NAME = "reporttype"
|
||||
|
||||
# Values that were already present BEFORE this migration
|
||||
OLD_VALUES = (
|
||||
"QUIDOS_PRESITE_NOTE",
|
||||
"CHARTED_SURVEYOR_REPORT",
|
||||
"ENERGY_PERFORMANCE_REPORT",
|
||||
"U_VALUE_CALCULATOR_REPORT",
|
||||
"OVERWRITING_U_VALUE_DECLARATION_FORM",
|
||||
"OSMOSIS_CONDITION_PAS_2035_REPORT",
|
||||
"DOMNA_CONDITION_PAS_2035_REPORT",
|
||||
)
|
||||
|
||||
# Values we are ADDING in this migration
|
||||
NEW_VALUES = (
|
||||
"DECENT_HOMES_RAW_DATA",
|
||||
"DECENT_HOMES_SUMMARY",
|
||||
"DECENT_HOMES_PROPERTY_META",
|
||||
)
|
||||
|
||||
def upgrade() -> None:
|
||||
for v in NEW_VALUES:
|
||||
op.execute(f"ALTER TYPE {ENUM_NAME} ADD VALUE IF NOT EXISTS '{v}'")
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
# 1) Create a replacement type with ONLY the old values
|
||||
old_vals = ", ".join(f"'{v}'" for v in OLD_VALUES)
|
||||
op.execute(f"CREATE TYPE {ENUM_NAME}_old AS ENUM ({old_vals})")
|
||||
|
||||
# 2) Move columns to the temporary type
|
||||
op.execute(
|
||||
f"ALTER TABLE documents ALTER COLUMN document_type TYPE {ENUM_NAME}_old "
|
||||
f"USING document_type::text::{ENUM_NAME}_old"
|
||||
)
|
||||
op.execute(
|
||||
f"ALTER TABLE uploaded_files ALTER COLUMN doc_type TYPE {ENUM_NAME}_old "
|
||||
f"USING doc_type::text::{ENUM_NAME}_old"
|
||||
)
|
||||
|
||||
# 3) Drop original type and rename the temp back
|
||||
op.execute(f"DROP TYPE {ENUM_NAME}")
|
||||
op.execute(f"ALTER TYPE {ENUM_NAME}_old RENAME TO {ENUM_NAME}")
|
||||
38
alembic/versions/ac8dba8cef50_added_more_report_type.py
Normal file
38
alembic/versions/ac8dba8cef50_added_more_report_type.py
Normal file
|
|
@ -0,0 +1,38 @@
|
|||
"""added more report type
|
||||
|
||||
Revision ID: ac8dba8cef50
|
||||
Revises: a8cc4a5fccb6
|
||||
Create Date: 2025-09-23 10:14:54.461633
|
||||
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = 'ac8dba8cef50'
|
||||
down_revision: Union[str, None] = 'a8cc4a5fccb6'
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
"""Upgrade schema."""
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
op.alter_column('uploaded_files', 'id',
|
||||
existing_type=sa.UUID(),
|
||||
server_default=None,
|
||||
existing_nullable=False)
|
||||
# ### end Alembic commands ###
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
"""Downgrade schema."""
|
||||
# ### commands auto generated by Alembic - please adjust! ###
|
||||
op.alter_column('uploaded_files', 'id',
|
||||
existing_type=sa.UUID(),
|
||||
server_default=sa.text('gen_random_uuid()'),
|
||||
existing_nullable=False)
|
||||
# ### end Alembic commands ###
|
||||
|
|
@ -8,7 +8,6 @@ from etl.fileReader.sitenotes import (
|
|||
SiteNotesExtractor,
|
||||
WarmHomesConditionReport
|
||||
)
|
||||
|
||||
from uuid import UUID
|
||||
import json
|
||||
from typing import Any
|
||||
|
|
|
|||
|
|
@ -6,8 +6,16 @@ import copy
|
|||
from collections import defaultdict
|
||||
from typing import List, Dict, Any, Union, Optional
|
||||
import boto3
|
||||
import datetime
|
||||
from urllib.parse import urlparse
|
||||
from decent_homes_pilot import decent_homes_calc
|
||||
import uuid
|
||||
from datetime import datetime, timezone, time, date
|
||||
from decimal import Decimal
|
||||
from sqlmodel import select
|
||||
from sqlalchemy import update
|
||||
from etl.models.topLevel import uploaded_files, ReportType
|
||||
from etl.db.db import get_db_session, init_db
|
||||
|
||||
|
||||
def process_complex(sheet_name, group_key="ADDRESS"):
|
||||
df = pd.read_excel("../../../../../home/Downloads/data.xlsx", sheet_name=sheet_name)
|
||||
|
|
@ -107,13 +115,14 @@ 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
|
||||
|
||||
def _json_default(o):
|
||||
from datetime import date, datetime, time
|
||||
from decimal import Decimal
|
||||
# datetimes → ISO 8601 strings
|
||||
if isinstance(o, (datetime, date, time)):
|
||||
return o.isoformat()
|
||||
|
|
@ -145,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.
|
||||
|
|
@ -171,7 +185,7 @@ def parse_s3_uri(uri: str):
|
|||
|
||||
return bucket, key
|
||||
|
||||
def upload_json_to_s3(json_obj, dest_uri: str) -> str:
|
||||
def upload_json_to_s3(json_obj, dest_uri: str, location="decent_homes/raw_data") -> str:
|
||||
"""
|
||||
Upload a JSON-serializable object to S3 at the given s3:// or https S3 URL.
|
||||
Returns the public-style HTTPS S3 URL (still private if bucket is private).
|
||||
|
|
@ -180,8 +194,8 @@ def upload_json_to_s3(json_obj, dest_uri: str) -> str:
|
|||
base_folder = os.path.dirname(pdf_key) # e.g. ".../report"
|
||||
|
||||
# Build jsonified folder + timestamp filename
|
||||
timestamp = datetime.datetime.utcnow().strftime("%Y%m%d_%H%M%S")
|
||||
json_key = f"{base_folder}/walthamforest_raw_data/jsonified/{timestamp}.json"
|
||||
timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
|
||||
json_key = f"{base_folder}/{location}/jsonified/{timestamp}.json"
|
||||
|
||||
# Same region/creds you used for download
|
||||
aws_access_key = "AKIAU5A36PPNJMZZ3KRW"
|
||||
|
|
@ -212,14 +226,54 @@ def upload_json_to_s3(json_obj, dest_uri: str) -> str:
|
|||
|
||||
|
||||
def generate_file_uri(UPRN):
|
||||
timestamp = datetime.datetime.utcnow().strftime("%Y%m%d_%H%M%S")
|
||||
timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
|
||||
file_uri = f"https://retrofit-energy-assessments-dev.s3.eu-west-2.amazonaws.com/documents/{UPRN}/"
|
||||
return file_uri
|
||||
|
||||
def create_or_update_uploaded_file_entry(
|
||||
db_session,
|
||||
uprn: str,
|
||||
doc_type: ReportType,
|
||||
json_uri: str,
|
||||
s3_file_uri: str
|
||||
):
|
||||
"""
|
||||
Create or update an entry in uploaded_files.
|
||||
- If a record with the same (uprn, doc_type) exists, update it.
|
||||
- Otherwise, insert a new record.
|
||||
Commits, refreshes, and returns the ORM object.
|
||||
"""
|
||||
existing = (
|
||||
db_session.query(uploaded_files)
|
||||
.filter(uploaded_files.uprn == str(uprn), uploaded_files.doc_type == doc_type)
|
||||
.one_or_none()
|
||||
)
|
||||
|
||||
if existing:
|
||||
# Update existing record
|
||||
existing.s3_json_uri = json_uri
|
||||
existing.s3_json_upload_timestamp = datetime.now(timezone.utc)
|
||||
existing.s3_file_uri = s3_file_uri
|
||||
obj = existing
|
||||
else:
|
||||
# Insert new record
|
||||
obj = uploaded_files(
|
||||
doc_type=doc_type,
|
||||
s3_json_uri=json_uri,
|
||||
s3_json_upload_timestamp=datetime.now(timezone.utc),
|
||||
s3_file_uri=s3_file_uri,
|
||||
uprn=str(uprn),
|
||||
)
|
||||
db_session.add(obj)
|
||||
|
||||
db_session.commit()
|
||||
db_session.refresh(obj)
|
||||
return obj
|
||||
|
||||
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")
|
||||
|
|
@ -232,9 +286,41 @@ def handler(event, context):
|
|||
house.update({"UPRN": uprn_mapping[pseudo_name.upper()]})
|
||||
|
||||
#upload to s3
|
||||
for house in houses:
|
||||
print(house["UPRN"])
|
||||
json_uri = upload_json_to_s3(house, generate_file_uri(house["UPRN"]))
|
||||
|
||||
|
||||
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")
|
||||
|
||||
# Save JSON locally
|
||||
filename = f"{uprn}.json"
|
||||
filepath = os.path.join("output", filename) # saves inside an "output" folder
|
||||
os.makedirs("output", exist_ok=True) # make sure folder exists
|
||||
|
||||
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)
|
||||
|
||||
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,
|
||||
)
|
||||
|
||||
# read data for flats
|
||||
assets = process_complex("Chingford Rd 236-256 Properties")
|
||||
|
|
@ -245,18 +331,50 @@ def handler(event, context):
|
|||
pseudo_name = house["ADDRESS"].split(",")[0]
|
||||
if pseudo_name.lower() in (k.lower() for k in uprn_mapping.keys()):
|
||||
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 house in flats:
|
||||
print(house["UPRN"])
|
||||
|
||||
for i,house in enumerate(flats):
|
||||
uprn = house["UPRN"]
|
||||
print(uprn)
|
||||
json_uri = upload_json_to_s3(house, generate_file_uri(house["UPRN"]))
|
||||
# with open("flat.json", "w") as f:
|
||||
# json.dump(houses[0], f, indent=2, ensure_ascii=False, default=_json_default)
|
||||
|
||||
# Save JSON locally
|
||||
filename = f"{house['UPRN']}.json"
|
||||
filepath = os.path.join("output", filename) # saves inside an "output" folder
|
||||
os.makedirs("output", exist_ok=True) # make sure folder exists
|
||||
|
||||
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)
|
||||
|
||||
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
|
||||
811
deployment/lambda/walthamforest_etl/docker/decent_homes_pilot.py
Normal file
811
deployment/lambda/walthamforest_etl/docker/decent_homes_pilot.py
Normal file
|
|
@ -0,0 +1,811 @@
|
|||
import json
|
||||
import os
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from docutils.nodes import table
|
||||
|
||||
|
||||
def years_between(d1, d2):
|
||||
# precise year difference (accounts for months/days)
|
||||
return (d1.year - d2.year) - ((d1.month, d1.day) < (d2.month, d2.day))
|
||||
|
||||
|
||||
def get_element(elements, label):
|
||||
"""Safely get an element dict by display label (your JSON keys)."""
|
||||
return elements.get(label)
|
||||
|
||||
|
||||
def append_result(decent_homes_meta, criteria, variable, sub_variable, result, install_date=None, expiry_date=None):
|
||||
decent_homes_meta.append({
|
||||
"criteria": criteria,
|
||||
"variable": variable,
|
||||
"sub_variable": sub_variable,
|
||||
"result": result,
|
||||
"hhsrs_rank": None,
|
||||
"hhsrs_score": None,
|
||||
"install_date": install_date,
|
||||
"expiry_date": expiry_date,
|
||||
})
|
||||
|
||||
|
||||
def decent_homes_calc(one_property):
|
||||
# Read in static json, which is transformed by Jun-te's script
|
||||
folder = "../../../../../home/Downloads/"
|
||||
fn = one_property
|
||||
|
||||
# filenames = ["flat 1.json", "house 1.json"]
|
||||
|
||||
houses_waltham_forest_data = pd.read_excel(
|
||||
os.path.join(folder, "data.xlsx"),
|
||||
sheet_name="Houses Asset Data"
|
||||
)
|
||||
flats_waltham_forest_data = pd.read_excel(
|
||||
os.path.join(folder, "data.xlsx"),
|
||||
sheet_name="CHINGFORD ROAD 236-254 Asset Bl"
|
||||
)
|
||||
|
||||
# Standardised variables which will form the enums in the db
|
||||
HHSRS_VARIABLES = [
|
||||
"damp_and_mould_growth",
|
||||
"excess_cold",
|
||||
"excess_heat",
|
||||
"asbestos_and_mm_fibres",
|
||||
"biocides",
|
||||
"carbon_monoxide_and_fuel_combustion_products",
|
||||
"lead",
|
||||
"radiation",
|
||||
"uncombusted_fuel_gas",
|
||||
"volatile_organic_compounds",
|
||||
"crowding_and_space",
|
||||
"entry_by_intruders",
|
||||
"lighting",
|
||||
"noise",
|
||||
"domestic_hygiene_pests_and_refuse",
|
||||
"food_safety",
|
||||
"personal_hygiene_sanitation_and_drainage",
|
||||
"water_supply",
|
||||
"falls_associated_with_baths",
|
||||
"falls_on_level_surfaces",
|
||||
"falls_on_stairs_and_steps",
|
||||
"falls_between_levels",
|
||||
"electrical_hazards",
|
||||
"fire",
|
||||
"flames_hot_surfaces_and_materials",
|
||||
"collision_and_entrapment",
|
||||
"explosions",
|
||||
"ergonomics",
|
||||
"structural_collapse_and_falling_elements"
|
||||
]
|
||||
|
||||
ELEMENT_CODE_TO_DESCRIPTION = {
|
||||
# One-to-one
|
||||
"HHSRSDAMP": "damp_and_mould_growth",
|
||||
"HHSRSCOLD": "excess_cold",
|
||||
"HHSRSHEAT": "excess_heat",
|
||||
"HHSRSASB": "asbestos_and_mm_fibres",
|
||||
"HHSRSBIOC": "biocides",
|
||||
"HHSRSLEAD": "lead",
|
||||
"HHSRSRADIA": "radiation",
|
||||
"HHSRSFUEL": "uncombusted_fuel_gas",
|
||||
"HHSRSORGAN": "volatile_organic_compounds",
|
||||
"HHSRSCROWD": "crowding_and_space",
|
||||
"HHSRSENTRY": "entry_by_intruders",
|
||||
"HHSRSLIGHT": "lighting",
|
||||
"HHSRSNOISE": "noise",
|
||||
"HHSRSDOMES": "domestic_hygiene_pests_and_refuse",
|
||||
"HHSRSFOOD": "food_safety",
|
||||
"HHSRSPERS": "personal_hygiene_sanitation_and_drainage",
|
||||
"HHSRSWATER": "water_supply",
|
||||
"HHSRSFBATH": "falls_associated_with_baths",
|
||||
"HHSRSFLEVE": "falls_on_level_surfaces",
|
||||
"HHSRSFSTAI": "falls_on_stairs_and_steps",
|
||||
"HHSRSFBETW": "falls_between_levels",
|
||||
"HHSRSELEC": "electrical_hazards",
|
||||
"HHSRSFIRE": "fire",
|
||||
"HHSRSFLAME": "flames_hot_surfaces_and_materials",
|
||||
"HHSRSEXPLO": "explosions",
|
||||
"HHSRSPOSI": "ergonomics",
|
||||
"HHSRSSTRUC": "structural_collapse_and_falling_elements",
|
||||
|
||||
# One-to-many expansions
|
||||
"HHSRSCO": "carbon_monoxide",
|
||||
"HHSRSSO2": "sulphur_dioxide_and_smoke",
|
||||
"HHSRSNO2": "nitrogen_dioxide",
|
||||
"HHSRSENTRP": "collision_and_entrapment",
|
||||
"HHSRSCLOW": "collision_hazards_and_low_headroom",
|
||||
}
|
||||
|
||||
CRITERION_B_VARIABLES = [
|
||||
"external_walls_structure", "lintels", "brickwork_spalling", "wall_finish", "roof_structure", "roof_finish",
|
||||
"chimneys", "windows", "external_doors", "kitchens", "bathrooms", "central_heating_boiler",
|
||||
"central_heating_distribution_system", "heating_other", "electrical_systems",
|
||||
]
|
||||
|
||||
CRITERION_C_VARIABLES = [
|
||||
"kitchen_less_than_20_years_old", "kitchen_adequate_space_and_layout", "bathroom_less_than_30_years_old",
|
||||
"bathroom_wc_appropriately_located", "adequate_external_noise_insulation", "adequate_common_entrance_areas",
|
||||
]
|
||||
|
||||
# Criterion C explicit age limits (different from component lifespans used elsewhere)
|
||||
CRITERION_C_AGE_LIMITS = {
|
||||
"kitchen_years_max": 20,
|
||||
"bathroom_years_max": 30,
|
||||
}
|
||||
|
||||
# Field labels as they appear in your JSON (based on your code)
|
||||
LABEL_KITCHEN = "Adequacy of Kitchen and Type in Property"
|
||||
LABEL_BATHROOM = "Adequacy of Bathroom Location in 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","SLIGHTRISK"],
|
||||
"no_data": ["TOBEASSESS"],
|
||||
}
|
||||
|
||||
# Criterion A - mapping of HHSRS variables to Waltham forest element codes
|
||||
HHSRS_MAPPING = {
|
||||
"damp_and_mould_growth": {"HHSRSDAMP": STANDARD_HHSRS_MAPPING},
|
||||
"excess_cold": {"HHSRSCOLD": STANDARD_HHSRS_MAPPING},
|
||||
"excess_heat": {"HHSRSHEAT": STANDARD_HHSRS_MAPPING},
|
||||
"asbestos_and_mm_fibres": {"HHSRSASB": STANDARD_HHSRS_MAPPING},
|
||||
"biocides": {"HHSRSBIOC": STANDARD_HHSRS_MAPPING},
|
||||
"carbon_monoxide_and_fuel_combustion_products": {
|
||||
"HHSRSCO": STANDARD_HHSRS_MAPPING,
|
||||
"HHSRSSO2": STANDARD_HHSRS_MAPPING,
|
||||
"HHSRSNO2": STANDARD_HHSRS_MAPPING
|
||||
},
|
||||
"lead": {"HHSRSLEAD": STANDARD_HHSRS_MAPPING},
|
||||
"radiation": {"HHSRSRADIA": STANDARD_HHSRS_MAPPING},
|
||||
"uncombusted_fuel_gas": {"HHSRSFUEL": STANDARD_HHSRS_MAPPING},
|
||||
"volatile_organic_compounds": {"HHSRSORGAN": STANDARD_HHSRS_MAPPING},
|
||||
"crowding_and_space": {"HHSRSCROWD": STANDARD_HHSRS_MAPPING},
|
||||
"entry_by_intruders": {"HHSRSENTRY": STANDARD_HHSRS_MAPPING},
|
||||
"lighting": {"HHSRSLIGHT": STANDARD_HHSRS_MAPPING},
|
||||
"noise": {"HHSRSNOISE": STANDARD_HHSRS_MAPPING},
|
||||
"domestic_hygiene_pests_and_refuse": {"HHSRSDOMES": STANDARD_HHSRS_MAPPING},
|
||||
"food_safety": {"HHSRSFOOD": STANDARD_HHSRS_MAPPING},
|
||||
"personal_hygiene_sanitation_and_drainage": {"HHSRSPERS": STANDARD_HHSRS_MAPPING},
|
||||
"water_supply": {"HHSRSWATER": STANDARD_HHSRS_MAPPING},
|
||||
"falls_associated_with_baths": {"HHSRSFBATH": STANDARD_HHSRS_MAPPING},
|
||||
"falls_on_level_surfaces": {"HHSRSFLEVE": STANDARD_HHSRS_MAPPING},
|
||||
"falls_on_stairs_and_steps": {"HHSRSFSTAI": STANDARD_HHSRS_MAPPING},
|
||||
"falls_between_levels": {"HHSRSFBETW": STANDARD_HHSRS_MAPPING},
|
||||
"electrical_hazards": {"HHSRSELEC": STANDARD_HHSRS_MAPPING},
|
||||
"fire": {"HHSRSFIRE": STANDARD_HHSRS_MAPPING},
|
||||
"flames_hot_surfaces_and_materials": {"HHSRSFLAME": STANDARD_HHSRS_MAPPING},
|
||||
"collision_and_entrapment": {"HHSRSENTRP": STANDARD_HHSRS_MAPPING, "HHSRSCLOW": STANDARD_HHSRS_MAPPING},
|
||||
"explosions": {"HHSRSEXPLO": STANDARD_HHSRS_MAPPING},
|
||||
"ergonomics": {"HHSRSPOSI": STANDARD_HHSRS_MAPPING},
|
||||
"structural_collapse_and_falling_elements": {"HHSRSSTRUC": STANDARD_HHSRS_MAPPING}
|
||||
}
|
||||
|
||||
# print(houses_waltham_forest_data[
|
||||
# houses_waltham_forest_data["ELEMENT CODE"] == "INTBTHADEQ"
|
||||
# ][["ATTRIBUTE CODE", "ATTRIBUTE CODE DESCRIPTION"]].drop_duplicates())
|
||||
|
||||
# print(flats_waltham_forest_data[
|
||||
# flats_waltham_forest_data["ELEMENT CODE"] == "INTBTHADEQ"
|
||||
# ][["ATTRIBUTE CODE", "ATTRIBUTE CODE DESCRIPTION"]].drop_duplicates())
|
||||
|
||||
|
||||
# Criterion B
|
||||
B_COMPONENT_LABELS = {
|
||||
# Key components
|
||||
"wall_structure": [
|
||||
"Wall Structure in External Area",
|
||||
],
|
||||
"lintels": [
|
||||
"Lintels in External Area",
|
||||
],
|
||||
"brickwork_spalling": [
|
||||
"Wall Spalling in External Area",
|
||||
],
|
||||
"wall_finish": [
|
||||
"Wall Finish 1 in External Area",
|
||||
"Wall Finish 2 in External Area",
|
||||
"External Decorations in External Area",
|
||||
"Brickwork Pointing in External Area",
|
||||
],
|
||||
"roof_structure": [
|
||||
"Roof Structure 1 in External Area",
|
||||
"Roof Structure 2 in External Area",
|
||||
"Roof Structure 3 in External Area",
|
||||
"Garage Roof in External Area",
|
||||
"Garage and Store Roofs in External Area",
|
||||
"Store Roof in External Area",
|
||||
"Fascia / Soffit / Bargeboard in External Area",
|
||||
"Gutters in External Area",
|
||||
"Downpipes in External Area",
|
||||
"Internal Downpipes in External Area"
|
||||
],
|
||||
"roof_finish": [
|
||||
"Roof Covering 1 in External Area",
|
||||
"Roof Covering 2 in External Area",
|
||||
"Roof Covering 3 in External Area",
|
||||
],
|
||||
"chimneys": [
|
||||
"Chimneys in External Area",
|
||||
],
|
||||
"windows": [
|
||||
"Windows in Property",
|
||||
"Windows 1 in External Area",
|
||||
"Windows 2 in External Area",
|
||||
"Garage and Store Windows in External Area",
|
||||
"Garage Windows in External Area",
|
||||
"Store Windows in External Area",
|
||||
],
|
||||
"external_doors": [
|
||||
"Type and Location of Front Door in Property",
|
||||
"Front Door Fire Rating in Property",
|
||||
"Patio and French Doors 1 in External Area",
|
||||
"Back and Side Doors 1 in External Area",
|
||||
"Back and Side Doors 2 in External Area",
|
||||
"Garage and Store Doors in External Area",
|
||||
"Garage Door in External Area",
|
||||
"Store Door in External Area",
|
||||
],
|
||||
"central_heating_boiler": [
|
||||
# "Heating Improvement Required in Property",
|
||||
"Boiler Fuel in Property",
|
||||
"Type of Water Heating in Property",
|
||||
],
|
||||
"heating_other": [
|
||||
# "Heating Distribution System in Property"
|
||||
"Boiler Fuel in Property",
|
||||
"Type of Water Heating in Property",
|
||||
],
|
||||
"electrical_systems": [
|
||||
"Electrics Required in Property",
|
||||
],
|
||||
# Other components
|
||||
"kitchen": [
|
||||
"Adequacy of Kitchen and Type in Property",
|
||||
],
|
||||
"bathroom": [
|
||||
"Adequacy of Bathroom Location in Property",
|
||||
],
|
||||
"central_heating_distribution_system": [
|
||||
"Heating Distribution System in Property",
|
||||
],
|
||||
}
|
||||
|
||||
KEY_COMPONENTS = {
|
||||
"wall_structure", "lintels", "brickwork_spalling", "wall_finish",
|
||||
"roof_structure", "roof_finish", "chimneys", "windows",
|
||||
"external_doors", "central_heating_boiler", "heating_other",
|
||||
"electrical_systems",
|
||||
}
|
||||
OTHER_COMPONENTS = {
|
||||
"kitchen", "bathroom", "central_heating_distribution_system",
|
||||
}
|
||||
|
||||
# Criterion C
|
||||
COMPONENT_LIFESPANS = {
|
||||
# Key components
|
||||
"wall_structure": {
|
||||
"house": 80, "flat_below_6_storeys": 80, "flat_above_6_storeys": 80
|
||||
},
|
||||
"lintels": {
|
||||
"house": 60, "flat_below_6_storeys": 60, "flat_above_6_storeys": 60
|
||||
},
|
||||
"brickwork_spalling": {
|
||||
"house": 30, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
|
||||
},
|
||||
"wall_finish": {
|
||||
"house": 60, "flat_below_6_storeys": 60, "flat_above_6_storeys": 30
|
||||
},
|
||||
"roof_structure": {
|
||||
"house": 50, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
|
||||
},
|
||||
"roof_finish": {
|
||||
"house": 50, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
|
||||
},
|
||||
"chimneys": {
|
||||
"house": 50, "flat_below_6_storeys": 50, "flat_above_6_storeys": None # N/A
|
||||
},
|
||||
"windows": {
|
||||
"house": 40, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
|
||||
},
|
||||
"external_doors": {
|
||||
"house": 40, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
|
||||
},
|
||||
"central_heating_boiler": {
|
||||
"house": 15, "flat_below_6_storeys": 15, "flat_above_6_storeys": 15
|
||||
},
|
||||
"heating_other": {
|
||||
"house": 30, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
|
||||
},
|
||||
"electrical_systems": {
|
||||
"house": 30, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
|
||||
},
|
||||
|
||||
# Other components
|
||||
"kitchen": {
|
||||
"house": 30, "flat_below_6_storeys": 30, "flat_above_6_storeys": 30
|
||||
},
|
||||
"bathroom": {
|
||||
"house": 40, "flat_below_6_storeys": 40, "flat_above_6_storeys": 40
|
||||
},
|
||||
"central_heating_distribution_system": {
|
||||
"house": 40, "flat_below_6_storeys": 40, "flat_above_6_storeys": 40
|
||||
},
|
||||
}
|
||||
|
||||
# Database design
|
||||
# creation_date, uprn, variable, result (pass/fail/nodata), hhsrs_score (optional, numeric), hhsrs_rank (A-J),
|
||||
# install_date (for components which expire, e.g. kitchen), remaining_life (for components which expire, e.g. kitchen),
|
||||
|
||||
# TODO: Add the criterion
|
||||
decent_homes_meta = []
|
||||
# Use to capture criterion A, B, C and D. Should be:
|
||||
# {"uprn": int, "creation_date": datetime, "criterion_a": bool, "criterion_b": bool, "criterion_c": bool,
|
||||
# "criterion_d": bool, "decent_homes": bool"}
|
||||
property_decent_homes = []
|
||||
with open(os.path.join(fn), "rb") as f:
|
||||
data = json.load(f)
|
||||
|
||||
today = pd.Timestamp.today().normalize()
|
||||
|
||||
property_info = data["property_info"]
|
||||
if property_info["PROP TYPE"] in ["HOU"]:
|
||||
property_type = "house"
|
||||
elif property_info["PROP TYPE"] == "FLA":
|
||||
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")
|
||||
|
||||
# ---------------- Criterion A ----------------
|
||||
# Critrion A: pass/fail
|
||||
# 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"] in mapping[element_code]["pass"]:
|
||||
result = "pass"
|
||||
elif v["ATTRIBUTE CODE"] in mapping[element_code]["fail"]:
|
||||
result = "fail"
|
||||
elif v["ATTRIBUTE CODE"] in mapping[element_code]["no_data"]:
|
||||
result = "no_data"
|
||||
else:
|
||||
raise ValueError(f"Unknown attribute code: '{v[element_code]}")
|
||||
check_pass.append(result)
|
||||
append_result(
|
||||
decent_homes_meta,
|
||||
criteria="A",
|
||||
variable=hhsrs_variable,
|
||||
sub_variable=ELEMENT_CODE_TO_DESCRIPTION[element_code],
|
||||
result=result,
|
||||
install_date=None,
|
||||
expiry_date=None,
|
||||
)
|
||||
|
||||
# We check if we have a pass, fail or no_data
|
||||
# if all([x == "pass" for x in check_pass]):
|
||||
# hhsrs_result = "pass"
|
||||
# elif any([x == "fail" for x in check_pass]):
|
||||
# hhsrs_result = "fail"
|
||||
# elif any([x == "no_data" for x in check_pass]):
|
||||
# hhsrs_result = "no_data"
|
||||
# else:
|
||||
# raise NotImplementedError("Mixed results not implemented")
|
||||
|
||||
# ---------------- Criterion B ----------------
|
||||
# Check each of the components
|
||||
|
||||
# ---------------- Criterion B ----------------
|
||||
property_boiler = get_element(data["elements"], "Boiler Fuel in Property")
|
||||
|
||||
for component, labels in B_COMPONENT_LABELS.items():
|
||||
for label in labels:
|
||||
label_data = get_element(data["elements"], label)
|
||||
|
||||
# Handle no-data or not-applicable
|
||||
if label_data["ATTRIBUTE CODE"] in ["UNKNOWN", "NONE", "UNKNOWNG", "UNKNOWNS", "UNKNOWNMAT"] and pd.isnull(label_data["INSTALL DATE"]):
|
||||
continue
|
||||
|
||||
# Special skip conditions for heating
|
||||
no_boiler_condition = (
|
||||
property_boiler["ATTRIBUTE CODE"] in ["NONENOCH"]
|
||||
and component == "central_heating_boiler"
|
||||
)
|
||||
other_heating_condition = (
|
||||
label_data["ATTRIBUTE CODE"] in ["NONENOCH"]
|
||||
and component == "heating_other"
|
||||
)
|
||||
if no_boiler_condition or other_heating_condition:
|
||||
# append_result(
|
||||
# decent_homes_meta,
|
||||
# criteria="B",
|
||||
# variable=component,
|
||||
# sub_variable=label,
|
||||
# result="pass",
|
||||
# install_date=None,
|
||||
# expiry_date=None,
|
||||
# )
|
||||
continue
|
||||
|
||||
# Normal case: evaluate install date + lifetime + remaining life
|
||||
install_date = pd.to_datetime(label_data["INSTALL DATE"])
|
||||
if pd.isnull(install_date):
|
||||
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
|
||||
|
||||
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 = today.to_pydatetime() + pd.DateOffset(years=label_data["REMAINING LIFE"])
|
||||
|
||||
component_result = "fail" if is_old and has_failed else "pass"
|
||||
|
||||
# Push into decent_homes_meta
|
||||
append_result(
|
||||
decent_homes_meta,
|
||||
criteria="B",
|
||||
variable=component,
|
||||
sub_variable=label,
|
||||
result=component_result,
|
||||
install_date=str(install_date),
|
||||
expiry_date=str(expiry_date),
|
||||
)
|
||||
|
||||
# ---------------- Criterion C ----------------
|
||||
|
||||
# Guard: property type string already set earlier
|
||||
is_flat = (property_info["PROP TYPE"] == "FLA")
|
||||
|
||||
# 1) Kitchen age ≤ 20 years
|
||||
kitchen = get_element(data["elements"], LABEL_KITCHEN)
|
||||
if kitchen:
|
||||
kit_install_raw = kitchen["INSTALL DATE"]
|
||||
kit_install = pd.to_datetime(kit_install_raw)
|
||||
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 = today.to_pydatetime() + pd.DateOffset(years=kitchen["REMAINING LIFE"])
|
||||
else:
|
||||
raise NotImplementedError("Kitchen data missing - pls check")
|
||||
append_result(
|
||||
decent_homes_meta,
|
||||
criteria="C",
|
||||
variable="kitchen_less_than_20_years_old",
|
||||
sub_variable="kitchen_less_than_20_years_old",
|
||||
result=kitchen_age_result,
|
||||
install_date=str(kit_install),
|
||||
expiry_date=str(kit_next_due)
|
||||
)
|
||||
|
||||
# 2) Kitchen adequate space/layout
|
||||
# Prefer explicit codes if you have them, fall back to text in ATTRIBUTE CODE DESCRIPTION
|
||||
if kitchen:
|
||||
kit_attr_desc = kitchen["ATTRIBUTE CODE"]
|
||||
if kit_attr_desc == "STDKITADQ":
|
||||
kitchen_adequacy_result = "pass"
|
||||
else:
|
||||
raise NotImplementedError("No other observed codes yet")
|
||||
else:
|
||||
raise NotImplementedError("Kitchen data missing - pls check")
|
||||
append_result(
|
||||
decent_homes_meta,
|
||||
criteria="C",
|
||||
variable="kitchen_adequate_space_and_layout",
|
||||
sub_variable="kitchen_adequate_space_and_layout",
|
||||
result=kitchen_adequacy_result,
|
||||
)
|
||||
|
||||
# 3) Bathroom age ≤ 30 years
|
||||
bath = get_element(data["elements"], LABEL_BATHROOM)
|
||||
if bath:
|
||||
bth_install_raw = bath["INSTALL DATE"]
|
||||
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 = today.to_pydatetime() + pd.DateOffset(years=bath["REMAINING LIFE"])
|
||||
else:
|
||||
raise NotImplementedError("Bathroom data missing - pls check")
|
||||
append_result(
|
||||
decent_homes_meta,
|
||||
criteria="C",
|
||||
variable="bathroom_less_than_30_years_old",
|
||||
sub_variable="bathroom_less_than_30_years_old",
|
||||
result=bathroom_age_result,
|
||||
install_date=str(bth_install),
|
||||
expiry_date=bth_next_due
|
||||
)
|
||||
|
||||
# 4) Bathroom/WC appropriately located
|
||||
if bath:
|
||||
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(f"No other observed codes yet {bth_attr_code}")
|
||||
else:
|
||||
raise NotImplementedError("Bathroom data missing - pls check")
|
||||
|
||||
append_result(
|
||||
decent_homes_meta,
|
||||
criteria="C",
|
||||
variable="bathroom_wc_appropriately_located",
|
||||
sub_variable="bathroom_wc_appropriately_located",
|
||||
result=bathroom_location_result
|
||||
)
|
||||
|
||||
# 5) Adequate external noise insulation
|
||||
noise = get_element(data["elements"], LABEL_NOISE)
|
||||
if noise:
|
||||
noise_code = noise["ATTRIBUTE CODE"]
|
||||
if noise_code in {"ADEQUATE"}:
|
||||
noise_result = "pass"
|
||||
else:
|
||||
raise NotImplementedError("No other observed codes yet")
|
||||
else:
|
||||
raise NotImplementedError("Noise insulation data missing - pls check")
|
||||
append_result(
|
||||
decent_homes_meta,
|
||||
criteria="C",
|
||||
variable="adequate_external_noise_insulation",
|
||||
sub_variable="adequate_external_noise_insulation",
|
||||
result=noise_result
|
||||
)
|
||||
|
||||
# 6) Adequate common entrance areas (flats only)
|
||||
if is_flat:
|
||||
common = get_element(data["elements"], LABEL_COMMON_CIRC)
|
||||
if common:
|
||||
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=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
|
||||
heating = get_element(data["elements"], "Heating Improvement Required in Property")
|
||||
if heating:
|
||||
heat_type_code = heating["ATTRIBUTE CODE"]
|
||||
if heat_type_code in {"NOTAPPLIC"}:
|
||||
heating_type_result = "pass"
|
||||
elif heat_type_code in {"WETINSFULL"}:
|
||||
heating_type_result = "fail"
|
||||
else:
|
||||
raise NotImplementedError("No other observed codes yet")
|
||||
else:
|
||||
raise NotImplementedError("Heating element missing in dataset")
|
||||
|
||||
append_result(
|
||||
decent_homes_meta,
|
||||
criteria="D",
|
||||
variable="efficient_heating_system_type",
|
||||
sub_variable="efficient_heating_system_type",
|
||||
result=heating_type_result
|
||||
)
|
||||
|
||||
# heating distribution
|
||||
heating_dist = get_element(data["elements"], "Heating Distribution System in Property")
|
||||
if heating_dist:
|
||||
dist_code = heating_dist["ATTRIBUTE CODE"]
|
||||
if dist_code == "UNKNOWN":
|
||||
# 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")
|
||||
|
||||
append_result(
|
||||
decent_homes_meta,
|
||||
criteria="D",
|
||||
variable="efficient_heating_distribution",
|
||||
sub_variable="efficient_heating_distribution",
|
||||
result=heating_dist_result
|
||||
)
|
||||
|
||||
# insulation
|
||||
loft = get_element(data["elements"], "Size in mm of Loft Insulation Thickness in Property")
|
||||
wall = get_element(data["elements"], "Wall Insulation Improvement in External Area")
|
||||
# To determine how much loft insulation is required
|
||||
|
||||
# Loft insulation check (example threshold: ≥ 270mm = pass)
|
||||
if loft:
|
||||
# We have a specific code, where further loft insulation is needed - It appears the heating type check has
|
||||
# already been completed in this dataset and so we just need to check the code
|
||||
loft_code = loft["ATTRIBUTE CODE"]
|
||||
if loft_code == "LOFTINSRQD":
|
||||
loft_result = "fail"
|
||||
elif loft_code.isnumeric():
|
||||
loft_result = "pass"
|
||||
elif loft_code == "UNKNOWN":
|
||||
loft_result = None
|
||||
else:
|
||||
raise NotImplementedError(f"Unknown loft insulation code - pls check {loft_code}")
|
||||
else:
|
||||
raise NotImplementedError("Loft insulation data missing - pls check")
|
||||
|
||||
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:
|
||||
wall_code = wall["ATTRIBUTE CODE"]
|
||||
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:
|
||||
raise NotImplementedError(f"No other observed codes yet {wall_code}")
|
||||
else:
|
||||
raise NotImplementedError("Wall insulation data missing - pls check")
|
||||
append_result(
|
||||
decent_homes_meta,
|
||||
criteria="D",
|
||||
variable="wall_insulation_sufficient",
|
||||
sub_variable="wall_insulation_sufficient",
|
||||
result=wall_result
|
||||
)
|
||||
|
||||
# ---------------- Criterion A overall ----------------
|
||||
a_vars = set(HHSRS_MAPPING.keys())
|
||||
latest_a_results = {r["variable"]: r["result"] for r in decent_homes_meta if r["variable"] in a_vars}
|
||||
|
||||
if any(v == "fail" for v in latest_a_results.values()):
|
||||
criterion_a_result = "fail"
|
||||
elif all(v == "pass" for v in latest_a_results.values()):
|
||||
criterion_a_result = "pass"
|
||||
else:
|
||||
criterion_a_result = "no_data"
|
||||
|
||||
# ---------------- Criterion B overall ----------------
|
||||
|
||||
component_results = {}
|
||||
|
||||
for component in B_COMPONENT_LABELS.keys():
|
||||
comp_rows = [r for r in decent_homes_meta if
|
||||
r["criteria"] == "B" and r["variable"] == component and r["sub_variable"] is not None]
|
||||
comp_sub_results = [r["result"] for r in comp_rows]
|
||||
|
||||
if not comp_sub_results: # no rows at all
|
||||
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):
|
||||
comp_result = "pass"
|
||||
else:
|
||||
comp_result = "no_data"
|
||||
|
||||
component_results[component] = comp_result
|
||||
|
||||
key_fails = [c for c, r in component_results.items() if c in KEY_COMPONENTS and r == "fail"]
|
||||
other_fails = [c for c, r in component_results.items() if c in OTHER_COMPONENTS and r == "fail"]
|
||||
|
||||
if key_fails:
|
||||
criterion_b_result = "fail"
|
||||
elif len(other_fails) >= 2:
|
||||
criterion_b_result = "fail"
|
||||
elif any(r == "no_data" for r in component_results.values()):
|
||||
criterion_b_result = "no_data"
|
||||
else:
|
||||
criterion_b_result = "pass"
|
||||
|
||||
# ---------------- Criterion C overall ----------------
|
||||
criterion_c_vars = [
|
||||
"kitchen_less_than_20_years_old",
|
||||
"kitchen_adequate_space_and_layout",
|
||||
"bathroom_less_than_30_years_old",
|
||||
"bathroom_wc_appropriately_located",
|
||||
"adequate_external_noise_insulation",
|
||||
]
|
||||
if is_flat:
|
||||
criterion_c_vars.append("adequate_common_entrance_areas")
|
||||
|
||||
latest_c_results = {r["variable"]: r["result"] for r in decent_homes_meta if r["variable"] in criterion_c_vars}
|
||||
|
||||
count_fails = sum(1 for v in latest_c_results.values() if v == "fail")
|
||||
# optionally count no_data too if you want strict interpretation
|
||||
criterion_c_result = "fail" if count_fails >= 3 else "pass"
|
||||
|
||||
# ---------------- Criterion D overall ----------------
|
||||
# Needs to have both efficient geating and distribution so all should pass
|
||||
criterion_d_vars = [
|
||||
"efficient_heating_system_type",
|
||||
"efficient_heating_distribution",
|
||||
"loft_insulation_sufficient",
|
||||
"wall_insulation_sufficient",
|
||||
]
|
||||
latest_d_results = {r["variable"]: r["result"] for r in decent_homes_meta if r["variable"] in criterion_d_vars}
|
||||
|
||||
if any(v == "fail" for v in latest_d_results.values()):
|
||||
criterion_d_result = "fail"
|
||||
elif all(v == "pass" for v in latest_d_results.values()):
|
||||
criterion_d_result = "pass"
|
||||
else:
|
||||
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(),
|
||||
"criterion_a": criterion_a_result,
|
||||
"criterion_b": criterion_b_result,
|
||||
"criterion_c": criterion_c_result,
|
||||
"criterion_d": criterion_d_result,
|
||||
"decent_homes": decent_homes_result,
|
||||
})
|
||||
|
||||
return property_decent_homes[0], decent_homes_meta,
|
||||
|
||||
|
|
@ -1,386 +1,386 @@
|
|||
# SQLModel mapping for ConditionReportModel using BaseModel
|
||||
from sqlmodel import SQLModel, Field, Relationship, Column, JSON
|
||||
from typing import Optional, List
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from etl.models.topLevel import BaseModel, Documents
|
||||
# # SQLModel mapping for ConditionReportModel using BaseModel
|
||||
# from sqlmodel import SQLModel, Field, Relationship, Column, JSON
|
||||
# from typing import Optional, List
|
||||
# import uuid
|
||||
# from datetime import datetime
|
||||
# from etl.models.topLevel import BaseModel, Documents
|
||||
|
||||
class AssessorDetails(BaseModel, table=True):
|
||||
assessor_name_and_id: str
|
||||
elmhurst_id: str
|
||||
# class AssessorDetails(BaseModel, table=True):
|
||||
# assessor_name_and_id: str
|
||||
# elmhurst_id: str
|
||||
|
||||
class InspectionAndProject(BaseModel, table=True):
|
||||
inspection_date: str
|
||||
# class InspectionAndProject(BaseModel, table=True):
|
||||
# inspection_date: str
|
||||
|
||||
class TheProperty(BaseModel, table=True):
|
||||
house_type: str
|
||||
on_which_floor_is_the_flat_located: str
|
||||
is_there_a_corridor: bool
|
||||
is_it_heated: bool
|
||||
it_there_a_balcony: bool
|
||||
classification_type: str
|
||||
orientation_front_elevation: str
|
||||
orientation_in_degrees_front_elevation: str
|
||||
exposure_zone: str
|
||||
main_wall_construction: str
|
||||
# class TheProperty(BaseModel, table=True):
|
||||
# house_type: str
|
||||
# on_which_floor_is_the_flat_located: str
|
||||
# is_there_a_corridor: bool
|
||||
# is_it_heated: bool
|
||||
# it_there_a_balcony: bool
|
||||
# classification_type: str
|
||||
# orientation_front_elevation: str
|
||||
# orientation_in_degrees_front_elevation: str
|
||||
# exposure_zone: str
|
||||
# main_wall_construction: str
|
||||
|
||||
class ElevationInfo(BaseModel, table=True):
|
||||
elevation_type: str
|
||||
cavity_wall_depth: str
|
||||
is_insulation_present: bool
|
||||
insulation_type: str
|
||||
# class ElevationInfo(BaseModel, table=True):
|
||||
# elevation_type: str
|
||||
# cavity_wall_depth: str
|
||||
# is_insulation_present: bool
|
||||
# insulation_type: str
|
||||
|
||||
main_elevation: Optional["MainElevation"] = Relationship(back_populates="elevation_info")
|
||||
# main_elevation: Optional["MainElevation"] = Relationship(back_populates="elevation_info")
|
||||
|
||||
elevation_id: Optional[uuid.UUID] = Field(foreign_key="elevation.id")
|
||||
elevation_table: Optional["Elevation"] = Relationship(back_populates="info")
|
||||
# elevation_id: Optional[uuid.UUID] = Field(foreign_key="elevation.id")
|
||||
# elevation_table: Optional["Elevation"] = Relationship(back_populates="info")
|
||||
|
||||
|
||||
class MainElevation(BaseModel, table=True):
|
||||
elevation_info_id: uuid.UUID = Field(foreign_key="elevationinfo.id")
|
||||
# class MainElevation(BaseModel, table=True):
|
||||
# elevation_info_id: uuid.UUID = Field(foreign_key="elevationinfo.id")
|
||||
|
||||
#SQLAlcemy things
|
||||
elevation_info: ElevationInfo = Relationship(back_populates="main_elevation")
|
||||
# #SQLAlcemy things
|
||||
# elevation_info: ElevationInfo = Relationship(back_populates="main_elevation")
|
||||
|
||||
class Elevation(BaseModel, table=True):
|
||||
protected_conservatory_or_aonb: bool
|
||||
material_type: str
|
||||
visible_signs_of_existing_wall_insulation: str
|
||||
ground_level_bridge_the_dpc: bool
|
||||
# class Elevation(BaseModel, table=True):
|
||||
# protected_conservatory_or_aonb: bool
|
||||
# material_type: str
|
||||
# visible_signs_of_existing_wall_insulation: str
|
||||
# ground_level_bridge_the_dpc: bool
|
||||
|
||||
info: List["ElevationInfo"] = Relationship(back_populates="elevation_table")
|
||||
# info: List["ElevationInfo"] = Relationship(back_populates="elevation_table")
|
||||
|
||||
class GeneralInformation(BaseModel, table=True):
|
||||
assessor_detail_id: uuid.UUID = Field(foreign_key="assessordetails.id")
|
||||
inspection_and_project_id: uuid.UUID = Field(foreign_key="inspectionandproject.id")
|
||||
the_property_id: uuid.UUID = Field(foreign_key="theproperty.id")
|
||||
main_elevation_id: uuid.UUID = Field(foreign_key="mainelevation.id")
|
||||
elevations_id: uuid.UUID = Field(foreign_key="elevation.id")
|
||||
# class GeneralInformation(BaseModel, table=True):
|
||||
# assessor_detail_id: uuid.UUID = Field(foreign_key="assessordetails.id")
|
||||
# inspection_and_project_id: uuid.UUID = Field(foreign_key="inspectionandproject.id")
|
||||
# the_property_id: uuid.UUID = Field(foreign_key="theproperty.id")
|
||||
# main_elevation_id: uuid.UUID = Field(foreign_key="mainelevation.id")
|
||||
# elevations_id: uuid.UUID = Field(foreign_key="elevation.id")
|
||||
|
||||
assessor_details: AssessorDetails = Relationship()
|
||||
inspection_and_project: InspectionAndProject = Relationship()
|
||||
the_property: TheProperty = Relationship()
|
||||
main_elevation: MainElevation = Relationship()
|
||||
elevations: Elevation = Relationship()
|
||||
# assessor_details: AssessorDetails = Relationship()
|
||||
# inspection_and_project: InspectionAndProject = Relationship()
|
||||
# the_property: TheProperty = Relationship()
|
||||
# main_elevation: MainElevation = Relationship()
|
||||
# elevations: Elevation = Relationship()
|
||||
|
||||
class PropertyAccess(BaseModel, table=True):
|
||||
are_there_any_road_restriction_in_the_locality: bool
|
||||
is_on_street_parking_available: bool
|
||||
are_there_any_overhead_wires_or_cables: bool
|
||||
is_the_access_gated: bool
|
||||
is_there_restricted_space_for_contractors_to_access_the_wall_area: bool
|
||||
is_there_restricted_space_for_contractors_to_access_the_roof_area: bool
|
||||
more_than_1_5_meters_in_width_to_fence_or__along_the_full_gable_elevation: bool
|
||||
is_access_to_the_rear_provided_by_use_of_a_ginnel: bool
|
||||
is_access_to_the_rear_provided_by_use_of_a_secured_alleyway: bool
|
||||
# class PropertyAccess(BaseModel, table=True):
|
||||
# are_there_any_road_restriction_in_the_locality: bool
|
||||
# is_on_street_parking_available: bool
|
||||
# are_there_any_overhead_wires_or_cables: bool
|
||||
# is_the_access_gated: bool
|
||||
# is_there_restricted_space_for_contractors_to_access_the_wall_area: bool
|
||||
# is_there_restricted_space_for_contractors_to_access_the_roof_area: bool
|
||||
# more_than_1_5_meters_in_width_to_fence_or__along_the_full_gable_elevation: bool
|
||||
# is_access_to_the_rear_provided_by_use_of_a_ginnel: bool
|
||||
# is_access_to_the_rear_provided_by_use_of_a_secured_alleyway: bool
|
||||
|
||||
|
||||
class ExternalElevation(BaseModel, table=True):
|
||||
structural_defects_of_elevation: str
|
||||
does_any_structural_defect_need_resolving_before_retrofit: bool
|
||||
any_signs_of_water_penetration_caused_by_failed_rainwater_goods_or_pipework: bool
|
||||
are_there_any_visible_signs_of_movement: bool
|
||||
are_there_any_visible_signs_of_cracking_to_the_existing_external_finish: bool
|
||||
# class ExternalElevation(BaseModel, table=True):
|
||||
# structural_defects_of_elevation: str
|
||||
# does_any_structural_defect_need_resolving_before_retrofit: bool
|
||||
# any_signs_of_water_penetration_caused_by_failed_rainwater_goods_or_pipework: bool
|
||||
# are_there_any_visible_signs_of_movement: bool
|
||||
# are_there_any_visible_signs_of_cracking_to_the_existing_external_finish: bool
|
||||
|
||||
class ExternalElevationFront(BaseModel, table=True):
|
||||
external_elevation_id: uuid.UUID = Field(foreign_key="externalelevation.id")
|
||||
external_elevation: ExternalElevation = Relationship()
|
||||
# class ExternalElevationFront(BaseModel, table=True):
|
||||
# external_elevation_id: uuid.UUID = Field(foreign_key="externalelevation.id")
|
||||
# external_elevation: ExternalElevation = Relationship()
|
||||
|
||||
class ExternalElevationRear(BaseModel, table=True):
|
||||
do_all_answers_for_the_front_elevation_apply_to_this_wall: bool
|
||||
external_elevation_id: Optional[uuid.UUID] = Field(foreign_key="externalelevation.id")
|
||||
external_elevation: Optional[ExternalElevation] = Relationship()
|
||||
# class ExternalElevationRear(BaseModel, table=True):
|
||||
# do_all_answers_for_the_front_elevation_apply_to_this_wall: bool
|
||||
# external_elevation_id: Optional[uuid.UUID] = Field(foreign_key="externalelevation.id")
|
||||
# external_elevation: Optional[ExternalElevation] = Relationship()
|
||||
|
||||
class ExternalElevationGableOne(BaseModel, table=True):
|
||||
do_all_answers_for_the_front_elevation_apply_to_this_wall: bool
|
||||
external_elevation_id: Optional[uuid.UUID] = Field(foreign_key="externalelevation.id")
|
||||
external_elevation: Optional[ExternalElevation] = Relationship()
|
||||
# class ExternalElevationGableOne(BaseModel, table=True):
|
||||
# do_all_answers_for_the_front_elevation_apply_to_this_wall: bool
|
||||
# external_elevation_id: Optional[uuid.UUID] = Field(foreign_key="externalelevation.id")
|
||||
# external_elevation: Optional[ExternalElevation] = Relationship()
|
||||
|
||||
class ExternalElevationGableTwo(BaseModel, table=True):
|
||||
is_there_a_fourth_external_elevation: bool
|
||||
external_elevation_id: Optional[uuid.UUID] = Field(foreign_key="externalelevation.id")
|
||||
# class ExternalElevationGableTwo(BaseModel, table=True):
|
||||
# is_there_a_fourth_external_elevation: bool
|
||||
# external_elevation_id: Optional[uuid.UUID] = Field(foreign_key="externalelevation.id")
|
||||
|
||||
class ConservatoryOrOutbuilding(BaseModel, table=True):
|
||||
is_there_a_conservatory: bool
|
||||
is_there_a_cellar_present: bool
|
||||
is_there_an_outbuilding: bool
|
||||
# class ConservatoryOrOutbuilding(BaseModel, table=True):
|
||||
# is_there_a_conservatory: bool
|
||||
# is_there_a_cellar_present: bool
|
||||
# is_there_an_outbuilding: bool
|
||||
|
||||
class AccessAndElevations(BaseModel, table=True):
|
||||
property_access_id: uuid.UUID = Field(foreign_key="propertyaccess.id")
|
||||
external_elevation_front_id: uuid.UUID = Field(foreign_key="externalelevationfront.id")
|
||||
external_elevation_back_id: uuid.UUID = Field(foreign_key="externalelevationrear.id")
|
||||
external_elevation_gable_one_id: uuid.UUID = Field(foreign_key="externalelevationgableone.id")
|
||||
external_elevation_gable_two_id: uuid.UUID = Field(foreign_key="externalelevationgabletwo.id")
|
||||
conservatory_or_out_building_id: uuid.UUID = Field(foreign_key="conservatoryoroutbuilding.id")
|
||||
# class AccessAndElevations(BaseModel, table=True):
|
||||
# property_access_id: uuid.UUID = Field(foreign_key="propertyaccess.id")
|
||||
# external_elevation_front_id: uuid.UUID = Field(foreign_key="externalelevationfront.id")
|
||||
# external_elevation_back_id: uuid.UUID = Field(foreign_key="externalelevationrear.id")
|
||||
# external_elevation_gable_one_id: uuid.UUID = Field(foreign_key="externalelevationgableone.id")
|
||||
# external_elevation_gable_two_id: uuid.UUID = Field(foreign_key="externalelevationgabletwo.id")
|
||||
# conservatory_or_out_building_id: uuid.UUID = Field(foreign_key="conservatoryoroutbuilding.id")
|
||||
|
||||
property_access: PropertyAccess = Relationship()
|
||||
external_elevation_front: ExternalElevationFront = Relationship()
|
||||
external_elevation_back: ExternalElevationRear = Relationship()
|
||||
external_elevation_gable_one: ExternalElevationGableOne = Relationship()
|
||||
external_elevation_gable_two: ExternalElevationGableTwo = Relationship()
|
||||
conservatory_or_out_building: ConservatoryOrOutbuilding = Relationship()
|
||||
# property_access: PropertyAccess = Relationship()
|
||||
# external_elevation_front: ExternalElevationFront = Relationship()
|
||||
# external_elevation_back: ExternalElevationRear = Relationship()
|
||||
# external_elevation_gable_one: ExternalElevationGableOne = Relationship()
|
||||
# external_elevation_gable_two: ExternalElevationGableTwo = Relationship()
|
||||
# conservatory_or_out_building: ConservatoryOrOutbuilding = Relationship()
|
||||
|
||||
class VentilationInfo(BaseModel, table=True):
|
||||
is_there_a_ventilation_system_present_in_the_room: bool
|
||||
any_damp_mould_or_excessive_condensation_within_the_room: bool
|
||||
are_there_sufficient_undercuts_on_the_closed_door: str
|
||||
is_there_any_open_flue_heating_appliances_within_the_room: bool
|
||||
# class VentilationInfo(BaseModel, table=True):
|
||||
# is_there_a_ventilation_system_present_in_the_room: bool
|
||||
# any_damp_mould_or_excessive_condensation_within_the_room: bool
|
||||
# are_there_sufficient_undercuts_on_the_closed_door: str
|
||||
# is_there_any_open_flue_heating_appliances_within_the_room: bool
|
||||
|
||||
class WindowsInfo(BaseModel, table=True):
|
||||
does_the_room_have_any_windows: bool
|
||||
condition_of_the_windows: Optional[str] = None
|
||||
do_the_windows_have_trickle_vents: Optional[bool] = None
|
||||
are_the_windows_openable: Optional[bool] = None
|
||||
input_trickle_vent_product_code_or_measurement: Optional[str] = None
|
||||
# class WindowsInfo(BaseModel, table=True):
|
||||
# does_the_room_have_any_windows: bool
|
||||
# condition_of_the_windows: Optional[str] = None
|
||||
# do_the_windows_have_trickle_vents: Optional[bool] = None
|
||||
# are_the_windows_openable: Optional[bool] = None
|
||||
# input_trickle_vent_product_code_or_measurement: Optional[str] = None
|
||||
|
||||
class RoomInfo(BaseModel, table=True):
|
||||
overall_condition_of_the_room: str
|
||||
does_the_room_have_any_defects: str
|
||||
are_there_any_sloped_ceiling_areas: Optional[bool] = None
|
||||
# class RoomInfo(BaseModel, table=True):
|
||||
# overall_condition_of_the_room: str
|
||||
# does_the_room_have_any_defects: str
|
||||
# are_there_any_sloped_ceiling_areas: Optional[bool] = None
|
||||
|
||||
windows_info_id: uuid.UUID = Field(foreign_key="windowsinfo.id")
|
||||
ventilation_info_id: uuid.UUID = Field(foreign_key="ventilationinfo.id")
|
||||
# windows_info_id: uuid.UUID = Field(foreign_key="windowsinfo.id")
|
||||
# ventilation_info_id: uuid.UUID = Field(foreign_key="ventilationinfo.id")
|
||||
|
||||
windows_info: WindowsInfo = Relationship()
|
||||
ventilation_info: VentilationInfo = Relationship()
|
||||
# windows_info: WindowsInfo = Relationship()
|
||||
# ventilation_info: VentilationInfo = Relationship()
|
||||
|
||||
class Hallway(BaseModel, table=True):
|
||||
is_there_a_hallway: bool
|
||||
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
room_info: Optional[RoomInfo] = Relationship()
|
||||
# class Hallway(BaseModel, table=True):
|
||||
# is_there_a_hallway: bool
|
||||
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
# room_info: Optional[RoomInfo] = Relationship()
|
||||
|
||||
class LivingRoom(BaseModel, table=True):
|
||||
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
room_info: Optional[RoomInfo] = Relationship()
|
||||
# class LivingRoom(BaseModel, table=True):
|
||||
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
# room_info: Optional[RoomInfo] = Relationship()
|
||||
|
||||
class DiningRoom(BaseModel, table=True):
|
||||
is_there_a_dining_room: bool
|
||||
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
room_info: Optional[RoomInfo] = Relationship()
|
||||
# class DiningRoom(BaseModel, table=True):
|
||||
# is_there_a_dining_room: bool
|
||||
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
# room_info: Optional[RoomInfo] = Relationship()
|
||||
|
||||
class Kitchen(BaseModel, table=True):
|
||||
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
room_info: Optional[RoomInfo] = Relationship()
|
||||
is_there_a_cooker_hood_present_in_the_room: bool
|
||||
# class Kitchen(BaseModel, table=True):
|
||||
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
# room_info: Optional[RoomInfo] = Relationship()
|
||||
# is_there_a_cooker_hood_present_in_the_room: bool
|
||||
|
||||
class Utility(BaseModel, table=True):
|
||||
is_there_a_utility_room: bool
|
||||
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
room_info: Optional[RoomInfo] = Relationship()
|
||||
# class Utility(BaseModel, table=True):
|
||||
# is_there_a_utility_room: bool
|
||||
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
# room_info: Optional[RoomInfo] = Relationship()
|
||||
|
||||
class WC(BaseModel, table=True):
|
||||
is_there_a_seperated_wc: bool
|
||||
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
room_info: Optional[RoomInfo] = Relationship()
|
||||
# class WC(BaseModel, table=True):
|
||||
# is_there_a_seperated_wc: bool
|
||||
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
# room_info: Optional[RoomInfo] = Relationship()
|
||||
|
||||
class Landing(BaseModel, table=True):
|
||||
is_there_a_landing: bool
|
||||
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
room_info: Optional[RoomInfo] = Relationship()
|
||||
# class Landing(BaseModel, table=True):
|
||||
# is_there_a_landing: bool
|
||||
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
# room_info: Optional[RoomInfo] = Relationship()
|
||||
|
||||
class LoftSpace(BaseModel, table=True):
|
||||
is_the_main_loft_space_accessible: str
|
||||
is_there_more_than_one_loft_space: bool
|
||||
# class LoftSpace(BaseModel, table=True):
|
||||
# is_the_main_loft_space_accessible: str
|
||||
# is_there_more_than_one_loft_space: bool
|
||||
|
||||
class RoomInRoof(BaseModel, table=True):
|
||||
is_there_a_room_in_roof: bool
|
||||
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
room_info: Optional[RoomInfo] = Relationship()
|
||||
# class RoomInRoof(BaseModel, table=True):
|
||||
# is_there_a_room_in_roof: bool
|
||||
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
# room_info: Optional[RoomInfo] = Relationship()
|
||||
|
||||
class Bedroom(BaseModel, table=True):
|
||||
double_or_single_bedroom: str
|
||||
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
room_info: Optional[RoomInfo] = Relationship()
|
||||
rooms_id: uuid.UUID = Field(foreign_key="rooms.id")
|
||||
rooms: Optional["Rooms"] = Relationship(back_populates="bedrooms")
|
||||
# class Bedroom(BaseModel, table=True):
|
||||
# double_or_single_bedroom: str
|
||||
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
# room_info: Optional[RoomInfo] = Relationship()
|
||||
# rooms_id: uuid.UUID = Field(foreign_key="rooms.id")
|
||||
# rooms: Optional["Rooms"] = Relationship(back_populates="bedrooms")
|
||||
|
||||
|
||||
class Bathroom(BaseModel, table=True):
|
||||
is_this_an_ensuite_bathroom: bool
|
||||
room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
room_info: Optional[RoomInfo] = Relationship()
|
||||
# class Bathroom(BaseModel, table=True):
|
||||
# is_this_an_ensuite_bathroom: bool
|
||||
# room_info_id: Optional[uuid.UUID] = Field(foreign_key="roominfo.id")
|
||||
# room_info: Optional[RoomInfo] = Relationship()
|
||||
|
||||
rooms_id: uuid.UUID = Field(foreign_key="rooms.id")
|
||||
rooms: Optional["Rooms"] = Relationship(back_populates="bathrooms")
|
||||
# rooms_id: uuid.UUID = Field(foreign_key="rooms.id")
|
||||
# rooms: Optional["Rooms"] = Relationship(back_populates="bathrooms")
|
||||
|
||||
|
||||
class Rooms(BaseModel, table=True):
|
||||
hallway_id: uuid.UUID = Field(foreign_key="hallway.id")
|
||||
hallway: Hallway = Relationship()
|
||||
# class Rooms(BaseModel, table=True):
|
||||
# hallway_id: uuid.UUID = Field(foreign_key="hallway.id")
|
||||
# hallway: Hallway = Relationship()
|
||||
|
||||
living_room_id: uuid.UUID = Field(foreign_key="livingroom.id")
|
||||
living_room: LivingRoom = Relationship()
|
||||
# living_room_id: uuid.UUID = Field(foreign_key="livingroom.id")
|
||||
# living_room: LivingRoom = Relationship()
|
||||
|
||||
dining_room_id: uuid.UUID = Field(foreign_key="diningroom.id")
|
||||
dining_room: DiningRoom = Relationship()
|
||||
# dining_room_id: uuid.UUID = Field(foreign_key="diningroom.id")
|
||||
# dining_room: DiningRoom = Relationship()
|
||||
|
||||
kitchen_id: uuid.UUID = Field(foreign_key="kitchen.id")
|
||||
kitchen: Kitchen = Relationship()
|
||||
# kitchen_id: uuid.UUID = Field(foreign_key="kitchen.id")
|
||||
# kitchen: Kitchen = Relationship()
|
||||
|
||||
utility_id: uuid.UUID = Field(foreign_key="utility.id")
|
||||
utility: Utility = Relationship()
|
||||
# utility_id: uuid.UUID = Field(foreign_key="utility.id")
|
||||
# utility: Utility = Relationship()
|
||||
|
||||
wash_chamber_id: uuid.UUID = Field(foreign_key="wc.id")
|
||||
wash_chamber: WC = Relationship()
|
||||
# wash_chamber_id: uuid.UUID = Field(foreign_key="wc.id")
|
||||
# wash_chamber: WC = Relationship()
|
||||
|
||||
landing_id: uuid.UUID = Field(foreign_key="landing.id")
|
||||
landing: Landing = Relationship()
|
||||
# landing_id: uuid.UUID = Field(foreign_key="landing.id")
|
||||
# landing: Landing = Relationship()
|
||||
|
||||
loft_space_id: uuid.UUID = Field(foreign_key="loftspace.id")
|
||||
loft_space: LoftSpace = Relationship()
|
||||
# loft_space_id: uuid.UUID = Field(foreign_key="loftspace.id")
|
||||
# loft_space: LoftSpace = Relationship()
|
||||
|
||||
room_in_roof_id: uuid.UUID = Field(foreign_key="roominroof.id")
|
||||
room_in_roof: RoomInRoof = Relationship()
|
||||
# room_in_roof_id: uuid.UUID = Field(foreign_key="roominroof.id")
|
||||
# room_in_roof: RoomInRoof = Relationship()
|
||||
|
||||
bedrooms: List[Bedroom] = Relationship(back_populates="rooms")
|
||||
bathrooms: List[Bathroom] = Relationship(back_populates="rooms")
|
||||
# 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 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 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 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 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 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 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 = Relationship()
|
||||
# class HeatingSystem(BaseModel, table=True):
|
||||
# general_condition_id: uuid.UUID = Field(foreign_key="generalconditionheatingsystem.id")
|
||||
# general_condition: GeneralConditionHeatingSystem = Relationship()
|
||||
|
||||
main_heating_one_id: uuid.UUID = Field(foreign_key="mainheatingone.id")
|
||||
main_heating_one: MainHeatingOne = Relationship()
|
||||
# main_heating_one_id: uuid.UUID = Field(foreign_key="mainheatingone.id")
|
||||
# main_heating_one: MainHeatingOne = Relationship()
|
||||
|
||||
main_heating_two_id: uuid.UUID = Field(foreign_key="mainheatingtwo.id")
|
||||
main_heating_two: MainHeatingTwo = Relationship()
|
||||
# main_heating_two_id: uuid.UUID = Field(foreign_key="mainheatingtwo.id")
|
||||
# main_heating_two: MainHeatingTwo = Relationship()
|
||||
|
||||
secondary_heating_id: uuid.UUID = Field(foreign_key="secondaryheating.id")
|
||||
secondary_heating: SecondaryHeating = Relationship()
|
||||
# secondary_heating_id: uuid.UUID = Field(foreign_key="secondaryheating.id")
|
||||
# secondary_heating: SecondaryHeating = Relationship()
|
||||
|
||||
heating_by_room_id: uuid.UUID = Field(foreign_key="heatingbyroom.id")
|
||||
heating_by_room: HeatingByRoom = Relationship()
|
||||
# heating_by_room_id: uuid.UUID = Field(foreign_key="heatingbyroom.id")
|
||||
# heating_by_room: HeatingByRoom = Relationship()
|
||||
|
||||
renewables_id: uuid.UUID = Field(foreign_key="renewables.id")
|
||||
renewables: Renewables = Relationship()
|
||||
# renewables_id: uuid.UUID = Field(foreign_key="renewables.id")
|
||||
# renewables: Renewables = Relationship()
|
||||
|
||||
class Occupant(BaseModel, table=True):
|
||||
name: str
|
||||
have_evidence_of_12_months_of_fuel_bill_data: bool
|
||||
total_number_of_occupants: int
|
||||
no_of_adult_occupants: int
|
||||
no_of_child_occupants: int
|
||||
no_of_occupant_of_a_pensionable_age: int
|
||||
are_there_any_vulnerable_people: bool
|
||||
is_there_anyone_with_a_disability: bool
|
||||
status_of_occupant: str
|
||||
landlord_wrote_that_the_tenent_agrees_assessment_been_supplied: bool
|
||||
# class Occupant(BaseModel, table=True):
|
||||
# name: str
|
||||
# have_evidence_of_12_months_of_fuel_bill_data: bool
|
||||
# total_number_of_occupants: int
|
||||
# no_of_adult_occupants: int
|
||||
# no_of_child_occupants: int
|
||||
# no_of_occupant_of_a_pensionable_age: int
|
||||
# are_there_any_vulnerable_people: bool
|
||||
# is_there_anyone_with_a_disability: bool
|
||||
# status_of_occupant: str
|
||||
# landlord_wrote_that_the_tenent_agrees_assessment_been_supplied: bool
|
||||
|
||||
|
||||
class EnergyUse(BaseModel, table=True):
|
||||
property_tenure: str
|
||||
who_is_the_electricity_payer: str
|
||||
# class EnergyUse(BaseModel, table=True):
|
||||
# property_tenure: str
|
||||
# who_is_the_electricity_payer: str
|
||||
|
||||
|
||||
class HeatingFromConditionReport(BaseModel, table=True):
|
||||
room_stat_in_temperature_in_celsius: Optional[str] = None
|
||||
room_stat_location: Optional[str] = None
|
||||
is_the_heating_pattern_known: Optional[str] = None
|
||||
# class HeatingFromConditionReport(BaseModel, table=True):
|
||||
# room_stat_in_temperature_in_celsius: Optional[str] = None
|
||||
# room_stat_location: Optional[str] = None
|
||||
# is_the_heating_pattern_known: Optional[str] = None
|
||||
|
||||
class ShowerAndBath(BaseModel, table=True):
|
||||
shower_type: str
|
||||
do_you_know_the_no_of_showers_per_day_per_week: bool
|
||||
please_input_no_of_showers_and_specify_a_day_or_a_week: str
|
||||
do_you_know_the_number_of_baths_per_day_or_per_week: str
|
||||
# class ShowerAndBath(BaseModel, table=True):
|
||||
# shower_type: str
|
||||
# do_you_know_the_no_of_showers_per_day_per_week: bool
|
||||
# please_input_no_of_showers_and_specify_a_day_or_a_week: str
|
||||
# do_you_know_the_number_of_baths_per_day_or_per_week: str
|
||||
|
||||
class FridgeAndFreezers(BaseModel, table=True):
|
||||
no_of_stand_alone_seperate_fridges: int
|
||||
no_of_stand_alone_seperate_freezers: int
|
||||
no_of_stand_alone_or_integrated_fridge_freezers: int
|
||||
# class FridgeAndFreezers(BaseModel, table=True):
|
||||
# no_of_stand_alone_seperate_fridges: int
|
||||
# no_of_stand_alone_seperate_freezers: int
|
||||
# no_of_stand_alone_or_integrated_fridge_freezers: int
|
||||
|
||||
class Cooker(BaseModel,table=True):
|
||||
range_fuel: str
|
||||
normal_large_range: str
|
||||
cooker_type: str
|
||||
# class Cooker(BaseModel,table=True):
|
||||
# range_fuel: str
|
||||
# normal_large_range: str
|
||||
# cooker_type: str
|
||||
|
||||
class TumbleDryer(BaseModel, table=True):
|
||||
percentage_of_annual_use: int
|
||||
space_for_outdoor_drying: bool
|
||||
# class TumbleDryer(BaseModel, table=True):
|
||||
# percentage_of_annual_use: int
|
||||
# space_for_outdoor_drying: bool
|
||||
|
||||
class OccupantAssessment(BaseModel, table=True):
|
||||
occupant_id: uuid.UUID = Field(foreign_key="occupant.id")
|
||||
occupant: Occupant = Relationship()
|
||||
# class OccupantAssessment(BaseModel, table=True):
|
||||
# occupant_id: uuid.UUID = Field(foreign_key="occupant.id")
|
||||
# occupant: Occupant = Relationship()
|
||||
|
||||
energy_use_id: uuid.UUID = Field(foreign_key="energyuse.id")
|
||||
energy_use: EnergyUse = Relationship()
|
||||
# energy_use_id: uuid.UUID = Field(foreign_key="energyuse.id")
|
||||
# energy_use: EnergyUse = Relationship()
|
||||
|
||||
heating_id: uuid.UUID = Field(foreign_key="heatingfromconditionreport.id")
|
||||
heating: HeatingFromConditionReport = Relationship()
|
||||
# heating_id: uuid.UUID = Field(foreign_key="heatingfromconditionreport.id")
|
||||
# heating: HeatingFromConditionReport = Relationship()
|
||||
|
||||
shower_and_bath_id: uuid.UUID = Field(foreign_key="showerandbath.id")
|
||||
shower_and_bath: ShowerAndBath = Relationship()
|
||||
# shower_and_bath_id: uuid.UUID = Field(foreign_key="showerandbath.id")
|
||||
# shower_and_bath: ShowerAndBath = Relationship()
|
||||
|
||||
# appliances: Optional[Appliances]
|
||||
# appliances_id
|
||||
# # appliances: Optional[Appliances]
|
||||
# # appliances_id
|
||||
|
||||
fridge_and_freezers_id: uuid.UUID = Field(foreign_key="fridgeandfreezers.id")
|
||||
fridge_and_freezers: FridgeAndFreezers = Relationship()
|
||||
# fridge_and_freezers_id: uuid.UUID = Field(foreign_key="fridgeandfreezers.id")
|
||||
# fridge_and_freezers: FridgeAndFreezers = Relationship()
|
||||
|
||||
cooker_id: uuid.UUID = Field(foreign_key="cooker.id")
|
||||
cooker: Cooker = Relationship()
|
||||
# cooker_id: uuid.UUID = Field(foreign_key="cooker.id")
|
||||
# cooker: Cooker = Relationship()
|
||||
|
||||
tumble_dryer_id: uuid.UUID = Field(foreign_key="tumbledryer.id")
|
||||
tumble_dryer: TumbleDryer = Relationship()
|
||||
# tumble_dryer_id: uuid.UUID = Field(foreign_key="tumbledryer.id")
|
||||
# tumble_dryer: TumbleDryer = Relationship()
|
||||
|
||||
class ConditionReportModel(BaseModel, table=True):
|
||||
project_site_name: str
|
||||
property_reference_code: str
|
||||
property_address: str
|
||||
postcode: str
|
||||
# class ConditionReportModel(BaseModel, table=True):
|
||||
# project_site_name: str
|
||||
# property_reference_code: str
|
||||
# property_address: str
|
||||
# postcode: str
|
||||
|
||||
general_information_id: uuid.UUID = Field(foreign_key="generalinformation.id")
|
||||
general_information: GeneralInformation = Relationship()
|
||||
# general_information_id: uuid.UUID = Field(foreign_key="generalinformation.id")
|
||||
# general_information: GeneralInformation = Relationship()
|
||||
|
||||
access_and_elevations_id: uuid.UUID = Field(foreign_key="accessandelevations.id")
|
||||
access_and_elevations: AccessAndElevations = Relationship()
|
||||
# access_and_elevations_id: uuid.UUID = Field(foreign_key="accessandelevations.id")
|
||||
# access_and_elevations: AccessAndElevations = Relationship()
|
||||
|
||||
rooms_id: uuid.UUID = Field(foreign_key="rooms.id")
|
||||
rooms: Rooms = Relationship()
|
||||
# rooms_id: uuid.UUID = Field(foreign_key="rooms.id")
|
||||
# rooms: Rooms = Relationship()
|
||||
|
||||
heating_system_id: uuid.UUID = Field(foreign_key="heatingsystem.id")
|
||||
heating_system: HeatingSystem = Relationship()
|
||||
# heating_system_id: uuid.UUID = Field(foreign_key="heatingsystem.id")
|
||||
# heating_system: HeatingSystem = Relationship()
|
||||
|
||||
occupancy_assessment_id: uuid.UUID = Field(foreign_key="occupantassessment.id")
|
||||
occupancy_assessment: OccupantAssessment = Relationship()
|
||||
# occupancy_assessment_id: uuid.UUID = Field(foreign_key="occupantassessment.id")
|
||||
# occupancy_assessment: OccupantAssessment = Relationship()
|
||||
|
|
|
|||
|
|
@ -1,335 +1,335 @@
|
|||
from sqlmodel import Field, SQLModel, Relationship
|
||||
import uuid
|
||||
from typing import Optional, List
|
||||
from datetime import datetime
|
||||
from pydantic import EmailStr
|
||||
from sqlalchemy import Column
|
||||
from sqlalchemy.dialects.postgresql import UUID
|
||||
from etl.models.topLevel import BaseModel, Documents
|
||||
# from sqlmodel import Field, SQLModel, Relationship
|
||||
# import uuid
|
||||
# from typing import Optional, List
|
||||
# from datetime import datetime
|
||||
# from pydantic import EmailStr
|
||||
# from sqlalchemy import Column
|
||||
# from sqlalchemy.dialects.postgresql import UUID
|
||||
# # from etl.models.topLevel import BaseModel, Documents
|
||||
|
||||
|
||||
class PreSiteNote(BaseModel, table=True):
|
||||
summary_info_id: uuid.UUID = Field(
|
||||
foreign_key="presitenotessummaryinfo.id",
|
||||
nullable=False
|
||||
)
|
||||
# class PreSiteNote(BaseModel, table=True):
|
||||
# summary_info_id: uuid.UUID = Field(
|
||||
# foreign_key="presitenotessummaryinfo.id",
|
||||
# nullable=False
|
||||
# )
|
||||
|
||||
summary_info: Optional["PreSiteNotesSummaryInfo"] = Relationship(back_populates="pre_site_notes")
|
||||
# summary_info: Optional["PreSiteNotesSummaryInfo"] = Relationship(back_populates="pre_site_notes")
|
||||
|
||||
|
||||
# Assessor Info
|
||||
assessor_id: uuid.UUID = Field(
|
||||
foreign_key="assessorinfo.id",
|
||||
nullable=False
|
||||
)
|
||||
# # Assessor Info
|
||||
# assessor_id: uuid.UUID = Field(
|
||||
# foreign_key="assessorinfo.id",
|
||||
# nullable=False
|
||||
# )
|
||||
|
||||
assessor: Optional["AssessorInfo"] = Relationship(back_populates="pre_site_notes")
|
||||
# assessor: Optional["AssessorInfo"] = Relationship(back_populates="pre_site_notes")
|
||||
|
||||
pre_site_note_description_id: uuid.UUID = Field(
|
||||
foreign_key="propertydescription.id",
|
||||
nullable=True
|
||||
)
|
||||
# pre_site_note_description_id: uuid.UUID = Field(
|
||||
# foreign_key="propertydescription.id",
|
||||
# nullable=True
|
||||
# )
|
||||
|
||||
pre_site_note_description: Optional["PropertyDescription"] = Relationship(back_populates="pre_site_notes")
|
||||
# pre_site_note_description: Optional["PropertyDescription"] = Relationship(back_populates="pre_site_notes")
|
||||
|
||||
|
||||
class Dimension(BaseModel, table=True):
|
||||
floor_area_m2: float
|
||||
room_height_m: float
|
||||
loss_perimeter_m: float
|
||||
party_wall_length_m: float
|
||||
property_detail_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
|
||||
property_detail: Optional["PropertyDetail"] = Relationship(back_populates="dimensions")
|
||||
# class Dimension(BaseModel, table=True):
|
||||
# floor_area_m2: float
|
||||
# room_height_m: float
|
||||
# loss_perimeter_m: float
|
||||
# party_wall_length_m: float
|
||||
# property_detail_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
|
||||
# property_detail: Optional["PropertyDetail"] = Relationship(back_populates="dimensions")
|
||||
|
||||
|
||||
class Walls(BaseModel, table=True):
|
||||
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 Walls(BaseModel, table=True):
|
||||
# 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, table=True):
|
||||
construction: str
|
||||
insulation_type: str
|
||||
insulation_thickness: str
|
||||
u_value_known: bool
|
||||
# class Roofs(BaseModel, table=True):
|
||||
# construction: str
|
||||
# insulation_type: str
|
||||
# insulation_thickness: str
|
||||
# u_value_known: bool
|
||||
|
||||
|
||||
class Floors(BaseModel, table=True):
|
||||
floor_type: str
|
||||
ground_floor_construction: str
|
||||
ground_floor_insulation_type: Optional[str] = ""
|
||||
floor_insulation_thickness_mm: Optional[float] = -1
|
||||
u_value_known: bool
|
||||
# class Floors(BaseModel, table=True):
|
||||
# floor_type: str
|
||||
# ground_floor_construction: str
|
||||
# ground_floor_insulation_type: Optional[str] = ""
|
||||
# floor_insulation_thickness_mm: Optional[float] = -1
|
||||
# u_value_known: bool
|
||||
|
||||
|
||||
class Windows(BaseModel, table=True):
|
||||
glazing_type: str
|
||||
area_m2: float
|
||||
roof_window: bool
|
||||
orientation: str
|
||||
u_value_w_m2_k: int
|
||||
g_value: int
|
||||
property_detail_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
|
||||
property_detail: Optional["PropertyDetail"] = Relationship(back_populates="windows")
|
||||
# class Windows(BaseModel, table=True):
|
||||
# glazing_type: str
|
||||
# area_m2: float
|
||||
# roof_window: bool
|
||||
# orientation: str
|
||||
# u_value_w_m2_k: int
|
||||
# g_value: int
|
||||
# property_detail_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
|
||||
# property_detail: Optional["PropertyDetail"] = Relationship(back_populates="windows")
|
||||
|
||||
|
||||
|
||||
class PropertyDetail(BaseModel, table=True):
|
||||
age_band: str
|
||||
wall_id: Optional[uuid.UUID] = Field(default=None, foreign_key="walls.id")
|
||||
roof_id: Optional[uuid.UUID] = Field(default=None, foreign_key="roofs.id")
|
||||
floor_id: Optional[uuid.UUID] = Field(default=None, foreign_key="floors.id")
|
||||
# class PropertyDetail(BaseModel, table=True):
|
||||
# age_band: str
|
||||
# wall_id: Optional[uuid.UUID] = Field(default=None, foreign_key="walls.id")
|
||||
# roof_id: Optional[uuid.UUID] = Field(default=None, foreign_key="roofs.id")
|
||||
# floor_id: Optional[uuid.UUID] = Field(default=None, foreign_key="floors.id")
|
||||
|
||||
# Relationships
|
||||
dimensions: List[Dimension] = Relationship(back_populates="property_detail")
|
||||
windows: List[Windows] = Relationship(back_populates="property_detail")
|
||||
# # Relationships
|
||||
# dimensions: List[Dimension] = Relationship(back_populates="property_detail")
|
||||
# windows: List[Windows] = Relationship(back_populates="property_detail")
|
||||
|
||||
|
||||
|
||||
|
||||
class Door(BaseModel, table=True):
|
||||
no_of_doors: int
|
||||
no_of_insulated_doors: int
|
||||
u_value_w_m2_k: Optional[str]
|
||||
# class Door(BaseModel, table=True):
|
||||
# no_of_doors: int
|
||||
# no_of_insulated_doors: int
|
||||
# u_value_w_m2_k: Optional[str]
|
||||
|
||||
property_description: Optional["PropertyDescription"] = Relationship(back_populates="door")
|
||||
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="door")
|
||||
|
||||
class VentilationAndCooling(BaseModel, table=True):
|
||||
no_of_open_fireplaces: int
|
||||
ventilation_type: str
|
||||
space_cooling_system_present: bool
|
||||
# class VentilationAndCooling(BaseModel, table=True):
|
||||
# no_of_open_fireplaces: int
|
||||
# ventilation_type: str
|
||||
# space_cooling_system_present: bool
|
||||
|
||||
property_description: Optional["PropertyDescription"] = Relationship(back_populates="ventilation_and_cooling")
|
||||
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="ventilation_and_cooling")
|
||||
|
||||
|
||||
|
||||
class Lighting(BaseModel, table=True):
|
||||
total_no_of_light_fittings: int
|
||||
total_no_of_lel_fittings: int
|
||||
# class Lighting(BaseModel, table=True):
|
||||
# total_no_of_light_fittings: int
|
||||
# total_no_of_lel_fittings: int
|
||||
|
||||
property_description: Optional["PropertyDescription"] = Relationship(back_populates="lighting")
|
||||
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="lighting")
|
||||
|
||||
|
||||
|
||||
class HeatingSystemControls(BaseModel, table=True):
|
||||
control_type: str
|
||||
flue_type: str
|
||||
fan_assisted_flue: bool
|
||||
heat_emitter_type: str
|
||||
electricity_meter_type: Optional[str] = ""
|
||||
mains_gas_available: Optional[bool] = False
|
||||
# class HeatingSystemControls(BaseModel, table=True):
|
||||
# control_type: str
|
||||
# flue_type: str
|
||||
# fan_assisted_flue: bool
|
||||
# heat_emitter_type: str
|
||||
# electricity_meter_type: Optional[str] = ""
|
||||
# mains_gas_available: Optional[bool] = False
|
||||
|
||||
|
||||
class HeatingFromPreSiteNotes(BaseModel, table=True):
|
||||
type: str
|
||||
heating_source: str
|
||||
efficiency_source: str
|
||||
heating_fuel: str
|
||||
brand_name: str
|
||||
model_name: str
|
||||
model_qualifer: str
|
||||
sap_2009_table: Optional[str] = ""
|
||||
percentage_of_heated_floor_area_served: Optional[str] = ""
|
||||
controls_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingsystemcontrols.id")
|
||||
# class HeatingFromPreSiteNotes(BaseModel, table=True):
|
||||
# type: str
|
||||
# heating_source: str
|
||||
# efficiency_source: str
|
||||
# heating_fuel: str
|
||||
# brand_name: str
|
||||
# model_name: str
|
||||
# model_qualifer: str
|
||||
# sap_2009_table: Optional[str] = ""
|
||||
# percentage_of_heated_floor_area_served: Optional[str] = ""
|
||||
# controls_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingsystemcontrols.id")
|
||||
|
||||
property_description: Optional["PropertyDescription"] = Relationship(
|
||||
back_populates="main_heating", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating_id]"}
|
||||
)
|
||||
property_description2: Optional["PropertyDescription"] = Relationship(
|
||||
back_populates="main_heating2", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating2_id]"}
|
||||
)
|
||||
# property_description: Optional["PropertyDescription"] = Relationship(
|
||||
# back_populates="main_heating", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating_id]"}
|
||||
# )
|
||||
# property_description2: Optional["PropertyDescription"] = Relationship(
|
||||
# back_populates="main_heating2", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating2_id]"}
|
||||
# )
|
||||
|
||||
|
||||
|
||||
class HeatingType(BaseModel, table=True):
|
||||
heating_type: str
|
||||
fuel_type: str
|
||||
# class HeatingType(BaseModel, table=True):
|
||||
# heating_type: str
|
||||
# fuel_type: str
|
||||
|
||||
property_description: Optional["PropertyDescription"] = Relationship(back_populates="secondary_heating_type")
|
||||
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="secondary_heating_type")
|
||||
|
||||
|
||||
class WaterHeating(BaseModel, table=True):
|
||||
heating_type: str
|
||||
fuel_type: str
|
||||
# class WaterHeating(BaseModel, table=True):
|
||||
# heating_type: str
|
||||
# fuel_type: str
|
||||
|
||||
property_description: Optional["PropertyDescription"] = Relationship(back_populates="water_heating")
|
||||
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="water_heating")
|
||||
|
||||
|
||||
|
||||
class HotWaterCylinder(BaseModel, table=True):
|
||||
volume: str
|
||||
insulation_type: str
|
||||
insulation_thickness: str
|
||||
thermostat: bool
|
||||
# class HotWaterCylinder(BaseModel, table=True):
|
||||
# volume: str
|
||||
# insulation_type: str
|
||||
# insulation_thickness: str
|
||||
# thermostat: bool
|
||||
|
||||
property_description: Optional["PropertyDescription"] = Relationship(back_populates="hot_water_cylinder")
|
||||
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="hot_water_cylinder")
|
||||
|
||||
|
||||
|
||||
class SolarWaterHeating(BaseModel, table=True):
|
||||
solar_water_heating_details_known: bool
|
||||
# class SolarWaterHeating(BaseModel, table=True):
|
||||
# solar_water_heating_details_known: bool
|
||||
|
||||
property_description: Optional["PropertyDescription"] = Relationship(back_populates="solar_water_heating")
|
||||
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="solar_water_heating")
|
||||
|
||||
|
||||
|
||||
class ShowerAndBaths(BaseModel, table=True):
|
||||
no_of_rooms_with_baths_and_or_shower: int
|
||||
no_of_rooms_with_mixer_shower_and_no_baths: int
|
||||
no_of_rooms_with_mixer_shower_and_baths: int
|
||||
# class ShowerAndBaths(BaseModel, table=True):
|
||||
# no_of_rooms_with_baths_and_or_shower: int
|
||||
# no_of_rooms_with_mixer_shower_and_no_baths: int
|
||||
# no_of_rooms_with_mixer_shower_and_baths: int
|
||||
|
||||
property_description: Optional["PropertyDescription"] = Relationship(back_populates="shower_and_baths")
|
||||
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="shower_and_baths")
|
||||
|
||||
|
||||
|
||||
class FlueGasHeatRecoverySystem(BaseModel, table=True):
|
||||
fghrs_present: bool
|
||||
# class FlueGasHeatRecoverySystem(BaseModel, table=True):
|
||||
# fghrs_present: bool
|
||||
|
||||
property_description: Optional["PropertyDescription"] = Relationship(back_populates="flue_gas_heat_recovery_system")
|
||||
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="flue_gas_heat_recovery_system")
|
||||
|
||||
|
||||
|
||||
class PhotovoltaicPanel(BaseModel, table=True):
|
||||
pvs_are_connected_to_dwelling_electricity_meter: bool
|
||||
percentage_of_external_roof_area_with_pvs: str
|
||||
# class PhotovoltaicPanel(BaseModel, table=True):
|
||||
# pvs_are_connected_to_dwelling_electricity_meter: bool
|
||||
# percentage_of_external_roof_area_with_pvs: str
|
||||
|
||||
property_description: Optional["PropertyDescription"] = Relationship(back_populates="photovoltaic_panel")
|
||||
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="photovoltaic_panel")
|
||||
|
||||
|
||||
|
||||
class WindTurbine(BaseModel, table=True):
|
||||
wind_turbine: bool
|
||||
# class WindTurbine(BaseModel, table=True):
|
||||
# wind_turbine: bool
|
||||
|
||||
property_description: Optional["PropertyDescription"] = Relationship(back_populates="wind_turbine")
|
||||
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="wind_turbine")
|
||||
|
||||
|
||||
|
||||
class OtherDetails(BaseModel, table=True):
|
||||
electricity_meter_type: str
|
||||
main_gas_avalible: bool
|
||||
# class OtherDetails(BaseModel, table=True):
|
||||
# electricity_meter_type: str
|
||||
# main_gas_avalible: bool
|
||||
|
||||
property_description: Optional["PropertyDescription"] = Relationship(back_populates="other_details")
|
||||
# property_description: Optional["PropertyDescription"] = Relationship(back_populates="other_details")
|
||||
|
||||
|
||||
|
||||
class PropertyDescription(BaseModel, table=True):
|
||||
built_form: str
|
||||
detachment_or_position: str
|
||||
no_of_main_property: int
|
||||
no_of_extension_1: Optional[int] = 0
|
||||
no_of_extension_2: Optional[int] = 0
|
||||
no_of_extension_3: Optional[int] = 0
|
||||
no_of_extension_4: Optional[int] = 0
|
||||
no_of_habitable_rooms: int
|
||||
no_of_heated_rooms: int
|
||||
heated_basement: bool
|
||||
conservatory_type: str
|
||||
percentage_of_draught_proofed: int
|
||||
terrain_type: str
|
||||
conservatory: bool
|
||||
# class PropertyDescription(BaseModel, table=True):
|
||||
# built_form: str
|
||||
# detachment_or_position: str
|
||||
# no_of_main_property: int
|
||||
# no_of_extension_1: Optional[int] = 0
|
||||
# no_of_extension_2: Optional[int] = 0
|
||||
# no_of_extension_3: Optional[int] = 0
|
||||
# no_of_extension_4: Optional[int] = 0
|
||||
# no_of_habitable_rooms: int
|
||||
# no_of_heated_rooms: int
|
||||
# heated_basement: bool
|
||||
# conservatory_type: str
|
||||
# percentage_of_draught_proofed: int
|
||||
# terrain_type: str
|
||||
# conservatory: bool
|
||||
|
||||
main_property_id: uuid.UUID = Field(foreign_key="propertydetail.id")
|
||||
ex1_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
|
||||
ex2_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
|
||||
ex3_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
|
||||
ex4_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
|
||||
# main_property_id: uuid.UUID = Field(foreign_key="propertydetail.id")
|
||||
# ex1_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
|
||||
# ex2_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
|
||||
# ex3_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
|
||||
# ex4_property_id: Optional[uuid.UUID] = Field(default=None, foreign_key="propertydetail.id")
|
||||
|
||||
door_id: Optional[uuid.UUID] = Field(default=None, foreign_key="door.id")
|
||||
ventilation_and_cooling_id: Optional[uuid.UUID] = Field(default=None, foreign_key="ventilationandcooling.id")
|
||||
lighting_id: Optional[uuid.UUID] = Field(default=None, foreign_key="lighting.id")
|
||||
water_heating_id: Optional[uuid.UUID] = Field(default=None, foreign_key="waterheating.id")
|
||||
hot_water_cylinder_id: Optional[uuid.UUID] = Field(default=None, foreign_key="hotwatercylinder.id")
|
||||
solar_water_heating_id: Optional[uuid.UUID] = Field(default=None, foreign_key="solarwaterheating.id")
|
||||
shower_and_baths_id: Optional[uuid.UUID] = Field(default=None, foreign_key="showerandbaths.id")
|
||||
flue_gas_heat_recovery_system_id: Optional[uuid.UUID] = Field(default=None, foreign_key="fluegasheatrecoverysystem.id")
|
||||
photovoltaic_panel_id: Optional[uuid.UUID] = Field(default=None, foreign_key="photovoltaicpanel.id")
|
||||
wind_turbine_id: Optional[uuid.UUID] = Field(default=None, foreign_key="windturbine.id")
|
||||
other_details_id: Optional[uuid.UUID] = Field(default=None, foreign_key="otherdetails.id")
|
||||
main_heating_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingfrompresitenotes.id")
|
||||
main_heating2_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingfrompresitenotes.id")
|
||||
secondary_heating_type_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingtype.id")
|
||||
# door_id: Optional[uuid.UUID] = Field(default=None, foreign_key="door.id")
|
||||
# ventilation_and_cooling_id: Optional[uuid.UUID] = Field(default=None, foreign_key="ventilationandcooling.id")
|
||||
# lighting_id: Optional[uuid.UUID] = Field(default=None, foreign_key="lighting.id")
|
||||
# water_heating_id: Optional[uuid.UUID] = Field(default=None, foreign_key="waterheating.id")
|
||||
# hot_water_cylinder_id: Optional[uuid.UUID] = Field(default=None, foreign_key="hotwatercylinder.id")
|
||||
# solar_water_heating_id: Optional[uuid.UUID] = Field(default=None, foreign_key="solarwaterheating.id")
|
||||
# shower_and_baths_id: Optional[uuid.UUID] = Field(default=None, foreign_key="showerandbaths.id")
|
||||
# flue_gas_heat_recovery_system_id: Optional[uuid.UUID] = Field(default=None, foreign_key="fluegasheatrecoverysystem.id")
|
||||
# photovoltaic_panel_id: Optional[uuid.UUID] = Field(default=None, foreign_key="photovoltaicpanel.id")
|
||||
# wind_turbine_id: Optional[uuid.UUID] = Field(default=None, foreign_key="windturbine.id")
|
||||
# other_details_id: Optional[uuid.UUID] = Field(default=None, foreign_key="otherdetails.id")
|
||||
# main_heating_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingfrompresitenotes.id")
|
||||
# main_heating2_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingfrompresitenotes.id")
|
||||
# secondary_heating_type_id: Optional[uuid.UUID] = Field(default=None, foreign_key="heatingtype.id")
|
||||
|
||||
# Relationships
|
||||
main_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_property_id]"})
|
||||
ex1_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex1_property_id]"})
|
||||
ex2_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex2_property_id]"})
|
||||
ex3_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex3_property_id]"})
|
||||
ex4_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex4_property_id]"})
|
||||
# # Relationships
|
||||
# main_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_property_id]"})
|
||||
# ex1_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex1_property_id]"})
|
||||
# ex2_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex2_property_id]"})
|
||||
# ex3_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex3_property_id]"})
|
||||
# ex4_property: Optional["PropertyDetail"] = Relationship(sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.ex4_property_id]"})
|
||||
|
||||
# Related Models
|
||||
door: Optional["Door"] = Relationship(back_populates="property_description")
|
||||
ventilation_and_cooling: Optional["VentilationAndCooling"] = Relationship(back_populates="property_description")
|
||||
lighting: Optional["Lighting"] = Relationship(back_populates="property_description")
|
||||
water_heating: Optional["WaterHeating"] = Relationship(back_populates="property_description")
|
||||
hot_water_cylinder: Optional["HotWaterCylinder"] = Relationship(back_populates="property_description")
|
||||
solar_water_heating: Optional["SolarWaterHeating"] = Relationship(back_populates="property_description")
|
||||
shower_and_baths: Optional["ShowerAndBaths"] = Relationship(back_populates="property_description")
|
||||
flue_gas_heat_recovery_system: Optional["FlueGasHeatRecoverySystem"] = Relationship(back_populates="property_description")
|
||||
photovoltaic_panel: Optional["PhotovoltaicPanel"] = Relationship(back_populates="property_description")
|
||||
wind_turbine: Optional["WindTurbine"] = Relationship(back_populates="property_description")
|
||||
other_details: Optional["OtherDetails"] = Relationship(back_populates="property_description")
|
||||
main_heating: Optional["HeatingFromPreSiteNotes"] = Relationship(back_populates="property_description", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating_id]"})
|
||||
main_heating2: Optional["HeatingFromPreSiteNotes"] = Relationship(back_populates="property_description", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating2_id]"})
|
||||
secondary_heating_type: Optional["HeatingType"] = Relationship(back_populates="property_description")
|
||||
# # Related Models
|
||||
# door: Optional["Door"] = Relationship(back_populates="property_description")
|
||||
# ventilation_and_cooling: Optional["VentilationAndCooling"] = Relationship(back_populates="property_description")
|
||||
# lighting: Optional["Lighting"] = Relationship(back_populates="property_description")
|
||||
# water_heating: Optional["WaterHeating"] = Relationship(back_populates="property_description")
|
||||
# hot_water_cylinder: Optional["HotWaterCylinder"] = Relationship(back_populates="property_description")
|
||||
# solar_water_heating: Optional["SolarWaterHeating"] = Relationship(back_populates="property_description")
|
||||
# shower_and_baths: Optional["ShowerAndBaths"] = Relationship(back_populates="property_description")
|
||||
# flue_gas_heat_recovery_system: Optional["FlueGasHeatRecoverySystem"] = Relationship(back_populates="property_description")
|
||||
# photovoltaic_panel: Optional["PhotovoltaicPanel"] = Relationship(back_populates="property_description")
|
||||
# wind_turbine: Optional["WindTurbine"] = Relationship(back_populates="property_description")
|
||||
# other_details: Optional["OtherDetails"] = Relationship(back_populates="property_description")
|
||||
# main_heating: Optional["HeatingFromPreSiteNotes"] = Relationship(back_populates="property_description", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating_id]"})
|
||||
# main_heating2: Optional["HeatingFromPreSiteNotes"] = Relationship(back_populates="property_description", sa_relationship_kwargs={"foreign_keys": "[PropertyDescription.main_heating2_id]"})
|
||||
# secondary_heating_type: Optional["HeatingType"] = Relationship(back_populates="property_description")
|
||||
|
||||
pre_site_notes: Optional["PreSiteNote"] = Relationship(back_populates="pre_site_note_description")
|
||||
# pre_site_notes: Optional["PreSiteNote"] = Relationship(back_populates="pre_site_note_description")
|
||||
|
||||
class AssessorInfo(BaseModel, table=True):
|
||||
accreditation_number: str
|
||||
name: str
|
||||
phone_number: Optional[str] = None
|
||||
email_address: Optional[EmailStr] = None
|
||||
# class AssessorInfo(BaseModel, table=True):
|
||||
# accreditation_number: str
|
||||
# name: str
|
||||
# phone_number: Optional[str] = None
|
||||
# email_address: Optional[EmailStr] = None
|
||||
|
||||
company_id: Optional[uuid.UUID] = Field(default=None, foreign_key="companyinfo.id")
|
||||
company: Optional["CompanyInfo"] = Relationship(back_populates="assessors")
|
||||
|
||||
pre_site_notes: List["PreSiteNote"] = Relationship(back_populates="assessor")
|
||||
documents: List["Documents"] = Relationship(back_populates="author")
|
||||
|
||||
|
||||
class PreSiteNotesSummaryInfo(BaseModel, table=True):
|
||||
reference_number: str
|
||||
epc_language: str
|
||||
uprn: Optional[str] = ""
|
||||
postcode: str
|
||||
region: str
|
||||
address: str
|
||||
town: str
|
||||
county: Optional[str] = ""
|
||||
property_tenure: str
|
||||
transaction_type: str
|
||||
inspection_date: datetime
|
||||
current_sap: str
|
||||
potential_sap: str
|
||||
current_ei: str
|
||||
potential_ei: str
|
||||
current_annual_emissions: str
|
||||
current_annual_emission_including_0925_multiplayer: str
|
||||
current_annual_energy_costs: str
|
||||
|
||||
pre_site_notes: List["PreSiteNote"] = Relationship(back_populates="summary_info")
|
||||
|
||||
class CompanyInfo(BaseModel, table=True):
|
||||
address: str
|
||||
trading_name: str
|
||||
post_code: str
|
||||
fax_number: Optional[str] = None
|
||||
related_party_disclosure: Optional[str] = None
|
||||
|
||||
assessors: List[AssessorInfo] = Relationship(back_populates="company")
|
||||
|
||||
|
||||
class Insulation(BaseModel, table=True):
|
||||
type: str
|
||||
|
||||
|
||||
|
||||
PreSiteNote.update_forward_refs()
|
||||
AssessorInfo.update_forward_refs()
|
||||
# company_id: Optional[uuid.UUID] = Field(default=None, foreign_key="companyinfo.id")
|
||||
# company: Optional["CompanyInfo"] = Relationship(back_populates="assessors")
|
||||
|
||||
# pre_site_notes: List["PreSiteNote"] = Relationship(back_populates="assessor")
|
||||
# documents: List["Documents"] = Relationship(back_populates="author")
|
||||
|
||||
|
||||
# class PreSiteNotesSummaryInfo(BaseModel, table=True):
|
||||
# reference_number: str
|
||||
# epc_language: str
|
||||
# uprn: Optional[str] = ""
|
||||
# postcode: str
|
||||
# region: str
|
||||
# address: str
|
||||
# town: str
|
||||
# county: Optional[str] = ""
|
||||
# property_tenure: str
|
||||
# transaction_type: str
|
||||
# inspection_date: datetime
|
||||
# current_sap: str
|
||||
# potential_sap: str
|
||||
# current_ei: str
|
||||
# potential_ei: str
|
||||
# current_annual_emissions: str
|
||||
# current_annual_emission_including_0925_multiplayer: str
|
||||
# current_annual_energy_costs: str
|
||||
|
||||
# pre_site_notes: List["PreSiteNote"] = Relationship(back_populates="summary_info")
|
||||
|
||||
# class CompanyInfo(BaseModel, table=True):
|
||||
# address: str
|
||||
# trading_name: str
|
||||
# post_code: str
|
||||
# fax_number: Optional[str] = None
|
||||
# related_party_disclosure: Optional[str] = None
|
||||
|
||||
# assessors: List[AssessorInfo] = Relationship(back_populates="company")
|
||||
|
||||
|
||||
# class Insulation(BaseModel, table=True):
|
||||
# type: str
|
||||
|
||||
|
||||
|
||||
# PreSiteNote.update_forward_refs()
|
||||
# AssessorInfo.update_forward_refs()
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
from sqlmodel import Field, SQLModel, Relationship, Column, text
|
||||
from sqlmodel import Field, SQLModel, Relationship, text
|
||||
import uuid
|
||||
from typing import Optional, List
|
||||
from datetime import datetime
|
||||
|
|
@ -10,41 +10,37 @@ from etl.fileReader.reportType import ReportType
|
|||
from sqlalchemy import DateTime
|
||||
from sqlalchemy.dialects.postgresql import JSON
|
||||
from sqlalchemy import Text
|
||||
from enum import Enum
|
||||
from sqlalchemy import Column
|
||||
|
||||
|
||||
class BaseModel(SQLModel):
|
||||
# Put primary_key=True in Column; don't pass primary_key to Field
|
||||
id: uuid.UUID = Field(
|
||||
sa_column=Column(
|
||||
UUID(as_uuid=True),
|
||||
primary_key=True,
|
||||
nullable=False,
|
||||
server_default=text("gen_random_uuid()"), # requires pgcrypto extension
|
||||
)
|
||||
)
|
||||
# Generate a fresh Column per table (no shared Column instance)
|
||||
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
||||
|
||||
class Buildings(BaseModel, table=True):
|
||||
address: str
|
||||
postcode: str
|
||||
UPRN: str
|
||||
landlord_id: str
|
||||
domna_id: str
|
||||
# class Buildings(BaseModel, table=True):
|
||||
# address: str
|
||||
# postcode: str
|
||||
# UPRN: str
|
||||
# landlord_id: str
|
||||
# domna_id: str
|
||||
|
||||
documents: List["Documents"] = Relationship(back_populates="building")
|
||||
# documents: List["Documents"] = Relationship(back_populates="building")
|
||||
|
||||
class Documents(BaseModel, table=True):
|
||||
assessor_id: uuid.UUID = Field(
|
||||
foreign_key="assessorinfo.id",
|
||||
nullable=False
|
||||
)
|
||||
author: Optional["AssessorInfo"] = Relationship(back_populates="documents")
|
||||
created_at: datetime
|
||||
document_type: ReportType
|
||||
# class Documents(BaseModel, table=True):
|
||||
# assessor_id: uuid.UUID = Field(
|
||||
# foreign_key="assessorinfo.id",
|
||||
# nullable=False
|
||||
# )
|
||||
# author: Optional["AssessorInfo"] = Relationship(back_populates="documents")
|
||||
# created_at: datetime
|
||||
# document_type: ReportType
|
||||
|
||||
building_id: uuid.UUID = Field(foreign_key="buildings.id", nullable=False)
|
||||
building: Optional["Buildings"] = Relationship(back_populates="documents")
|
||||
# building_id: uuid.UUID = Field(foreign_key="buildings.id", nullable=False)
|
||||
# building: Optional["Buildings"] = Relationship(back_populates="documents")
|
||||
|
||||
target_table: str
|
||||
target_id: uuid.UUID
|
||||
# target_table: str
|
||||
# target_id: uuid.UUID
|
||||
|
||||
class ReportType(str, Enum):
|
||||
QUIDOS_PRESITE_NOTE = "QUIDOS_PRESITE_NOTE"
|
||||
|
|
@ -54,6 +50,10 @@ class ReportType(str, Enum):
|
|||
OVERWRITING_U_VALUE_DECLARATION_FORM = "OVERWRITING_U_VALUE_DECLARATION_FORM"
|
||||
OSMOSIS_CONDITION_PAS_2035_REPORT = "OSMOSIS_CONDITION_PAS_2035_REPORT"
|
||||
DOMNA_CONDITION_PAS_2035_REPORT = "DOMNA_CONDITION_PAS_2035_REPORT"
|
||||
# Decent Homes Things
|
||||
DECENT_HOMES_RAW_DATA = "DECENT_HOMES_RAW_DATA"
|
||||
DECENT_HOMES_SUMMARY = "DECENT_HOMES_SUMMARY"
|
||||
DECENT_HOMES_PROPERTY_META = "DECENT_HOMES_PROPERTY_META"
|
||||
|
||||
class uploaded_files(BaseModel, table=True):
|
||||
__tablename__ = "uploaded_files"
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
#poetry run alembic revision --autogenerate -m "json_uri is a string"
|
||||
#poetry run alembic revision --autogenerate -m "added more enums"
|
||||
|
||||
poetry run alembic upgrade head
|
||||
|
||||
|
|
|
|||
16
poetry.lock
generated
16
poetry.lock
generated
|
|
@ -1,4 +1,4 @@
|
|||
# This file is automatically @generated by Poetry 2.1.3 and should not be changed by hand.
|
||||
# This file is automatically @generated by Poetry 2.1.4 and should not be changed by hand.
|
||||
|
||||
[[package]]
|
||||
name = "alembic"
|
||||
|
|
@ -479,6 +479,18 @@ idna = ["idna (>=3.7)"]
|
|||
trio = ["trio (>=0.23)"]
|
||||
wmi = ["wmi (>=1.5.1)"]
|
||||
|
||||
[[package]]
|
||||
name = "docutils"
|
||||
version = "0.22.2"
|
||||
description = "Docutils -- Python Documentation Utilities"
|
||||
optional = false
|
||||
python-versions = ">=3.9"
|
||||
groups = ["main"]
|
||||
files = [
|
||||
{file = "docutils-0.22.2-py3-none-any.whl", hash = "sha256:b0e98d679283fc3bb0ead8a5da7f501baa632654e7056e9c5846842213d674d8"},
|
||||
{file = "docutils-0.22.2.tar.gz", hash = "sha256:9fdb771707c8784c8f2728b67cb2c691305933d68137ef95a75db5f4dfbc213d"},
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "email-validator"
|
||||
version = "2.2.0"
|
||||
|
|
@ -2097,4 +2109,4 @@ files = [
|
|||
[metadata]
|
||||
lock-version = "2.1"
|
||||
python-versions = ">=3.12"
|
||||
content-hash = "dfda98ea4e00851a83a2f67c231b59476d407a1e38006610722c64842976e736"
|
||||
content-hash = "326b28eb13e9454717ce89fe4100053066e927e5698895690f71c14cfe0c1d6c"
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ dependencies = [
|
|||
"hubspot-api-client (>=12.0.0,<13.0.0)",
|
||||
"boto3 (>=1.39.6,<2.0.0)",
|
||||
"psycopg2-binary (>=2.9.10,<3.0.0)",
|
||||
"docutils (>=0.22.2,<0.23.0)",
|
||||
]
|
||||
|
||||
[tool.poetry]
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue