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

rewrote classifying new declies without pandas

Browse source
This commit is contained in:
Khalim Conn-Kowlessar 2023-07-20 15:48:26 +01:00
parent 7b8c46242c
commit 5a5c0d5f18
13 changed files with 199 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.idea/Model.iml generated
View file

@ -6,7 +6,7 @@
<sourceFolder url="file://$MODULE_DIR$/model_data" isTestSource="false" />
<sourceFolder url="file://$MODULE_DIR$/open_uprn" isTestSource="false" />
</content>
<orderEntry type="jdk" jdkName="Python 3.10 (model_data)" jdkType="Python SDK" />
<orderEntry type="jdk" jdkName="Python 3.10 (hestia-data)" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
</module>

2
.idea/misc.xml generated
View file

@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.10 (model_data)" project-jdk-type="Python SDK" />
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.10 (hestia-data)" project-jdk-type="Python SDK" />
<component name="PythonCompatibilityInspectionAdvertiser">
<option name="version" value="3" />
</component>

28
backend/app/plan/router.py
View file

@ -6,6 +6,10 @@ from backend.app.config import get_settings
from model_data.Property import Property
from epc_api.client import EpcClient
from utils.logger import setup_logger
from recommendations.FloorRecommendations import FloorRecommendations
from recommendations.WallRecommendations import WallRecommendations
from utils.uvalue_estimates import classify_decile_newvalues
# TODO: This is placeholder until data is stored in DB
from backend.app.plan.temp_cleaned_data import cleaned
from backend.app.plan.uvalue_estimates_walls import uvalue_estimates_walls
@ -51,6 +55,17 @@ in_conservation_area_data = [
{'uprn': 200003489276, 'is_in_conservation_area': 'in_conservation_area'}
]
# TODO: db
floors_decile_data = {
'decile_labels': ['Decile 1', 'Decile 2', 'Decile 3', 'Decile 4', 'Decile 5', 'Decile 6', 'Decile 7', 'Decile 8',
'Decile 9', 'Decile 10'], 'decile_boundaries': [6., 50., 56., 69., 77.6, 87., 98., 112.,
127., 150., 2279.]}
walls_decile_data = {
'decile_labels': ['Decile 1', 'Decile 2', 'Decile 3', 'Decile 4', 'Decile 5', 'Decile 6', 'Decile 7', 'Decile 8',
'Decile 9', 'Decile 10'], 'decile_boundaries': [6., 49., 51., 55., 64., 71., 76., 83., 96.,
120., 2279.]}
@router.post("/trigger")
async def trigger_plan(body: PlanTriggerRequest):
@ -85,7 +100,20 @@ async def trigger_plan(body: PlanTriggerRequest):
)
p.set_is_in_conservation_area(in_conservation_area)
logger.info("Getting components and properties recommendations")
for p in input_properties:
# For each property, classiy floor area decide
total_floor_area_group_decile = classify_decile_newvalues(
decile_boundaries=floors_decile_data["decile_boundaries"],
decile_labels=floors_decile_data["decile_labels"],
new_values=[float(p.data["total-floor-area"])],
)[0]
print("hey")
print(total_floor_area_group_decile)
for p in input_properties:
p.get_components(cleaned)
# floor_recommendations = FloorRecommendations(property_instance=p, uvalue_estimates=uvalue_estimates_floors)
return {"message": "Plan complete"}

2
model_data/app.py
View file

@ -365,3 +365,5 @@ def app():
uvalue_estimates = UvalueEstimations(data=data)
uvalue_estimates.get_estimates(cleaner=cleaner)
# TODO: Store these to a db
uvalue_estimates.floors_decile_data

48
model_data/tests/test_utils.py
View file

@ -1,53 +1,7 @@
import logging
from io import StringIO
from unittest.mock import patch
from model_data.utils import is_percentage_or_number, correct_spelling
from utils.logger import setup_logger
class TestLogger:
def test_setup_logger_default(self):
log_stream = StringIO()
handler = logging.StreamHandler(log_stream)
logger = setup_logger()
logger.addHandler(handler)
# log something
logger.info("Hello World!")
log_stream.seek(0)
# assert that log was written
assert log_stream.read() == "Hello World!\n"
# remove the handler after use
logger.removeHandler(handler)
@patch('logging.FileHandler')
def test_setup_logger_file(self, mock_file_handler):
# setup the logger
logger = setup_logger(log_file='test.log', overwrite_handler=True)
# assert FileHandler was called correctly
mock_file_handler.assert_called_once_with('test.log')
# clean up after use
for handler in logger.handlers[:]:
handler.close()
logger.removeHandler(handler)
def test_setup_logger_loglevel(self):
log_stream = StringIO()
handler = logging.StreamHandler(log_stream)
logger = setup_logger(level=logging.DEBUG)
logger.addHandler(handler)
# log something
logger.debug("Hello World!")
log_stream.seek(0)
# assert that log was written
assert log_stream.read() == "Hello World!\n"
# remove the handler after use
logger.removeHandler(handler)
class TestUtils:
def test_is_percentage_or_number(self):
assert is_percentage_or_number("88")

17
model_data/recommendations/FloorRecommendations.py → recommendations/FloorRecommendations.py
View file

@ -1,9 +1,8 @@
import math
from model_data.BaseUtility import BaseUtility
from model_data.Property import Property
from model_data.analysis.UvalueEstimations import UvalueEstimations
from model_data.rdsap_tables import default_wall_thickness, age_band_data
from model_data.recommendations.recommendation_utils import (
from recommendations.recommendation_utils import (
r_value_per_mm_to_u_value, calculate_u_value_uplift, is_diminishing_returns, update_lowest_selected_u_value,
get_recommended_part
)
@ -92,7 +91,7 @@ class FloorRecommendations(BaseUtility):
"4th": 4
}
def __init__(self, property_instance: Property, uvalue_estimates: UvalueEstimations):
def __init__(self, property_instance: Property, uvalue_estimates):
self.property = property_instance
self.uvalue_estimates = uvalue_estimates
# For audit purposes, when estimating u values we'll store it
@ -253,7 +252,7 @@ class FloorRecommendations(BaseUtility):
# Given the U-value, we recommend solid floor insulation options which are usually solid foam
self.recommend_floor_insulation(u_value=u_value, parts=solid_floor_insulation_parts)
def _get_floors_uvalue_estimate(self):
def _get_floors_uvalue_estimate(self, total_floor_area_group_decile):
"""
Wrapper function which contains the methodology to extract a property's walls u-value estimate
@ -261,11 +260,11 @@ class FloorRecommendations(BaseUtility):
:return:
"""
total_floor_area_group_decile = self.uvalue_estimates.classify_decile_newvalues(
decile_boundaries=self.uvalue_estimates.floors_decile_data["decile_boundaries"],
decile_labels=self.uvalue_estimates.floors_decile_data["decile_labels"],
new_values=[float(self.property.data["total-floor-area"])],
)[0]
# total_floor_area_group_decile = UvalueEstimations.classify_decile_newvalues(
# decile_boundaries=self.uvalue_estimates.floors_decile_data["decile_boundaries"],
# decile_labels=self.uvalue_estimates.floors_decile_data["decile_labels"],
# new_values=[float(self.property.data["total-floor-area"])],
# )[0]
u_value_estimate = self.uvalue_estimates.floors[
(self.uvalue_estimates.floors["local-authority"] == self.property.data["local-authority"]) &

5
model_data/recommendations/WallRecommendations.py → recommendations/WallRecommendations.py
View file

@ -2,9 +2,8 @@ import itertools
import math
from model_data.Property import Property
from model_data.analysis.UvalueEstimations import UvalueEstimations
from model_data.BaseUtility import BaseUtility
from model_data.recommendations.recommendation_utils import (
from recommendations.recommendation_utils import (
r_value_per_mm_to_u_value, calculate_u_value_uplift, is_diminishing_returns, update_lowest_selected_u_value,
get_recommended_part
)
@ -217,7 +216,7 @@ class WallRecommendations(BaseUtility):
"solid_brick": 2,
}
def __init__(self, property_instance: Property, uvalue_estimates: UvalueEstimations):
def __init__(self, property_instance: Property, uvalue_estimates):
self.property = property_instance
self.uvalue_estimates = uvalue_estimates
# For audit purposes, when estimating u values we'll store it

0
model_data/recommendations/recommendation_utils.py → recommendations/recommendation_utils.py
View file

0
model_data/tests/test_floor_recommendations.py → recommendations/tests/test_floor_recommendations.py
View file

0
model_data/tests/test_wall_recommendations.py → recommendations/tests/test_wall_recommendations.py
View file

49
utils/tests/test_logger.py Normal file
View file

@ -0,0 +1,49 @@
import logging
from io import StringIO
from unittest.mock import patch
from utils.logger import setup_logger
class TestLogger:
def test_setup_logger_default(self):
log_stream = StringIO()
handler = logging.StreamHandler(log_stream)
logger = setup_logger()
logger.addHandler(handler)
# log something
logger.info("Hello World!")
log_stream.seek(0)
# assert that log was written
assert log_stream.read() == "Hello World!\n"
# remove the handler after use
logger.removeHandler(handler)
@patch('logging.FileHandler')
def test_setup_logger_file(self, mock_file_handler):
# setup the logger
logger = setup_logger(log_file='test.log', overwrite_handler=True)
# assert FileHandler was called correctly
mock_file_handler.assert_called_once_with('test.log')
# clean up after use
for handler in logger.handlers[:]:
handler.close()
logger.removeHandler(handler)
def test_setup_logger_loglevel(self):
log_stream = StringIO()
handler = logging.StreamHandler(log_stream)
logger = setup_logger(level=logging.DEBUG)
logger.addHandler(handler)
# log something
logger.debug("Hello World!")
log_stream.seek(0)
# assert that log was written
assert log_stream.read() == "Hello World!\n"
# remove the handler after use
logger.removeHandler(handler)

80
utils/tests/test_uvalue_estimates.py Normal file
View file

@ -0,0 +1,80 @@
from utils.uvalue_estimates import classify_decile_newvalues
def test_classify_decile_newvalues_edge_cases():
decile_labels = [f"Decile {i + 1}" for i in range(10)]
decile_boundaries = list(range(11))
# Test with values at the exact boundaries
assert classify_decile_newvalues(decile_boundaries, decile_labels, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) == ['Decile 1',
'Decile 2',
'Decile 3',
'Decile 4',
'Decile 5',
'Decile 6',
'Decile 7',
'Decile 8',
'Decile 9',
'Decile 10']
# Test with values at the exact boundaries, but in reverse order
assert classify_decile_newvalues(decile_boundaries, decile_labels, [10, 9, 8, 7, 6, 5, 4, 3, 2, 1]) == ['Decile 10',
'Decile 9',
'Decile 8',
'Decile 7',
'Decile 6',
'Decile 5',
'Decile 4',
'Decile 3',
'Decile 2',
'Decile 1']
# Test with values just below the boundaries
assert classify_decile_newvalues(decile_boundaries, decile_labels, [x - 0.5 for x in range(2, 12)]) == ['Decile 1',
'Decile 2',
'Decile 3',
'Decile 4',
'Decile 5',
'Decile 6',
'Decile 7',
'Decile 8',
'Decile 9',
'Decile 10']
# Test with values just above the boundaries
assert classify_decile_newvalues(decile_boundaries, decile_labels, [x + 0.5 for x in range(1, 11)]) == ['Decile 2',
'Decile 3',
'Decile 4',
'Decile 5',
'Decile 6',
'Decile 7',
'Decile 8',
'Decile 9',
'Decile 10',
None]
# Test with empty list
assert classify_decile_newvalues(decile_boundaries, decile_labels, []) == []
# Test with a single value
assert classify_decile_newvalues(decile_boundaries, decile_labels, [5.5]) == ['Decile 6']
# Test with all values the same
assert classify_decile_newvalues(decile_boundaries, decile_labels, [5, 5, 5, 5, 5]) == ['Decile 5', 'Decile 5',
'Decile 5', 'Decile 5',
'Decile 5']
# Test with values out of order
assert classify_decile_newvalues(decile_boundaries, decile_labels, [10, 5, 1, 7, 3]) == ['Decile 10', 'Decile 5',
'Decile 1', 'Decile 7',
'Decile 3']
# Test with negative decile boundaries
decile_boundaries = list(range(-10, 1))
assert classify_decile_newvalues(decile_boundaries, decile_labels, [-9, -5, -1]) == ['Decile 2', 'Decile 6',
'Decile 10']
# Test with floating point decile boundaries
decile_boundaries = [x / 10 for x in range(11)]
assert classify_decile_newvalues(decile_boundaries, decile_labels, [0.35, 0.55, 0.75]) == ['Decile 4', 'Decile 6',
'Decile 8']

27
utils/uvalue_estimates.py Normal file
View file

@ -0,0 +1,27 @@
from typing import List
from bisect import bisect_left
def classify_decile_newvalues(
decile_boundaries: List[float], decile_labels: List[str], new_values: List[float]
) -> List[str]:
"""
This is a complementary function to UvalueEstimations.classify_decile_newvalues, which does not use pandas
so that we can use this function inside of our lamdbas, without having to import pandas.
:param decile_boundaries:
:param decile_labels:
:param new_values:
:return:
"""
classifications = []
for value in new_values:
if value < decile_boundaries[0] or value > decile_boundaries[-1]:
classifications.append(None)
else:
i = bisect_left(decile_boundaries, value)
if i:
i -= 1
classifications.append(decile_labels[i])
return classifications