hemna/aprsd
1
0
Fork 0
mirror of https://github.com/craigerl/aprsd.git synced 2025-06-13 20:02:26 -04:00
Code Issues Projects Releases Wiki Activity

Removed LocationPlugin from aprsd core

Browse Source 
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
This commit is contained in:
Hemna 2024-12-10 17:30:17 -05:00
parent f0c02606eb
commit 3bba8a19da
6 changed files with 2 additions and 405 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
aprsd/cmds/list_plugins.py
View File

@ -20,9 +20,7 @@ from thesmuggler import smuggle
from aprsd import cli_helper
from aprsd import plugin as aprsd_plugin
from aprsd.main import cli
from aprsd.plugins import (
fortune, location, notify, ping, time, version, weather,
)
from aprsd.plugins import fortune, notify, ping, time, version, weather
LOG = logging.getLogger("APRSD")
@ -122,7 +120,7 @@ def get_installed_extensions():
def show_built_in_plugins(console):
modules = [fortune, location, notify, ping, time, version, weather]
modules = [fortune, notify, ping, time, version, weather]
plugins = []
for module in modules:

108
aprsd/conf/plugin_common.py
View File

@ -18,11 +18,6 @@ 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",
@ -60,106 +55,6 @@ 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)
@ -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,
}

181
aprsd/plugins/location.py
View File

@ -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

1
requirements.in
View File

@ -3,7 +3,6 @@ aprslib>=0.7.0
beautifulsoup4
click
dataclasses-json
geopy
kiss3
loguru
oslo.config

2
requirements.txt
View File

@ -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

109
tests/plugins/test_location.py
View File

@ -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)