diff --git a/weather_helpers/util.py b/weather_helpers/util.py index 367f3c72..925e4030 100644 --- a/weather_helpers/util.py +++ b/weather_helpers/util.py @@ -17,151 +17,10 @@ from typing import List import pytz -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.location import get_geolocation as _get_geo 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.""" @@ -236,7 +95,7 @@ def get_geolocation(location: str): Raises: LocationNotFound error if the API returns no results. """ - geolocation = _get_geolocation(location) + geolocation = _get_geo(location) if geolocation is None: raise LocationNotFoundError(f"Location {location} is unknown")