Add or update type hints

pynws/const.py

@@ -2,32 +2,39 @@
 Constants for pynws
 """
 import os
+import sys
 from enum import unique
 
 from .backports.enum import StrEnum
 from .version import __version__
 
+if sys.version_info >= (3, 8):
+    from typing import Final
+else:
+    from typing_extensions import Final
+
+
 file_dir = os.path.join(os.path.dirname(__file__), "..")
 
 version = __version__
 
-API_URL = "https://api.weather.gov/"
-API_POINTS_STATIONS = "points/{},{}/stations"
-API_STATIONS_OBSERVATIONS = "stations/{}/observations/"
-API_STATIONS_OBSERVATIONS_LATEST = "stations/{}/observations/latest"
-API_ACCEPT = "application/geo+json"
-API_USER = "pynws {}"
-API_DETAILED_FORECAST = "gridpoints/{}/{},{}"
-API_GRIDPOINTS_FORECAST = "gridpoints/{}/{},{}/forecast"
-API_GRIDPOINTS_FORECAST_HOURLY = "gridpoints/{}/{},{}/forecast/hourly"
-API_POINTS = "points/{},{}"
-API_ALERTS_ACTIVE_ZONE = "alerts/active/zone/{}"
+API_URL: Final = "https://api.weather.gov/"
+API_POINTS_STATIONS: Final = "points/{},{}/stations"
+API_STATIONS_OBSERVATIONS: Final = "stations/{}/observations/"
+API_STATIONS_OBSERVATIONS_LATEST: Final = "stations/{}/observations/latest"
+API_ACCEPT: Final = "application/geo+json"
+API_USER: Final = "pynws {}"
+API_DETAILED_FORECAST: Final = "gridpoints/{}/{},{}"
+API_GRIDPOINTS_FORECAST: Final = "gridpoints/{}/{},{}/forecast"
+API_GRIDPOINTS_FORECAST_HOURLY: Final = "gridpoints/{}/{},{}/forecast/hourly"
+API_POINTS: Final = "points/{},{}"
+API_ALERTS_ACTIVE_ZONE: Final = "alerts/active/zone/{}"
 
-DEFAULT_USERID = "CODEemail@address"
+DEFAULT_USERID: Final = "CODEemail@address"
 
-ALERT_ID = "id"
+ALERT_ID: Final = "id"
 
-API_WEATHER_CODE = {
+API_WEATHER_CODE: Final = {
     "skc": "Fair/clear",
     "few": "A few clouds",
     "sct": "Partly cloudy",

pynws/forecast.py

@@ -3,12 +3,12 @@
 
 import re
 from datetime import datetime, timedelta, timezone
-from typing import Any, Dict, Generator, Iterable, List, Tuple, Union
+from typing import Any, Dict, Iterable, Iterator, List, Tuple, Union
 
-from .const import Detail
+from .const import Detail, Final
 from .units import get_converter
 
-ISO8601_PERIOD_REGEX = re.compile(
+ISO8601_PERIOD_REGEX: Final = re.compile(
     r"^P"
     r"((?P<weeks>\d+)W)?"
     r"((?P<days>\d+)D)?"
@@ -19,10 +19,10 @@
     r")?$"
 )
 
-ONE_HOUR = timedelta(hours=1)
+ONE_HOUR: Final = timedelta(hours=1)
 
 DetailValue = Union[int, float, list, str, None]
-_TimeValues = List[Tuple[datetime, datetime, DetailValue]]
+_TimeValue = Tuple[datetime, datetime, DetailValue]
 
 
 class DetailedForecast:
@@ -33,7 +33,7 @@ def __init__(self: DetailedForecast, properties: Dict[str, Any]):
             raise TypeError(f"{properties!r} is not a dictionary")
 
         self.update_time = datetime.fromisoformat(properties["updateTime"])
-        self.details: Dict[Detail, _TimeValues] = {}
+        self.details: Dict[Detail, List[_TimeValue]] = {}
 
         for prop_name, prop_value in properties.items():
             try:
@@ -44,7 +44,7 @@ def __init__(self: DetailedForecast, properties: Dict[str, Any]):
             unit_code = prop_value.get("uom")
             converter = get_converter(unit_code) if unit_code else None
 
-            time_values: _TimeValues = []
+            time_values: List[_TimeValue] = []
 
             for value in prop_value["values"]:
                 isodatetime, duration_str = value["validTime"].split("/")
@@ -82,7 +82,9 @@ def last_update(self: DetailedForecast) -> datetime:
         return self.update_time
 
     @staticmethod
-    def _get_value_for_time(when, time_values: _TimeValues) -> DetailValue:
+    def _get_value_for_time(
+        when: datetime, time_values: List[_TimeValue]
+    ) -> DetailValue:
         for start_time, end_time, value in time_values:
             if start_time <= when < end_time:
                 return value
@@ -100,7 +102,7 @@ def get_details_for_time(
             TypeError: If 'when' argument is not a 'datetime'.
 
         Returns:
-            dict[Detail, DetailValue]: All forecast details for the specified time.
+            Dict[Detail, DetailValue]: All forecast details for the specified time.
         """
         if not isinstance(when, datetime):
             raise TypeError(f"{when!r} is not a datetime")
@@ -114,7 +116,7 @@ def get_details_for_time(
 
     def get_details_for_times(
         self: DetailedForecast, iterable_when: Iterable[datetime]
-    ) -> Generator[Dict[Detail, DetailValue], None, None]:
+    ) -> Iterator[Dict[Detail, DetailValue]]:
         """Retrieve all forecast details for a list of times.
 
         Args:
@@ -124,7 +126,7 @@ def get_details_for_times(
             TypeError: If 'iterable_when' argument is not a collection.
 
         Yields:
-            Generator[dict[Detail, DetailValue]]: Sequence of forecast details
+            Iterator[Dict[Detail, DetailValue]]: Sequence of forecast details
             corresponding with the list of times to retrieve.
         """
         if not isinstance(iterable_when, Iterable):
@@ -143,7 +145,7 @@ def get_detail_for_time(
             when (datetime): Point in time of requested forecast detail.
 
         Raises:
-            TypeError: If 'detail' argument is not a 'Detail'.
+            TypeError: If 'detail' argument is not a 'Detail' or 'str'.
             TypeError: If 'when' argument is not a 'datetime'.
 
         Returns:
@@ -161,7 +163,7 @@ def get_detail_for_time(
 
     def get_details_by_hour(
         self: DetailedForecast, start_time: datetime, hours: int = 24
-    ) -> Generator[Dict[Detail, DetailValue], None, None]:
+    ) -> Iterator[Dict[Detail, DetailValue]]:
         """Retrieve a sequence of hourly forecast details
 
         Args:
@@ -172,7 +174,7 @@ def get_details_by_hour(
             TypeError: If 'start_time' argument is not a 'datetime'.
 
         Yields:
-            Generator[dict[Detail, DetailValue]]: Sequence of forecast detail
+            Iterator[Dict[Detail, DetailValue]]: Sequence of forecast detail
             values with one details dictionary per requested hour.
         """
         if not isinstance(start_time, datetime):

pynws/nws.py

@@ -1,6 +1,11 @@
 """pynws module."""
 from __future__ import annotations
 
+from datetime import datetime
+from typing import Any, Dict, List, Optional, Tuple, cast
+
+from aiohttp import ClientSession
+
 from .forecast import DetailedForecast
 from .raw_data import (
     raw_alerts_active_zone,
@@ -17,56 +22,75 @@
 class NwsError(Exception):
     """Error in Nws Class"""
 
-    def __init__(self, message):
+    def __init__(self: NwsError, message: str):
         super().__init__(message)
         self.message = message
 
 
 class Nws:
     """Class to more easily get data for one location."""
 
-    def __init__(self, session, userid, latlon=None, station=None):
-        self.session = session
-        self.latlon = latlon
-        self.station = station
-        self.userid = userid
-
-        self.wfo = None
-        self.x = None
-        self.y = None
-
-        self.forecast_zone = None
-        self.county_zone = None
-        self.fire_weather_zone = None
-
-    async def get_points_stations(self):
+    def __init__(
+        self: Nws,
+        session: ClientSession,
+        userid: str,
+        latlon: Optional[Tuple[float, float]] = None,
+        station: Optional[str] = None,
+    ):
+        if not session:
+            raise NwsError(f"{session!r} is required")
+        if not isinstance(userid, str) or not userid:
+            raise NwsError(f"{userid!r} is required")
+        if latlon and (not isinstance(latlon, tuple) or len(latlon) != 2):
+            raise NwsError(f"{latlon!r} is required to be tuple[float, float]")
+
+        self.session: ClientSession = session
+        self.userid: str = userid
+        self.latlon: Optional[Tuple[float, float]] = latlon
+        self.station: Optional[str] = station
+
+        self.wfo: Optional[str] = None
+        self.x: Optional[int] = None
+        self.y: Optional[int] = None
+
+        self.forecast_zone: Optional[str] = None
+        self.county_zone: Optional[str] = None
+        self.fire_weather_zone: Optional[str] = None
+
+    async def get_points_stations(self: Nws) -> List[str]:
         """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"]]
 
-    async def get_stations_observations(self, limit=0, start_time=None):
+    async def get_stations_observations(
+        self: Nws, limit: int = 0, start_time: Optional[datetime] = None
+    ) -> List[Dict[str, Any]]:
         """Returns observation list"""
         if self.station is None:
             raise NwsError("Need to set station")
         res = await raw_stations_observations(
             self.station, self.session, self.userid, limit, start_time
         )
         observations = [o["properties"] for o in res["features"]]
-        return sorted(observations, key=lambda o: o.get("timestamp"), reverse=True)
+        return sorted(
+            observations, key=lambda o: cast(str, o.get("timestamp")), reverse=True
+        )
 
-    async def get_stations_observations_latest(self):
+    async def get_stations_observations_latest(self: Nws) -> Dict[str, Any]:
         """Returns latest observation"""
         if self.station is None:
             raise NwsError("Need to set station")
         res = await raw_stations_observations_latest(
             self.station, self.session, self.userid
         )
-        return res.get("properties")
+        return cast(Dict[str, Any], res.get("properties"))
 
-    async def get_points(self):
+    async def get_points(self: Nws) -> None:
         """Saves griddata from latlon."""
+        if self.latlon is None:
+            raise NwsError("Latitude and longitude are required")
         data = await raw_points(*self.latlon, self.session, self.userid)
 
         properties = data.get("properties")
@@ -87,48 +111,60 @@ async def get_detailed_forecast(self: Nws) -> DetailedForecast:
         """
         if self.wfo is None:
             await self.get_points()
+        if self.wfo is None or self.x is None or self.y is None:
+            raise NwsError("Error retrieving points")
         raw_forecast = await raw_detailed_forecast(
             self.wfo, self.x, self.y, self.session, self.userid
         )
         return DetailedForecast(raw_forecast["properties"])
 
-    async def get_gridpoints_forecast(self):
+    async def get_gridpoints_forecast(self: Nws) -> List[Dict[str, Any]]:
         """Return daily forecast from grid."""
         if self.wfo is None:
             await self.get_points()
+        if self.wfo is None or self.x is None or self.y is None:
+            raise NwsError("Error retrieving points")
         raw_forecast = await raw_gridpoints_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_gridpoints_forecast_hourly(self: Nws) -> List[Dict[str, Any]]:
         """Return hourly forecast from grid."""
         if self.wfo is None:
             await self.get_points()
+        if self.wfo is None or self.x is None or self.y is None:
+            raise NwsError("Error retrieving points")
         raw_forecast = await raw_gridpoints_forecast_hourly(
             self.wfo, self.x, self.y, self.session, self.userid
         )
         return raw_forecast["properties"]["periods"]
 
-    async def get_alerts_active_zone(self, zone):
+    async def get_alerts_active_zone(self: Nws, zone: str) -> List[Dict[str, Any]]:
         """Returns alerts dict for zone."""
         alerts = await raw_alerts_active_zone(zone, self.session, self.userid)
         return [alert["properties"] for alert in alerts["features"]]
 
-    async def get_alerts_forecast_zone(self):
+    async def get_alerts_forecast_zone(self: Nws) -> List[Dict[str, Any]]:
         """Returns alerts dict for forecast zone."""
         if self.forecast_zone is None:
             await self.get_points()
+        if self.forecast_zone is None:
+            raise NwsError("Error retrieving points")
         return await self.get_alerts_active_zone(self.forecast_zone)
 
-    async def get_alerts_county_zone(self):
+    async def get_alerts_county_zone(self: Nws) -> List[Dict[str, Any]]:
         """Returns alerts dict for county zone."""
         if self.county_zone is None:
             await self.get_points()
+        if self.county_zone is None:
+            raise NwsError("Error retrieving points")
         return await self.get_alerts_active_zone(self.county_zone)
 
-    async def get_alerts_fire_weather_zone(self):
+    async def get_alerts_fire_weather_zone(self: Nws) -> List[Dict[str, Any]]:
         """Returns alerts dict for fire weather zone."""
         if self.fire_weather_zone is None:
             await self.get_points()
-            return await self.get_alerts_active_zone(self.fire_weather_zone)
+        if self.fire_weather_zone is None:
+            raise NwsError("Error retrieving points")
+        return await self.get_alerts_active_zone(self.fire_weather_zone)