From a8f73610fed76702454937931553509c4cd0345a Mon Sep 17 00:00:00 2001 From: Hemna Date: Wed, 22 Nov 2023 20:39:06 -0500 Subject: [PATCH] Rework Location Plugin This Patch updates the location plugin to allow configuring which geopy library's supported geocoders. This patch also adds a fake geopy geocoder class that uses the us government's API for location. --- aprsd/conf/plugin_common.py | 108 ++++++++++++++++++++++++++++++++++++ aprsd/plugins/location.py | 94 +++++++++++++++++++++++++++++-- 2 files changed, 196 insertions(+), 6 deletions(-) diff --git a/aprsd/conf/plugin_common.py b/aprsd/conf/plugin_common.py index 4d43f3e..778cf17 100644 --- a/aprsd/conf/plugin_common.py +++ b/aprsd/conf/plugin_common.py @@ -18,6 +18,11 @@ owm_wx_group = cfg.OptGroup( title="Options for the OWMWeatherPlugin", ) +location_group = cfg.OptGroup( + name="location_plugin", + title="Options for the LocationPlugin", +) + aprsfi_opts = [ cfg.StrOpt( "apiKey", @@ -62,6 +67,106 @@ avwx_opts = [ ), ] +location_opts = [ + cfg.StrOpt( + "geopy_geocoder", + choices=[ + "ArcGIS", "AzureMaps", "Baidu", "Bing", "GoogleV3", "HERE", + "Nominatim", "OpenCage", "TomTom", "USGov", "What3Words", "Woosmap", + ], + default="Nominatim", + help="The geopy geocoder to use. Default is Nominatim." + "See https://geopy.readthedocs.io/en/stable/#module-geopy.geocoders" + "for more information.", + ), + cfg.StrOpt( + "user_agent", + default="APRSD", + help="The user agent to use for the Nominatim geocoder." + "See https://geopy.readthedocs.io/en/stable/#module-geopy.geocoders" + "for more information.", + ), + cfg.StrOpt( + "arcgis_username", + default=None, + help="The username to use for the ArcGIS geocoder." + "See https://geopy.readthedocs.io/en/latest/#arcgis" + "for more information." + "Only used for the ArcGIS geocoder.", + ), + cfg.StrOpt( + "arcgis_password", + default=None, + help="The password to use for the ArcGIS geocoder." + "See https://geopy.readthedocs.io/en/latest/#arcgis" + "for more information." + "Only used for the ArcGIS geocoder.", + ), + cfg.StrOpt( + "azuremaps_subscription_key", + help="The subscription key to use for the AzureMaps geocoder." + "See https://geopy.readthedocs.io/en/latest/#azuremaps" + "for more information." + "Only used for the AzureMaps geocoder.", + ), + cfg.StrOpt( + "baidu_api_key", + help="The API key to use for the Baidu geocoder." + "See https://geopy.readthedocs.io/en/latest/#baidu" + "for more information." + "Only used for the Baidu geocoder.", + ), + cfg.StrOpt( + "bing_api_key", + help="The API key to use for the Bing geocoder." + "See https://geopy.readthedocs.io/en/latest/#bing" + "for more information." + "Only used for the Bing geocoder.", + ), + cfg.StrOpt( + "google_api_key", + help="The API key to use for the Google geocoder." + "See https://geopy.readthedocs.io/en/latest/#googlev3" + "for more information." + "Only used for the Google geocoder.", + ), + cfg.StrOpt( + "here_api_key", + help="The API key to use for the HERE geocoder." + "See https://geopy.readthedocs.io/en/latest/#here" + "for more information." + "Only used for the HERE geocoder.", + ), + cfg.StrOpt( + "opencage_api_key", + help="The API key to use for the OpenCage geocoder." + "See https://geopy.readthedocs.io/en/latest/#opencage" + "for more information." + "Only used for the OpenCage geocoder.", + ), + cfg.StrOpt( + "tomtom_api_key", + help="The API key to use for the TomTom geocoder." + "See https://geopy.readthedocs.io/en/latest/#tomtom" + "for more information." + "Only used for the TomTom geocoder.", + ), + cfg.StrOpt( + "what3words_api_key", + help="The API key to use for the What3Words geocoder." + "See https://geopy.readthedocs.io/en/latest/#what3words" + "for more information." + "Only used for the What3Words geocoder.", + ), + cfg.StrOpt( + "woosmap_api_key", + help="The API key to use for the Woosmap geocoder." + "See https://geopy.readthedocs.io/en/latest/#woosmap" + "for more information." + "Only used for the Woosmap geocoder.", + ), +] + def register_opts(config): config.register_group(aprsfi_group) @@ -72,6 +177,8 @@ def register_opts(config): config.register_opts(owm_wx_opts, group=owm_wx_group) config.register_group(avwx_group) config.register_opts(avwx_opts, group=avwx_group) + config.register_group(location_group) + config.register_opts(location_opts, group=location_group) def list_opts(): @@ -80,4 +187,5 @@ def list_opts(): query_group.name: query_plugin_opts, owm_wx_group.name: owm_wx_opts, avwx_group.name: avwx_opts, + location_group.name: location_opts, } diff --git a/aprsd/plugins/location.py b/aprsd/plugins/location.py index 5b21a2b..4855979 100644 --- a/aprsd/plugins/location.py +++ b/aprsd/plugins/location.py @@ -2,7 +2,8 @@ import logging import re import time -from geopy.geocoders import Nominatim +from geopy.geocoders import ArcGIS, AzureMaps, Baidu, Bing, GoogleV3 +from geopy.geocoders import HereV7, Nominatim, OpenCage, TomTom, What3WordsV3, Woosmap from oslo_config import cfg from aprsd import packets, plugin, plugin_utils @@ -13,6 +14,82 @@ CONF = cfg.CONF LOG = logging.getLogger("APRSD") +class UsLocation: + raw = {} + + def __init__(self, info): + self.info = info + + def __str__(self): + return self.info + + +class USGov: + """US Government geocoder that uses the geopy API. + + This is a dummy class the implements the geopy reverse API, + so the factory can return an object that conforms to the API. + """ + def reverse(self, coordinates): + """Reverse geocode a coordinate.""" + LOG.info(f"USGov reverse geocode {coordinates}") + coords = coordinates.split(",") + lat = float(coords[0]) + lon = float(coords[1]) + result = plugin_utils.get_weather_gov_for_gps(lat, lon) + # LOG.info(f"WEATHER: {result}") + # LOG.info(f"area description {result['location']['areaDescription']}") + if 'location' in result: + loc = UsLocation(result['location']['areaDescription']) + else: + loc = UsLocation("Unknown Location") + + LOG.info(f"USGov reverse geocode LOC {loc}") + return loc + + +def geopy_factory(): + """Factory function for geopy geocoders.""" + geocoder = CONF.location_plugin.geopy_geocoder + LOG.info(f"Using geocoder: {geocoder}") + user_agent = CONF.location_plugin.user_agent + LOG.info(f"Using user_agent: {user_agent}") + + if geocoder == "Nominatim": + return Nominatim(user_agent=user_agent) + elif geocoder == "USGov": + return USGov() + elif geocoder == "ArcGIS": + return ArcGIS( + username=CONF.location_plugin.arcgis_username, + password=CONF.location_plugin.arcgis_password, + user_agent=user_agent, + ) + elif geocoder == "AzureMaps": + return AzureMaps( + user_agent=user_agent, + subscription_key=CONF.location_plugin.azuremaps_subscription_key, + ) + elif geocoder == "Baidu": + return Baidu(user_agent=user_agent, api_key=CONF.location_plugin.baidu_api_key) + elif geocoder == "Bing": + return Bing(user_agent=user_agent, api_key=CONF.location_plugin.bing_api_key) + elif geocoder == "GoogleV3": + return GoogleV3(user_agent=user_agent, api_key=CONF.location_plugin.google_api_key) + elif geocoder == "HERE": + return HereV7(user_agent=user_agent, api_key=CONF.location_plugin.here_api_key) + elif geocoder == "OpenCage": + return OpenCage(user_agent=user_agent, api_key=CONF.location_plugin.opencage_api_key) + elif geocoder == "TomTom": + return TomTom(user_agent=user_agent, api_key=CONF.location_plugin.tomtom_api_key) + elif geocoder == "What3Words": + return What3WordsV3(user_agent=user_agent, api_key=CONF.location_plugin.what3words_api_key) + elif geocoder == "Woosmap": + return Woosmap(user_agent=user_agent, api_key=CONF.location_plugin.woosmap_api_key) + else: + raise ValueError(f"Unknown geocoder: {geocoder}") + + class LocationPlugin(plugin.APRSDRegexCommandPluginBase, plugin.APRSFIKEYMixin): """Location!""" @@ -57,19 +134,24 @@ class LocationPlugin(plugin.APRSDRegexCommandPluginBase, plugin.APRSFIKEYMixin): # Get some information about their location try: tic = time.perf_counter() - geolocator = Nominatim(user_agent="APRSD") + geolocator = geopy_factory() + LOG.info(f"Using GEOLOCATOR: {geolocator}") coordinates = f"{lat:0.6f}, {lon:0.6f}" location = geolocator.reverse(coordinates) address = location.raw.get("address") + LOG.debug(f"GEOLOCATOR address: {address}") toc = time.perf_counter() if address: LOG.info(f"Geopy address {address} took {toc - tic:0.4f}") - if address.get("country_code") == "us": - area_info = f"{address.get('county')}, {address.get('state')}" + if address.get("country_code") == "us": + area_info = f"{address.get('county')}, {address.get('state')}" + else: + # what to do for address for non US? + area_info = f"{address.get('country'), 'Unknown'}" else: - # what to do for address for non US? - area_info = f"{address.get('country'), 'Unknown'}" + area_info = str(location) except Exception as ex: + LOG.error(ex) LOG.error(f"Failed to fetch Geopy address {ex}") area_info = "Unknown Location"