-
Notifications
You must be signed in to change notification settings - Fork 8
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* API refactoring * Fix pylint and black warnings * Cleanup * Standardize forecast method names * Improve exception messages for Forecast class * Use enum for layers * Fix pylint warning * Really fixed pylint warning * Fixed black warning * Standardize naming * Retrieve all forecast layers for a given point in time * API cleanup * Rollback type hinting change * Cleanup * Revert API changes that would affect HA integration * Revert unrelated changes * Cleanup forecast units * Add method to retrieve a single forecast layer for a given time Add lint type hints * Backward compatible with old python versions * Compatibility with old Python versions * Use 'self' instead of class name * Add method to generate a range of hourly forecasts * Add method to retrieve multiple forecasts when passed a list of datetimes * Cleanup
- Loading branch information
Showing
7 changed files
with
280 additions
and
6 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1 @@ | ||
"""Backports from newer Python versions.""" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,131 @@ | ||
"""Forecast class""" | ||
from __future__ import annotations | ||
|
||
import re | ||
from datetime import datetime, timedelta, timezone | ||
from typing import Any, Generator, Iterable | ||
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_for_times( | ||
self, iterable_when: Iterable[datetime] | ||
) -> Generator[dict[str, Any]]: | ||
"""Retrieve all forecast layers for a list of times.""" | ||
|
||
if not isinstance(iterable_when, Iterable): | ||
raise TypeError(f"{iterable_when!r} is not an Iterable") | ||
|
||
for when in iterable_when: | ||
yield self.get_forecast_for_time(when) | ||
|
||
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""" | ||
|
||
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 hour in range(hours): | ||
yield self.get_forecast_for_time(start_time + timedelta(hours=hour)) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,7 +1,4 @@ | ||
import os | ||
from unittest.mock import patch | ||
|
||
import aiohttp | ||
import pytest | ||
|
||
|
||
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters