Return detailed forecast values in implied metric units

pynws/backports/enum.py

@@ -2,15 +2,17 @@
 from __future__ import annotations
 
 from enum import Enum
-from typing import Any, TypeVar
+from typing import Any, List, TypeVar
 
-T = TypeVar("T", bound="StrEnum")
+_StrEnumT = TypeVar("_StrEnumT", 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:
+    def __new__(
+        cls: type[_StrEnumT], value: str, *args: Any, **kwargs: Any
+    ) -> _StrEnumT:
         """Create a new StrEnum instance."""
         if not isinstance(value, str):
             raise TypeError(f"{value!r} is not a string")
@@ -22,7 +24,7 @@ def __str__(self) -> str:
 
     @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]
+        name: str, start: int, count: int, last_values: List[Any]
     ) -> Any:
         """
         Make `auto()` explicitly unsupported.

pynws/const.py

@@ -1,6 +1,7 @@
 """
 Constants for pynws
 """
+from enum import unique
 import os
 
 from .backports.enum import StrEnum
@@ -64,6 +65,7 @@
 }
 
 
+@unique
 class Detail(StrEnum):
     """Detailed forecast value names"""
 

pynws/forecast.py

@@ -3,8 +3,9 @@
 
 import re
 from datetime import datetime, timedelta, timezone
-from typing import Any, Generator, Iterable
-from pynws.const import Detail
+from typing import Any, Generator, Iterable, List, Tuple, Union
+from .const import Detail
+from .units import get_converter
 
 
 ISO8601_PERIOD_REGEX = re.compile(
@@ -20,118 +21,169 @@
 
 ONE_HOUR = timedelta(hours=1)
 
+DetailValue = Union[int, float, list, str, None]
+_TimeValues = List[Tuple[datetime, datetime, DetailValue]]
+
 
 class DetailedForecast:
     """Class to retrieve forecast values for a point in time."""
 
-    def __init__(self, properties: dict[str, Any]):
+    def __init__(self: DetailedForecast, 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.details = details = {}
+        self.details: dict[Detail, _TimeValues] = {}
 
         for prop_name, prop_value in properties.items():
-            if not isinstance(prop_value, dict) or "values" not in prop_value:
+            try:
+                detail = Detail(prop_name)
+            except ValueError:
                 continue
 
-            time_values = []
+            unit_code = prop_value.get("uom")
+            converter = get_converter(unit_code) if unit_code else None
+
+            time_values: _TimeValues = []
 
             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)
-                time_values.append((start_time, end_time, value["value"]))
+                value = value["value"]
+                if converter and value:
+                    value = converter(value)
+                time_values.append((start_time, end_time, value))
 
-            units = prop_value.get("uom")
-            units = units.split(":")[-1] if units else None
-            details[prop_name] = time_values, units
+            self.details[detail] = time_values
 
     @staticmethod
     def _parse_duration(duration_str: str) -> timedelta:
         match = ISO8601_PERIOD_REGEX.match(duration_str)
+        if not match:
+            raise ValueError(f"{duration_str!r} is not an ISO 8601 string")
         groups = match.groupdict()
 
+        values: dict[str, float] = {}
         for key, val in groups.items():
-            groups[key] = int(val or "0")
+            values[key] = float(val or "0")
 
         return timedelta(
-            weeks=groups["weeks"],
-            days=groups["days"],
-            hours=groups["hours"],
-            minutes=groups["minutes"],
-            seconds=groups["seconds"],
+            weeks=values["weeks"],
+            days=values["days"],
+            hours=values["hours"],
+            minutes=values["minutes"],
+            seconds=values["seconds"],
         )
 
     @property
-    def last_update(self) -> datetime:
-        """When the forecast was last updated"""
+    def last_update(self: DetailedForecast) -> datetime:
+        """When the forecast was last updated."""
         return self.update_time
 
     @staticmethod
-    def _find_detail_for_time(
-        when, time_values: tuple[datetime, datetime, Any], units: str | None
-    ) -> tuple[Any, str | None]:
+    def _get_value_for_time(when, time_values: _TimeValues) -> DetailValue:
         for start_time, end_time, value in time_values:
             if start_time <= when < end_time:
-                return value, units
-        return None, None
+                return value
+        return None
 
     def get_details_for_time(
-        self, when: datetime
-    ) -> dict[Detail, tuple[Any, str | None]]:
-        """Retrieve all forecast details for a point in time."""
+        self: DetailedForecast, when: datetime
+    ) -> dict[Detail, DetailValue]:
+        """Retrieve all forecast details for a point in time.
+
+        Args:
+            when (datetime): Point in time of requested forecast.
 
+        Raises:
+            TypeError: If 'when' argument is not a 'datetime'.
+
+        Returns:
+            dict[Detail, DetailValue]: All forecast details for the specified time.
+        """
         if not isinstance(when, datetime):
             raise TypeError(f"{when!r} is not a datetime")
 
         when = when.astimezone(timezone.utc)
-        details = {}
-        for detail, (time_values, units) in self.details.items():
-            details[detail] = self._find_detail_for_time(when, time_values, units)
+
+        details: dict[Detail, DetailValue] = {}
+        for detail, time_values in self.details.items():
+            details[detail] = self._get_value_for_time(when, time_values)
         return details
 
     def get_details_for_times(
-        self, iterable_when: Iterable[datetime]
-    ) -> Generator[dict[Detail, tuple[Any, Any]]]:
-        """Retrieve all forecast details for a list of times."""
+        self: DetailedForecast, iterable_when: Iterable[datetime]
+    ) -> Generator[dict[Detail, DetailValue], None, None]:
+        """Retrieve all forecast details for a list of times.
 
+        Args:
+            iterable_when (Iterable[datetime]): List of times to retrieve.
+
+        Raises:
+            TypeError: If 'iterable_when' argument is not a collection.
+
+        Yields:
+            Generator[dict[Detail, DetailValue]]: Sequence of forecast details
+            corresponding with the list of times to retrieve.
+        """
         if not isinstance(iterable_when, Iterable):
-            raise TypeError(f"{iterable_when!r} is not an Iterable")
+            raise TypeError(f"{iterable_when!r} is not iterable")
 
         for when in iterable_when:
             yield self.get_details_for_time(when)
 
     def get_detail_for_time(
-        self, detail: Detail, when: datetime
-    ) -> tuple[Any, str | None]:
-        """Retrieve single forecast detail for a point in time."""
-
-        if not isinstance(detail, Detail):
-            raise TypeError(f"{detail!r} is not a Detail")
+        self: DetailedForecast, detail_arg: Union[Detail, str], when: datetime
+    ) -> DetailValue:
+        """Retrieve single forecast detail for a point in time.
+
+        Args:
+            detail_arg (Union[Detail, str]): Forecast detail to retrieve.
+            when (datetime): Point in time of requested forecast detail.
+
+        Raises:
+            TypeError: If 'detail' argument is not a 'Detail'.
+            TypeError: If 'when' argument is not a 'datetime'.
+
+        Returns:
+            DetailValue: Requested forecast detail value for the specified time.
+        """
+        if not isinstance(detail_arg, Detail) and not isinstance(detail_arg, str):
+            raise TypeError(f"{detail_arg!r} is not a Detail or str")
         if not isinstance(when, datetime):
             raise TypeError(f"{when!r} is not a datetime")
 
         when = when.astimezone(timezone.utc)
-        time_values, units = self.details.get(detail)
-        if time_values and units:
-            return self._find_detail_for_time(when, time_values, units)
-        return None, None
+        detail = detail_arg if isinstance(detail_arg, Detail) else Detail(detail_arg)
+        time_values = self.details.get(detail)
+        return self._get_value_for_time(when, time_values) if time_values else None
 
     def get_details_by_hour(
-        self, start_time: datetime, hours: int = 12
-    ) -> Generator[dict[Detail, Any]]:
-        """Retrieve a sequence of hourly forecast details"""
+        self: DetailedForecast, start_time: datetime, hours: int = 24
+    ) -> Generator[dict[Detail, DetailValue], None, None]:
+        """Retrieve a sequence of hourly forecast details
+
+        Args:
+            start_time (datetime): First time to retrieve.
+            hours (int, optional): Number of hours to retrieve.
+
+        Raises:
+            TypeError: If 'start_time' argument is not a 'datetime'.
 
+        Yields:
+            Generator[dict[Detail, DetailValue]]: Sequence of forecast detail
+            values with one details dictionary per requested hour.
+        """
         if not isinstance(start_time, datetime):
             raise TypeError(f"{start_time!r} is not a datetime")
 
         start_time = start_time.replace(minute=0, second=0, microsecond=0)
         for _ in range(hours):
             end_time = start_time + ONE_HOUR
-            details = {
-                str(Detail.START_TIME): datetime.isoformat(start_time),
-                str(Detail.END_TIME): datetime.isoformat(end_time),
+            details: dict[Detail, DetailValue] = {
+                Detail.START_TIME: datetime.isoformat(start_time),
+                Detail.END_TIME: datetime.isoformat(end_time),
             }
             details.update(self.get_details_for_time(start_time))
             yield details

pynws/nws.py

@@ -1,5 +1,6 @@
 """pynws module."""
-from pynws.raw_data import (
+from __future__ import annotations
+from .raw_data import (
     raw_alerts_active_zone,
     raw_detailed_forecast,
     raw_gridpoints_forecast,
@@ -9,7 +10,7 @@
     raw_stations_observations,
     raw_stations_observations_latest,
 )
-from pynws.forecast import DetailedForecast
+from .forecast import DetailedForecast
 
 
 class NwsError(Exception):
@@ -77,8 +78,12 @@ async def get_points(self):
             self.fire_weather_zone = properties.get("fireWeatherZone").split("/")[-1]
         return properties
 
-    async def get_detailed_forecast(self):
-        """Return all forecast data from grid."""
+    async def get_detailed_forecast(self: Nws) -> DetailedForecast:
+        """Return all forecast data from grid.
+
+        Returns:
+            DetailedForecast: Object with all forecast details for all available times.
+        """
         if self.wfo is None:
             await self.get_points()
         raw_forecast = await raw_detailed_forecast(

pynws/raw_data.py

@@ -2,8 +2,8 @@
 from datetime import datetime
 import logging
 
-from pynws.const import API_ACCEPT, API_USER
-import pynws.urls
+from . import urls
+from .const import API_ACCEPT, API_USER
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -38,55 +38,55 @@ 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)
     return await _make_request(websession, url, header, params)
 
 
 async def raw_stations_observations_latest(station, websession, userid):
     """Get observation response from station"""
-    url = pynws.urls.stations_observations_latest_url(station)
+    url = urls.stations_observations_latest_url(station)
     header = get_header(userid)
     return await _make_request(websession, url, header)
 
 
 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)
     return await _make_request(websession, url, header)
 
 
 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)
     return await _make_request(websession, url, header)
 
 
 async def raw_detailed_forecast(wfo, x, y, websession, userid):
     """Return griddata response."""
-    url = pynws.urls.detailed_forecast_url(wfo, x, y)
+    url = urls.detailed_forecast_url(wfo, x, y)
     header = get_header(userid)
     return await _make_request(websession, url, header)
 
 
 async def raw_gridpoints_forecast(wfo, x, y, websession, userid):
     """Return griddata response."""
-    url = pynws.urls.gridpoints_forecast_url(wfo, x, y)
+    url = urls.gridpoints_forecast_url(wfo, x, y)
     header = get_header(userid)
     return await _make_request(websession, url, header)
 
 
 async def raw_gridpoints_forecast_hourly(wfo, x, y, websession, userid):
     """Return griddata response."""
-    url = pynws.urls.gridpoints_forecast_hourly_url(wfo, x, y)
+    url = urls.gridpoints_forecast_hourly_url(wfo, x, y)
     header = get_header(userid)
     return await _make_request(websession, url, header)
 
 
 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)
     return await _make_request(websession, url, header)