Source code for i3pystatus.weather

import json
import re
import threading
import time
from urllib.request import Request, urlopen

from i3pystatus import SettingsBase, IntervalModule, formatp
from i3pystatus.core.util import user_open, internet, require


class WeatherBackend(SettingsBase):
    settings = ()

    @require(internet)
    def http_request(self, url, headers=None):
        req = Request(url, headers=headers or {})
        with urlopen(req) as content:
            try:
                content_type = dict(content.getheaders())['Content-Type']
                charset = re.search(r'charset=(.*)', content_type).group(1)
            except AttributeError:
                charset = 'utf-8'
            return content.read().decode(charset)

    @require(internet)
    def api_request(self, url, headers=None):
        self.logger.debug(f'Making API request to {url}')
        try:
            response_json = self.http_request(url, headers=headers).strip()
            if not response_json:
                self.logger.debug(f'JSON response from {url} was blank')
                return {}
            try:
                response = json.loads(response_json)
            except json.decoder.JSONDecodeError as exc:
                self.logger.error(f'Error loading JSON: {exc}')
                self.logger.debug(f'JSON text that failed to load: {response_json}')
                return {}
            self.logger.log(5, f'API response: {response}')
            error = self.check_response(response)
            if error:
                self.logger.error(f'Error in JSON response: {error}')
                return {}
            return response
        except Exception as exc:
            self.logger.exception(f'Failed to make API request to {url}')
            return {}

    def check_response(self, response):
        return False


[docs]class Weather(IntervalModule): ''' This is a generic weather-checker which must use a configured weather backend. For list of all available backends see :ref:`weatherbackends`. Double-clicking on the module will launch the forecast page for the location being checked, and single-clicking will trigger an update. .. _weather-formatters: .. rubric:: Available formatters * `{city}` — Location of weather observation * `{condition}` — Current weather condition (Rain, Snow, Overcast, etc.) * `{icon}` — Icon representing the current weather condition * `{observation_time}` — Time of weather observation (supports strftime format flags) * `{current_temp}` — Current temperature, excluding unit * `{low_temp}` — Forecasted low temperature, excluding unit * `{high_temp}` — Forecasted high temperature, excluding unit (may be empty in the late afternoon) * `{temp_unit}` — Either ``°C`` or ``°F``, depending on whether metric or * `{feelslike}` — "Feels Like" temperature, excluding unit * `{dewpoint}` — Dewpoint temperature, excluding unit imperial units are being used * `{wind_speed}` — Wind speed, excluding unit * `{wind_unit}` — Either ``kph`` or ``mph``, depending on whether metric or imperial units are being used * `{wind_direction}` — Wind direction * `{wind_gust}` — Speed of wind gusts in mph/kph, excluding unit * `{pressure}` — Barometric pressure, excluding unit * `{pressure_unit}` — ``mb`` or ``in``, depending on whether metric or imperial units are being used * `{pressure_trend}` — ``+`` if rising, ``-`` if falling, or an empty string if the pressure is steady (neither rising nor falling) * `{visibility}` — Visibility distance, excluding unit * `{visibility_unit}` — Either ``km`` or ``mi``, depending on whether metric or imperial units are being used * `{humidity}` — Current humidity, excluding percentage symbol * `{uv_index}` — UV Index * `{update_error}` — When the configured weather backend encounters an error during an update, this formatter will be set to the value of the backend's **update_error** config value. Otherwise, this formatter will be an empty string. This module supports the :ref:`formatp <formatp>` extended string format syntax. This allows for values to be hidden when they evaluate as False. The default **format** string value for this module makes use of this syntax to conditionally show the value of the **update_error** config value when the backend encounters an error during an update. See the following links for usage examples for the available weather backends: - :ref:`Weather.com <weather-usage-weathercom>` - :ref:`Weather Underground <weather-usage-wunderground>` .. rubric:: Troubleshooting If an error is encountered while updating, the ``{update_error}`` formatter will be set, and (provided it is in your ``format`` string) will show up next to the forecast to alert you to the error. The error message will (by default be logged to ``~/.i3pystatus-<pid>`` where ``<pid>`` is the PID of the update thread. However, it may be more convenient to manually set the logfile to make the location of the log data predictable and avoid clutter in your home directory. Additionally, using the ``DEBUG`` log level can be helpful in revealing why the module is not working as expected. For example: .. code-block:: python import logging from i3pystatus import Status from i3pystatus.weather import weathercom status = Status(logfile='/home/username/var/i3pystatus.log') status.register( 'weather', format='{condition} {current_temp}{temp_unit}[ {icon}][ Hi: {high_temp}][ Lo: {low_temp}][ {update_error}]', colorize=True, hints={'markup': 'pango'}, update_error='<span color="#ff0000">!</span>', log_level=logging.DEBUG, backend=weathercom.Weathercom( location_code='94107:4:US', units='imperial', log_level=logging.DEBUG, ), ) .. note:: The log level must be set separately in both the module and backend contexts. ''' settings = ( ('colorize', 'Vary the color depending on the current conditions.'), ('color_icons', 'Dictionary mapping weather conditions to tuples ' 'containing a UTF-8 code for the icon, and the color ' 'to be used.'), ('color', 'Display color (or fallback color if ``colorize`` is True). ' 'If not specified, falls back to default i3bar color.'), ('backend', 'Weather backend instance'), ('refresh_icon', 'Text to display (in addition to any text currently ' 'shown by the module) when refreshing weather data. ' '**NOTE:** Depending on how quickly the update is ' 'performed, the icon may not be displayed.'), ('online_interval', 'seconds between updates when online (defaults to interval)'), ('offline_interval', 'seconds between updates when offline (default: 300)'), 'format', ) required = ('backend',) colorize = False color_icons = { 'Fair': (u'\u263c', '#ffcc00'), 'Fog': (u'', '#949494'), 'Cloudy': (u'\u2601', '#f8f8ff'), 'Partly Cloudy': (u'\u2601', '#f8f8ff'), # \u26c5 is not in many fonts 'Rainy': (u'\u26c8', '#cbd2c0'), 'Thunderstorm': (u'\u26a1', '#cbd2c0'), 'Sunny': (u'\u2600', '#ffff00'), 'Snow': (u'\u2603', '#ffffff'), 'default': ('', None), } color = None backend = None interval = 1800 offline_interval = 300 online_interval = None refresh_icon = '⟳' format = '{current_temp}{temp_unit}[ {update_error}]' output = {'full_text': ''} on_doubleleftclick = ['launch_web'] on_leftclick = ['check_weather'] def launch_web(self): if self.backend.conditions_url and self.backend.conditions_url != 'N/A': self.logger.debug(f'Launching {self.backend.conditions_url} in browser') user_open(self.backend.conditions_url) def init(self): if self.online_interval is None: self.online_interval = int(self.interval) if self.backend is None: raise RuntimeError('A backend is required') self.backend.data = { 'city': '', 'condition': '', 'observation_time': '', 'current_temp': '', 'low_temp': '', 'high_temp': '', 'temp_unit': '', 'feelslike': '', 'dewpoint': '', 'wind_speed': '', 'wind_unit': '', 'wind_direction': '', 'wind_gust': '', 'pressure': '', 'pressure_unit': '', 'pressure_trend': '', 'visibility': '', 'visibility_unit': '', 'humidity': '', 'uv_index': '', 'update_error': '', } self.backend.init() self.condition = threading.Condition() self.thread = threading.Thread(target=self.update_thread, daemon=True) self.thread.start() def update_thread(self): if internet(): self.interval = self.online_interval else: self.interval = self.offline_interval try: self.check_weather() while True: with self.condition: self.condition.wait(self.interval) self.check_weather() except Exception: msg = 'Exception in {thread} at {time}, module {name}'.format( thread=threading.current_thread().name, time=time.strftime('%c'), name=self.__class__.__name__, ) self.logger.error(msg, exc_info=True)
[docs] def check_weather(self): ''' Check the weather using the configured backend ''' self.output['full_text'] = \ self.refresh_icon + self.output.get('full_text', '') self.backend.check_weather() self.refresh_display()
[docs] def get_color_data(self, condition): ''' Disambiguate similarly-named weather conditions, and return the icon and color that match. ''' if condition not in self.color_icons: # Check for similarly-named conditions if no exact match found condition_lc = condition.lower() if 'cloudy' in condition_lc or 'clouds' in condition_lc: if 'partly' in condition_lc: condition = 'Partly Cloudy' else: condition = 'Cloudy' elif condition_lc == 'overcast': condition = 'Cloudy' elif 'thunder' in condition_lc or 't-storm' in condition_lc: condition = 'Thunderstorm' elif 'snow' in condition_lc: condition = 'Snow' elif 'rain' in condition_lc or 'showers' in condition_lc: condition = 'Rainy' elif 'sunny' in condition_lc: condition = 'Sunny' elif 'clear' in condition_lc or 'fair' in condition_lc: condition = 'Fair' elif 'fog' in condition_lc: condition = 'Fog' return self.color_icons['default'] \ if condition not in self.color_icons \ else self.color_icons[condition]
def refresh_display(self): self.logger.debug(f'Weather data: {self.backend.data}') self.backend.data['icon'], condition_color = \ self.get_color_data(self.backend.data['condition']) color = condition_color if self.colorize else self.color self.output = { 'full_text': formatp(self.format, **self.backend.data).strip(), 'color': color, } def run(self): pass