From d1d91a0da72bc89f690dc2ed72822a4d1fccec14 Mon Sep 17 00:00:00 2001 From: Hemna Date: Tue, 10 Dec 2024 17:30:17 -0500 Subject: [PATCH] Removed LocationPlugin from aprsd core This eliminates the requirement for geopy library and all it's deps. The new location for the LocationPlugin is here: https://github.com/hemna/aprsd-location-plugin --- aprsd/conf/plugin_common.py | 108 -------------------- aprsd/plugins/location.py | 181 --------------------------------- requirements.in | 1 - requirements.txt | 2 - tests/plugins/test_location.py | 109 -------------------- 5 files changed, 401 deletions(-) delete mode 100644 aprsd/plugins/location.py delete mode 100644 tests/plugins/test_location.py diff --git a/aprsd/conf/plugin_common.py b/aprsd/conf/plugin_common.py index b1d96e30..7101b824 100644 --- a/aprsd/conf/plugin_common.py +++ b/aprsd/conf/plugin_common.py @@ -18,11 +18,6 @@ title="Options for the OWMWeatherPlugin", ) -location_group = cfg.OptGroup( - name="location_plugin", - title="Options for the LocationPlugin", -) - aprsfi_opts = [ cfg.StrOpt( "apiKey", @@ -60,106 +55,6 @@ ), ] -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) @@ -169,8 +64,6 @@ 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(): @@ -178,5 +71,4 @@ def list_opts(): aprsfi_group.name: aprsfi_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 deleted file mode 100644 index 879f12b7..00000000 --- a/aprsd/plugins/location.py +++ /dev/null @@ -1,181 +0,0 @@ -import logging -import re -import time - -from geopy.geocoders import ( - ArcGIS, AzureMaps, Baidu, Bing, GoogleV3, HereV7, Nominatim, OpenCage, - TomTom, What3WordsV3, Woosmap, -) -from oslo_config import cfg - -from aprsd import packets, plugin, plugin_utils -from aprsd.utils import trace - - -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!""" - - command_regex = r"^([l]|[l]\s|location)" - command_name = "location" - short_description = "Where in the world is a CALLSIGN's last GPS beacon?" - - def setup(self): - self.ensure_aprs_fi_key() - - @trace.trace - def process(self, packet: packets.MessagePacket): - LOG.info("Location Plugin") - fromcall = packet.from_call - message = packet.get("message_text", None) - - api_key = CONF.aprs_fi.apiKey - - # optional second argument is a callsign to search - a = re.search(r"^.*\s+(.*)", message) - if a is not None: - searchcall = a.group(1) - searchcall = searchcall.upper() - else: - # if no second argument, search for calling station - searchcall = fromcall - - try: - aprs_data = plugin_utils.get_aprs_fi(api_key, searchcall) - except Exception as ex: - LOG.error(f"Failed to fetch aprs.fi '{ex}'") - return "Failed to fetch aprs.fi location" - - LOG.debug(f"LocationPlugin: aprs_data = {aprs_data}") - if not len(aprs_data["entries"]): - LOG.error("Didn't get any entries from aprs.fi") - return "Failed to fetch aprs.fi location" - - lat = float(aprs_data["entries"][0]["lat"]) - lon = float(aprs_data["entries"][0]["lng"]) - - # Get some information about their location - try: - tic = time.perf_counter() - 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')}" - else: - # what to do for address for non US? - area_info = f"{address.get('country'), 'Unknown'}" - else: - area_info = str(location) - except Exception as ex: - LOG.error(ex) - LOG.error(f"Failed to fetch Geopy address {ex}") - area_info = "Unknown Location" - - try: # altitude not always provided - alt = float(aprs_data["entries"][0]["altitude"]) - except Exception: - alt = 0 - altfeet = int(alt * 3.28084) - aprs_lasttime_seconds = aprs_data["entries"][0]["lasttime"] - # aprs_lasttime_seconds = aprs_lasttime_seconds.encode( - # "ascii", errors="ignore" - # ) # unicode to ascii - delta_seconds = time.time() - int(aprs_lasttime_seconds) - delta_hours = delta_seconds / 60 / 60 - - reply = "{}: {} {}' {},{} {}h ago".format( - searchcall, - area_info, - str(altfeet), - f"{lat:0.2f}", - f"{lon:0.2f}", - str("%.1f" % round(delta_hours, 1)), - ).rstrip() - - return reply diff --git a/requirements.in b/requirements.in index d4fb12a1..55694eb3 100644 --- a/requirements.in +++ b/requirements.in @@ -3,7 +3,6 @@ aprslib>=0.7.0 beautifulsoup4 click dataclasses-json -geopy kiss3 loguru oslo.config diff --git a/requirements.txt b/requirements.txt index e7445065..1e7fae0f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -15,8 +15,6 @@ click==8.1.7 # via -r requirements.in commonmark==0.9.1 # via rich dataclasses-json==0.6.7 # via -r requirements.in debtcollector==3.0.0 # via oslo-config -geographiclib==2.0 # via geopy -geopy==2.4.1 # via -r requirements.in idna==3.10 # via requests importlib-metadata==8.5.0 # via ax253, kiss3 kiss3==8.0.0 # via -r requirements.in diff --git a/tests/plugins/test_location.py b/tests/plugins/test_location.py deleted file mode 100644 index 6c609983..00000000 --- a/tests/plugins/test_location.py +++ /dev/null @@ -1,109 +0,0 @@ -from unittest import mock - -from oslo_config import cfg - -from aprsd import conf # noqa: F401 -from aprsd.plugins import location as location_plugin - -from .. import fake, test_plugin - - -CONF = cfg.CONF - - -class TestLocationPlugin(test_plugin.TestPlugin): - - def test_location_not_enabled_missing_aprs_fi_key(self): - # When the aprs.fi api key isn't set, then - # the LocationPlugin will be disabled. - CONF.callsign = fake.FAKE_TO_CALLSIGN - CONF.aprs_fi.apiKey = None - fortune = location_plugin.LocationPlugin() - expected = "LocationPlugin isn't enabled" - packet = fake.fake_packet(message="location") - actual = fortune.filter(packet) - self.assertEqual(expected, actual) - - @mock.patch("aprsd.plugin_utils.get_aprs_fi") - def test_location_failed_aprs_fi_location(self, mock_check): - # When the aprs.fi api key isn't set, then - # the LocationPlugin will be disabled. - mock_check.side_effect = Exception - CONF.callsign = fake.FAKE_TO_CALLSIGN - fortune = location_plugin.LocationPlugin() - expected = "Failed to fetch aprs.fi location" - packet = fake.fake_packet(message="location") - actual = fortune.filter(packet) - self.assertEqual(expected, actual) - - @mock.patch("aprsd.plugin_utils.get_aprs_fi") - def test_location_failed_aprs_fi_location_no_entries(self, mock_check): - # When the aprs.fi api key isn't set, then - # the LocationPlugin will be disabled. - mock_check.return_value = {"entries": []} - CONF.callsign = fake.FAKE_TO_CALLSIGN - fortune = location_plugin.LocationPlugin() - expected = "Failed to fetch aprs.fi location" - packet = fake.fake_packet(message="location") - actual = fortune.filter(packet) - self.assertEqual(expected, actual) - - @mock.patch("aprsd.plugin_utils.get_aprs_fi") - @mock.patch("geopy.geocoders.Nominatim.reverse") - @mock.patch("time.time") - def test_location_unknown_gps(self, mock_time, mock_geocode, mock_check_aprs): - # When the aprs.fi api key isn't set, then - # the LocationPlugin will be disabled. - mock_check_aprs.return_value = { - "entries": [ - { - "lat": 1, - "lng": 1, - "lasttime": 10, - }, - ], - } - mock_geocode.side_effect = Exception - mock_time.return_value = 10 - CONF.callsign = fake.FAKE_TO_CALLSIGN - fortune = location_plugin.LocationPlugin() - expected = "KFAKE: Unknown Location 0' 1.00,1.00 0.0h ago" - packet = fake.fake_packet(message="location") - actual = fortune.filter(packet) - self.assertEqual(expected, actual) - - @mock.patch("aprsd.plugin_utils.get_aprs_fi") - @mock.patch("geopy.geocoders.Nominatim.reverse") - @mock.patch("time.time") - def test_location_works(self, mock_time, mock_geocode, mock_check_aprs): - # When the aprs.fi api key isn't set, then - # the LocationPlugin will be disabled. - mock_check_aprs.return_value = { - "entries": [ - { - "lat": 1, - "lng": 1, - "lasttime": 10, - }, - ], - } - expected = "Appomattox" - state = "VA" - - class TempLocation: - raw = { - "address": { - "county": expected, - "country_code": "us", - "state": state, - "country": "United States", - }, - } - mock_geocode.return_value = TempLocation() - mock_time.return_value = 10 - CONF.callsign = fake.FAKE_TO_CALLSIGN - fortune = location_plugin.LocationPlugin() - expected = f"KFAKE: {expected}, {state} 0' 1.00,1.00 0.0h ago" - packet = fake.fake_packet(message="location") - actual = fortune.filter(packet) - self.assertEqual(expected, actual)