Leave location validation to specific carbon API

cats/CI_api_interface.py

@@ -1,11 +1,15 @@
+import sys
 from collections import namedtuple
 from datetime import datetime
 
 from .forecast import CarbonIntensityPointEstimate
 
 
+class InvalidLocationError(Exception):
+    pass
+
+
 APIInterface = namedtuple('APIInterface', ['get_request_url', 'parse_response_data'])
-# TODO add a validation function to check the validity of the --location argument
 
 def ciuk_request_url(timestamp: datetime, postcode: str):
     # This transformation is specific to the CI-UK API.
@@ -20,6 +24,11 @@ def ciuk_request_url(timestamp: datetime, postcode: str):
     else:
         dt = timestamp.replace(minute=1, second=0, microsecond=0)
 
+    if (len(postcode) > 4):
+        sys.stderr.write(f"Warning: truncating postcode {postcode} to ")
+        postcode = postcode[:-3].strip()
+        sys.stderr.write(f"{postcode}.\n")
+
     return (
         "https://api.carbonintensity.org.uk/regional/intensity/"
         + dt.strftime("%Y-%m-%dT%H:%MZ")
@@ -39,6 +48,20 @@ def ciuk_parse_response_data(response: dict):
     :param response:
     :return:
     """
+    def invalid_code(r: dict):
+        try:
+            return "postcode" in r['error']['message']
+        except KeyError:
+            return False
+
+    # carbonintensity.org.uk API's response behavior is inconsistent
+    # with regards to bad postcodes. Passing a single character in the
+    # URL will give a 400 error code with a useful message.  However
+    # giving a longer string, not matching a valid postcode, lead to
+    # no response at all.
+    if (not response) or invalid_code(response):
+        raise InvalidLocationError
+
     datefmt = "%Y-%m-%dT%H:%MZ"
     return [
         CarbonIntensityPointEstimate(

cats/__init__.py

@@ -4,9 +4,9 @@
 import yaml
 import sys
 
-from .check_clean_arguments import validate_jobinfo, validate_duration, validate_location
+from .check_clean_arguments import validate_jobinfo, validate_duration
 from .optimise_starttime import get_avg_estimates  # noqa: F401
-from .CI_api_interface import API_interfaces
+from .CI_api_interface import API_interfaces, InvalidLocationError
 from .CI_api_query import get_CI_forecast  # noqa: F401
 from .carbonFootprint import greenAlgorithmsCalculator
 
@@ -90,15 +90,15 @@ def main(arguments=None):
 
     ## Location
     if args.location:
-        location = validate_location(args.location, choice_CI_API)
+        location = args.location
         sys.stderr.write(f"Using location provided: {location}\n")
     elif "location" in config.keys():
-        location = validate_location(config["location"], choice_CI_API)
+        location = config["location"]
         sys.stderr.write(f"Using location from config file: {location}\n")
     else:
         r = requests.get("https://ipapi.co/json").json()
         postcode = r["postal"]
-        location = validate_location(postcode, choice_CI_API)
+        location = postcode
         sys.stderr.write(f"WARNING: location not provided. Estimating location from IP address: {location}.\n")
 
     ## Duration
@@ -109,7 +109,15 @@ def main(arguments=None):
     ########################
 
     CI_API_interface = API_interfaces[choice_CI_API]
-    CI_forecast = get_CI_forecast(location, CI_API_interface)
+    try:
+        CI_forecast = get_CI_forecast(location, CI_API_interface)
+    except InvalidLocationError:
+        sys.stderr.write(f"Error: unknown location {location}\n")
+        sys.stderr.write(
+            "Location should be be specified as the outward code,\n"
+            "for example 'SW7' for postcode 'SW7 EAZ'.\n"
+        )
+        sys.exit(1)
 
     #############################
     ## Find optimal start time ##

cats/check_clean_arguments.py

@@ -54,16 +54,3 @@ def validate_duration(duration):
         raise ValueError("--duration needs to be positive (number of minutes)")
 
     return duration_int
-
-def validate_location(location, choice_CI_API):
-    if choice_CI_API == 'carbonintensity.org.uk':
-        # in case the long format of the postcode is provided:
-        loc_cleaned = location.split()[0]
-
-        # check that it's between 2 and 4 characters long
-        # TODO Check that it's a valid postcode for the API/country of interest
-        if (len(loc_cleaned) < 2) or (len(loc_cleaned) > 4):
-            raise ValueError(f"{location} is an invalid UK postcode. Only the first part of the postcode is expected (e.g. M15).")
-    else:
-        loc_cleaned = location
-    return loc_cleaned

tests/test_api_query.py

@@ -1,5 +1,7 @@
+import pytest
+
 import cats
-from cats.CI_api_interface import API_interfaces
+from cats.CI_api_interface import API_interfaces, InvalidLocationError
 from cats.forecast import CarbonIntensityPointEstimate
 
 def test_api_call():
@@ -13,3 +15,12 @@ def test_api_call():
     assert isinstance(response, list)
     for item in response:
         assert isinstance(item, CarbonIntensityPointEstimate)
+
+def test_bad_postcode():
+    api_interface = API_interfaces["carbonintensity.org.uk"]
+
+    with pytest.raises(InvalidLocationError):
+        response = cats.CI_api_query.get_CI_forecast('OX48', api_interface)
+
+    with pytest.raises(InvalidLocationError):
+        response = cats.CI_api_query.get_CI_forecast('A', api_interface)