Add Forecast Class

example.py

@@ -13,10 +13,10 @@ async def example():
         nws = pynws.SimpleNWS(*PHILLY, USERID, session)
         await nws.set_station()
         await nws.update_observation()
-        await nws.update_forecast()
+        await nws.update_daily_forecast()
         await nws.update_alerts_forecast_zone()
         print(nws.observation)
-        print(nws.forecast[0])
+        print(nws.daily_forecast[0])
         print(nws.alerts_forecast_zone)
 
 

pynws/backports/__init__.py

@@ -0,0 +1 @@
+"""Backports from newer Python versions."""

pynws/backports/enum.py

@@ -0,0 +1,33 @@
+"""Python 3.11 Enum backports from https://github.com/home-assistant/core/tree/dev/homeassistant/backports"""
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any, TypeVar
+
+T = TypeVar("T", bound="StrEnum")
+
+
+class StrEnum(str, Enum):
+    """Partial backport of Python 3.11's StrEnum for our basic use cases."""
+
+    def __new__(cls: type[T], value: str, *args: Any, **kwargs: Any) -> T:
+        """Create a new StrEnum instance."""
+        if not isinstance(value, str):
+            raise TypeError(f"{value!r} is not a string")
+        return super().__new__(cls, value, *args, **kwargs)
+
+    def __str__(self) -> str:
+        """Return self.value."""
+        return str(self.value)
+
+    @staticmethod
+    def _generate_next_value_(  # pylint: disable=arguments-differ # https://github.com/PyCQA/pylint/issues/5371
+        name: str, start: int, count: int, last_values: list[Any]
+    ) -> Any:
+        """
+        Make `auto()` explicitly unsupported.
+
+        We may revisit this when it's very clear that Python 3.11's
+        `StrEnum.auto()` behavior will no longer change.
+        """
+        raise TypeError("auto() is not supported by this implementation")

pynws/const.py

@@ -1,6 +1,4 @@
-"""
-Constants for pynws
-"""
+"""Constants for pynws"""
 import os
 
 from .version import __version__
@@ -14,9 +12,9 @@
 API_STATIONS_OBSERVATIONS = "stations/{}/observations/"
 API_ACCEPT = "application/geo+json"
 API_USER = "pynws {}"
-API_FORECAST_ALL = "gridpoints/{}/{},{}"
-API_GRIDPOINTS_FORECAST = "gridpoints/{}/{},{}/forecast"
-API_GRIDPOINTS_FORECAST_HOURLY = "gridpoints/{}/{},{}/forecast/hourly"
+API_ALL_FORECAST = "gridpoints/{}/{},{}"
+API_DAILY_FORECAST = "gridpoints/{}/{},{}/forecast"
+API_HOURLY_FORECAST = "gridpoints/{}/{},{}/forecast/hourly"
 API_POINTS = "points/{},{}"
 API_ALERTS_ACTIVE_ZONE = "alerts/active/zone/{}"
 

pynws/forecast.py

@@ -0,0 +1,80 @@
+"""Forecast class"""
+import re
+from datetime import datetime, timedelta, timezone
+
+
+ISO8601_PERIOD_REGEX = re.compile(
+    r"^P"
+    r"((?P<weeks>\d+)W)?"
+    r"((?P<days>\d+)D)?"
+    r"((?:T)"
+    r"((?P<hours>\d+)H)?"
+    r"((?P<minutes>\d+)M)?"
+    r"((?P<seconds>\d+)S)?"
+    r")?$"
+)
+
+
+class Forecast:  # pylint: disable=too-few-public-methods
+    """Class to retrieve forecast layer values for a point in time."""
+
+    def __init__(self, properties):
+        if not isinstance(properties, dict):
+            raise TypeError(f"{properties!r} is not a dictionary")
+
+        self.update_time = datetime.fromisoformat(properties["updateTime"])
+        self.layers = layers = {}
+
+        for prop_name, prop_value in properties.items():
+            if not isinstance(prop_value, dict) or "values" not in prop_value:
+                continue
+
+            layer_values = []
+
+            for value in prop_value["values"]:
+                isodatetime, duration_str = value["validTime"].split("/")
+                start_time = datetime.fromisoformat(isodatetime)
+                end_time = start_time + Forecast._parse_duration(duration_str)
+                layer_values.append((start_time, end_time, value["value"]))
+
+            units = prop_value.get("uom", None)
+            layers[prop_name] = (layer_values, units)
+
+    @staticmethod
+    def _parse_duration(duration_str):
+        match = ISO8601_PERIOD_REGEX.match(duration_str)
+        groups = match.groupdict()
+
+        for key, val in groups.items():
+            groups[key] = int(val or "0")
+
+        return timedelta(
+            weeks=groups["weeks"],
+            days=groups["days"],
+            hours=groups["hours"],
+            minutes=groups["minutes"],
+            seconds=groups["seconds"],
+        )
+
+    @property
+    def last_update(self):
+        """When the forecast was last updated"""
+        return self.update_time
+
+    def get_forecast_for_time(self, when):
+        """Retrieve all forecast layers for a point in time."""
+
+        if not isinstance(when, datetime):
+            raise TypeError(f"{when!r} is not a datetime")
+
+        when = when.astimezone(timezone.utc)
+
+        result = {}
+
+        for layer_name, (layer_values, units) in self.layers.items():
+            for start_time, end_time, value in layer_values:
+                if start_time <= when < end_time:
+                    result[layer_name] = (value, units)
+                    break
+
+        return result

pynws/layer.py

@@ -0,0 +1,64 @@
+"""Layer enums"""
+from .backports.enum import StrEnum
+
+
+class Layer(StrEnum):
+    """Forecast layers"""
+
+    TEMPERATURE = "temperature"
+    DEWPOINT = "dewpoint"
+    MAX_TEMPERATURE = "maxTemperature"
+    MIN_TEMPERATURE = "minTemperature"
+    RELATIVE_HUMIDITY = "relativeHumidity"
+    APPARENT_TEMPERATURE = "apparentTemperature"
+    HEAT_INDEX = "heatIndex"
+    WIND_CHILL = "windChill"
+    SKY_COVER = "skyCover"
+    WIND_DIRECTION = "windDirection"
+    WIND_SPEED = "windSpeed"
+    WIND_GUST = "windGust"
+    WEATHER = "weather"
+    HAZARDS = "hazards"
+    PROBABILITY_OF_PRECIPITATION = "probabilityOfPrecipitation"
+    QUANTITATIVE_PRECIPITATION = "quantitativePrecipitation"
+    ICE_ACCUMULATION = "iceAccumulation"
+    SNOWFALL_AMOUNT = "snowfallAmount"
+    SNOW_LEVEL = "snowLevel"
+    CEILING_HEIGHT = "ceilingHeight"
+    VISIBILITY = "visibility"
+    TRANSPORT_WIND_SPEED = "transportWindSpeed"
+    TRANSPORT_WIND_DIRECTION = "transportWindDirection"
+    MIXING_HEIGHT = "mixingHeight"
+    HAINES_INDEX = "hainesIndex"
+    LIGHTNING_ACTIVITY_LEVEL = "lightningActivityLevel"
+    TWENTY_FOOT_WIND_SPEED = "twentyFootWindSpeed"
+    TWENTY_FOOT_WIND_DIRECTION = "twentyFootWindDirection"
+    WAVE_HEIGHT = "waveHeight"
+    WAVE_PERIOD = "wavePeriod"
+    WAVE_DIRECTION = "waveDirection"
+    PRIMARY_SWELL_HEIGHT = "primarySwellHeight"
+    PRIMARY_SWELL_DIRECTION = "primarySwellDirection"
+    SECONDARY_SWELL_HEIGHT = "secondarySwellHeight"
+    SECONDARY_SWELL_DIRECTION = "secondarySwellDirection"
+    WAVE_PERIOD2 = "wavePeriod2"
+    WIND_WAVE_HEIGHT = "windWaveHeight"
+    DISPERSION_INDEX = "dispersionIndex"
+    PRESSURE = "pressure"
+    PROBABILITY_OF_TROPICAL_STORM_WINDS = "probabilityOfTropicalStormWinds"
+    PROBABILITY_OF_HURRICANE_WINDS = "probabilityOfHurricaneWinds"
+    POTENTIAL_OF_15MPH_WINDS = "potentialOf15mphWinds"
+    POTENTIAL_OF_25MPH_WINDS = "potentialOf25mphWinds"
+    POTENTIAL_OF_35MPH_WINDS = "potentialOf35mphWinds"
+    POTENTIAL_OF_45MPH_WINDS = "potentialOf45mphWinds"
+    POTENTIAL_OF_20MPH_WIND_GUSTS = "potentialOf20mphWindGusts"
+    POTENTIAL_OF_30MPH_WIND_GUSTS = "potentialOf30mphWindGusts"
+    POTENTIAL_OF_40MPH_WIND_GUSTS = "potentialOf40mphWindGusts"
+    POTENTIAL_OF_50MPH_WIND_GUSTS = "potentialOf50mphWindGusts"
+    POTENTIAL_OF_60MPH_WIND_GUSTS = "potentialOf60mphWindGusts"
+    GRASSLAND_FIRE_DANGER_INDEX = "grasslandFireDangerIndex"
+    PROBABILITY_OF_THUNDER = "probabilityOfThunder"
+    DAVIS_STABILITY_INDEX = "davisStabilityIndex"
+    ATMOSPHERIC_DISPERSION_INDEX = "atmosphericDispersionIndex"
+    LOW_VISIBILITY_OCCURRENCE_RISK_INDEX = "lowVisibilityOccurrenceRiskIndex"
+    STABILITY = "stability"
+    RED_FLAG_THREAT_INDEX = "redFlagThreatIndex"

pynws/nws.py

@@ -1,13 +1,14 @@
 """pynws module."""
-from pynws.raw_data import (
+from .raw_data import (
     raw_alerts_active_zone,
-    raw_forecast_all,
-    raw_gridpoints_forecast,
-    raw_gridpoints_forecast_hourly,
+    raw_all_forecast,
+    raw_daily_forecast,
+    raw_hourly_forecast,
     raw_points,
     raw_points_stations,
     raw_stations_observations,
 )
+from .forecast import Forecast
 
 
 class NwsError(Exception):
@@ -38,7 +39,7 @@ def __init__(self, session, userid, latlon=None, station=None):
     async def get_points_stations(self):
         """Returns station list"""
         if self.latlon is None:
-            raise NwsError("Need to set lattitude and longitude")
+            raise NwsError("Need to set latitude and longitude")
         res = await raw_points_stations(*self.latlon, self.session, self.userid)
         return [s["properties"]["stationIdentifier"] for s in res["features"]]
 
@@ -65,29 +66,29 @@ async def get_points(self):
             self.fire_weather_zone = properties.get("fireWeatherZone").split("/")[-1]
         return properties
 
-    async def get_forecast_all(self):
+    async def get_all_forecast(self):
         """Return all forecast data from grid."""
         if self.wfo is None:
             await self.get_points()
-        raw_forecast = await raw_forecast_all(
+        raw_forecast = await raw_all_forecast(
             self.wfo, self.x, self.y, self.session, self.userid
         )
-        return raw_forecast["properties"]
+        return Forecast(raw_forecast["properties"])
 
-    async def get_gridpoints_forecast(self):
+    async def get_daily_forecast(self):
         """Return daily forecast from grid."""
         if self.wfo is None:
             await self.get_points()
-        raw_forecast = await raw_gridpoints_forecast(
+        raw_forecast = await raw_daily_forecast(
             self.wfo, self.x, self.y, self.session, self.userid
         )
         return raw_forecast["properties"]["periods"]
 
-    async def get_gridpoints_forecast_hourly(self):
+    async def get_hourly_forecast(self):
         """Return hourly forecast from grid."""
         if self.wfo is None:
             await self.get_points()
-        raw_forecast = await raw_gridpoints_forecast_hourly(
+        raw_forecast = await raw_hourly_forecast(
             self.wfo, self.x, self.y, self.session, self.userid
         )
         return raw_forecast["properties"]["periods"]

pynws/raw_data.py

@@ -1,8 +1,8 @@
 """Functions to retrieve raw data."""
 from datetime import datetime
 
-from pynws.const import API_ACCEPT, API_USER
-import pynws.urls
+from .const import API_ACCEPT, API_USER
+from . import urls
 
 
 def get_header(userid):
@@ -24,7 +24,7 @@ async def raw_stations_observations(station, websession, userid, limit=0, start=
             raise ValueError
         params["start"] = start.isoformat(timespec="seconds")
 
-    url = pynws.urls.stations_observations_url(station)
+    url = urls.stations_observations_url(station)
     header = get_header(userid)
     async with websession.get(url, headers=header, params=params) as res:
         res.raise_for_status()
@@ -34,7 +34,7 @@ async def raw_stations_observations(station, websession, userid, limit=0, start=
 
 async def raw_points_stations(lat, lon, websession, userid):
     """Get list of stations for lat/lon"""
-    url = pynws.urls.points_stations_url(lat, lon)
+    url = urls.points_stations_url(lat, lon)
     header = get_header(userid)
     async with websession.get(url, headers=header) as res:
         res.raise_for_status()
@@ -44,37 +44,37 @@ async def raw_points_stations(lat, lon, websession, userid):
 
 async def raw_points(lat, lon, websession, userid):
     """Return griddata response."""
-    url = pynws.urls.points_url(lat, lon)
+    url = urls.points_url(lat, lon)
     header = get_header(userid)
     async with websession.get(url, headers=header) as res:
         res.raise_for_status()
         jres = await res.json()
     return jres
 
 
-async def raw_forecast_all(wfo, x, y, websession, userid):
+async def raw_all_forecast(wfo, x, y, websession, userid):
     """Return griddata response."""
-    url = pynws.urls.forecast_all_url(wfo, x, y)
+    url = urls.all_forecast_url(wfo, x, y)
     header = get_header(userid)
     async with websession.get(url, headers=header) as res:
         res.raise_for_status()
         jres = await res.json()
     return jres
 
 
-async def raw_gridpoints_forecast(wfo, x, y, websession, userid):
+async def raw_daily_forecast(wfo, x, y, websession, userid):
     """Return griddata response."""
-    url = pynws.urls.gridpoints_forecast_url(wfo, x, y)
+    url = urls.daily_forecast_url(wfo, x, y)
     header = get_header(userid)
     async with websession.get(url, headers=header) as res:
         res.raise_for_status()
         jres = await res.json()
     return jres
 
 
-async def raw_gridpoints_forecast_hourly(wfo, x, y, websession, userid):
+async def raw_hourly_forecast(wfo, x, y, websession, userid):
     """Return griddata response."""
-    url = pynws.urls.gridpoints_forecast_hourly_url(wfo, x, y)
+    url = urls.hourly_forecast_url(wfo, x, y)
     header = get_header(userid)
     async with websession.get(url, headers=header) as res:
         res.raise_for_status()
@@ -84,7 +84,7 @@ async def raw_gridpoints_forecast_hourly(wfo, x, y, websession, userid):
 
 async def raw_alerts_active_zone(zone, websession, userid):
     """Return griddata response."""
-    url = pynws.urls.alerts_active_zone_url(zone)
+    url = urls.alerts_active_zone_url(zone)
     header = get_header(userid)
     async with websession.get(url, headers=header) as res:
         res.raise_for_status()