cached/weather_forecast.py

   1 #!/usr/bin/env python3
   2
   3 import datetime
   4 import logging
   5 import os
   6 import urllib.request
   7 from dataclasses import dataclass
   8 from typing import Any
   9
  10 import astral  # type: ignore
  11 import pytz
  12 from astral.sun import sun  # type: ignore
  13 from bs4 import BeautifulSoup  # type: ignore
  14 from overrides import overrides
  15
  16 import argparse_utils
  17 import config
  18 import dateparse.dateparse_utils as dp
  19 import datetime_utils
  20 import persistent
  21 import smart_home.thermometers as temps
  22 import text_utils
  23
  24 logger = logging.getLogger(__name__)
  25
  26 cfg = config.add_commandline_args(
  27     f'Cached Weather Forecast ({__file__})',
  28     'Arguments controlling detailed weather rendering',
  29 )
  30 cfg.add_argument(
  31     '--weather_forecast_cachefile',
  32     type=str,
  33     default=f'{os.environ["HOME"]}/cache/.weather_forecast_cache',
  34     metavar='FILENAME',
  35     help='File in which to cache weather data',
  36 )
  37 cfg.add_argument(
  38     '--weather_forecast_stalest_acceptable',
  39     type=argparse_utils.valid_duration,
  40     default=datetime.timedelta(seconds=7200),  # 2 hours
  41     metavar='DURATION',
  42     help='Maximum acceptable age of cached data.  If zero, forces a refetch',
  43 )
  44
  45
  46 @dataclass
  47 class WeatherForecast:
  48     date: datetime.date  # The date
  49     sunrise: datetime.datetime  # Sunrise datetime
  50     sunset: datetime.datetime  # Sunset datetime
  51     description: str  # Textual description of weather
  52
  53
  54 @persistent.persistent_autoloaded_singleton()  # type: ignore
  55 class CachedDetailedWeatherForecast(persistent.Persistent):
  56     def __init__(self, forecasts=None):
  57         if forecasts is not None:
  58             self.forecasts = forecasts
  59             return
  60
  61         now = datetime_utils.now_pacific()
  62         self.forecasts = {}
  63
  64         # Ask the raspberry pi about the outside temperature.
  65         current_temp = temps.ThermometerRegistry().read_temperature(
  66             'house_outside', convert_to_fahrenheit=True
  67         )
  68
  69         # Get a weather forecast for Bellevue.
  70         www = urllib.request.urlopen(
  71             "https://forecast.weather.gov/MapClick.php?lat=47.652775&lon=-122.170716"
  72         )
  73         forecast_response = www.read()
  74         www.close()
  75
  76         soup = BeautifulSoup(forecast_response, "html.parser")
  77         forecast = soup.find(id='detailed-forecast-body')
  78         parser = dp.DateParser()
  79
  80         said_temp = False
  81         last_dt = now
  82         dt = now
  83         for (day, txt) in zip(
  84             forecast.find_all('b'),
  85             forecast.find_all(class_='col-sm-10 forecast-text'),
  86         ):
  87             last_dt = dt
  88             try:
  89                 dt = parser.parse(day.get_text())
  90             except Exception:
  91                 dt = last_dt
  92             assert dt is not None
  93
  94             # Compute sunrise/sunset times on dt.
  95             city = astral.LocationInfo("Bellevue", "USA", "US/Pacific", 47.653, -122.171)
  96             s = sun(city.observer, date=dt, tzinfo=pytz.timezone("US/Pacific"))
  97             sunrise = s['sunrise']
  98             sunset = s['sunset']
  99
 100             if dt.date() == now.date() and not said_temp and current_temp is not None:
 101                 blurb = f'{day.get_text()}: The current outside tempterature is {current_temp}. '
 102                 blurb += txt.get_text()
 103                 said_temp = True
 104             else:
 105                 blurb = f'{day.get_text()}: {txt.get_text()}'
 106             blurb = text_utils.wrap_string(blurb, 80)
 107
 108             if dt.date() in self.forecasts:
 109                 self.forecasts[dt.date()].description += '\n' + blurb
 110             else:
 111                 self.forecasts[dt.date()] = WeatherForecast(
 112                     date=dt,
 113                     sunrise=sunrise,
 114                     sunset=sunset,
 115                     description=blurb,
 116                 )
 117
 118     @classmethod
 119     @overrides
 120     def load(cls) -> Any:
 121         if persistent.was_file_written_within_n_seconds(
 122             config.config['weather_forecast_cachefile'],
 123             config.config['weather_forecast_stalest_acceptable'].total_seconds(),
 124         ):
 125             import pickle
 126
 127             with open(config.config['weather_forecast_cachefile'], 'rb') as rf:
 128                 weather_data = pickle.load(rf)
 129                 return cls(weather_data)
 130         return None
 131
 132     @overrides
 133     def save(self) -> bool:
 134         import pickle
 135
 136         with open(config.config['weather_forecast_cachefile'], 'wb') as wf:
 137             pickle.dump(
 138                 self.forecasts,
 139                 wf,
 140                 pickle.HIGHEST_PROTOCOL,
 141             )
 142         return True