1
0
mirror of https://github.com/craigerl/aprsd.git synced 2024-11-10 10:33:31 -05:00

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.
This commit is contained in:
Hemna 2023-11-22 20:39:06 -05:00
parent c0e2ef1199
commit a8f73610fe
2 changed files with 196 additions and 6 deletions

View File

@ -18,6 +18,11 @@ owm_wx_group = cfg.OptGroup(
title="Options for the OWMWeatherPlugin", title="Options for the OWMWeatherPlugin",
) )
location_group = cfg.OptGroup(
name="location_plugin",
title="Options for the LocationPlugin",
)
aprsfi_opts = [ aprsfi_opts = [
cfg.StrOpt( cfg.StrOpt(
"apiKey", "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): def register_opts(config):
config.register_group(aprsfi_group) 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_opts(owm_wx_opts, group=owm_wx_group)
config.register_group(avwx_group) config.register_group(avwx_group)
config.register_opts(avwx_opts, 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(): def list_opts():
@ -80,4 +187,5 @@ def list_opts():
query_group.name: query_plugin_opts, query_group.name: query_plugin_opts,
owm_wx_group.name: owm_wx_opts, owm_wx_group.name: owm_wx_opts,
avwx_group.name: avwx_opts, avwx_group.name: avwx_opts,
location_group.name: location_opts,
} }

View File

@ -2,7 +2,8 @@ import logging
import re import re
import time 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 oslo_config import cfg
from aprsd import packets, plugin, plugin_utils from aprsd import packets, plugin, plugin_utils
@ -13,6 +14,82 @@ CONF = cfg.CONF
LOG = logging.getLogger("APRSD") 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): class LocationPlugin(plugin.APRSDRegexCommandPluginBase, plugin.APRSFIKEYMixin):
"""Location!""" """Location!"""
@ -57,19 +134,24 @@ class LocationPlugin(plugin.APRSDRegexCommandPluginBase, plugin.APRSFIKEYMixin):
# Get some information about their location # Get some information about their location
try: try:
tic = time.perf_counter() 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}" coordinates = f"{lat:0.6f}, {lon:0.6f}"
location = geolocator.reverse(coordinates) location = geolocator.reverse(coordinates)
address = location.raw.get("address") address = location.raw.get("address")
LOG.debug(f"GEOLOCATOR address: {address}")
toc = time.perf_counter() toc = time.perf_counter()
if address: if address:
LOG.info(f"Geopy address {address} took {toc - tic:0.4f}") LOG.info(f"Geopy address {address} took {toc - tic:0.4f}")
if address.get("country_code") == "us": if address.get("country_code") == "us":
area_info = f"{address.get('county')}, {address.get('state')}" 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: else:
# what to do for address for non US? area_info = str(location)
area_info = f"{address.get('country'), 'Unknown'}"
except Exception as ex: except Exception as ex:
LOG.error(ex)
LOG.error(f"Failed to fetch Geopy address {ex}") LOG.error(f"Failed to fetch Geopy address {ex}")
area_info = "Unknown Location" area_info = "Unknown Location"