kaylee/kayleevc/plugins/darksky.py

# This is part of Kaylee
# -- this code is licensed GPLv3
# Copyright 2015-2016 Clayton G. Hobbs
# Portions Copyright 2013 Jezra

"""Dark Sky weather plugin for Kaylee

This plugin provides weather information `powered by Dark Sky
<https://darksky.net/poweredby/>`__.  The user must provide a key for
the Dark Sky API, as well as decimal latitude and longitude::

    ".darksky": {
        "api_key": "USER_API_KEY",
        "latitude": USER_LATITUDE,
        "longitude": USER_LONGITUDE
    }
"""

import json
import os
import time

import requests

from .pluginbase import PluginBase


class Plugin(PluginBase):
    """Main class for the Dark Sky weather plugin"""

    def __init__(self, config, name):
        """Initialize the Dark Sky weather plugin"""
        super().__init__(config, name)

        self._cache_filename = os.path.join(config.cache_dir, 'darksky.json')
        self._weather_url = 'https://api.darksky.net/forecast/{}/{},{}'.format(
            self.options['api_key'],
            self.options['latitude'],
            self.options['longitude']
        )
        self._cache_max_age = 1800

        self.commands = {
            'whats the temperature': self._temperature,
            'whats todays low': self._todays_low,
            'whats todays high': self._todays_high,
            'whats tomorrows high': self._tomorrows_high,
            'whats the humidity': self._relative_humidity,
            'whats the weather': self._current_conditions,
            'whens sunset': self._sunset_time,
            'hoos your weather provider': self._provider_credits
        }

        self.corpus_strings.update(self.commands)

    def _temperature(self):
        return '{:.1f} degrees Fahrenheit'.format(
            self.cache['currently']['temperature'])

    def _todays_low(self):
        return '{:.1f} degrees Fahrenheit'.format(
            self.cache['daily']['data'][0]['temperatureMin'])

    def _todays_high(self):
        return '{:.1f} degrees Fahrenheit'.format(
            self.cache['daily']['data'][0]['temperatureMax'])

    def _tomorrows_high(self):
        return '{:.1f} degrees Fahrenheit'.format(
            self.cache['daily']['data'][1]['temperatureMax'])

    def _relative_humidity(self):
        return '{:.0f} percent'.format(
            100 * self.cache['currently']['humidity'])

    def _current_conditions(self):
        return '{}, {}, relative humidity {}'.format(
            self.cache['currently']['summary'],
            self._temperature(),
            self._relative_humidity())

    def _sunset_time(self):
        # Get the time in a more useful structure
        t = time.localtime(self.cache['daily']['data'][0]['sunsetTime'])

        # Format each part of the time to be pronounced nicely
        hour = time.strftime('%I', t)
        if hour[0] == '0':
            hour = hour[1]

        minute = time.strftime('%M', t)
        if minute[0] == '0':
            if minute[1] == '0':
                minute = "o'clock"
            else:
                minute = 'O' + minute[1]

        ante_post = time.strftime('%p', t)
        if ante_post[0] == 'A':
            ante_post = 'AE ' + ante_post[1]

        # Put the parts together
        return '{} {} {}'.format(hour, minute, ante_post)

    def _provider_credits(self):
        return 'Powered by Dark Sky, https://darksky.net/poweredby/.'

    def _update_cache_if_stale(self):
        try:
            st = os.stat(self._cache_filename)
        except FileNotFoundError:
            # If there's no cache file, we need to get one
            self._update_cache()
        else:
            if time.time() - st.st_mtime > self._cache_max_age:
                self._update_cache()

    def _update_cache(self):
        r = requests.get(self._weather_url, stream=True)
        if r.status_code == 200:
            with open(self._cache_filename, 'wb') as f:
                for chunk in r:
                    f.write(chunk)

    def _read_cache(self):
        with open(self._cache_filename, 'r') as f:
            return json.load(f)

    def recognizer_finished(self, recognizer, text):
        """Speak reasonably up-to-date weather information"""
        # Is there a matching command?
        if text in self.corpus_strings:
            self.emit('processed', text)
            self._update_cache_if_stale()
            self.cache = self._read_cache()
            self.emit('tts', self.commands[text]())
            return True
        else:
            return False