deprecate/backend_client

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

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"]