junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-08 11:17:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #707 from Hestia-Homes/main

Browse source 
Deployment Feb 13th
This commit is contained in:
KhalimCK 2026-02-13 12:59:46 +00:00 • committed by GitHub
commit cf14ff6f0c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 428 additions and 102 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.devcontainer/asset_list/requirements.txt
View file

@ -7,7 +7,7 @@ mangum==0.19.0
# AWS
boto3==1.35.44
# Data
openpyxl==3.1.2
openpyxl==3.1.5
# Basic
pytz
uvicorn[standard]

2
.github/workflows/_build_image.yml vendored
View file

@ -104,4 +104,4 @@ jobs:
--image-ids imageTag=${GITHUB_SHA} \
--query 'imageDetails[0].imageDigest' \
--output text)
echo "image_digest=$DIGEST" >> "$GITHUB_OUTPUT"
echo "image_digest=$DIGEST" >> "$GITHUB_OUTPUT"

18
.github/workflows/_deploy_lambda.yml vendored
View file

@ -23,6 +23,18 @@ on:
required: true
type: string
terraform_apply:
required: false
type: string
default: 'false'
# can only be 'true' or 'false'
terraform_destroy:
required: false
type: string
default: 'false'
# can only be 'true' or 'false'
secrets:
AWS_ACCESS_KEY_ID:
required: true
@ -87,5 +99,11 @@ jobs:
-out=lambdaplan
- name: Terraform Apply
if: inputs.terraform_apply == 'true' && inputs.terraform_destroy != 'true'
working-directory: ${{ inputs.lambda_path }}
run: terraform apply -auto-approve lambdaplan
- name: Terraform Destroy
if: inputs.terraform_destroy == 'true' && inputs.terraform_apply != 'true'
working-directory: ${{ inputs.lambda_path }}
run: terraform destroy -auto-approve

33
.github/workflows/deploy_terraform.yml vendored
View file

@ -4,29 +4,43 @@ on:
push:
branches:
- "**"
paths:
- 'infrastructure/terraform/**'
- '.github/workflows/deploy_terraform.yml'
- '.github/workflows/_build_image.yml'
- '.github/workflows/_deploy_lambda.yml'
jobs:
determine_stage:
runs-on: ubuntu-latest
outputs:
stage: ${{ steps.set-stage.outputs.stage }}
terraform_apply: ${{ steps.set-stage.outputs.terraform_apply }}
env:
AWS_ACCESS_KEY_ID: ${{ secrets.DEV_AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }}
AWS_REGION: ${{ secrets.DEV_AWS_REGION }}
DEV_DB_HOST: ${{ secrets.DEV_DB_HOST }}
steps:
- name: Determine stage from branch
id: set-stage
shell: bash
run: |
env
BRANCH="${GITHUB_REF_NAME}"
if [[ "$BRANCH" == "prod" ]]; then
echo "stage=prod" >> "$GITHUB_OUTPUT"
echo "terraform_apply=false" >> "$GITHUB_OUTPUT"
elif [[ "$BRANCH" == "dev" ]]; then
echo "stage=dev" >> "$GITHUB_OUTPUT"
echo "terraform_apply=true" >> "$GITHUB_OUTPUT"
else
# Feature branch
echo "stage=dev" >> "$GITHUB_OUTPUT"
echo "terraform_apply=false" >> "$GITHUB_OUTPUT"
fi
# ============================================================
@ -93,6 +107,7 @@ jobs:
stage: ${{ needs.determine_stage.outputs.stage }}
ecr_repo: address2uprn-${{ needs.determine_stage.outputs.stage }}
image_digest: ${{ needs.address2uprn_image.outputs.image_digest }}
terraform_apply: ${{ needs.determine_stage.outputs.terraform_apply }}
secrets:
AWS_ACCESS_KEY_ID: ${{ secrets.DEV_AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }}
@ -109,10 +124,17 @@ jobs:
ecr_repo: postcode_splitter-${{ needs.determine_stage.outputs.stage }}
dockerfile_path: backend/postcode_splitter/handler/Dockerfile
build_context: .
build_args: |
DEV_DB_HOST=$DEV_DB_HOST
DEV_DB_PORT=$DEV_DB_PORT
DEV_DB_NAME=$DEV_DB_NAME
secrets:
AWS_ACCESS_KEY_ID: ${{ secrets.DEV_AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }}
AWS_REGION: ${{ secrets.DEV_AWS_REGION }}
DEV_DB_HOST: ${{ secrets.DEV_DB_HOST }}
DEV_DB_PORT: ${{ secrets.DEV_DB_PORT }}
DEV_DB_NAME: ${{ secrets.DEV_DB_NAME }}
# ============================================================
# 3️⃣ Deploy Postcode Splitter Lambda
@ -126,6 +148,7 @@ jobs:
stage: ${{ needs.determine_stage.outputs.stage }}
ecr_repo: postcode_splitter-${{ needs.determine_stage.outputs.stage }}
image_digest: ${{ needs.postcodeSplitter_image.outputs.image_digest }}
terraform_apply: ${{ needs.determine_stage.outputs.terraform_apply }}
secrets:
AWS_ACCESS_KEY_ID: ${{ secrets.DEV_AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }}
@ -165,8 +188,8 @@ jobs:
stage: ${{ needs.determine_stage.outputs.stage }}
ecr_repo: condition-etl-${{ needs.determine_stage.outputs.stage }}
image_digest: ${{ needs.condition_etl_image.outputs.image_digest }}
terraform_apply: ${{ needs.determine_stage.outputs.terraform_apply }}
secrets:
AWS_ACCESS_KEY_ID: ${{ secrets.DEV_AWS_ACCESS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }}
AWS_REGION: ${{ secrets.DEV_AWS_REGION }}
AWS_REGION: ${{ secrets.DEV_AWS_REGION }}

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

@ -4,9 +4,6 @@ on:
pull_request:
branches:
- "**"
push:
branches:
- "**"
jobs:
@ -30,4 +27,4 @@ jobs:
env:
EPC_AUTH_TOKEN: ${{ secrets.DEV_EPC_AUTH_TOKEN }}
run: |
make test
make test

1
.idea/inspectionProfiles/profiles_settings.xml generated
View file

@ -1,5 +1,6 @@
<component name="InspectionProjectProfileManager">
<settings>
<option name="PROJECT_PROFILE" value="Default" />
<option name="USE_PROJECT_PROFILE" value="false" />
<version value="1.0" />
</settings>

18
asset_list/app.py
View file

@ -69,24 +69,24 @@ def app():
Property UPRN
"""
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/Aspire"
data_filename = "ASPIRE ASSET LIST.xlsx"
sheet_name = "Asset List"
postcode_column = "Postcode"
data_folder = "/Users/khalimconn-kowlessar/Documents/hestia/Customers/West Kent"
data_filename = "West Kent Asset List.xlsx"
sheet_name = "Sheet1"
postcode_column = "POSTCODE"
address1_column = None
address1_method = "house_number_extraction"
fulladdress_column = "Address"
fulladdress_column = "ADDRESS"
address_cols_to_concat = []
missing_postcodes_method = None
landlord_year_built = None
landlord_os_uprn = None
landlord_property_type = "Property Type"
landlord_property_type = "PROPERTY TYPE"
landlord_built_form = None
landlord_wall_construction = None
landlord_roof_construction = None
landlord_wall_construction = "wall combined"
landlord_roof_construction = "HEATING SYSTEM"
landlord_heating_system = None
landlord_existing_pv = None
landlord_property_id = "LLUPRN"
landlord_property_id = "UPRN"
landlord_sap = None
outcomes_filename = None
outcomes_sheetname = None

14
asset_list/mappings/roof.py
View file

@ -308,6 +308,18 @@ ROOF_CONSTRUCTION_MAPPINGS = {
'Flat: No Insulation': 'flat uninsulated',
'AnotherDwellingAbove: Unknown, PitchedNormalLoftAccess: 250mm': 'another dwelling above',
'PitchedNormalLoftAccess: 175mm': 'pitched insulated',
'AnotherDwellingAbove: 300mm': 'another dwelling above'
'AnotherDwellingAbove: 300mm': 'another dwelling above',
'Another dwelling above, As built': 'another dwelling above',
'Pitched (slates or tiles) no loft access, 400mm+': 'pitched insulated',
'Pitched (slates or tiles) access to loft, 400mm+': 'pitched insulated',
'Pitched (slates or tiles) access to loft, 300mm': 'pitched insulated',
'Pitched (slates or tiles) access to loft, 75mm': 'pitched less than 100mm insulation',
'Pitched (slates or tiles) no loft access, 300mm': 'pitched insulated',
'Pitched (slates or tiles) access to loft, 270mm': 'pitched insulated',
'Pitched (slates or tiles) access to loft, 100mm': 'pitched insulated',
'Pitched (slates or tiles) no loft access, 200mm': 'pitched insulated',
'Pitched (slates or tiles) access to loft, 200mm': 'pitched insulated',
'Pitched (slates or tiles) access to loft, 50mm': 'pitched less than 100mm insulation'
}

8
asset_list/mappings/walls.py
View file

@ -363,6 +363,12 @@ WALL_CONSTRUCTION_MAPPINGS = {
'Timber Frame, As Built': 'timber frame unknown insulation',
'Solid Brick, Internal Insulation': 'insulated solid brick',
'Granite or Whinstone, As Built': 'granite or whinstone unknown insulation',
'Solid Brick, External': 'insulated solid brick'
'Solid Brick, External': 'insulated solid brick',
'Cavity, Filled cavity': 'filled cavity',
'Solid Brick, As built': 'solid brick unknown insulation',
'System built, As built': 'system built unknown insulation',
'Timber frame, As built': 'timber frame unknown insulation',
'Cavity, As built': 'cavity unknown insulation'
}

6
asset_list/utils.py
View file

@ -220,7 +220,7 @@ def get_data(
searcher.find_property(skip_os=True)
# Check if we have a flat or appartment
if searcher.newest_epc is None and uprn is None:
if not searcher.newest_epc and uprn is None:
# Try again:
if SearchEpc.get_house_number(address=str(house_number), postcode=postcode) is None:
# Backup
@ -252,12 +252,12 @@ def get_data(
searcher.find_property(skip_os=True)
# As a final resort, we estimate the EPC
if property_type is not None and searcher.newest_epc is None:
if property_type is not None and not searcher.newest_epc:
searcher.ordnance_survey_client.property_type = property_type
searcher.ordnance_survey_client.built_form = built_form
searcher.find_property(skip_os=True)
if searcher.newest_epc is None:
if not searcher.newest_epc:
no_epc.append(home[row_id_name])
continue

4
backend/addresses/Addresses.py
View file

@ -1,3 +1,4 @@
from typing import Iterator
from backend.addresses.Address import Address
@ -12,6 +13,9 @@ class Addresses:
def __len__(self) -> int:
return len(self._addresses)
def __iter__(self) -> Iterator[Address]:
return iter(self._addresses)
@classmethod
def from_plan_input(cls, plan_input: list[dict], body) -> "Addresses":
addresses = []

2
backend/engine/engine.py
View file

@ -865,7 +865,7 @@ async def model_engine(body: PlanTriggerRequest):
check_duplicate_property_ids(input_properties)
logger.info("Inserting property data")
# We now bulk upload all of the EPC data
# We now bulk upload all the EPC data
with db_session() as session:
db_funcs.epc_functions.EpcStoreService.bulk_upsert_epc_data(session, epc_upserts)

6
etl/epc_clean/epc_attributes/RoofAttributes.py
View file

@ -74,6 +74,8 @@ class RoofAttributes(Definitions):
"insulation_thickness",
]
NODATA_NULLS = ["insulation_thickness", "thermal_transmittance", "thermal_transmittance_unit"]
def __init__(self, description: str):
"""
:param description: Description of the roof.
@ -153,6 +155,10 @@ class RoofAttributes(Definitions):
if self.nodata:
for key in self.DEFAULT_KEYS:
result[key] = False
# Insulation thickness, thermal transmittance and thermal transmittance unit are set to None for nodata
# cases
for k in self.NODATA_NULLS:
result[k] = None
return result
description = self.description

14
etl/epc_clean/tests/test_roof_attributes.py
View file

@ -26,11 +26,10 @@ class TestRoofAttributes:
def test_empty_str(self):
# Test initialization with an empty description
assert RoofAttributes('').process() == {
'thermal_transmittance': False, 'thermal_transmittance_unit': False, 'is_pitched': False,
'thermal_transmittance': None, 'thermal_transmittance_unit': None, 'is_pitched': False,
'is_roof_room': False, 'is_loft': False, 'is_flat': False, 'is_thatched': False, 'is_at_rafters': False,
'is_assumed': False, 'has_dwelling_above': False, 'is_valid': False, 'insulation_thickness': False
'is_assumed': False, 'has_dwelling_above': False, 'is_valid': False, 'insulation_thickness': None
}
assert set(list(RoofAttributes('').process().values())) == {False}
def test_clean_roof(self):
result = RoofAttributes('Pitched, 270 mm loft insulation').process()
@ -92,15 +91,6 @@ class TestRoofAttributes:
with pytest.raises(ValueError):
RoofAttributes('nonsense string').process()
def test_clean_roof_no_description(self):
roof = RoofAttributes('').process()
assert roof == {
'thermal_transmittance': False, 'thermal_transmittance_unit': False, 'is_pitched': False,
'is_roof_room': False, 'is_loft': False, 'is_flat': False, 'is_thatched': False,
'is_at_rafters': False, 'is_assumed': False, 'has_dwelling_above': False, 'is_valid': False,
'insulation_thickness': False
}
def test_clean_roof_edge_cases(self):
# Insulation thickness edge case
assert RoofAttributes('Pitched, 99999 mm loft insulation').process()['insulation_thickness'] == "99999"

18
recommendations/Recommendations.py
View file

@ -768,6 +768,24 @@ class Recommendations:
# Update the current phase values
current_phase_values["sap"] = previous_phase_values["sap"] + property_phase_impact["sap"]
# This is very much an edge case but we also the end result taking the property
# below a SAP rating of 1, which is the minimum SAP rating
if previous_phase_values["sap"] + property_phase_impact["sap"] < 1:
sap_adjustment = 1 - (previous_phase_values["sap"] + property_phase_impact["sap"])
adjustments.append(
{
"recommendation_id": rec["recommendation_id"],
"phase": rec["phase"],
"sap_adjustment": sap_adjustment,
}
)
# The new impact should be the current impact plus the adjustment
property_phase_impact["sap"] = property_phase_impact["sap"] + sap_adjustment
# Update the current phase values
current_phase_values["sap"] = previous_phase_values["sap"] + property_phase_impact["sap"]
elif rec["type"] == "loft_insulation":
# When we have a loft insulation recommendation, where there is an extension and the existing
# amount of loft insulation is already good, we limit the SAP points

10
recommendations/RoofRecommendations.py
View file

@ -59,9 +59,9 @@ class RoofRecommendations:
# Extract the insulation thickness from the roof, which is used throughout this method
self.insulation_thickness = convert_thickness_to_numeric(
self.property.roof["insulation_thickness"],
self.property.roof["is_pitched"],
self.property.roof["is_flat"]
string_thickness=self.property.roof["insulation_thickness"],
is_pitched=self.property.roof["is_pitched"],
is_flat=self.property.roof["is_flat"]
)
@classmethod
@ -300,6 +300,10 @@ class RoofRecommendations:
):
return False
if self.property.roof["original_description"] is None:
# There is no description so we cannot make an assessment
return False
return True
@staticmethod

41
recommendations/optimiser/funding_optimiser.py
View file

@ -10,6 +10,7 @@ In the future, we will adapt this into a class-based structure to allow for more
from copy import deepcopy
import pandas as pd
import numpy as np
from typing import Mapping, Union
from itertools import product
from backend.app.plan.schemas import (
@ -823,21 +824,23 @@ def optimise_with_scenarios(
# No special path; just exclude ASHP from options and allow us to optimise.
measures_no_heat_pump = exclude_measure_types(optimisation_measures, ["air_source_heat_pump"])
picked, total_cost, total_gain = run_optimizer(
measures_no_heat_pump,
budget=budget,
sub_target_gain=target_gain,
)
if target_gain > 0:
# If we don't have any gain, we don't actually need to do this
picked, total_cost, total_gain = run_optimizer(
measures_no_heat_pump,
budget=budget,
sub_target_gain=target_gain,
)
if picked is not None:
solutions.append({
"scenario": "no_heat_pump",
"items": picked,
"fixed_items": [],
"total_cost": total_cost,
"total_gain": total_gain,
"already_installed_gain": sum([x["gain"] for x in picked if x["already_installed"]])
})
if picked is not None:
solutions.append({
"scenario": "no_heat_pump",
"items": picked,
"fixed_items": [],
"total_cost": total_cost,
"total_gain": total_gain,
"already_installed_gain": sum([x["gain"] for x in picked if x["already_installed"]])
})
solutions_df = append_solution_metrics(solutions, target_gain, p, already_installed_sap)
@ -1101,7 +1104,12 @@ def contributes_min_insulation(opt_types):
})
def run_optimizer(input_measures, budget=None, sub_target_gain=None, allow_slack=False):
def run_optimizer(
input_measures: list[list[Mapping[str, int | float | str]]],
budget: Union[float, None] = None,
sub_target_gain: Union[float, None] = None,
allow_slack: bool = False
):
"""
Thin wrapper over your optimisers.
Returns: list[dict] selected_options
@ -1112,7 +1120,7 @@ def run_optimizer(input_measures, budget=None, sub_target_gain=None, allow_slack
if budget is not None:
opt = GainOptimiser(
input_measures, max_cost=budget, max_gain=(sub_target_gain or float("inf")),
input_measures, max_cost=budget, max_gain=0 if sub_target_gain == 0 else (sub_target_gain or float("inf")),
allow_slack=allow_slack
)
else:
@ -1123,6 +1131,7 @@ def run_optimizer(input_measures, budget=None, sub_target_gain=None, allow_slack
opt.setup()
opt.solve()
cost = sum([x["cost"] for x in opt.solution])
return opt.solution, cost, opt.solution_gain

141
recommendations/tests/test_optimisers.py
View file

@ -1,6 +1,7 @@
import pytest
from recommendations.optimiser.funding_optimiser import build_heat_pump_paths
from recommendations.optimiser.funding_optimiser import run_optimizer
class DummyProp:
@ -68,3 +69,143 @@ def test_build_heat_pump_paths():
assert eg2 == [{'AND': ['internal_wall_insulation', 'loft_insulation', 'air_source_heat_pump']},
{'AND': ['external_wall_insulation', 'loft_insulation', 'air_source_heat_pump']}]
def test_run_optimizer_empty_input():
solution, cost, gain = run_optimizer([])
assert solution is None
assert cost == 0.0
assert gain == 0.0
def test_uses_gain_optimiser_when_budget_provided(monkeypatch):
captured_args = {}
class FakeGainOptimiser:
def __init__(self, measures, max_cost, max_gain, allow_slack):
captured_args["measures"] = measures
captured_args["max_cost"] = max_cost
captured_args["max_gain"] = max_gain
captured_args["allow_slack"] = allow_slack
self.solution = [{"cost": 100}]
self.solution_gain = 5
def setup(self):
pass
def solve(self):
pass
monkeypatch.setattr(
"recommendations.optimiser.funding_optimiser.GainOptimiser",
FakeGainOptimiser
)
measures = [[{"cost": 100, "gain": 5}]]
solution, cost, gain = run_optimizer(
measures,
budget=500,
sub_target_gain=10,
allow_slack=True
)
assert captured_args["max_cost"] == 500
assert captured_args["max_gain"] == 10
assert captured_args["allow_slack"] is True
assert cost == 100
assert gain == 5
def test_sub_target_gain_zero_sets_max_gain_zero(monkeypatch):
captured_args = {}
class FakeGainOptimiser:
def __init__(self, measures, max_cost, max_gain, allow_slack):
captured_args["max_gain"] = max_gain
self.solution = []
self.solution_gain = 0
def setup(self):
pass
def solve(self):
pass
monkeypatch.setattr(
"recommendations.optimiser.funding_optimiser.GainOptimiser",
FakeGainOptimiser
)
measures = [[{"cost": 100, "gain": 5}]]
run_optimizer(
measures,
budget=500,
sub_target_gain=0
)
assert captured_args["max_gain"] == 0
def test_sub_target_gain_none_sets_max_gain_infinity(monkeypatch):
captured_args = {}
class FakeGainOptimiser:
def __init__(self, measures, max_cost, max_gain, allow_slack):
captured_args["max_gain"] = max_gain
self.solution = []
self.solution_gain = 0
def setup(self):
pass
def solve(self):
pass
monkeypatch.setattr(
"recommendations.optimiser.funding_optimiser.GainOptimiser",
FakeGainOptimiser
)
measures = [[{"cost": 100, "gain": 5}]]
run_optimizer(
measures,
budget=500,
sub_target_gain=None
)
assert captured_args["max_gain"] == float("inf")
def test_uses_cost_optimiser_when_no_budget(monkeypatch):
captured_args = {}
class FakeCostOptimiser:
def __init__(self, measures, min_gain):
captured_args["min_gain"] = min_gain
self.solution = [{"cost": 50}]
self.solution_gain = 10
def setup(self):
pass
def solve(self):
pass
monkeypatch.setattr(
"recommendations.optimiser.funding_optimiser.CostOptimiser",
FakeCostOptimiser
)
measures = [[{"cost": 50, "gain": 10}]]
solution, cost, gain = run_optimizer(
measures,
sub_target_gain=10
)
assert captured_args["min_gain"] == 10
assert cost == 50
assert gain == 10

126
recommendations/tests/test_recommendations.py
View file

@ -347,21 +347,21 @@ def property_instance():
"input_data, expected",
[
(
[
{"recommendation_id": "a", "phase": 0, "sap_adjustment": 1.7},
{"recommendation_id": "b", "phase": 0, "sap_adjustment": 1.7},
],
[{"recommendation_id": "a", "phase": 0, "sap_adjustment": 1.7}],
[
{"recommendation_id": "a", "phase": 0, "sap_adjustment": 1.7},
{"recommendation_id": "b", "phase": 0, "sap_adjustment": 1.7},
],
[{"recommendation_id": "a", "phase": 0, "sap_adjustment": 1.7}],
),
(
[
{"recommendation_id": "a", "phase": 1, "sap_adjustment": 2},
{"recommendation_id": "b", "phase": 2, "sap_adjustment": 3},
],
[
{"recommendation_id": "a", "phase": 1, "sap_adjustment": 2},
{"recommendation_id": "b", "phase": 2, "sap_adjustment": 3},
],
[
{"recommendation_id": "a", "phase": 1, "sap_adjustment": 2},
{"recommendation_id": "b", "phase": 2, "sap_adjustment": 3},
],
[
{"recommendation_id": "a", "phase": 1, "sap_adjustment": 2},
{"recommendation_id": "b", "phase": 2, "sap_adjustment": 3},
],
),
],
)
@ -1478,3 +1478,103 @@ def test_lighting_and_loft_adjustment_combined(property_instance, heat_demand_pr
{'recommendation_id': '0_phase=0', 'phase': 0, 'sap_adjustment': np.float64(1.7)},
{'recommendation_id': '4_phase=2', 'phase': 2, 'sap_adjustment': np.float64(4.0)}
]
def test_mechanical_ventilation_sap_floor(property_instance):
rec = {
"type": "mechanical_ventilation",
"recommendation_id": "mv_test",
"phase": 1,
}
previous_phase_values = {"sap": 2.0}
current_phase_values = {"sap": 0.5} # model prediction already below 1
property_phase_impact = {"sap": -1.5, "carbon": 0, "heat_demand": 0}
adjustments = []
updated_impact, updated_current, updated_adjustments = (
Recommendations._apply_measure_specific_rules(
rec=rec,
property_phase_impact=property_phase_impact,
previous_phase_values=previous_phase_values,
current_phase_values=current_phase_values,
adjustments=adjustments,
property_instance=property_instance
)
)
# SAP should be clamped to minimum 1
assert updated_current["sap"] == 1.0
# Original final SAP would have been 0.5 → so adjustment = 1 - 0.5 = 0.5
assert updated_adjustments == [
{
"recommendation_id": "mv_test",
"phase": 1,
"sap_adjustment": 0.5,
}
]
# Impact should now reflect new clamped SAP
assert updated_impact["sap"] == -1.0 # 2.0 → 1.0
def test_mechanical_ventilation_no_floor_adjustment(property_instance):
rec = {
"type": "mechanical_ventilation",
"recommendation_id": "mv_test",
"phase": 1,
}
previous_phase_values = {"sap": 5.0}
current_phase_values = {"sap": 3.0}
property_phase_impact = {"sap": -2.0, "carbon": 0, "heat_demand": 0}
adjustments = []
updated_impact, updated_current, updated_adjustments = (
Recommendations._apply_measure_specific_rules(
rec=rec,
property_phase_impact=property_phase_impact,
previous_phase_values=previous_phase_values,
current_phase_values=current_phase_values,
adjustments=adjustments,
property_instance=property_instance
)
)
# No adjustment expected
assert updated_adjustments == []
# SAP unchanged
assert updated_current["sap"] == 3.0
assert updated_impact["sap"] == -2.0
def test_mechanical_ventilation_exactly_one_no_adjustment(property_instance):
# Test when SAP = 1
rec = {
"type": "mechanical_ventilation",
"recommendation_id": "mv_test",
"phase": 1,
}
previous_phase_values = {"sap": 2.0}
current_phase_values = {"sap": 1.0}
property_phase_impact = {"sap": -1.0, "carbon": 0, "heat_demand": 0}
adjustments = []
updated_impact, updated_current, updated_adjustments = (
Recommendations._apply_measure_specific_rules(
rec=rec,
property_phase_impact=property_phase_impact,
previous_phase_values=previous_phase_values,
current_phase_values=current_phase_values,
adjustments=adjustments,
property_instance=property_instance
)
)
# Exactly 1 → no adjustment
assert updated_adjustments == []
assert updated_current["sap"] == 1.0
assert updated_impact["sap"] == -1.0

24
recommendations/tests/test_roof_recommendations.py
View file

@ -8,6 +8,30 @@ from recommendations.tests.test_data.materials import materials
class TestRoofRecommendations:
def test_null_roof_description(self):
epc_record = EPCRecord()
epc_record.prepared_epc = {
"county": "Cambridgeshire",
}
property_instance = Property(id=0, address="fake", postcode="fake", epc_record=epc_record)
property_instance.age_band = "F"
property_instance.insulation_floor_area = 100
property_instance.roof = {
'original_description': None,
'clean_description': None,
'thermal_transmittance': None,
'thermal_transmittance_unit': None,
'is_pitched': True, 'is_roof_room': False, 'is_loft': False, 'is_flat': False, 'is_thatched': False,
'is_at_rafters': False, 'is_assumed': True, 'has_dwelling_above': False, 'is_valid': True,
'insulation_thickness': 'none', 'roof_thermal_transmittance': None, 'roof_insulation_thickness': None
}
property_instance.already_installed = []
roof_recommender = RoofRecommendations(property_instance=property_instance, materials=materials)
roof_recommender.recommend(phase=0)
assert not roof_recommender.recommendations
def test_loft_insulation_recommendation_no_insulation(self):
epc_record = EPCRecord()
epc_record.prepared_epc = {

37
sfr/principal_pitch/2_export_data.py
View file

@ -28,12 +28,12 @@ from sqlalchemy import func
# PORTFOLIO_ID = 206
# SCENARIOS = [389]
PORTFOLIO_ID = 524
PORTFOLIO_ID = 568
SCENARIOS = [
1009,
1059,
]
scenario_names = {
1009: "EPC C; Most Economic",
1059: "EPC C - 10k budget",
}
@ -230,7 +230,7 @@ for scenario_id in SCENARIOS:
# Get recs for this scenario
recommended_measures_df = recommendations_df[
recommendations_df["scenario_id"] == scenario_id
][["property_id", "measure_type", "estimated_cost", "default"]]
][["property_id", "measure_type", "estimated_cost", "default"]]
recommended_measures_df = recommended_measures_df[
recommended_measures_df["default"]
]
@ -238,7 +238,7 @@ for scenario_id in SCENARIOS:
post_install_sap = recommendations_df[
recommendations_df["scenario_id"] == scenario_id
][["property_id", "default", "sap_points"]]
][["property_id", "default", "sap_points"]]
post_install_sap = post_install_sap[post_install_sap["default"]]
# Sum up the sap points by property id
post_install_sap = (
@ -301,33 +301,6 @@ for scenario_id in SCENARIOS:
)
df["uprn"] = df["uprn"].astype(str)
relevant_plans = plans_df[plans_df["scenario_id"] == scenario_id]
df2 = df.merge(
relevant_plans[["property_id", "post_sap_points", "post_epc_rating"]],
how="left",
on="property_id",
suffixes=("", "_plan"),
)
print(df2["predicted_post_works_epc"].value_counts())
print(df2["post_epc_rating"].value_counts())
z = df2[
(df2["predicted_post_works_epc"] != "D")
& (df2["post_epc_rating"].astype(str) == "Epc.D")
]
df2["predicted_post_works_epc"].value_counts()
df2["post_epc_rating"].astype(str).value_counts()
df2[df2["total_retrofit_cost"] > 0].shape
getting_works = df[df["total_retrofit_cost"] > 0]
getting_works["predicted_post_works_epc"].value_counts()
32565 / getting_works.shape[0]
df[df["predicted_post_works_sap"] == ""]
# Create excel to store to
filename = f"{scenario_names[scenario_id]} - 20250113 final.xlsx"
with pd.ExcelWriter(filename) as writer: