Basic functionality for #5: Implement Text forecast

datapoint/Manager.py

@@ -15,6 +15,8 @@
 from .Day import Day
 from .Timestep import Timestep
 from .Element import Element
+from .regions.RegionManager import RegionManager
+
 
 if (sys.version_info > (3, 0)):
     long = int
@@ -84,6 +86,8 @@ def __init__(self, api_key=""):
         self.sites_last_request = None
         self.sites_update_time = 3600
 
+        self.regions = RegionManager(self.api_key)
+
     def __call_api(self, path, params=None):
         """
         Call the datapoint api using the requests module

datapoint/regions/RegionManager.py

@@ -0,0 +1,66 @@
+import json
+from time import time
+
+import requests
+
+from datapoint.Site import Site
+from datapoint.regions.region_names import REGION_NAMES
+
+REGIONS_BASE_URL = 'http://datapoint.metoffice.gov.uk/public/data/txt/wxfcs/regionalforecast/json'
+
+class RegionManager(object):
+    '''
+    Datapoint Manager for national and regional text forecasts 
+    '''
+    def __init__(self, api_key, base_url=None):
+        self.api_key = api_key
+        self.all_regions_path = '/sitelist'
+        if not base_url:
+            self.base_url = REGIONS_BASE_URL
+
+        # The list of regions changes infrequently so limit to requesting it
+        # every hour.
+        self.regions_last_update = 0
+        self.regions_last_request = None
+        self.regions_update_time = 3600
+
+    def call_api(self, path, **kwargs):
+        '''
+        Call datapoint api
+        '''
+        if 'key' not in kwargs:
+            kwargs['key'] = self.api_key
+        req = requests.get('{}{}'.format(self.base_url, path), params=kwargs)
+
+        if req.status_code != requests.codes.ok:
+            req.raise_for_status()
+
+        return req.json()
+
+    def get_all_regions(self):
+        '''
+        Request a list of regions from Datapoint. Returns each Region 
+        as a Site object. Regions rarely change, so we cache the response 
+        for one hour to minimise requests to API. 
+        '''
+        if (time() - self.regions_last_update) < self.regions_update_time:
+            return self.regions_last_request
+
+        response = self.call_api(self.all_regions_path)
+        regions = []
+        for location in response['Locations']['Location']:
+            region = Site()
+            region.id = location['@id']
+            region.region = location['@name']
+            region.name = REGION_NAMES[location['@name']]
+            regions.append(region)
+
+        self.regions_last_update = time()
+        self.regions_last_request = regions
+        return regions
+
+    def get_raw_forecast(self, region_id):
+        '''
+        Request unformatted forecast for a specific region_id.
+        '''
+        return self.call_api('/{}'.format(region_id)) 

datapoint/regions/region_names.py

@@ -0,0 +1,19 @@
+REGION_NAMES = {
+    'os': 'Orkney & Shetland',
+    'he': 'Highland & Eilean Siar',
+    'gr': 'Grampian',
+    'ta': 'Tayside',
+    'st': 'Strathclyde',
+    'dg': 'Dumfries, Galloway, Lothian',
+    'ni': 'Northern Ireland',
+    'yh': 'Yorkshire & the Humber',
+    'ne': 'Northeast England',
+    'em': 'East Midlands',
+    'ee': 'East of England',
+    'se': 'London & Southeast England',
+    'nw': 'Northwest England',
+    'wm': 'West Midlands',
+    'sw': 'Southwest England',
+    'wl': 'Wales',
+    'uk': 'UK',
+}

examples/text_forecast/text_forecast.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python
+"""
+This example will print out the 30 day text forecast for a region of the UK.
+"""
+
+import datapoint
+
+# Create datapoint connection
+conn = datapoint.Manager(api_key="aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee")
+
+# Get all regions and print out their details
+regions = conn.regions.get_all_regions()
+for region in regions:
+    print (region.name, region.id, region.region)
+
+# Get all forecasts for a specific region
+my_region = regions[0]
+forecast = conn.regions.get_raw_forecast(my_region.id)['RegionalFcst']
+
+# Print the forecast details
+print 'Forecast for {} (issued at {}):'.format(my_region.name, forecast['issuedAt'])
+
+sections = forecast['FcstPeriods']['Period']
+for section in forecast['FcstPeriods']['Period']:
+    paragraph = []
+    content = section['Paragraph']
+
+    # Some paragraphs have multiple sections
+    if isinstance(content, dict):
+        paragraph.append(content)
+    else:
+        paragraph = content
+
+    for line in paragraph:
+            print '{}\n{}\n'.format(line['title'], line['$'])

setup.py

@@ -60,7 +60,7 @@
       maintainer_email='jacob@jacobtomlinson.co.uk',
       url='https://github.com/jacobtomlinson/datapoint-python',
       license='GPLv3',
-      packages=['datapoint'],
+      packages=['datapoint', 'datapoint.regions'],
       classifiers=[
           'Development Status :: 3 - Alpha',
           'Programming Language :: Python :: 2.6',

tests/integration/regions_test.py

@@ -0,0 +1,41 @@
+import os
+
+from nose.tools import *
+from requests import HTTPError
+
+import datapoint
+
+
+class TestRegions(object):
+    def __init__(self):
+        self.manager = datapoint.Manager(api_key=os.environ['API_KEY'])
+        self.regions = self.manager.regions
+
+    def test_key(self):
+        assert self.regions.api_key == os.environ['API_KEY']
+
+    def test_call_api(self):
+        assert (
+            u'RegionalFcst' in self.regions.call_api('/500'))
+        assert_raises(
+            HTTPError, self.regions.call_api, '/fake_path')
+        assert_raises(
+            HTTPError, self.regions.call_api, '/500', key='fake_key')
+
+    def test_get_all_regions(self):
+        all_regions = self.regions.get_all_regions()
+        sample_region = filter(lambda x: x.id == '515', all_regions)[0]
+        assert (sample_region.name == 'UK')
+        assert (sample_region.region == 'uk')
+
+    def test_get_raw_forecast(self):
+        sample_region = self.regions.get_all_regions()[0]
+        response = self.regions.get_raw_forecast(
+            sample_region.id)['RegionalFcst']
+        assert (response['regionId'] == sample_region.region)
+
+        # Based on what Datapoint serves at time of writing...
+        forecast_periods = response['FcstPeriods']['Period']
+        forecast_ids = [period['id'] for period in forecast_periods]
+        expected_ids = ['day1to2', 'day3to5', 'day6to15', 'day16to30']
+        assert (forecast_ids == expected_ids)