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

run all tests

Browse source
This commit is contained in:
Jun-te Kim 2026-01-28 15:54:38 +00:00
parent d5f4799675
commit ef40363a69
5 changed files with 114 additions and 121 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.devcontainer/backend/devcontainer.json
View file

@ -21,7 +21,8 @@
"jgclark.vscode-todo-highlight", "jgclark.vscode-todo-highlight",
"corentinartaud.pdfpreview", "corentinartaud.pdfpreview",
"ms-python.vscode-python-envs", "ms-python.vscode-python-envs",
"ms-python.black-formatter" "ms-python.black-formatter",
"waderyan.gitblame"
], ],
"settings": { "settings": {
"files.defaultWorkspace": "/workspaces/model", "files.defaultWorkspace": "/workspaces/model",

3
.github/workflows/unit_tests.yml vendored
View file

@ -2,6 +2,9 @@ name: Run unit tests
on: on:
pull_request: pull_request:
branches:
- '*'
jobs: jobs:
test: test:

213
backend/address2UPRN/main.py
View file

@ -7,8 +7,10 @@ from tqdm import tqdm
import re import re
EPC_AUTH_TOKEN = os.getenv("EPC_AUTH_TOKEN", "a2Nvbm5rb3dsZXNzYXJAZ21haWwuY29tOjY5MGJiMWM0NmIyOGI5ZDUxYzAxMzQzYzNiZGNlZGJjZDNmODQwMzA=") EPC_AUTH_TOKEN = os.getenv(
client = EpcClient(auth_token=EPC_AUTH_TOKEN) "EPC_AUTH_TOKEN",
"a2Nvbm5rb3dsZXNzYXJAZ21haWwuY29tOjY5MGJiMWM0NmIyOGI5ZDUxYzAxMzQzYzNiZGNlZGJjZDNmODQwMzA=",
)
import re import re
from difflib import SequenceMatcher from difflib import SequenceMatcher
@ -31,8 +33,6 @@ def levenshtein(a: str, b: str) -> float:
def extract_numbers(s: str) -> Set[str]: def extract_numbers(s: str) -> Set[str]:
return set(extract_number_sequence(s)) return set(extract_number_sequence(s))
def tokenise(s: str) -> Set[str]: def tokenise(s: str) -> Set[str]:
return set(s.split()) return set(s.split())
@ -54,9 +54,10 @@ def levenshtein(a: str, b: str) -> float:
seq_a = extract_number_sequence(a_norm) seq_a = extract_number_sequence(a_norm)
seq_b = extract_number_sequence(b_norm) seq_b = extract_number_sequence(b_norm)
has_flat_token_user = any(tok in a_norm for tok in ("flat", "apt", "apartment", "unit")) has_flat_token_user = any(
has_flat_token_epc = "flat" in b_norm tok in a_norm for tok in ("flat", "apt", "apartment", "unit")
)
has_flat_token_epc = "flat" in b_norm
if ( if (
len(seq_a) == 2 len(seq_a) == 2
@ -67,7 +68,6 @@ def levenshtein(a: str, b: str) -> float:
): ):
return 0.0 return 0.0
# --- token similarity (order-independent) --- # --- token similarity (order-independent) ---
toks_a = tokenise(a_norm) toks_a = tokenise(a_norm)
toks_b = tokenise(b_norm) toks_b = tokenise(b_norm)
@ -82,8 +82,7 @@ def levenshtein(a: str, b: str) -> float:
# --- weighted blend --- # --- weighted blend ---
return round( return round(
0.65 * token_score + 0.65 * token_score + 0.35 * char_score,
0.35 * char_score,
4, 4,
) )
@ -114,13 +113,11 @@ def normalise_address(s: str) -> str:
"cres": "crescent", "cres": "crescent",
"ct": "court", "ct": "court",
"dr": "drive", "dr": "drive",
# flats / units # flats / units
"apt": "flat", "apt": "flat",
"apartment": "flat", "apartment": "flat",
"unit": "flat", "unit": "flat",
"ste": "suite", "ste": "suite",
# numbering noise # numbering noise
"no": "", "no": "",
"no.": "", "no.": "",
@ -155,15 +152,15 @@ def score_addresses(
if column not in df.columns: if column not in df.columns:
raise ValueError(f"Missing column: {column}") raise ValueError(f"Missing column: {column}")
return df[column].apply( return df[column].apply(lambda x: levenshtein(user_address, x))
lambda x: levenshtein(user_address, x)
)
def get_epc_data_with_postcode(postcode, size=500, attempt=1, max_attempts=3): def get_epc_data_with_postcode(postcode, size=500, attempt=1, max_attempts=3):
""" """
Recursively fetch EPC data by postcode. Recursively fetch EPC data by postcode.
If results hit the size limit, retry with double size up to max_attempts. If results hit the size limit, retry with double size up to max_attempts.
""" """
client = EpcClient(auth_token=EPC_AUTH_TOKEN)
url = os.path.join(client.domestic.host, "search") url = os.path.join(client.domestic.host, "search")
@ -176,10 +173,7 @@ def get_epc_data_with_postcode(postcode, size=500, attempt=1, max_attempts=3):
params={"postcode": postcode}, params={"postcode": postcode},
) )
results_df = pd.DataFrame( results_df = pd.DataFrame(search_resp["rows"], columns=search_resp["column-names"])
search_resp["rows"],
columns=search_resp["column-names"]
)
row_count = len(results_df) row_count = len(results_df)
@ -217,13 +211,7 @@ def df_has_single_uprn(df: pd.DataFrame, uprn: str, column: str = "uprn") -> boo
return False return False
# Drop nulls and normalise to string # Drop nulls and normalise to string
uprns = ( uprns = df[column].dropna().astype(str).str.strip().unique()
df[column]
.dropna()
.astype(str)
.str.strip()
.unique()
)
# No valid UPRNs to compare # No valid UPRNs to compare
if len(uprns) == 0: if len(uprns) == 0:
@ -256,23 +244,13 @@ def get_uprn_candidates(
user_norm = normalise_address(user_address) user_norm = normalise_address(user_address)
out["lexiscore"] = out[address_column].apply( out["lexiscore"] = out[address_column].apply(lambda x: levenshtein(user_norm, x))
lambda x: levenshtein(user_norm, x)
)
# Normalise UPRN to string # Normalise UPRN to string
out[uprn_column] = ( out[uprn_column] = out[uprn_column].astype(str).str.replace(r"\.0$", "", regex=True)
out[uprn_column]
.astype(str)
.str.replace(r"\.0$", "", regex=True)
)
# Rank: 1 = best match # Rank: 1 = best match
out["lexirank"] = ( out["lexirank"] = out["lexiscore"].rank(method="dense", ascending=False).astype(int)
out["lexiscore"]
.rank(method="dense", ascending=False)
.astype(int)
)
return out.sort_values( return out.sort_values(
["lexirank", "lexiscore"], ["lexirank", "lexiscore"],
@ -307,6 +285,7 @@ def get_uprn(user_inputed_address: str, postcode: str):
# Safe to return the agreed UPRN # Safe to return the agreed UPRN
return top_rank_df.iloc[0]["uprn"] return top_rank_df.iloc[0]["uprn"]
def resolve_uprns_for_postcode_group( def resolve_uprns_for_postcode_group(
group_df: pd.DataFrame, group_df: pd.DataFrame,
epc_df: pd.DataFrame, epc_df: pd.DataFrame,
@ -332,46 +311,54 @@ def resolve_uprns_for_postcode_group(
) )
if scored_df.empty: if scored_df.empty:
results.append({ results.append(
"found_uprn": None, {
"best_match_uprn": None, "found_uprn": None,
"best_match_address": None, "best_match_uprn": None,
"best_match_lexiscore": None, "best_match_address": None,
"status": "no_epc_candidates", "best_match_lexiscore": None,
}) "status": "no_epc_candidates",
}
)
continue continue
best_score = scored_df.iloc[0]["lexiscore"] best_score = scored_df.iloc[0]["lexiscore"]
if best_score <= 0: if best_score <= 0:
results.append({ results.append(
"found_uprn": None, {
"best_match_uprn": None, "found_uprn": None,
"best_match_address": None, "best_match_uprn": None,
"best_match_lexiscore": best_score, "best_match_address": None,
"status": "zero_score", "best_match_lexiscore": best_score,
}) "status": "zero_score",
}
)
continue continue
top_rank_df = scored_df[scored_df["lexirank"] == 1] top_rank_df = scored_df[scored_df["lexirank"] == 1]
if not df_has_single_uprn(top_rank_df, top_rank_df.iloc[0]["uprn"]): if not df_has_single_uprn(top_rank_df, top_rank_df.iloc[0]["uprn"]):
results.append({ results.append(
"found_uprn": None, {
"best_match_uprn": top_rank_df.iloc[0]["uprn"], "found_uprn": None,
"best_match_address": top_rank_df.iloc[0]["address"], "best_match_uprn": top_rank_df.iloc[0]["uprn"],
"best_match_lexiscore": best_score, "best_match_address": top_rank_df.iloc[0]["address"],
"status": "ambiguous", "best_match_lexiscore": best_score,
}) "status": "ambiguous",
}
)
continue continue
results.append({ results.append(
"found_uprn": str(top_rank_df.iloc[0]["uprn"]), {
"best_match_uprn": str(top_rank_df.iloc[0]["uprn"]), "found_uprn": str(top_rank_df.iloc[0]["uprn"]),
"best_match_address": top_rank_df.iloc[0]["address"], "best_match_uprn": str(top_rank_df.iloc[0]["uprn"]),
"best_match_lexiscore": best_score, "best_match_address": top_rank_df.iloc[0]["address"],
"status": "matched", "best_match_lexiscore": best_score,
}) "status": "matched",
}
)
return pd.concat( return pd.concat(
[group_df.reset_index(drop=True), pd.DataFrame(results)], [group_df.reset_index(drop=True), pd.DataFrame(results)],
@ -379,8 +366,7 @@ def resolve_uprns_for_postcode_group(
) )
def test(a, b):
def test(a,b):
assert a == b, f"erorr: {a}{type(a)} != {b}: {type(b)}" assert a == b, f"erorr: {a}{type(a)} != {b}: {type(b)}"
@ -398,16 +384,23 @@ def run_all_test():
test(get_uprn("28A", "se6 4tf"), "100023278633") test(get_uprn("28A", "se6 4tf"), "100023278633")
test(get_uprn("6 Aitken Close", "E8 4SQ"), False) test(get_uprn("6 Aitken Close", "E8 4SQ"), False)
# unique case # unique case
test(get_uprn("Flat 5, 1, Semley Gate", "e9 5nh"), "10008238198") test(get_uprn("Flat 5, 1, Semley Gate", "e9 5nh"), "10008238198")
test(get_uprn("5 , 1 Semley Gate", "e9 5nh"), "10008238198") test(get_uprn("5 , 1 Semley Gate", "e9 5nh"), "10008238198")
test(get_uprn("5 Semley Gate", "e9 5nh"), "10008238198" ) test(get_uprn("5 Semley Gate", "e9 5nh"), "10008238198")
test(get_uprn("1, 5 Semley Gate", "e9 5nh"), False) test(get_uprn("1, 5 Semley Gate", "e9 5nh"), False)
test(get_uprn("1 Semley Gate", "e9 5nh"), "10008238188") # this one return "flat 1, in 1 semley gate" test(
test(get_uprn("48 Oswald Street", "E5 0BT"), False) # this one return "flat 1, in 1 semley gate" get_uprn("1 Semley Gate", "e9 5nh"), "10008238188"
test(get_uprn("42 Oswald Street", "E5 0BT"), False) # this one return "flat 1, in 1 semley gate" ) # this one return "flat 1, in 1 semley gate"
test(get_uprn("46 Oswald Street", "E5 0BT"), False) # this one return "flat 1, in 1 semley gate" test(
get_uprn("48 Oswald Street", "E5 0BT"), False
) # this one return "flat 1, in 1 semley gate"
test(
get_uprn("42 Oswald Street", "E5 0BT"), False
) # this one return "flat 1, in 1 semley gate"
test(
get_uprn("46 Oswald Street", "E5 0BT"), False
) # this one return "flat 1, in 1 semley gate"
get_uprn_candidates(get_epc_data_with_postcode("e5 0bt"), "48 Oswald Street") get_uprn_candidates(get_epc_data_with_postcode("e5 0bt"), "48 Oswald Street")
@ -430,24 +423,22 @@ if __name__ == "__main__":
input_address = str(row[ADDRESS_COL]).strip() input_address = str(row[ADDRESS_COL]).strip()
postcode = str(row[POSTCODE_COL]).strip() postcode = str(row[POSTCODE_COL]).strip()
expected_uprn = ( expected_uprn = None if pd.isna(row[UPRN_COL]) else str(int(row[UPRN_COL]))
None
if pd.isna(row[UPRN_COL])
else str(int(row[UPRN_COL]))
)
try: try:
epc_df = get_epc_data_with_postcode(postcode) epc_df = get_epc_data_with_postcode(postcode)
if epc_df.empty: if epc_df.empty:
failures.append({ failures.append(
**row.to_dict(), {
"found_uprn": None, **row.to_dict(),
"best_match_uprn": None, "found_uprn": None,
"best_match_address": None, "best_match_uprn": None,
"best_match_lexiscore": None, "best_match_address": None,
"status": "no_epc_results", "best_match_lexiscore": None,
}) "status": "no_epc_results",
}
)
continue continue
scored_df = get_uprn_candidates( scored_df = get_uprn_candidates(
@ -464,34 +455,32 @@ if __name__ == "__main__":
found_uprn = get_uprn(input_address, postcode) found_uprn = get_uprn(input_address, postcode)
except Exception as e: except Exception as e:
failures.append({ failures.append(
**row.to_dict(), {
"found_uprn": None, **row.to_dict(),
"best_match_uprn": None, "found_uprn": None,
"best_match_address": None, "best_match_uprn": None,
"best_match_lexiscore": None, "best_match_address": None,
"status": "exception", "best_match_lexiscore": None,
"error": str(e), "status": "exception",
}) "error": str(e),
}
)
continue continue
found_uprn_norm = ( found_uprn_norm = None if not found_uprn else str(found_uprn)
None if not found_uprn else str(found_uprn)
)
if found_uprn_norm != expected_uprn: if found_uprn_norm != expected_uprn:
failures.append({ failures.append(
**row.to_dict(), {
"found_uprn": found_uprn_norm, **row.to_dict(),
"best_match_uprn": best_match_uprn, "found_uprn": found_uprn_norm,
"best_match_address": best_match_address, "best_match_uprn": best_match_uprn,
"best_match_lexiscore": best_match_lexiscore, "best_match_address": best_match_address,
"status": ( "best_match_lexiscore": best_match_lexiscore,
"no_match" "status": ("no_match" if found_uprn_norm is None else "mismatch"),
if found_uprn_norm is None }
else "mismatch" )
),
})
failures_df = pd.DataFrame(failures) failures_df = pd.DataFrame(failures)

2
conftest.py
View file

@ -8,7 +8,7 @@ DEFAULT_ENV = {
"DATA_BUCKET": "test", "DATA_BUCKET": "test",
"PLAN_TRIGGER_BUCKET": "test", "PLAN_TRIGGER_BUCKET": "test",
"ENGINE_SQS_URL": "test", "ENGINE_SQS_URL": "test",
"EPC_AUTH_TOKEN": "test", # overridden in GitHub Actions "EPC_AUTH_TOKEN": "a2Nvbm5rb3dsZXNzYXJAZ21haWwuY29tOjY5MGJiMWM0NmIyOGI5ZDUxYzAxMzQzYzNiZGNlZGJjZDNmODQwMzA=", # overridden in GitHub Actions
"GOOGLE_SOLAR_API_KEY": "test", "GOOGLE_SOLAR_API_KEY": "test",
"DB_HOST": "localhost", "DB_HOST": "localhost",
"DB_USERNAME": "test", "DB_USERNAME": "test",

2
pytest.ini
View file

@ -1,4 +1,4 @@
[pytest] [pytest]
pythonpath = . pythonpath = .
addopts = --cov-report term-missing --cov=etl/epc --cov=recommendations --cov=backend --cov=etl/epc_clean --cov=etl/spatial addopts = --cov-report term-missing --cov=etl/epc --cov=recommendations --cov=backend --cov=etl/epc_clean --cov=etl/spatial
testpaths = recommendations/tests backend/tests etl/epc/tests etl/epc_clean/tests etl/spatial/tests backend/condition/tests testpaths = recommendations/tests backend/tests etl/epc/tests etl/epc_clean/tests etl/spatial/tests backend/condition/tests backend/address2UPRN/tests