Source code for

import re
from datetime import datetime
from urllib.request import Request, urlopen

from i3pystatus.core.util import internet, require
from import WeatherBackend

[docs]class Wunderground(WeatherBackend): ''' This module retrieves weather data from Weather Underground. .. note:: Previous versions of this module required an API key to work. Weather Underground has since discontinued their API, and this module has been rewritten to reflect that. .. rubric:: Finding your weather station To use this module, you must provide a weather station code (as the ``location_code`` option). To find your weather station, first search for your city and click to view the current conditions. Below the city name you will see the station name, and to the right of that a ``CHANGE`` link. Clicking that link will display a map, where you can find the station closest to you. Clicking on that station will take you back to the current conditions page. The weather station code will now be the last part of the URL. For example: .. code-block:: text In this case, the weather station code would be ``KMACAMBR4``. .. _weather-usage-wunderground: .. rubric:: Usage example .. code-block:: python from i3pystatus import Status from import wunderground 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'}, backend=wunderground.Wunderground( location_code='KMACAMBR4', units='imperial', update_error='<span color="#ff0000">!</span>', ), ) See :ref:`here <weather-formatters>` for a list of formatters which can be used. ''' settings = ( ('location_code', 'Location code from'), ('units', '\'metric\' or \'imperial\''), ('update_error', 'Value for the ``{update_error}`` formatter when an ' 'error is encountered while checking weather data'), ) required = ('location_code',) location_code = None units = 'metric' update_error = '!' url_template = '{location_code}' # This will be set in the init based on the passed location code forecast_url = None summary_url = '{api_key}&stationId={location_code}&format=json&units={units_type}' observation_url = '{api_key}&stationId={location_code}&format=json&units={units_type}' overview_url = ';v3-wx-observations-current;v3-location-point?apiKey={api_key}&geocodes={lat:.2f}%2C{lon:.2f}&language=en-US&units=e&format=json' headers = { 'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:80.0) Gecko/20100101 Firefox/80.0', 'Referer': '{location_code}', 'Accept': 'application/json, text/plain, */*', 'Accept-Language': 'en-US,en;q=0.5', 'Connection': 'keep-alive', } def init(self): self.units_type = 'm' if self.units == 'metric' else 'e' self.forecast_url = self.url_template.format(**vars(self)) @require(internet)
[docs] def get_api_key(self): ''' Grab the API key out of the page source from the home page ''' url = '' try: page_source = self.http_request( url, headers={ 'User-Agent': self.headers['User-Agent'], 'Accept-Language': self.headers['Accept-Language'], 'Conncetion': self.headers['Connection'], }, ) except Exception as exc: self.logger.exception('Failed to load %s', url) else: try: return'apiKey=([0-9a-f]+)', page_source).group(1) except AttributeError: self.logger.error('Failed to find API key in mainpage source')
@require(internet) def api_request(self, url, headers=None): if headers is None: headers = {} return super(Wunderground, self).api_request( url, headers=dict([(k, v.format(**vars(self))) for k, v in headers.items()])) @require(internet)
[docs] def check_weather(self): ''' Query the desired station and return the weather data ''' # Get the API key from the page source self.api_key = self.get_api_key() if self.api_key is None:['update_error'] = self.update_error return['update_error'] = '' try: try: summary = self.api_request(self.summary_url.format(**vars(self)))['summaries'][0] except (IndexError, KeyError): self.logger.error( 'Failed to retrieve summary data from API response. ' 'Run module with debug logging to get more information.' )['update_error'] = self.update_error return try: observation = self.api_request(self.observation_url.format(**vars(self)))['observations'][0] except (IndexError, KeyError): self.logger.error( 'Failed to retrieve observation data from API response. ' 'Run module with debug logging to get more information.' )['update_error'] = self.update_error return = observation['lat'] self.lon = observation['lon'] try: overview = self.api_request(self.overview_url.format(**vars(self)))[0] except IndexError: self.logger.error( 'Failed to retrieve overview data from API response. ' 'Run module with debug logging to get more information.' )['update_error'] = self.update_error return if self.units == 'metric': temp_unit = '°C' speed_unit = 'kph' distance_unit = 'km' pressure_unit = 'mb' else: temp_unit = '°F' speed_unit = 'mph' distance_unit = 'mi' pressure_unit = 'in' try: observation_time_str = observation.get('obsTimeLocal', '') observation_time = datetime.strptime(observation_time_str, '%Y-%m-%d %H:%M:%S') except (ValueError, AttributeError): observation_time = datetime.fromtimestamp(0) def _find(path, data, default=''): ptr = data try: for item in path.split(':'): if item == 'units': item = self.units ptr = ptr[item] except (KeyError, IndexError, TypeError): return default return str(ptr) pressure_tendency = _find( 'v3-wx-observations-current:pressureTendencyTrend', overview).lower() pressure_trend = '+' if pressure_tendency == 'rising' else '-'['city'] = _find('v3-location-point:location:city', overview)['condition'] = _find('v3-wx-observations-current:wxPhraseMedium', overview)['observation_time'] = observation_time['current_temp'] = _find('units:temp', observation, '0')['low_temp'] = _find('units:tempLow', summary)['high_temp'] = _find('units:tempHigh', summary)['temp_unit'] = temp_unit['feelslike'] = _find('units:heatIndex', observation)['dewpoint'] = _find('units:dewpt', observation)['wind_speed'] = _find('units:windSpeed', observation)['wind_unit'] = speed_unit['wind_direction'] = _find('v3-wx-observations-current:windDirectionCardinal', overview)['wind_gust'] = _find('units:windGust', observation)['pressure'] = _find('units:pressure', observation)['pressure_unit'] = pressure_unit['pressure_trend'] = pressure_trend['visibility'] = _find('v3-wx-observations-current:visibility', overview)['visibility_unit'] = distance_unit['humidity'] = _find('humidity', observation)['uv_index'] = _find('uv', observation) except Exception: # Don't let an uncaught exception kill the update thread self.logger.error( 'Uncaught error occurred while checking weather. ' 'Exception follows:', exc_info=True )['update_error'] = self.update_error