diff --git a/datapoint/Manager.py b/datapoint/Manager.py index 1ce9aa7..8c203ec 100644 --- a/datapoint/Manager.py +++ b/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 diff --git a/datapoint/regions/RegionManager.py b/datapoint/regions/RegionManager.py new file mode 100644 index 0000000..446fff8 --- /dev/null +++ b/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)) diff --git a/datapoint/regions/__init__.py b/datapoint/regions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/datapoint/regions/region_names.py b/datapoint/regions/region_names.py new file mode 100644 index 0000000..8dd5d45 --- /dev/null +++ b/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', +} \ No newline at end of file diff --git a/examples/text_forecast/text_forecast.py b/examples/text_forecast/text_forecast.py new file mode 100644 index 0000000..61ae08e --- /dev/null +++ b/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['$']) diff --git a/setup.py b/setup.py index 7a99af6..81c8dbf 100644 --- a/setup.py +++ b/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', diff --git a/tests/integration/regions_test.py b/tests/integration/regions_test.py new file mode 100644 index 0000000..246c15a --- /dev/null +++ b/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) \ No newline at end of file