diff --git a/requirements.txt b/requirements.txt index d30bfacf..65fb7525 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,4 +5,3 @@ ovos-workshop>=2.2.0,<3.0.0 ovos-number-parser>=0.0.1,<1.0.0 ovos-date-parser>=0.0.2,<1.0.0 ovos-utterance-normalizer>=0.0.1,<1.0.0 -ovos-backend-client>=1.0.1,<2.0.0 diff --git a/weather_helpers/util.py b/weather_helpers/util.py index 3cdb4762..367f3c72 100644 --- a/weather_helpers/util.py +++ b/weather_helpers/util.py @@ -13,15 +13,155 @@ # limitations under the License. """Utility functions for the weather skill.""" from datetime import datetime, timedelta, tzinfo -from typing import List from itertools import islice +from typing import List import pytz - -from ovos_date_parser import nice_date, extract_datetime -from ovos_backend_client.api import GeolocationApi +import requests +from ovos_config import Configuration +from ovos_date_parser import nice_date, extract_datetime +from ovos_utils import timed_lru_cache from ovos_utils.time import now_local, to_local +try: + from timezonefinder import TimezoneFinder +except ImportError: + TimezoneFinder = None + + +# Geolocation Api +@timed_lru_cache(seconds=600) # cache results for 10 mins +def _get_geolocation(location): + """Call the geolocation endpoint. + + Args: + location (str): the location to lookup (e.g. Kansas City Missouri) + + Returns: + str: JSON structure with lookup results + """ + url = "https://nominatim.openstreetmap.org/search" + + data = requests.get(url, params={"q": location, "format": "json", "limit": 1}, + headers={"User-Agent": "OVOS/1.0"}).json()[0] + lat = data.get("lat") + lon = data.get("lon") + + if lat and lon: + return get_reverse_geolocation(lat, lon) + + url = "https://nominatim.openstreetmap.org/details.php" + details = requests.get(url, params={"osmid": data['osm_id'], "osmtype": data['osm_type'][0].upper(), + "format": "json"}, + headers={"User-Agent": "OVOS/1.0"}).json() + + # if no addresstags are present for the location an empty list is sent instead of a dict + tags = details.get("addresstags") or {} + + place_type = details["extratags"].get("linked_place") or details.get("category") or data.get( + "type") or data.get("class") + name = details["localname"] or details["names"].get("name") or details["names"].get("official_name") or data[ + "display_name"] + cc = details["country_code"] or tags.get("country") or details["extratags"].get('ISO3166-1:alpha2') or "" + # TODO - lang support, official name is reported in various langs + location = { + "address": data["display_name"], + "city": { + "code": tags.get("postcode") or + details["calculated_postcode"] or "", + "name": name if place_type == "city" else "", + "state": { + "code": tags.get("state_code") or + details["calculated_postcode"] or "", + "name": name if place_type == "state" else tags.get("state"), + "country": { + "code": cc.upper(), + "name": name if place_type == "country" else "" # TODO - country code to name + } + } + }, + "coordinate": { + "latitude": lat, + "longitude": lon + } + } + if "timezone" not in location: + location["timezone"] = _get_timezone(lon=lon, lat=lat) + return location + + +@timed_lru_cache(seconds=600) # cache results for 10 mins +def get_reverse_geolocation(lat, lon): + """Call the reverse geolocation endpoint. + + Args: + lat (float): latitude + lon (float): longitude + + Returns: + str: JSON structure with lookup results + """ + url = "https://nominatim.openstreetmap.org/reverse" + details = requests.get(url, params={"lat": lat, "lon": lon, "format": "json"}, + headers={"User-Agent": "OVOS/1.0"}).json() + address = details.get("address") + location = { + "address": details["display_name"], + "city": { + "code": address.get("postcode") or "", + "name": address.get("city") or + address.get("village") or + address.get("town") or + address.get("hamlet") or + address.get("county") or "", + "state": { + "code": address.get("state_code") or + address.get("ISO3166-2-lvl4") or + address.get("ISO3166-2-lvl6") + or "", + "name": address.get("state") or + address.get("county") + or "", + "country": { + "code": address.get("country_code", "").upper() or "", + "name": address.get("country") or "", + } + } + }, + "coordinate": { + "latitude": details.get("lat") or lat, + "longitude": details.get("lon") or lon + } + } + if "timezone" not in location: + location["timezone"] = _get_timezone( + lat=details.get("lat") or lat, + lon=details.get("lon") or lon) + return location + + +def _get_lat_lon(**kwargs): + lat = kwargs.get("latitude") or kwargs.get("lat") + lon = kwargs.get("longitude") or kwargs.get("lon") or kwargs.get("lng") + if not lat or not lon: + cfg = Configuration().get("location", {}).get("coordinate", {}) + lat = cfg.get("latitude") + lon = cfg.get("longitude") + return lat, lon + + +def _get_timezone(**kwargs): + if TimezoneFinder: + lat, lon = _get_lat_lon(**kwargs) + tz = TimezoneFinder().timezone_at(lng=float(lon), lat=float(lat)) + return { + "name": tz.replace("/", " "), + "code": tz + } + else: + cfg = Configuration().get("location", {}).get("timezone") + return cfg or {"name": "UTC", "code": "UTC"} + class LocationNotFoundError(ValueError): """Raise when the API cannot find the requested location.""" @@ -96,8 +236,7 @@ def get_geolocation(location: str): Raises: LocationNotFound error if the API returns no results. """ - geolocation_api = GeolocationApi() - geolocation = geolocation_api.get_geolocation(location) + geolocation = _get_geolocation(location) if geolocation is None: raise LocationNotFoundError(f"Location {location} is unknown") @@ -106,7 +245,7 @@ def get_geolocation(location: str): return { "city": geolocation["city"]["name"], "region": geolocation["city"]["state"]["name"], - "country": geolocation["city"]["state"]["country"]["name"], + "country": geolocation["city"]["state"]["country"]["name"], "latitude": geolocation["coordinate"]["latitude"], "longitude": geolocation["coordinate"]["longitude"], "timezone": geolocation["timezone"]["code"]