re add

OrdnanceSurvey.py

@@ -0,0 +1,139 @@
+from functools import lru_cache
+import urllib.parse
+import requests
+from utils.logger import setup_logger
+
+logger = setup_logger()
+
+
+class OrdnanceSuveyClient:
+
+    def __init__(self, address, postcode, api_key):
+        """
+        This class is tasked with interaction with the ordnance survey API.
+        :param address: The address for the property to search for
+        :param postcode: The postcode for the property to search for
+        """
+
+        self.address = address
+        self.postcode = postcode
+        self.full_address = ", ".join([self.address, self.postcode])
+        self.api_key = api_key
+
+        self.results = None
+
+        self.most_relevant_result = None
+        self.property_type = None
+        self.built_form = None
+        # This will be postcode and address, as returned by the ordnance survey
+        self.address_os = None
+        self.postcode_os = None
+
+    def set_places_address(self):
+        """
+        Given a response from the places api, this function will set the address and postcode of the property
+        """
+
+        if self.most_relevant_result is None:
+            raise ValueError("No results found - run get_places_api first")
+
+        self.address_os = self.most_relevant_result["ADDRESS"]
+
+        if "POSTCODE" in self.most_relevant_result:
+            self.postcode_os = self.most_relevant_result["POSTCODE"]
+        else:
+            self.postcode_os = self.most_relevant_result["POSTCODE_LOCATOR"]
+        # We strip out the postcode from the address as this is already stored separately
+        self.address_os = self.address_os.replace(self.postcode_os, "").strip()
+        # Remove trailing comma
+        self.address_os = self.address_os.rstrip(",").strip()
+        # Convert to title case
+        self.address_os = self.address_os.title()
+        # Make sure postcode is upper case
+        self.postcode_os = self.postcode_os.upper()
+
+    @lru_cache(maxsize=128)
+    def get_places_api(self, filter_by_postcode=False):
+        """
+        This method is tasked with getting the places api from the Ordnance Survey.
+        """
+
+        if not self.api_key:
+            raise ValueError("Ordnance Survey API key not specified")
+
+        encoded_address_query = urllib.parse.quote(self.full_address)
+
+        url = (
+            f"https://api.os.uk/search/places/v1/find?query={encoded_address_query}&dataset=DPA,LPI&matchprecision=10"
+            f"&key={self.api_key}"
+        )
+
+        response = requests.get(url)
+        if response.status_code == 200:
+            data = response.json()
+            res = data["results"]
+
+            if filter_by_postcode:
+                results = []
+                for r in res:
+                    if "DPA" in r:
+                        if r["DPA"]["POSTCODE"] == self.postcode:
+                            results.append(r)
+                    elif "LPI" in r:
+                        if r["LPI"]["POSTCODE_LOCATOR"] == self.postcode:
+                            results.append(r)
+                    else:
+                        raise ValueError("Could not find postcode in either DPA or LPI")
+            else:
+                results = res
+
+            self.results = results
+
+            # Extract some details about the best match
+            self.most_relevant_result = (
+                self.results[0]["DPA"]
+                if "DPA" in self.results[0]
+                else self.results[0]["LPI"]
+            )
+
+            self.parse_classification_code(
+                self.most_relevant_result["CLASSIFICATION_CODE"]
+            )
+            self.set_places_address()
+
+        else:
+            logger.info(
+                "Could not find any results for the provided address and postcode"
+            )
+
+        return {"status": response.status_code}
+
+    def parse_classification_code(self, classification_code: str):
+        """
+        This function will convert the classification code, returned by the OS places api, to a property type that is
+        compatible with the EPC database.
+
+        The various classifications cane be found here:
+        https://osdatahub.os.uk/docs/places/technicalSpecification
+
+        Under LPI Output, CLASSIFICATION_CODE is described, and a link is provided to the full table of classifications
+        For these purposes, we do not need the full classification as this includes non-residential properties. We only
+        parse the ones of interest to us
+        :return:
+        """
+
+        value_map = {
+            # In the OS api, "RD" is a "Dwelling" however this is not valid property type in the EPC database
+            "RD": {},
+            "RD02": {"property_type": "House", "built_form": "Detached"},
+            "RD03": {"property_type": "House", "built_form": "Semi-Detached"},
+            "RD04": {"property_type": "House", "built_form": "Mid-Terrace"},
+            "RD06": {"property_type": "Flat"},
+        }
+        # Other classifications can be found in here:
+        # https://osdatahub.os.uk/docs/places/technicalSpecification in the CLASSIFICATION_CODE description.
+        # A lookup table csv can be downloaded which contains all of the codes
+
+        mapped = value_map.get(classification_code, {})
+        self.property_type = mapped.get("property_type", "")
+        self.built_form = mapped.get("built_form", "")