mirror of https://github.com/craigerl/aprsd.git
refactor Plugin objects to plugins directory
This patch moves all of the plugins out of plugin.py into their own separate plugins/<plugin>.py file. This makes it easier to maintain each plugin. NOTE: You will have to update your ~/.config/aprsd/aprsd.yml to change the python location path for each plugin enabled. For example: OLD: enabled_plugins: - aprsd.plugin.EmailPlugin TO NEW enabled_plugins: - aprsd.plugins.email.EmailPlugin
This commit is contained in:
parent
df7a4b1fd7
commit
d5a34b4d11
2
Makefile
2
Makefile
|
|
@ -55,4 +55,4 @@ check: .venv/bin/tox # Code format check with isort and black
|
|||
tox -epep8
|
||||
|
||||
fix: .venv/bin/tox # fixes code formatting with isort and black
|
||||
tox -efmt
|
||||
tox -efmt
|
||||
|
|
|
|||
431
aprsd/plugin.py
431
aprsd/plugin.py
|
|
@ -3,19 +3,11 @@ import abc
|
|||
import fnmatch
|
||||
import importlib
|
||||
import inspect
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import time
|
||||
|
||||
import aprsd
|
||||
from aprsd import email, messaging
|
||||
from aprsd.fuzzyclock import fuzzy
|
||||
import pluggy
|
||||
import requests
|
||||
from thesmuggler import smuggle
|
||||
|
||||
# setup the global logger
|
||||
|
|
@ -25,17 +17,61 @@ hookspec = pluggy.HookspecMarker("aprsd")
|
|||
hookimpl = pluggy.HookimplMarker("aprsd")
|
||||
|
||||
CORE_PLUGINS = [
|
||||
"aprsd.plugin.EmailPlugin",
|
||||
"aprsd.plugin.FortunePlugin",
|
||||
"aprsd.plugin.LocationPlugin",
|
||||
"aprsd.plugin.PingPlugin",
|
||||
"aprsd.plugin.QueryPlugin",
|
||||
"aprsd.plugin.TimePlugin",
|
||||
"aprsd.plugin.WeatherPlugin",
|
||||
"aprsd.plugin.VersionPlugin",
|
||||
"aprsd.plugins.email.EmailPlugin",
|
||||
"aprsd.plugins.fortune.FortunePlugin",
|
||||
"aprsd.plugins.location.LocationPlugin",
|
||||
"aprsd.plugins.ping.PingPlugin",
|
||||
"aprsd.plugins.query.QueryPlugin",
|
||||
"aprsd.plugins.time.TimePlugin",
|
||||
"aprsd.plugins.weather.WeatherPlugin",
|
||||
"aprsd.plugins.version.VersionPlugin",
|
||||
]
|
||||
|
||||
|
||||
class APRSDCommandSpec:
|
||||
"""A hook specification namespace."""
|
||||
|
||||
@hookspec
|
||||
def run(self, fromcall, message, ack):
|
||||
"""My special little hook that you can customize."""
|
||||
pass
|
||||
|
||||
|
||||
class APRSDPluginBase(metaclass=abc.ABCMeta):
|
||||
def __init__(self, config):
|
||||
"""The aprsd config object is stored."""
|
||||
self.config = config
|
||||
|
||||
@property
|
||||
def command_name(self):
|
||||
"""The usage string help."""
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
def command_regex(self):
|
||||
"""The regex to match from the caller"""
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
def version(self):
|
||||
"""Version"""
|
||||
raise NotImplementedError
|
||||
|
||||
@hookimpl
|
||||
def run(self, fromcall, message, ack):
|
||||
if re.search(self.command_regex, message):
|
||||
return self.command(fromcall, message, ack)
|
||||
|
||||
@abc.abstractmethod
|
||||
def command(self, fromcall, message, ack):
|
||||
"""This is the command that runs when the regex matches.
|
||||
|
||||
To reply with a message over the air, return a string
|
||||
to send.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class PluginManager:
|
||||
# The singleton instance object for this class
|
||||
_instance = None
|
||||
|
|
@ -197,366 +233,3 @@ class PluginManager:
|
|||
|
||||
def get_plugins(self):
|
||||
return self._pluggy_pm.get_plugins()
|
||||
|
||||
|
||||
class APRSDCommandSpec:
|
||||
"""A hook specification namespace."""
|
||||
|
||||
@hookspec
|
||||
def run(self, fromcall, message, ack):
|
||||
"""My special little hook that you can customize."""
|
||||
pass
|
||||
|
||||
|
||||
class APRSDPluginBase(metaclass=abc.ABCMeta):
|
||||
def __init__(self, config):
|
||||
"""The aprsd config object is stored."""
|
||||
self.config = config
|
||||
|
||||
@property
|
||||
def command_name(self):
|
||||
"""The usage string help."""
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
def command_regex(self):
|
||||
"""The regex to match from the caller"""
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
def version(self):
|
||||
"""Version"""
|
||||
raise NotImplementedError
|
||||
|
||||
@hookimpl
|
||||
def run(self, fromcall, message, ack):
|
||||
if re.search(self.command_regex, message):
|
||||
return self.command(fromcall, message, ack)
|
||||
|
||||
@abc.abstractmethod
|
||||
def command(self, fromcall, message, ack):
|
||||
"""This is the command that runs when the regex matches.
|
||||
|
||||
To reply with a message over the air, return a string
|
||||
to send.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
||||
class FortunePlugin(APRSDPluginBase):
|
||||
"""Fortune."""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^[fF]"
|
||||
command_name = "fortune"
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("FortunePlugin")
|
||||
reply = None
|
||||
|
||||
fortune_path = shutil.which("fortune")
|
||||
if not fortune_path:
|
||||
reply = "Fortune command not installed"
|
||||
return reply
|
||||
|
||||
try:
|
||||
process = subprocess.Popen(
|
||||
[fortune_path, "-s", "-n 60"],
|
||||
stdout=subprocess.PIPE,
|
||||
)
|
||||
reply = process.communicate()[0]
|
||||
reply = reply.decode(errors="ignore").rstrip()
|
||||
except Exception as ex:
|
||||
reply = "Fortune command failed '{}'".format(ex)
|
||||
LOG.error(reply)
|
||||
|
||||
return reply
|
||||
|
||||
|
||||
class LocationPlugin(APRSDPluginBase):
|
||||
"""Location!"""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^[lL]"
|
||||
command_name = "location"
|
||||
|
||||
config_items = {"apikey": "aprs.fi api key here"}
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("Location Plugin")
|
||||
# get last location of a callsign, get descriptive name from weather service
|
||||
try:
|
||||
# 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
|
||||
url = (
|
||||
"http://api.aprs.fi/api/get?name="
|
||||
+ searchcall
|
||||
+ "&what=loc&apikey=104070.f9lE8qg34L8MZF&format=json"
|
||||
)
|
||||
response = requests.get(url)
|
||||
# aprs_data = json.loads(response.read())
|
||||
aprs_data = json.loads(response.text)
|
||||
LOG.debug("LocationPlugin: aprs_data = {}".format(aprs_data))
|
||||
lat = aprs_data["entries"][0]["lat"]
|
||||
lon = aprs_data["entries"][0]["lng"]
|
||||
try: # altitude not always provided
|
||||
alt = 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
|
||||
url2 = (
|
||||
"https://forecast.weather.gov/MapClick.php?lat="
|
||||
+ str(lat)
|
||||
+ "&lon="
|
||||
+ str(lon)
|
||||
+ "&FcstType=json"
|
||||
)
|
||||
response2 = requests.get(url2)
|
||||
wx_data = json.loads(response2.text)
|
||||
|
||||
reply = "{}: {} {}' {},{} {}h ago".format(
|
||||
searchcall,
|
||||
wx_data["location"]["areaDescription"],
|
||||
str(altfeet),
|
||||
str(alt),
|
||||
str(lon),
|
||||
str("%.1f" % round(delta_hours, 1)),
|
||||
).rstrip()
|
||||
except Exception as e:
|
||||
LOG.debug("Locate failed with: " + "%s" % str(e))
|
||||
reply = "Unable to find station " + searchcall + ". Sending beacons?"
|
||||
|
||||
return reply
|
||||
|
||||
|
||||
class PingPlugin(APRSDPluginBase):
|
||||
"""Ping."""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^[pP]"
|
||||
command_name = "ping"
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("PINGPlugin")
|
||||
stm = time.localtime()
|
||||
h = stm.tm_hour
|
||||
m = stm.tm_min
|
||||
s = stm.tm_sec
|
||||
reply = (
|
||||
"Pong! " + str(h).zfill(2) + ":" + str(m).zfill(2) + ":" + str(s).zfill(2)
|
||||
)
|
||||
return reply.rstrip()
|
||||
|
||||
|
||||
class QueryPlugin(APRSDPluginBase):
|
||||
"""Query command."""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = r"^\?.*"
|
||||
command_name = "query"
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("Query COMMAND")
|
||||
|
||||
tracker = messaging.MsgTrack()
|
||||
reply = "Pending Messages ({})".format(len(tracker))
|
||||
|
||||
searchstring = "^" + self.config["ham"]["callsign"] + ".*"
|
||||
# only I can do admin commands
|
||||
if re.search(searchstring, fromcall):
|
||||
r = re.search(r"^\?-\*", message)
|
||||
if r is not None:
|
||||
if len(tracker) > 0:
|
||||
reply = "Resend ALL Delayed msgs"
|
||||
LOG.debug(reply)
|
||||
tracker.restart_delayed()
|
||||
else:
|
||||
reply = "No Delayed Msgs"
|
||||
LOG.debug(reply)
|
||||
return reply
|
||||
|
||||
r = re.search(r"^\?-[fF]!", message)
|
||||
if r is not None:
|
||||
reply = "Deleting ALL Delayed msgs."
|
||||
LOG.debug(reply)
|
||||
tracker.flush()
|
||||
return reply
|
||||
|
||||
return reply
|
||||
|
||||
|
||||
class TimePlugin(APRSDPluginBase):
|
||||
"""Time command."""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^[tT]"
|
||||
command_name = "time"
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("TIME COMMAND")
|
||||
stm = time.localtime()
|
||||
h = stm.tm_hour
|
||||
m = stm.tm_min
|
||||
cur_time = fuzzy(h, m, 1)
|
||||
reply = "{} ({}:{} PDT) ({})".format(
|
||||
cur_time,
|
||||
str(h),
|
||||
str(m).rjust(2, "0"),
|
||||
message.rstrip(),
|
||||
)
|
||||
return reply
|
||||
|
||||
|
||||
class WeatherPlugin(APRSDPluginBase):
|
||||
"""Weather Command"""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^[wW]"
|
||||
command_name = "weather"
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("Weather Plugin")
|
||||
try:
|
||||
url = (
|
||||
"http://api.aprs.fi/api/get?"
|
||||
"&what=loc&apikey=104070.f9lE8qg34L8MZF&format=json"
|
||||
"&name=%s" % fromcall
|
||||
)
|
||||
response = requests.get(url)
|
||||
# aprs_data = json.loads(response.read())
|
||||
aprs_data = json.loads(response.text)
|
||||
lat = aprs_data["entries"][0]["lat"]
|
||||
lon = aprs_data["entries"][0]["lng"]
|
||||
url2 = (
|
||||
"https://forecast.weather.gov/MapClick.php?lat=%s"
|
||||
"&lon=%s&FcstType=json" % (lat, lon)
|
||||
)
|
||||
response2 = requests.get(url2)
|
||||
# wx_data = json.loads(response2.read())
|
||||
wx_data = json.loads(response2.text)
|
||||
reply = (
|
||||
"%sF(%sF/%sF) %s. %s, %s."
|
||||
% (
|
||||
wx_data["currentobservation"]["Temp"],
|
||||
wx_data["data"]["temperature"][0],
|
||||
wx_data["data"]["temperature"][1],
|
||||
wx_data["data"]["weather"][0],
|
||||
wx_data["time"]["startPeriodName"][1],
|
||||
wx_data["data"]["weather"][1],
|
||||
)
|
||||
).rstrip()
|
||||
LOG.debug("reply: '{}' ".format(reply))
|
||||
except Exception as e:
|
||||
LOG.debug("Weather failed with: " + "%s" % str(e))
|
||||
reply = "Unable to find you (send beacon?)"
|
||||
|
||||
return reply
|
||||
|
||||
|
||||
class EmailPlugin(APRSDPluginBase):
|
||||
"""Email Plugin."""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^-.*"
|
||||
command_name = "email"
|
||||
|
||||
# message_number:time combos so we don't resend the same email in
|
||||
# five mins {int:int}
|
||||
email_sent_dict = {}
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("Email COMMAND")
|
||||
reply = None
|
||||
|
||||
searchstring = "^" + self.config["ham"]["callsign"] + ".*"
|
||||
# only I can do email
|
||||
if re.search(searchstring, fromcall):
|
||||
# digits only, first one is number of emails to resend
|
||||
r = re.search("^-([0-9])[0-9]*$", message)
|
||||
if r is not None:
|
||||
LOG.debug("RESEND EMAIL")
|
||||
email.resend_email(r.group(1), fromcall)
|
||||
reply = messaging.NULL_MESSAGE
|
||||
# -user@address.com body of email
|
||||
elif re.search(r"^-([A-Za-z0-9_\-\.@]+) (.*)", message):
|
||||
# (same search again)
|
||||
a = re.search(r"^-([A-Za-z0-9_\-\.@]+) (.*)", message)
|
||||
if a is not None:
|
||||
to_addr = a.group(1)
|
||||
content = a.group(2)
|
||||
|
||||
email_address = email.get_email_from_shortcut(to_addr)
|
||||
if not email_address:
|
||||
reply = "Bad email address"
|
||||
return reply
|
||||
|
||||
# send recipient link to aprs.fi map
|
||||
if content == "mapme":
|
||||
content = "Click for my location: http://aprs.fi/{}".format(
|
||||
self.config["ham"]["callsign"],
|
||||
)
|
||||
too_soon = 0
|
||||
now = time.time()
|
||||
# see if we sent this msg number recently
|
||||
if ack in self.email_sent_dict:
|
||||
# BUG(hemna) - when we get a 2 different email command
|
||||
# with the same ack #, we don't send it.
|
||||
timedelta = now - self.email_sent_dict[ack]
|
||||
if timedelta < 300: # five minutes
|
||||
too_soon = 1
|
||||
if not too_soon or ack == 0:
|
||||
LOG.info("Send email '{}'".format(content))
|
||||
send_result = email.send_email(to_addr, content)
|
||||
reply = messaging.NULL_MESSAGE
|
||||
if send_result != 0:
|
||||
reply = "-{} failed".format(to_addr)
|
||||
# messaging.send_message(fromcall, "-" + to_addr + " failed")
|
||||
else:
|
||||
# clear email sent dictionary if somehow goes over 100
|
||||
if len(self.email_sent_dict) > 98:
|
||||
LOG.debug(
|
||||
"DEBUG: email_sent_dict is big ("
|
||||
+ str(len(self.email_sent_dict))
|
||||
+ ") clearing out.",
|
||||
)
|
||||
self.email_sent_dict.clear()
|
||||
self.email_sent_dict[ack] = now
|
||||
else:
|
||||
LOG.info(
|
||||
"Email for message number "
|
||||
+ ack
|
||||
+ " recently sent, not sending again.",
|
||||
)
|
||||
else:
|
||||
reply = "Bad email address"
|
||||
# messaging.send_message(fromcall, "Bad email address")
|
||||
|
||||
return reply
|
||||
|
||||
|
||||
class VersionPlugin(APRSDPluginBase):
|
||||
"""Version of APRSD Plugin."""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^[vV]"
|
||||
command_name = "version"
|
||||
|
||||
# message_number:time combos so we don't resend the same email in
|
||||
# five mins {int:int}
|
||||
email_sent_dict = {}
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("Version COMMAND")
|
||||
return "APRSD version '{}'".format(aprsd.__version__)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,88 @@
|
|||
import logging
|
||||
import re
|
||||
import time
|
||||
|
||||
from aprsd import email, messaging, plugin
|
||||
|
||||
LOG = logging.getLogger("APRSD")
|
||||
|
||||
|
||||
class EmailPlugin(plugin.APRSDPluginBase):
|
||||
"""Email Plugin."""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^-.*"
|
||||
command_name = "email"
|
||||
|
||||
# message_number:time combos so we don't resend the same email in
|
||||
# five mins {int:int}
|
||||
email_sent_dict = {}
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("Email COMMAND")
|
||||
reply = None
|
||||
|
||||
searchstring = "^" + self.config["ham"]["callsign"] + ".*"
|
||||
# only I can do email
|
||||
if re.search(searchstring, fromcall):
|
||||
# digits only, first one is number of emails to resend
|
||||
r = re.search("^-([0-9])[0-9]*$", message)
|
||||
if r is not None:
|
||||
LOG.debug("RESEND EMAIL")
|
||||
email.resend_email(r.group(1), fromcall)
|
||||
reply = messaging.NULL_MESSAGE
|
||||
# -user@address.com body of email
|
||||
elif re.search(r"^-([A-Za-z0-9_\-\.@]+) (.*)", message):
|
||||
# (same search again)
|
||||
a = re.search(r"^-([A-Za-z0-9_\-\.@]+) (.*)", message)
|
||||
if a is not None:
|
||||
to_addr = a.group(1)
|
||||
content = a.group(2)
|
||||
|
||||
email_address = email.get_email_from_shortcut(to_addr)
|
||||
if not email_address:
|
||||
reply = "Bad email address"
|
||||
return reply
|
||||
|
||||
# send recipient link to aprs.fi map
|
||||
if content == "mapme":
|
||||
content = "Click for my location: http://aprs.fi/{}".format(
|
||||
self.config["ham"]["callsign"],
|
||||
)
|
||||
too_soon = 0
|
||||
now = time.time()
|
||||
# see if we sent this msg number recently
|
||||
if ack in self.email_sent_dict:
|
||||
# BUG(hemna) - when we get a 2 different email command
|
||||
# with the same ack #, we don't send it.
|
||||
timedelta = now - self.email_sent_dict[ack]
|
||||
if timedelta < 300: # five minutes
|
||||
too_soon = 1
|
||||
if not too_soon or ack == 0:
|
||||
LOG.info("Send email '{}'".format(content))
|
||||
send_result = email.send_email(to_addr, content)
|
||||
reply = messaging.NULL_MESSAGE
|
||||
if send_result != 0:
|
||||
reply = "-{} failed".format(to_addr)
|
||||
# messaging.send_message(fromcall, "-" + to_addr + " failed")
|
||||
else:
|
||||
# clear email sent dictionary if somehow goes over 100
|
||||
if len(self.email_sent_dict) > 98:
|
||||
LOG.debug(
|
||||
"DEBUG: email_sent_dict is big ("
|
||||
+ str(len(self.email_sent_dict))
|
||||
+ ") clearing out.",
|
||||
)
|
||||
self.email_sent_dict.clear()
|
||||
self.email_sent_dict[ack] = now
|
||||
else:
|
||||
LOG.info(
|
||||
"Email for message number "
|
||||
+ ack
|
||||
+ " recently sent, not sending again.",
|
||||
)
|
||||
else:
|
||||
reply = "Bad email address"
|
||||
# messaging.send_message(fromcall, "Bad email address")
|
||||
|
||||
return reply
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
import logging
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
from aprsd import plugin
|
||||
|
||||
LOG = logging.getLogger("APRSD")
|
||||
|
||||
|
||||
class FortunePlugin(plugin.APRSDPluginBase):
|
||||
"""Fortune."""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^[fF]"
|
||||
command_name = "fortune"
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("FortunePlugin")
|
||||
reply = None
|
||||
|
||||
fortune_path = shutil.which("fortune")
|
||||
if not fortune_path:
|
||||
reply = "Fortune command not installed"
|
||||
return reply
|
||||
|
||||
try:
|
||||
process = subprocess.Popen(
|
||||
[fortune_path, "-s", "-n 60"],
|
||||
stdout=subprocess.PIPE,
|
||||
)
|
||||
reply = process.communicate()[0]
|
||||
reply = reply.decode(errors="ignore").rstrip()
|
||||
except Exception as ex:
|
||||
reply = "Fortune command failed '{}'".format(ex)
|
||||
LOG.error(reply)
|
||||
|
||||
return reply
|
||||
|
|
@ -0,0 +1,77 @@
|
|||
import json
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
|
||||
from aprsd import plugin
|
||||
import requests
|
||||
|
||||
LOG = logging.getLogger("APRSD")
|
||||
|
||||
|
||||
class LocationPlugin(plugin.APRSDPluginBase):
|
||||
"""Location!"""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^[lL]"
|
||||
command_name = "location"
|
||||
|
||||
config_items = {"apikey": "aprs.fi api key here"}
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("Location Plugin")
|
||||
# get last location of a callsign, get descriptive name from weather service
|
||||
try:
|
||||
# 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
|
||||
url = (
|
||||
"http://api.aprs.fi/api/get?name="
|
||||
+ searchcall
|
||||
+ "&what=loc&apikey=104070.f9lE8qg34L8MZF&format=json"
|
||||
)
|
||||
response = requests.get(url)
|
||||
# aprs_data = json.loads(response.read())
|
||||
aprs_data = json.loads(response.text)
|
||||
LOG.debug("LocationPlugin: aprs_data = {}".format(aprs_data))
|
||||
lat = aprs_data["entries"][0]["lat"]
|
||||
lon = aprs_data["entries"][0]["lng"]
|
||||
try: # altitude not always provided
|
||||
alt = 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
|
||||
url2 = (
|
||||
"https://forecast.weather.gov/MapClick.php?lat="
|
||||
+ str(lat)
|
||||
+ "&lon="
|
||||
+ str(lon)
|
||||
+ "&FcstType=json"
|
||||
)
|
||||
response2 = requests.get(url2)
|
||||
wx_data = json.loads(response2.text)
|
||||
|
||||
reply = "{}: {} {}' {},{} {}h ago".format(
|
||||
searchcall,
|
||||
wx_data["location"]["areaDescription"],
|
||||
str(altfeet),
|
||||
str(alt),
|
||||
str(lon),
|
||||
str("%.1f" % round(delta_hours, 1)),
|
||||
).rstrip()
|
||||
except Exception as e:
|
||||
LOG.debug("Locate failed with: " + "%s" % str(e))
|
||||
reply = "Unable to find station " + searchcall + ". Sending beacons?"
|
||||
|
||||
return reply
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
import logging
|
||||
import time
|
||||
|
||||
from aprsd import plugin
|
||||
|
||||
LOG = logging.getLogger("APRSD")
|
||||
|
||||
|
||||
class PingPlugin(plugin.APRSDPluginBase):
|
||||
"""Ping."""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^[pP]"
|
||||
command_name = "ping"
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("PINGPlugin")
|
||||
stm = time.localtime()
|
||||
h = stm.tm_hour
|
||||
m = stm.tm_min
|
||||
s = stm.tm_sec
|
||||
reply = (
|
||||
"Pong! " + str(h).zfill(2) + ":" + str(m).zfill(2) + ":" + str(s).zfill(2)
|
||||
)
|
||||
return reply.rstrip()
|
||||
|
|
@ -0,0 +1,43 @@
|
|||
import logging
|
||||
import re
|
||||
|
||||
from aprsd import messaging, plugin
|
||||
|
||||
LOG = logging.getLogger("APRSD")
|
||||
|
||||
|
||||
class QueryPlugin(plugin.APRSDPluginBase):
|
||||
"""Query command."""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = r"^\?.*"
|
||||
command_name = "query"
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("Query COMMAND")
|
||||
|
||||
tracker = messaging.MsgTrack()
|
||||
reply = "Pending Messages ({})".format(len(tracker))
|
||||
|
||||
searchstring = "^" + self.config["ham"]["callsign"] + ".*"
|
||||
# only I can do admin commands
|
||||
if re.search(searchstring, fromcall):
|
||||
r = re.search(r"^\?-\*", message)
|
||||
if r is not None:
|
||||
if len(tracker) > 0:
|
||||
reply = "Resend ALL Delayed msgs"
|
||||
LOG.debug(reply)
|
||||
tracker.restart_delayed()
|
||||
else:
|
||||
reply = "No Delayed Msgs"
|
||||
LOG.debug(reply)
|
||||
return reply
|
||||
|
||||
r = re.search(r"^\?-[fF]!", message)
|
||||
if r is not None:
|
||||
reply = "Deleting ALL Delayed msgs."
|
||||
LOG.debug(reply)
|
||||
tracker.flush()
|
||||
return reply
|
||||
|
||||
return reply
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
import logging
|
||||
import time
|
||||
|
||||
from aprsd import fuzzyclock, plugin
|
||||
|
||||
LOG = logging.getLogger("APRSD")
|
||||
|
||||
|
||||
class TimePlugin(plugin.APRSDPluginBase):
|
||||
"""Time command."""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^[tT]"
|
||||
command_name = "time"
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("TIME COMMAND")
|
||||
stm = time.localtime()
|
||||
h = stm.tm_hour
|
||||
m = stm.tm_min
|
||||
cur_time = fuzzyclock.fuzzy(h, m, 1)
|
||||
reply = "{} ({}:{} PDT) ({})".format(
|
||||
cur_time,
|
||||
str(h),
|
||||
str(m).rjust(2, "0"),
|
||||
message.rstrip(),
|
||||
)
|
||||
return reply
|
||||
|
|
@ -0,0 +1,22 @@
|
|||
import logging
|
||||
|
||||
import aprsd
|
||||
from aprsd import plugin
|
||||
|
||||
LOG = logging.getLogger("APRSD")
|
||||
|
||||
|
||||
class VersionPlugin(plugin.APRSDPluginBase):
|
||||
"""Version of APRSD Plugin."""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^[vV]"
|
||||
command_name = "version"
|
||||
|
||||
# message_number:time combos so we don't resend the same email in
|
||||
# five mins {int:int}
|
||||
email_sent_dict = {}
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("Version COMMAND")
|
||||
return "APRSD version '{}'".format(aprsd.__version__)
|
||||
|
|
@ -0,0 +1,53 @@
|
|||
import json
|
||||
import logging
|
||||
|
||||
from aprsd import plugin
|
||||
import requests
|
||||
|
||||
LOG = logging.getLogger("APRSD")
|
||||
|
||||
|
||||
class WeatherPlugin(plugin.APRSDPluginBase):
|
||||
"""Weather Command"""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^[wW]"
|
||||
command_name = "weather"
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("Weather Plugin")
|
||||
try:
|
||||
url = (
|
||||
"http://api.aprs.fi/api/get?"
|
||||
"&what=loc&apikey=104070.f9lE8qg34L8MZF&format=json"
|
||||
"&name=%s" % fromcall
|
||||
)
|
||||
response = requests.get(url)
|
||||
# aprs_data = json.loads(response.read())
|
||||
aprs_data = json.loads(response.text)
|
||||
lat = aprs_data["entries"][0]["lat"]
|
||||
lon = aprs_data["entries"][0]["lng"]
|
||||
url2 = (
|
||||
"https://forecast.weather.gov/MapClick.php?lat=%s"
|
||||
"&lon=%s&FcstType=json" % (lat, lon)
|
||||
)
|
||||
response2 = requests.get(url2)
|
||||
# wx_data = json.loads(response2.read())
|
||||
wx_data = json.loads(response2.text)
|
||||
reply = (
|
||||
"%sF(%sF/%sF) %s. %s, %s."
|
||||
% (
|
||||
wx_data["currentobservation"]["Temp"],
|
||||
wx_data["data"]["temperature"][0],
|
||||
wx_data["data"]["temperature"][1],
|
||||
wx_data["data"]["weather"][0],
|
||||
wx_data["time"]["startPeriodName"][1],
|
||||
wx_data["data"]["weather"][1],
|
||||
)
|
||||
).rstrip()
|
||||
LOG.debug("reply: '{}' ".format(reply))
|
||||
except Exception as e:
|
||||
LOG.debug("Weather failed with: " + "%s" % str(e))
|
||||
reply = "Unable to find you (send beacon?)"
|
||||
|
||||
return reply
|
||||
|
|
@ -2,8 +2,11 @@ import unittest
|
|||
from unittest import mock
|
||||
|
||||
import aprsd
|
||||
from aprsd import plugin
|
||||
from aprsd.fuzzyclock import fuzzy
|
||||
from aprsd.plugins import fortune as fortune_plugin
|
||||
from aprsd.plugins import ping as ping_plugin
|
||||
from aprsd.plugins import time as time_plugin
|
||||
from aprsd.plugins import version as version_plugin
|
||||
|
||||
|
||||
class TestPlugin(unittest.TestCase):
|
||||
|
|
@ -14,17 +17,17 @@ class TestPlugin(unittest.TestCase):
|
|||
|
||||
@mock.patch("shutil.which")
|
||||
def test_fortune_fail(self, mock_which):
|
||||
fortune_plugin = plugin.FortunePlugin(self.config)
|
||||
fortune = fortune_plugin.FortunePlugin(self.config)
|
||||
mock_which.return_value = None
|
||||
message = "fortune"
|
||||
expected = "Fortune command not installed"
|
||||
actual = fortune_plugin.run(self.fromcall, message, self.ack)
|
||||
actual = fortune.run(self.fromcall, message, self.ack)
|
||||
self.assertEqual(expected, actual)
|
||||
|
||||
@mock.patch("subprocess.Popen")
|
||||
@mock.patch("shutil.which")
|
||||
def test_fortune_success(self, mock_which, mock_popen):
|
||||
fortune_plugin = plugin.FortunePlugin(self.config)
|
||||
fortune = fortune_plugin.FortunePlugin(self.config)
|
||||
mock_which.return_value = "/usr/bin/games"
|
||||
|
||||
mock_process = mock.MagicMock()
|
||||
|
|
@ -33,7 +36,7 @@ class TestPlugin(unittest.TestCase):
|
|||
|
||||
message = "fortune"
|
||||
expected = "Funny fortune"
|
||||
actual = fortune_plugin.run(self.fromcall, message, self.ack)
|
||||
actual = fortune.run(self.fromcall, message, self.ack)
|
||||
self.assertEqual(expected, actual)
|
||||
|
||||
@mock.patch("time.localtime")
|
||||
|
|
@ -43,13 +46,13 @@ class TestPlugin(unittest.TestCase):
|
|||
m = fake_time.tm_min = 12
|
||||
fake_time.tm_sec = 55
|
||||
mock_time.return_value = fake_time
|
||||
time_plugin = plugin.TimePlugin(self.config)
|
||||
time = time_plugin.TimePlugin(self.config)
|
||||
|
||||
fromcall = "KFART"
|
||||
message = "location"
|
||||
ack = 1
|
||||
|
||||
actual = time_plugin.run(fromcall, message, ack)
|
||||
actual = time.run(fromcall, message, ack)
|
||||
self.assertEqual(None, actual)
|
||||
|
||||
cur_time = fuzzy(h, m, 1)
|
||||
|
|
@ -61,7 +64,7 @@ class TestPlugin(unittest.TestCase):
|
|||
str(m).rjust(2, "0"),
|
||||
message.rstrip(),
|
||||
)
|
||||
actual = time_plugin.run(fromcall, message, ack)
|
||||
actual = time.run(fromcall, message, ack)
|
||||
self.assertEqual(expected, actual)
|
||||
|
||||
@mock.patch("time.localtime")
|
||||
|
|
@ -72,7 +75,7 @@ class TestPlugin(unittest.TestCase):
|
|||
s = fake_time.tm_sec = 55
|
||||
mock_time.return_value = fake_time
|
||||
|
||||
ping = plugin.PingPlugin(self.config)
|
||||
ping = ping_plugin.PingPlugin(self.config)
|
||||
|
||||
fromcall = "KFART"
|
||||
message = "location"
|
||||
|
|
@ -102,19 +105,19 @@ class TestPlugin(unittest.TestCase):
|
|||
|
||||
def test_version(self):
|
||||
expected = "APRSD version '{}'".format(aprsd.__version__)
|
||||
version_plugin = plugin.VersionPlugin(self.config)
|
||||
version = version_plugin.VersionPlugin(self.config)
|
||||
|
||||
fromcall = "KFART"
|
||||
message = "No"
|
||||
ack = 1
|
||||
|
||||
actual = version_plugin.run(fromcall, message, ack)
|
||||
actual = version.run(fromcall, message, ack)
|
||||
self.assertEqual(None, actual)
|
||||
|
||||
message = "version"
|
||||
actual = version_plugin.run(fromcall, message, ack)
|
||||
actual = version.run(fromcall, message, ack)
|
||||
self.assertEqual(expected, actual)
|
||||
|
||||
message = "Version"
|
||||
actual = version_plugin.run(fromcall, message, ack)
|
||||
actual = version.run(fromcall, message, ack)
|
||||
self.assertEqual(expected, actual)
|
||||
|
|
|
|||
Loading…
Reference in New Issue