Add Forecast Class

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/forecast.py

@@ -0,0 +1,81 @@
+"""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")
+            units = units.split(":")[-1] if units else 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 pynws.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

@@ -8,6 +8,7 @@
     raw_points_stations,
     raw_stations_observations,
 )
+from pynws.forecast import Forecast
 
 
 class NwsError(Exception):
@@ -72,7 +73,7 @@ async def get_forecast_all(self):
         raw_forecast = await raw_forecast_all(
             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):
         """Return daily forecast from grid."""

tests/conftest.py

@@ -1,7 +1,4 @@
-import os
 from unittest.mock import patch
-
-import aiohttp
 import pytest
 
 

tests/test_nws.py

@@ -1,4 +1,6 @@
-from pynws import Nws, NwsError
+from pynws import Nws, NwsError, Forecast
+from pynws.layer import Layer
+from datetime import datetime
 import pytest
 
 from tests.helpers import setup_app
@@ -55,7 +57,15 @@ async def test_nws_forecast_all(aiohttp_client, loop, mock_urls):
     forecast = await nws.get_forecast_all()
     assert nws.wfo
     assert forecast
-    assert isinstance(forecast, dict)
+    assert isinstance(forecast, Forecast)
+
+    when = datetime.fromisoformat("2022-02-04T03:15:00+00:00")
+
+    values = forecast.get_forecast_for_time(when)
+    assert isinstance(values, dict)
+    assert values[Layer.TEMPERATURE] == (18.88888888888889, "degC")
+    assert values[Layer.RELATIVE_HUMIDITY] == (97.0, "percent")
+    assert values[Layer.WIND_SPEED] == (12.964, "km_h-1")
 
 
 async def test_nws_gridpoints_forecast(aiohttp_client, loop, mock_urls):
@@ -68,6 +78,13 @@ async def test_nws_gridpoints_forecast(aiohttp_client, loop, mock_urls):
     assert forecast
     assert isinstance(forecast, list)
 
+    values = forecast[0]
+    assert isinstance(values, dict)
+    assert values["startTime"] == "2019-10-13T14:00:00-04:00"
+    assert values["temperature"] == 41
+    assert values["temperatureUnit"] == "F"
+    assert values["windSpeed"] == "10 mph"
+
 
 async def test_nws_gridpoints_forecast_hourly(aiohttp_client, loop, mock_urls):
     app = setup_app()
@@ -79,6 +96,13 @@ async def test_nws_gridpoints_forecast_hourly(aiohttp_client, loop, mock_urls):
     assert forecast
     assert isinstance(forecast, list)
 
+    values = forecast[0]
+    assert isinstance(values, dict)
+    assert values["startTime"] == "2019-10-14T20:00:00-04:00"
+    assert values["temperature"] == 78
+    assert values["temperatureUnit"] == "F"
+    assert values["windSpeed"] == "0 mph"
+
 
 async def test_nws_alerts_active_zone(aiohttp_client, loop, mock_urls):
     app = setup_app()