path: root/cached/weather_data.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# © Copyright 2021-2022, Scott Gasch

"""How's the weather?"""

import datetime
import json
import logging
import os
import urllib.request
from dataclasses import dataclass
from typing import Any, Dict, List, Optional

from overrides import overrides

import argparse_utils
import config
import datetime_utils
import list_utils
import persistent
import scott_secrets
import type_utils

logger = logging.getLogger(__name__)

cfg = config.add_commandline_args(
    f'Cached Weather Data List ({__file__})',
    'Arguments controlling cached weather data',
)
cfg.add_argument(
    '--weather_data_cachefile',
    type=str,
    default=f'{os.environ["HOME"]}/cache/.weather_summary_cache',
    metavar='FILENAME',
    help='File in which to cache weather data',
)
cfg.add_argument(
    '--weather_data_stalest_acceptable',
    type=argparse_utils.valid_duration,
    default=datetime.timedelta(seconds=7200),  # 2 hours
    metavar='DURATION',
    help='Maximum acceptable age of cached data.  If zero, forces a refetch',
)


@dataclass
class WeatherData:
    date: datetime.date  # The date
    high: float  # The predicted high in F
    low: float  # The predicted low in F
    precipitation_inches: float  # Number of inches of precipitation / day
    conditions: List[str]  # Conditions per ~3h window
    most_common_condition: str  # The most common condition
    icon: str  # An icon to represent it


@persistent.persistent_autoloaded_singleton()  # type: ignore
class CachedWeatherData(persistent.Persistent):
    def __init__(self, weather_data: Dict[datetime.date, WeatherData] = None):
        """C'tor.  Do not pass a dict except for testing purposes.

        The @persistent_autoloaded_singleton decorator handles
        invoking our load and save methods at construction time for
        you.
        """

        if weather_data is not None:
            self.weather_data = weather_data
            return
        self.weather_data = {}
        icon_by_condition = {
            "Thunderstorm": "⚡",
            "Drizzle": "",
            "Rain": "☂️",
            "Snow": "❄️",
            "Clear": "☀️",
            "Clouds": "⛅",
            "Mist": "🌫",
            "Smoke": "🚬",
            "Haze": "🌥️",
            "Dust": "💨",
            "Fog": "🌁",
            "Sand": "🏜️",
            "Ash": "🌋",
            "Squall": "🌬",
            "Tornado": "🌪️",
        }
        now = datetime.datetime.now()
        dates = set()
        highs: Dict[datetime.date, Optional[float]] = {}
        lows: Dict[datetime.date, Optional[float]] = {}
        conditions: Dict[datetime.date, List[str]] = {}
        precip: Dict[datetime.date, float] = {}
        param = "id=5786882"  # Bellevue, WA
        key = scott_secrets.OPEN_WEATHER_MAP_KEY
        www = urllib.request.urlopen(
            f'http://api.openweathermap.org/data/2.5/weather?zip=98005,us&APPID={key}&units=imperial'
        )
        response = www.read()
        www.close()
        parsed_json = json.loads(response)
        logger.debug(parsed_json)
        dt = datetime.datetime.fromtimestamp(parsed_json["dt"]).date()
        dates.add(dt)
        condition = parsed_json["weather"][0]["main"]
        icon = icon_by_condition.get(condition, '?')
        p = 0.0
        if 'rain' in parsed_json:
            if '3h' in parsed_json['rain']:
                p += float(parsed_json['rain']['3h'])
            elif '1h' in parsed_json['rain']:
                p += float(parsed_json['rain']['1h'])
        if 'snow' in parsed_json:
            if '3h' in parsed_json['snow']:
                p += float(parsed_json['snow']['3h'])
            elif '1h' in parsed_json['snow']:
                p += float(parsed_json['snow']['1h'])
        if dt == now.date() and now.hour > 18 and condition == 'Clear':
            icon = '🌙'
        self.weather_data[dt] = WeatherData(
            date=dt,
            high=float(parsed_json["main"]["temp_max"]),
            low=float(parsed_json["main"]["temp_min"]),
            precipitation_inches=p / 25.4,
            conditions=[condition],
            most_common_condition=condition,
            icon=icon,
        )

        www = urllib.request.urlopen(
            f"http://api.openweathermap.org/data/2.5/forecast?{param}&APPID={key}&units=imperial"
        )
        response = www.read()
        www.close()
        parsed_json = json.loads(response)
        logger.debug(parsed_json)
        count = parsed_json["cnt"]
        for x in range(count):
            data = parsed_json["list"][x]
            dt = datetime.datetime.strptime(data['dt_txt'], '%Y-%m-%d %H:%M:%S')
            dt = dt.date()
            dates.add(dt)
            if dt not in highs:
                highs[dt] = None
                lows[dt] = None
                conditions[dt] = []
            for temp in (
                data["main"]["temp"],
                data['main']['temp_min'],
                data['main']['temp_max'],
            ):
                if highs[dt] is None or temp > highs[dt]:
                    highs[dt] = temp
                if lows[dt] is None or temp < lows[dt]:
                    lows[dt] = temp
            cond = data["weather"][0]["main"]
            precip[dt] = 0.0
            if 'rain' in parsed_json:
                if '3h' in parsed_json['rain']:
                    precip[dt] += float(parsed_json['rain']['3h'])
                elif '1h' in parsed_json['rain']:
                    precip[dt] += float(parsed_json['rain']['1h'])
            if 'snow' in parsed_json:
                if '3h' in parsed_json['snow']:
                    precip[dt] += float(parsed_json['snow']['3h'])
                elif '1h' in parsed_json['snow']:
                    precip[dt] += float(parsed_json['snow']['1h'])
            conditions[dt].append(cond)

        today = datetime_utils.now_pacific().date()
        for dt in sorted(dates):
            if dt == today:
                high = highs.get(dt, None)
                if high is not None and self.weather_data[today].high < high:
                    self.weather_data[today].high = high
                continue
            most_common_condition = list_utils.most_common(conditions[dt])
            icon = icon_by_condition.get(most_common_condition, '?')
            if dt == now.date() and now.hour > 18 and condition == 'Clear':
                icon = '🌙'
            self.weather_data[dt] = WeatherData(
                date=dt,
                high=type_utils.unwrap_optional(highs[dt]),
                low=type_utils.unwrap_optional(lows[dt]),
                precipitation_inches=precip[dt] / 25.4,
                conditions=conditions[dt],
                most_common_condition=most_common_condition,
                icon=icon,
            )

    @classmethod
    @overrides
    def load(cls) -> Any:

        """Depending on whether we have fresh data persisted either uses that
        data to instantiate the class or makes an HTTP request to fetch the
        necessary data.

        Note that because this is a subclass of Persistent this is taken
        care of automatically.
        """

        if persistent.was_file_written_within_n_seconds(
            config.config['weather_data_cachefile'],
            config.config['weather_data_stalest_acceptable'].total_seconds(),
        ):
            import pickle

            with open(config.config['weather_data_cachefile'], 'rb') as rf:
                weather_data = pickle.load(rf)
                return cls(weather_data)
        return None

    @overrides
    def save(self) -> bool:
        """
        Saves the current data to disk if required.  Again, because this is
        a subclass of Persistent this is taken care of for you.
        """

        import pickle

        with open(config.config['weather_data_cachefile'], 'wb') as wf:
            pickle.dump(
                self.weather_data,
                wf,
                pickle.HIGHEST_PROTOCOL,
            )
        return True