iicd/Inkycal
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add owm 3.0 API capabilities to get UVI reading into the fullscreen weather (again)

Browse Source
This commit is contained in:
mrbwburns
2024-02-04 10:01:49 -08:00
parent a88663defe
commit d4f9a7a845
2 changed files with 87 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
inkycal/custom/openweathermap_wrapper.py
View File

@@ -17,6 +17,10 @@ from dateutil import tz
TEMP_UNITS = Literal["celsius", "fahrenheit"]
WIND_UNITS = Literal["meters_sec", "km_hour", "miles_hour", "knots", "beaufort"]
WEATHER_TYPE = Literal["current", "forecast"]
API_VERSIONS = Literal["2.5", "3.0"]
API_BASE_URL = "https://api.openweathermap.org/data"
logger = logging.getLogger(__name__)
logger.setLevel(level=logging.INFO)
@@ -27,43 +31,72 @@ def is_timestamp_within_range(timestamp: datetime, start_time: datetime, end_tim
return start_time <= timestamp <= end_time
def get_json_from_url(request_url):
response = requests.get(request_url)
if not response.ok:
raise AssertionError(
f"Failure getting the current weather: code {response.status_code}. Reason: {response.text}"
)
return json.loads(response.text)
class OpenWeatherMap:
def __init__(
self,
api_key: str,
city_id: int,
city_id: int = None,
lat: float = None,
lon: float = None,
api_version: API_VERSIONS = "2.5",
temp_unit: TEMP_UNITS = "celsius",
wind_unit: WIND_UNITS = "meters_sec",
language: str = "en",
tz_name: str = "UTC"
tz_name: str = "UTC",
) -> None:
self.api_key = api_key
self.city_id = city_id
self.temp_unit = temp_unit
self.wind_unit = wind_unit
self.language = language
self._api_version = "2.5"
self._base_url = f"https://api.openweathermap.org/data/{self._api_version}"
self._api_version = api_version
if self._api_version == "3.0":
assert type(lat) is float and type(lon) is float
self.location_substring = (
f"lat={str(lat)}&lon={str(lon)}" if (lat is not None and lon is not None) else f"id={str(city_id)}"
)
self.tz_zone = tz.gettz(tz_name)
logger.info(f"OWM wrapper initialized for city id {self.city_id}, language {self.language} and timezone {tz_name}.")
logger.info(
f"OWM wrapper initialized for API version {self._api_version}, language {self.language} and timezone {tz_name}."
)
def get_weather_data_from_owm(self, weather: WEATHER_TYPE):
# Gets current weather or forecast from the configured OWM API.
if weather == "current":
# Gets current weather status from the 2.5 API: https://openweathermap.org/current
# This is primarily using the 2.5 API since the 3.0 API actually has less info
weather_url = f"{API_BASE_URL}/2.5/weather?{self.location_substring}&appid={self.api_key}&units=Metric&lang={self.language}"
weather_data = get_json_from_url(weather_url)
# Only if we do have a 3.0 API-enabled key, we can also get the UVI reading from that endpoint: https://openweathermap.org/api/one-call-3
if self._api_version == "3.0":
weather_url = f"{API_BASE_URL}/3.0/onecall?{self.location_substring}&appid={self.api_key}&exclude=minutely,hourly,daily&units=Metric&lang={self.language}"
weather_data["uvi"] = get_json_from_url(weather_url)["current"]["uvi"]
elif weather == "forecast":
# Gets weather forecasts from the 2.5 API: https://openweathermap.org/forecast5
# This is only using the 2.5 API since the 3.0 API actually has less info
weather_url = f"{API_BASE_URL}/2.5/forecast?{self.location_substring}&appid={self.api_key}&units=Metric&lang={self.language}"
weather_data = get_json_from_url(weather_url)["list"]
return weather_data
def get_current_weather(self) -> Dict:
"""
Gets current weather status from this API: https://openweathermap.org/current
Decodes the OWM current weather data for our purposes
:return:
Current weather as dictionary
"""
# Gets weather forecast from this API:
current_weather_url = (
f"{self._base_url}/weather?id={self.city_id}&appid={self.api_key}&units=Metric&lang={self.language}"
)
response = requests.get(current_weather_url)
if not response.ok:
raise AssertionError(
f"Failure getting the current weather: code {response.status_code}. Reason: {response.text}"
)
current_data = json.loads(response.text)
current_data = self.get_weather_data_from_owm(weather="current")
current_weather = {}
current_weather["detailed_status"] = current_data["weather"][0]["description"]
current_weather["weather_icon_name"] = current_data["weather"][0]["icon"]
@@ -80,10 +113,17 @@ class OpenWeatherMap:
if "gust" in current_data["wind"]:
current_weather["wind_gust"] = self.get_converted_windspeed(current_data["wind"]["gust"])
else:
logger.info(f"OpenWeatherMap response did not contain a wind gust speed. Using base wind: {current_weather['wind']} m/s.")
logger.info(
f"OpenWeatherMap response did not contain a wind gust speed. Using base wind: {current_weather['wind']} m/s."
)
current_weather["wind_gust"] = current_weather["wind"]
current_weather["uvi"] = None # TODO: this is no longer supported with 2.5 API, find alternative
current_weather["sunrise"] = datetime.fromtimestamp(current_data["sys"]["sunrise"], tz=self.tz_zone) # unix timestamp -> to our timezone
if "uvi" in current_data: # this is only supported in v3.0 API
current_weather["uvi"] = current_data["uvi"]
else:
current_weather["uvi"] = None
current_weather["sunrise"] = datetime.fromtimestamp(
current_data["sys"]["sunrise"], tz=self.tz_zone
) # unix timestamp -> to our timezone
current_weather["sunset"] = datetime.fromtimestamp(current_data["sys"]["sunset"], tz=self.tz_zone)
self.current_weather = current_weather
@@ -92,21 +132,13 @@ class OpenWeatherMap:
def get_weather_forecast(self) -> List[Dict]:
"""
Gets weather forecasts from this API: https://openweathermap.org/forecast5
Decodes the OWM weather forecast for our purposes
What you get is a list of 40 forecasts for 3-hour time slices, totaling to 5 days.
:return:
Forecasts data dictionary
"""
#
forecast_url = (
f"{self._base_url}/forecast?id={self.city_id}&appid={self.api_key}&units=Metric&lang={self.language}"
)
response = requests.get(forecast_url)
if not response.ok:
raise AssertionError(
f"Failure getting the current weather: code {response.status_code}. Reason: {response.text}"
)
forecast_data = json.loads(response.text)["list"]
forecast_data = self.get_weather_data_from_owm(weather="forecast")
# Add forecast data to hourly_data_dict list of dictionaries
hourly_forecasts = []
@@ -134,10 +166,12 @@ class OpenWeatherMap:
"precip_probability": forecast["pop"]
* 100.0, # OWM value is unitless, directly converting to % scale
"icon": forecast["weather"][0]["icon"],
"datetime": datetime.fromtimestamp(forecast["dt"], tz=self.tz_zone)
"datetime": datetime.fromtimestamp(forecast["dt"], tz=self.tz_zone),
}
)
logger.debug(f"Added rain forecast at {datetime.fromtimestamp(forecast['dt'], tz=self.tz_zone)}: {precip_mm}")
logger.debug(
f"Added rain forecast at {datetime.fromtimestamp(forecast['dt'], tz=self.tz_zone)}: {precip_mm}"
)
self.hourly_forecasts = hourly_forecasts
@@ -155,7 +189,7 @@ class OpenWeatherMap:
"""
# Make sure hourly forecasts are up to date
_ = self.get_weather_forecast()
# Calculate the start and end times for the specified number of days from now
current_time = datetime.now(tz=self.tz_zone)
start_time = (
@@ -292,8 +326,9 @@ def main():
current_weather = owm.get_current_weather()
print(current_weather)
hourly_forecasts = owm.get_weather_forecast()
_ = owm.get_weather_forecast()
print(owm.get_forecast_for_day(days_from_today=2))
if __name__ == "__main__":
main()