Add or update type hints

pynws/forecast.py

@@ -3,7 +3,7 @@
 
 import re
 from datetime import datetime, timedelta, timezone
-from typing import Any, Generator, Iterable, List, Tuple, Union
+from typing import Any, Dict, Generator, Iterable, List, Tuple, Union
 from .const import Detail
 from .units import get_converter
 
@@ -28,12 +28,12 @@
 class DetailedForecast:
     """Class to retrieve forecast values for a point in time."""
 
-    def __init__(self: DetailedForecast, 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: dict[Detail, _TimeValues] = {}
+        self.details: Dict[Detail, _TimeValues] = {}
 
         for prop_name, prop_value in properties.items():
             try:
@@ -64,7 +64,7 @@ def _parse_duration(duration_str: str) -> timedelta:
             raise ValueError(f"{duration_str!r} is not an ISO 8601 string")
         groups = match.groupdict()
 
-        values: dict[str, float] = {}
+        values: Dict[str, float] = {}
         for key, val in groups.items():
             values[key] = float(val or "0")
 
@@ -90,7 +90,7 @@ def _get_value_for_time(when, time_values: _TimeValues) -> DetailValue:
 
     def get_details_for_time(
         self: DetailedForecast, when: datetime
-    ) -> dict[Detail, DetailValue]:
+    ) -> Dict[Detail, DetailValue]:
         """Retrieve all forecast details for a point in time.
 
         Args:
@@ -100,21 +100,21 @@ 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")
 
         when = when.astimezone(timezone.utc)
 
-        details: dict[Detail, DetailValue] = {}
+        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: DetailedForecast, iterable_when: Iterable[datetime]
-    ) -> Generator[dict[Detail, DetailValue], None, None]:
+    ) -> Generator[Dict[Detail, DetailValue], None, None]:
         """Retrieve all forecast details for a list of times.
 
         Args:
@@ -124,7 +124,7 @@ def get_details_for_times(
             TypeError: If 'iterable_when' argument is not a collection.
 
         Yields:
-            Generator[dict[Detail, DetailValue]]: Sequence of forecast details
+            Generator[Dict[Detail, DetailValue]]: Sequence of forecast details
             corresponding with the list of times to retrieve.
         """
         if not isinstance(iterable_when, Iterable):
@@ -161,7 +161,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]:
+    ) -> Generator[Dict[Detail, DetailValue], None, None]:
         """Retrieve a sequence of hourly forecast details
 
         Args:
@@ -172,7 +172,7 @@ def get_details_by_hour(
             TypeError: If 'start_time' argument is not a 'datetime'.
 
         Yields:
-            Generator[dict[Detail, DetailValue]]: Sequence of forecast detail
+            Generator[Dict[Detail, DetailValue]]: Sequence of forecast detail
             values with one details dictionary per requested hour.
         """
         if not isinstance(start_time, datetime):
@@ -181,7 +181,7 @@ def get_details_by_hour(
         start_time = start_time.replace(minute=0, second=0, microsecond=0)
         for _ in range(hours):
             end_time = start_time + ONE_HOUR
-            details: dict[Detail, DetailValue] = {
+            details: Dict[Detail, DetailValue] = {
                 Detail.START_TIME: datetime.isoformat(start_time),
                 Detail.END_TIME: datetime.isoformat(end_time),
             }

pynws/nws.py

@@ -1,5 +1,8 @@
 """pynws module."""
 from __future__ import annotations
+from typing import Any, Dict, List, Optional, Tuple, cast
+from datetime import datetime
+from aiohttp import ClientSession
 from .raw_data import (
     raw_alerts_active_zone,
     raw_detailed_forecast,
@@ -24,38 +27,55 @@ def __init__(self, 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: Tuple[float, float],
+        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 not isinstance(latlon, tuple) or len(latlon) != 2:
+            raise NwsError(f"{latlon!r} is required to be tuple[lat, lon]")
+
+        self.session: ClientSession = session
+        self.userid: str = userid
+        self.latlon: 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: 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")
@@ -64,7 +84,7 @@ async def get_stations_observations_latest(self):
         )
         return res.get("properties")
 
-    async def get_points(self):
+    async def get_points(self: Nws) -> None:
         """Saves griddata from latlon."""
         data = await raw_points(*self.latlon, self.session, self.userid)
 
@@ -86,48 +106,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)

pynws/raw_data.py

@@ -1,22 +1,24 @@
 """Functions to retrieve raw data."""
 from datetime import datetime
 import logging
+from typing import Any, Dict
+from aiohttp import ClientSession
 
 from . import urls
 from .const import API_ACCEPT, API_USER
 
 _LOGGER = logging.getLogger(__name__)
 
 
-def get_header(userid):
+def get_header(userid: str) -> Dict[str, str]:
     """Get header.
 
     NWS recommends including an email in userid.
     """
     return {"accept": API_ACCEPT, "User-Agent": API_USER.format(userid)}
 
 
-async def _make_request(websession, url, header, params=None):
+async def _make_request(websession: ClientSession, url: str, header, params=None):
     """Make request."""
     async with websession.get(url, headers=header, params=params) as res:
         _LOGGER.debug("Request for %s returned code: %s", url, res.status)
@@ -27,65 +29,81 @@ async def _make_request(websession, url, header, params=None):
     return obs
 
 
-async def raw_stations_observations(station, websession, userid, limit=0, start=None):
+async def raw_stations_observations(
+    station: str,
+    websession: ClientSession,
+    userid: str,
+    limit: int = 0,
+    start_time: datetime = None,
+):
     """Get observation response from station"""
-    params = {}
+    params: Dict[str, Any] = {}
     if limit > 0:
         params["limit"] = limit
 
-    if start:
-        if not isinstance(start, datetime):
+    if start_time:
+        if not isinstance(start_time, datetime):
             raise ValueError
-        params["start"] = start.isoformat(timespec="seconds")
+        params["start"] = start_time.isoformat(timespec="seconds")
 
     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):
+async def raw_stations_observations_latest(
+    station: str, websession: ClientSession, userid: str
+):
     """Get observation response from 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):
+async def raw_points_stations(
+    lat: float, lon: float, websession: ClientSession, userid: str
+):
     """Get list of stations for 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):
+async def raw_points(lat: float, lon: float, websession: ClientSession, userid: str):
     """Return griddata response."""
     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):
+async def raw_detailed_forecast(
+    wfo: str, x: int, y: int, websession: ClientSession, userid: str
+):
     """Return griddata response."""
     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):
+async def raw_gridpoints_forecast(
+    wfo: str, x: int, y: int, websession: ClientSession, userid: str
+):
     """Return griddata response."""
     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):
+async def raw_gridpoints_forecast_hourly(
+    wfo: str, x: int, y: int, websession: ClientSession, userid: str
+):
     """Return griddata response."""
     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):
+async def raw_alerts_active_zone(zone: str, websession: ClientSession, userid: str):
     """Return griddata response."""
     url = urls.alerts_active_zone_url(zone)
     header = get_header(userid)