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,123 @@
+"""Forecast class"""
+from __future__ import annotations
+
+import re
+from datetime import datetime, timedelta, timezone
+from typing import Any, Generator
+from pynws.layer import Layer
+
+
+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")?$"
+)
+
+ONE_HOUR = timedelta(hours=1)
+
+
+class Forecast:
+    """Class to retrieve forecast layer values for a point in time."""
+
+    def __init__(self, properties: dict[str, Any]):
+        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 + self._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: str) -> timedelta:
+        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) -> datetime:
+        """When the forecast was last updated"""
+        return self.update_time
+
+    @staticmethod
+    def _get_layer_value_for_time(
+        when, layer_values: tuple[datetime, datetime, Any], units: str | None
+    ) -> tuple[Any, str | None]:
+        for start_time, end_time, value in layer_values:
+            if start_time <= when < end_time:
+                return (value, units)
+        return (None, None)
+
+    def get_forecast_for_time(self, when: datetime) -> dict[Any, str | None]:
+        """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)
+        forecast = {}
+        for layer_name, (layer_values, units) in self.layers.items():
+            forecast[layer_name] = self._get_layer_value_for_time(
+                when, layer_values, units
+            )
+        return forecast
+
+    def get_forecast_layer_for_time(
+        self, layer: Layer, when: datetime
+    ) -> tuple[Any, str | None]:
+        """Retrieve single forecast layer for a point in time."""
+
+        if not isinstance(layer, Layer):
+            raise TypeError(f"{layer!r} is not a Layer")
+        if not isinstance(when, datetime):
+            raise TypeError(f"{when!r} is not a datetime")
+
+        when = when.astimezone(timezone.utc)
+        values_and_unit = self.layers.get(layer)
+        if values_and_unit:
+            return self._get_layer_value_for_time(when, *values_and_unit)
+        return (None, None)
+
+    def get_hourly_forecasts(
+        self, start_time: datetime, hours: int = 12
+    ) -> Generator[dict[str, Any]]:
+        """Retrieve a sequence of hourly forecasts with all layers"""
+
+        current_time = start_time.replace(minute=0, second=0, microsecond=0)
+        for _ in range(hours):
+            hourly = self.get_forecast_for_time(current_time)
+            if not hourly:
+                break
+            hourly["startTime"] = current_time.isoformat()
+            current_time += ONE_HOUR
+            hourly["endTime"] = current_time.isoformat()
+            yield hourly

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,7 @@
-from pynws import Nws, NwsError
+from pynws import Nws, NwsError, Forecast
+from pynws.layer import Layer
+from datetime import datetime
+from types import GeneratorType
 import pytest
 
 from tests.helpers import setup_app
@@ -55,7 +58,28 @@ 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")
+
+    # get_forecast_for_time tests
+    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")
+
+    # get_forecast_layer_for_time tests
+    value = forecast.get_forecast_layer_for_time(Layer.TEMPERATURE, when)
+    assert value == (18.88888888888889, "degC")
+
+    # get_hourly_forecasts tests
+    hourly_forecasts = forecast.get_hourly_forecasts(when, 15)
+    assert isinstance(hourly_forecasts, GeneratorType)
+    hourly_forecasts = list(hourly_forecasts)
+    assert len(hourly_forecasts) == 15
+    for hourly_forecast in hourly_forecasts:
+        assert isinstance(hourly_forecast, dict)
 
 
 async def test_nws_gridpoints_forecast(aiohttp_client, loop, mock_urls):
@@ -68,6 +92,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 +110,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()