Cached 7-day Hourly Forecast

forecast.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
 
from __future__ import unicode_literals
 
from datetime import datetime, timedelta
import requests
 
 
class ForecastException(Exception):
    """There was an ambiguous exception that occurred."""
 
 
class ForecastNotLoadedError(ForecastException):
    """The forecast wasn't loaded before being requested."""
 
 
class ForecastOutOfRangeError(ForecastException):
    """There's no forecast available for that time."""
 
 
class Forecast(object):
    FORECASTIO_ENDPOINT = 'https://api.forecast.io/forecast'
    FORECASTIO_ARGS = 'units=si&exclude=currently,minutely,daily&extend=hourly'
 
    def __init__(self, api_key=""):
        self._api_key = api_key
        self._cached_weather_events = {}  # a cache using datetimes as keys
 
    def load(self, lat, lon):
        """Load a 7 day Forecast for a given location."""
        # empty the cache
        self._cached_weather_events = {}
 
        # load in new events
        r = requests.get("%s/%s/%s,%s?%s" % (self.FORECASTIO_ENDPOINT,
                                             self._api_key,
                                             lat, lon,
                                             self.FORECASTIO_ARGS))
        # TODO: request error handling
 
        data = r.json()
        hourly = data["hourly"]
        for hour in hourly["data"]:
            weather = Weather(hour)
            self._cached_weather_events[weather.time] = Weather(hour)
 
    def forecast(self, datetime):
        """Fetch the forecast for a given datetime."""
        nearest_hour = self._nearest_hour(datetime)
 
        return self._cached_weather_events[nearest_hour]
 
    def _nearest_hour(self, dt):
        """Determine the nearest full hour."""
        cutoff = datetime(dt.year, dt.month, dt.day, 23, 30)
        if dt >= cutoff:
            """ We'll use tomorrow's 00Z forecast. """
            return datetime(dt.year, dt.month, dt.day, 0, 0) + timedelta(1)
        if dt.minute >= 30:
            return datetime(dt.year, dt.month, dt.day, dt.hour, 0) + \
                timedelta(hours=1)
 
        return datetime(dt.year, dt.month, dt.day, dt.hour, 0)
 
 
class Weather(object):
    SKY_STATE = ['clear', 'scattered cloud', 'cloudy with breaks', 'cloudy']
 
    """A class to represent the weather at a given point in time."""
    def __init__(self, response):
        self.text_summary = response['summary']
        self.time = datetime.utcfromtimestamp(response['time'])
        self.temperature = response['temperature']
        self.apparent_temperature = response['apparentTemperature']
        self.wind_speed = response['windSpeed']
        self.wind_bearing = response['windBearing']
        self.precipitation_intensity = response['precipIntensity']
        self.precipitation_probability = response['precipProbability']
        self.pressure = response['pressure']
        self.dew_point = response['dewPoint']
        self.humidity = response['humidity']
        self.ozone = response['ozone']
        self.cloud_cover = response['cloudCover']
        self.cloud_summary = self._cloud_to_summary(self.cloud_cover)
 
    def _cloud_to_summary(self, cloud_cover):
        if cloud_cover < .1:
            return self.SKY_STATE[0]
        if cloud_cover < .4:
            return self.SKY_STATE[1]
        if cloud_cover < .75:
            return self.SKY_STATE[2]
        return self.SKY_STATE[3]
 
    def __str__(self):
        return "%s (%sºC, feels like: %sºC)" % (self.text_summary,
                                                  self.temperature,
                                                  self.apparent_temperature)
 
if __name__ == "__main__":
    # configure a Forecast object
    forecaster = Forecast("")
 
    # loads into memory a the 7 day hourly forecast
    forecaster.load(51.585278, -0.078056)
 
    # loads a Weather object for a given time
    now = datetime.now()
    weather = forecaster.forecast(now)
    print "Weather: %s at %s" % (weather, now)