Implemented conservation area logic

model_data/ConservationAreaClient.py

@@ -0,0 +1,105 @@
+import pandas as pd
+import geopandas as gpd
+from shapely.geometry import Point
+from model_data.utils import setup_logger
+
+logger = setup_logger()
+
+
+class ConservationAreaClient:
+    """
+    Class to interact and manupulate convervation area data. The historic england data
+    can be found at the following location:
+    https://opendata-historicengland.hub.arcgis.com/datasets/historicengland::conservation-areas/about
+
+    We also use a separate government conservation area dataset which can be found here:
+    https://www.planning.data.gov.uk/dataset/conservation-area
+    """
+
+    SOURCES = ["historic_england"]
+    IN_CONSERVATION_AREA = "in_conservation_area"
+    NOT_IN_CONSERVATION_AREA = "not_in_conservation_area"
+    UNKNOWN = "unknown"
+
+    def __init__(self, historic_england_path, gov_path):
+        self.historic_england_path = historic_england_path
+        self.gov_path = gov_path
+
+        self.historic_england_data = None
+        self.gov_data = None
+
+    def read(self):
+        """
+        Read the data
+        """
+        logger.info("Reading in historic england conservation area shapefile")
+        self.historic_england_data = gpd.read_file(self.historic_england_path)
+
+        logger.info("Reading in Govenment conservation area geojson")
+        self.gov_data = gpd.read_file(self.gov_path)
+        self.gov_data = self.gov_data.drop(columns=["dataset"])
+
+    def is_in_conservation_area_historic_england(self, x_bng: float, y_bng: float) -> str:
+        """
+        Check if a property is in a conservation area
+        :param x_bng: x coordinate in british national grid coordinates
+        :param y_bng: y coordinate in british national grid coordinates
+        """
+
+        point = Point(x_bng, y_bng)
+
+        within_areas = self.historic_england_data.contains(point)
+
+        if within_areas.any():
+            names = self.historic_england_data.loc[within_areas, "NAME"]
+            # We want to deduce if we actually have data on this area
+
+            if all(names.values == "No data available for publication by HE"):
+                return self.UNKNOWN
+
+            return self.IN_CONSERVATION_AREA
+
+        return self.NOT_IN_CONSERVATION_AREA
+
+    def is_in_conservation_area_historic_gov(self, longitude: float, latitude: float) -> str:
+        """
+        Check if a property is in a conservation area
+        :param longitude: longtitude coordinate
+        :param latitude: latitude coordinate
+        """
+
+        point = Point(longitude, latitude)
+
+        return self.gov_data.contains(point).any()
+
+    def calculate_distance_to_nearest_conservation_area(self, x: float, y: float, source: str) -> float:
+        if source == "historic_england":
+            return self._distance_to_nearest_conservation_area_historic_england(x, y, self.historic_england_data)
+
+    @staticmethod
+    def _distance_to_nearest_conservation_area_historic_england(
+        x: float, y: float, conservation_areas: gpd.GeoDataFrame
+    ) -> float:
+        """
+        Calculate the distance from a given point to the nearest conservation area.
+
+        :param x: The x-coordinate of the point.
+        :param y: The y-coordinate of the point.
+        :param conservation_areas: A GeoDataFrame containing the conservation areas polygons.
+
+        :return: The distance in the same units as the coordinate system of the conservation areas.
+
+        :raises FileNotFoundError: If the conservation areas GeoDataFrame is not found.
+        :raises IndexError: If no nearest conservation area is found.
+        """
+
+        # Convert the point coordinates to a Shapely Point object
+        point_geom = Point(x, y)
+
+        # Calculate the distance between the point and the conservation areas
+        distances = conservation_areas.geometry.distance(point_geom)
+
+        # Find the minimum distance. Since the data uses british national grid, the units are meters.
+        distance_meters = distances.min()
+
+        return distance_meters

model_data/Property.py

@@ -3,6 +3,7 @@ from model_data.config import EPC_AUTH_TOKEN
 from model_data.OpenUprnClient import OpenUprnClient
 from model_data.EpcClean import EpcClean
 from model_data.BaseUtility import BaseUtility
+from model_data.ConservationAreaClient import ConservationAreaClient
 
 
 class Property(BaseUtility):
@@ -33,6 +34,7 @@ class Property(BaseUtility):
         self.address1 = address1
         self.data = data
         self.full_sap_epc = None
+        self.in_conservation_area = None
 
         if epc_client:
             self.epc_client = epc_client
@@ -110,3 +112,26 @@ class Property(BaseUtility):
             if len(attributes) != 1:
                 raise ValueError("Either No attributes or multiple found for %s" % description)
             setattr(self, self.ATTRIBUTE_MAP[description], attributes[0])
+
+    def set_is_in_conservation_area(self, conservation_area_client: ConservationAreaClient):
+
+        if not self.coordinates:
+            raise ValueError("Coordinates have not been set, run get_coordinates() first")
+
+        is_in_conservation_area = conservation_area_client.is_in_conservation_area_historic_england(
+            x_bng=self.coordinates["x_coordinate"],
+            y_bng=self.coordinates["y_coordinate"]
+        )
+
+        self.in_conservation_area = is_in_conservation_area
+        if is_in_conservation_area == "unknown":
+            # We double check the secondary data source
+            backup = conservation_area_client.is_in_conservation_area_historic_gov(
+                longitude=self.coordinates["longitude"],
+                latitude=self.coordinates["latitude"]
+            )
+
+            if backup:
+                self.in_conservation_area = ConservationAreaClient.IN_CONSERVATION_AREA
+            else:
+                self.in_conservation_area = ConservationAreaClient.UNKNOWN

model_data/app.py

@@ -2,6 +2,7 @@ from tqdm import tqdm
 import os
 from model_data.BoreholeClient import BoreholeClient
 from model_data.LandRegistryClient import LandRegistryClient
+from model_data.ConservationAreaClient import ConservationAreaClient
 
 from model_data.temp_inputs import input_data
 from model_data.Property import Property
@@ -46,12 +47,25 @@ def handler():
     )
     open_uprn_client.read()
 
-    # What's going on here?
     # We're using Ordinance Survey Open Uprn data
     # to find the coordinates of each address, which we will then be able to use at a later stage
     for p in input_properties:
         p.get_coordinates(open_uprn_client)
 
+    conservation_area_client = ConservationAreaClient(
+        historic_england_path=os.path.abspath(
+            os.path.dirname(__file__)
+        ) + "/model_data/local_data/Historic_Eng_Conservation_Areas/Conservation_Areas.shp",
+        gov_path=os.path.abspath(
+            os.path.dirname(__file__)
+        ) + "/model_data/local_data/gov-conservation-area.geojson"
+    )
+    conservation_area_client.read()
+
+    # Check if the property is in a conversation area
+    for p in input_properties:
+        p.set_is_in_conservation_area(conservation_area_client)
+
     local_authorities = {p.data['local-authority'] for p in input_properties}
 
     data = []
@@ -104,10 +118,10 @@ def handler():
         [{"address1": p.address1, **p.walls} for p in input_properties]
     )
 
-    input_properties[1].data["address1"]
-    input_properties[1].data["postcode"]
-    walls_df["address1"].values[1]
-    walls_df["original_description"].values[1]
+    input_properties[6].data["address1"]
+    input_properties[6].data["postcode"]
+    walls_df["address1"].values[6]
+    walls_df["original_description"].values[6]
     # Walls
     # Property 0
     # '28 Distillery Wharf', 'Average thermal transmittance 0.16 W/m-¦K'
@@ -122,6 +136,29 @@ def handler():
     # Since the wall is solid brick (therefore no cavity), we can recommend the following:
     # External wall insulation
     # Internal wall insulation
+    # Property 2
+    # '49, Elderfield Road', Solid brick, as built, no insulation (assumed)
+    # Same as property 1
+    # Property 3
+    # 26, Stanhope Road', 'Average thermal transmittance 0.14 W/m-¦K'
+    # Same as property 0
+    # Property 4
+    # 'Flat 3 Frederick Building' 'Solid brick, as built, no insulation (assumed)'
+    # Same as property 1
+    # 'Flat 4 Frederick Building' 'Solid brick, as built, no insulation (assumed)'
+    # Same as property 1
+    # 'Flat 28, 22 Adelina Grove' 'Solid brick, as built, insulated (assumed)'
 
     from model_data.recommendations.WallRecommendations import WallRecommendations
-    self = WallRecommendations(property_instance=input_properties[1])
+    self = WallRecommendations(property_instance=input_properties[6])
+
+    # We need to deduce a U-value for "Good" energy effieciency
+
+    df = pd.DataFrame(data)
+    df = df[df["walls-description"].str.contains("Average thermal transmittance")]
+
+    mainheating = pd.DataFrame(
+        [{"address1": p.address1, "postcode": p.postcode, **p.main_heating} for p in input_properties])
+    hotwater = pd.DataFrame([{"address1": p.address1, **p.hotwater} for p in input_properties])
+
+    mainheating[["address1", "postcode"]]

model_data/recommendations/WallRecommendations.py

@@ -234,6 +234,7 @@ class WallRecommendations:
         is_cavity_wall = self.property.walls["is_cavity_wall"]
         is_solid_brick = self.property.walls["is_solid_brick"]
         insulation_thickness = self.property.walls["insulation_thickness"]
+        wall_energy_efficiency = self.property.data["walls-energy-eff"]
 
         if u_value:
             if self.property.walls["thermal_transmittance_unit"] != self.U_VALUE_UNIT:
@@ -259,9 +260,6 @@ class WallRecommendations:
 
         if is_solid_brick and insulation_thickness == "none":
 
-            # TODO: what if we recommend both internal and external wall insulation? Individually, they might not
-            #       get the wall to the required u-value, but together they might. We need to handle this case
-
             # This is an estimated figure based on industry standards
             u_value = self.DEFAULT_U_VALUES["solid_brick"]
 
@@ -319,6 +317,11 @@ class WallRecommendations:
                             ]
                             self.recommendations.append(recommendation)
 
+        if is_solid_brick and insulation_thickness == "average" and wall_energy_efficiency in ["Good", "Average"]:
+
+            if self.in_converation_area:
+                blah
+
         raise NotImplementedError("Not implemented yet")
 
     @staticmethod

model_data/requirements.txt

@@ -11,4 +11,5 @@ fuzzywuzzy
 python-Levenshtein
 dbfread
 pyproj
-pint
+pint
+geopandas