mirror of
https://github.com/craigerl/aprsd.git
synced 2025-04-09 21:18:37 -04:00
Added aprsd-dev plugin test cli and WxPlugin
This patch adds a new CLI app called aprsd-dev. arpsd-dev is used specifically for developing plugins. It allows you to run a plugin directly without the need to run aprsd server. This patch also adds the Weather Metar plugin called WxPlugin. You can use it to fetch METAR from the nearest station for a callsign or from a known METAR station id. Call WxPlugin with a message of 'wx' for closest metar station or 'wx KAUN' for metar at KAUN wx station
This commit is contained in:
parent
658e6b6202
commit
cc8fd178ce
@ -15,7 +15,7 @@ cd aprsd
|
||||
pip install -e .
|
||||
cd ~/.venv_aprsd/bin
|
||||
|
||||
./aprsd sample-config # generates a config.yml template
|
||||
./aprsd sample-config # generates a config.yml template
|
||||
|
||||
vi ~/.config/aprsd/config.yml # copy/edit config here
|
||||
|
||||
|
||||
197
aprsd/dev.py
Normal file
197
aprsd/dev.py
Normal file
@ -0,0 +1,197 @@
|
||||
#
|
||||
# Dev.py is used to help develop plugins
|
||||
#
|
||||
#
|
||||
# python included libs
|
||||
import logging
|
||||
from logging import NullHandler
|
||||
from logging.handlers import RotatingFileHandler
|
||||
import os
|
||||
import sys
|
||||
|
||||
# local imports here
|
||||
import aprsd
|
||||
from aprsd import client, email, plugin, utils
|
||||
import click
|
||||
import click_completion
|
||||
|
||||
# setup the global logger
|
||||
# logging.basicConfig(level=logging.DEBUG) # level=10
|
||||
LOG = logging.getLogger("APRSD")
|
||||
|
||||
LOG_LEVELS = {
|
||||
"CRITICAL": logging.CRITICAL,
|
||||
"ERROR": logging.ERROR,
|
||||
"WARNING": logging.WARNING,
|
||||
"INFO": logging.INFO,
|
||||
"DEBUG": logging.DEBUG,
|
||||
}
|
||||
|
||||
CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
|
||||
|
||||
|
||||
def custom_startswith(string, incomplete):
|
||||
"""A custom completion match that supports case insensitive matching."""
|
||||
if os.environ.get("_CLICK_COMPLETION_COMMAND_CASE_INSENSITIVE_COMPLETE"):
|
||||
string = string.lower()
|
||||
incomplete = incomplete.lower()
|
||||
return string.startswith(incomplete)
|
||||
|
||||
|
||||
click_completion.core.startswith = custom_startswith
|
||||
click_completion.init()
|
||||
|
||||
|
||||
cmd_help = """Shell completion for click-completion-command
|
||||
Available shell types:
|
||||
\b
|
||||
%s
|
||||
Default type: auto
|
||||
""" % "\n ".join(
|
||||
"{:<12} {}".format(k, click_completion.core.shells[k])
|
||||
for k in sorted(click_completion.core.shells.keys())
|
||||
)
|
||||
|
||||
|
||||
@click.group(help=cmd_help, context_settings=CONTEXT_SETTINGS)
|
||||
@click.version_option()
|
||||
def main():
|
||||
pass
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.option(
|
||||
"-i",
|
||||
"--case-insensitive/--no-case-insensitive",
|
||||
help="Case insensitive completion",
|
||||
)
|
||||
@click.argument(
|
||||
"shell",
|
||||
required=False,
|
||||
type=click_completion.DocumentedChoice(click_completion.core.shells),
|
||||
)
|
||||
def show(shell, case_insensitive):
|
||||
"""Show the click-completion-command completion code"""
|
||||
extra_env = (
|
||||
{"_CLICK_COMPLETION_COMMAND_CASE_INSENSITIVE_COMPLETE": "ON"}
|
||||
if case_insensitive
|
||||
else {}
|
||||
)
|
||||
click.echo(click_completion.core.get_code(shell, extra_env=extra_env))
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.option(
|
||||
"--append/--overwrite",
|
||||
help="Append the completion code to the file",
|
||||
default=None,
|
||||
)
|
||||
@click.option(
|
||||
"-i",
|
||||
"--case-insensitive/--no-case-insensitive",
|
||||
help="Case insensitive completion",
|
||||
)
|
||||
@click.argument(
|
||||
"shell",
|
||||
required=False,
|
||||
type=click_completion.DocumentedChoice(click_completion.core.shells),
|
||||
)
|
||||
@click.argument("path", required=False)
|
||||
def install(append, case_insensitive, shell, path):
|
||||
"""Install the click-completion-command completion"""
|
||||
extra_env = (
|
||||
{"_CLICK_COMPLETION_COMMAND_CASE_INSENSITIVE_COMPLETE": "ON"}
|
||||
if case_insensitive
|
||||
else {}
|
||||
)
|
||||
shell, path = click_completion.core.install(
|
||||
shell=shell,
|
||||
path=path,
|
||||
append=append,
|
||||
extra_env=extra_env,
|
||||
)
|
||||
click.echo("{} completion installed in {}".format(shell, path))
|
||||
|
||||
|
||||
# Setup the logging faciility
|
||||
# to disable logging to stdout, but still log to file
|
||||
# use the --quiet option on the cmdln
|
||||
def setup_logging(config, loglevel, quiet):
|
||||
log_level = LOG_LEVELS[loglevel]
|
||||
LOG.setLevel(log_level)
|
||||
log_format = "[%(asctime)s] [%(threadName)-12s] [%(levelname)-5.5s]" " %(message)s"
|
||||
date_format = "%m/%d/%Y %I:%M:%S %p"
|
||||
log_formatter = logging.Formatter(fmt=log_format, datefmt=date_format)
|
||||
log_file = config["aprs"].get("logfile", None)
|
||||
if log_file:
|
||||
fh = RotatingFileHandler(log_file, maxBytes=(10248576 * 5), backupCount=4)
|
||||
else:
|
||||
fh = NullHandler()
|
||||
|
||||
fh.setFormatter(log_formatter)
|
||||
LOG.addHandler(fh)
|
||||
|
||||
if not quiet:
|
||||
sh = logging.StreamHandler(sys.stdout)
|
||||
sh.setFormatter(log_formatter)
|
||||
LOG.addHandler(sh)
|
||||
|
||||
|
||||
@main.command()
|
||||
@click.option(
|
||||
"--loglevel",
|
||||
default="DEBUG",
|
||||
show_default=True,
|
||||
type=click.Choice(
|
||||
["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"],
|
||||
case_sensitive=False,
|
||||
),
|
||||
show_choices=True,
|
||||
help="The log level to use for aprsd.log",
|
||||
)
|
||||
@click.option(
|
||||
"-c",
|
||||
"--config",
|
||||
"config_file",
|
||||
show_default=True,
|
||||
default=utils.DEFAULT_CONFIG_FILE,
|
||||
help="The aprsd config file to use for options.",
|
||||
)
|
||||
@click.option(
|
||||
"-p",
|
||||
"--plugin",
|
||||
"plugin_path",
|
||||
show_default=True,
|
||||
default="aprsd.plugins.wx.WxPlugin",
|
||||
help="The plugin to run",
|
||||
)
|
||||
@click.argument("fromcall")
|
||||
@click.argument("message", nargs=-1, required=True)
|
||||
def test_plugin(
|
||||
loglevel,
|
||||
config_file,
|
||||
plugin_path,
|
||||
fromcall,
|
||||
message,
|
||||
):
|
||||
"""APRSD Plugin test app."""
|
||||
|
||||
config = utils.parse_config(config_file)
|
||||
email.CONFIG = config
|
||||
|
||||
setup_logging(config, loglevel, False)
|
||||
LOG.info("Test APRSD PLugin version: {}".format(aprsd.__version__))
|
||||
if type(message) is tuple:
|
||||
message = " ".join(message)
|
||||
LOG.info("P'{}' F'{}' C'{}'".format(plugin_path, fromcall, message))
|
||||
client.Client(config)
|
||||
|
||||
pm = plugin.PluginManager(config)
|
||||
obj = pm._create_class(plugin_path, plugin.APRSDPluginBase, config=config)
|
||||
|
||||
reply = obj.run(fromcall, message, 1)
|
||||
LOG.info("Result = '{}'".format(reply))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@ -24,6 +24,7 @@ CORE_PLUGINS = [
|
||||
"aprsd.plugins.query.QueryPlugin",
|
||||
"aprsd.plugins.time.TimePlugin",
|
||||
"aprsd.plugins.weather.WeatherPlugin",
|
||||
"aprsd.plugins.weather.WxPlugin",
|
||||
"aprsd.plugins.version.VersionPlugin",
|
||||
]
|
||||
|
||||
|
||||
@ -1,5 +1,6 @@
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
|
||||
from aprsd import plugin
|
||||
import requests
|
||||
@ -52,3 +53,107 @@ class WeatherPlugin(plugin.APRSDPluginBase):
|
||||
reply = "Unable to find you (send beacon?)"
|
||||
|
||||
return reply
|
||||
|
||||
|
||||
class WxPlugin(plugin.APRSDPluginBase):
|
||||
"""METAR Command"""
|
||||
|
||||
version = "1.0"
|
||||
command_regex = "^[wx]"
|
||||
command_name = "wx (Metar)"
|
||||
|
||||
def get_aprs(self, fromcall):
|
||||
LOG.debug("Fetch aprs.fi location for '{}'".format(fromcall))
|
||||
api_key = self.config["aprs.fi"]["apiKey"]
|
||||
try:
|
||||
url = (
|
||||
"http://api.aprs.fi/api/get?"
|
||||
"&what=loc&apikey={}&format=json"
|
||||
"&name={}".format(api_key, fromcall)
|
||||
)
|
||||
response = requests.get(url)
|
||||
except Exception:
|
||||
raise Exception("Failed to get aprs.fi location")
|
||||
else:
|
||||
response.raise_for_status()
|
||||
return response
|
||||
|
||||
def get_station(self, lat, lon):
|
||||
LOG.debug("Fetch station at {}, {}".format(lat, lon))
|
||||
try:
|
||||
url2 = (
|
||||
"https://forecast.weather.gov/MapClick.php?lat=%s"
|
||||
"&lon=%s&FcstType=json" % (lat, lon)
|
||||
)
|
||||
response = requests.get(url2)
|
||||
except Exception:
|
||||
raise Exception("Failed to get metar station")
|
||||
else:
|
||||
response.raise_for_status()
|
||||
return response
|
||||
|
||||
def get_metar(self, station):
|
||||
LOG.debug("Fetch metar for station '{}'".format(station))
|
||||
try:
|
||||
url = "https://api.weather.gov/stations/{}/observations/latest".format(
|
||||
station,
|
||||
)
|
||||
response = requests.get(url)
|
||||
except Exception:
|
||||
raise Exception("Failed to fetch metar")
|
||||
else:
|
||||
response.raise_for_status()
|
||||
return response
|
||||
|
||||
def command(self, fromcall, message, ack):
|
||||
LOG.info("WX Plugin '{}'".format(message))
|
||||
a = re.search(r"^.*\s+(.*)", message)
|
||||
if a is not None:
|
||||
searchcall = a.group(1)
|
||||
station = searchcall.upper()
|
||||
try:
|
||||
resp = self.get_metar(station)
|
||||
except Exception as e:
|
||||
LOG.debug("Weather failed with: {}".format(str(e)))
|
||||
reply = "Unable to find station METAR"
|
||||
else:
|
||||
station_data = json.loads(resp.text)
|
||||
reply = station_data["properties"]["rawMessage"]
|
||||
|
||||
return reply
|
||||
else:
|
||||
# if no second argument, search for calling station
|
||||
fromcall = fromcall
|
||||
try:
|
||||
resp = self.get_aprs(fromcall)
|
||||
except Exception as e:
|
||||
LOG.debug("Weather failed with: {}".format(str(e)))
|
||||
reply = "Unable to find you (send beacon?)"
|
||||
else:
|
||||
aprs_data = json.loads(resp.text)
|
||||
lat = aprs_data["entries"][0]["lat"]
|
||||
lon = aprs_data["entries"][0]["lng"]
|
||||
|
||||
try:
|
||||
resp = self.get_station(lat, lon)
|
||||
except Exception as e:
|
||||
LOG.debug("Weather failed with: {}".format(str(e)))
|
||||
reply = "Unable to find you (send beacon?)"
|
||||
else:
|
||||
wx_data = json.loads(resp.text)
|
||||
|
||||
if wx_data["location"]["metar"]:
|
||||
station = wx_data["location"]["metar"]
|
||||
try:
|
||||
resp = self.get_metar(station)
|
||||
except Exception as e:
|
||||
LOG.debug("Weather failed with: {}".format(str(e)))
|
||||
reply = "Failed to get Metar"
|
||||
else:
|
||||
station_data = json.loads(resp.text)
|
||||
reply = station_data["properties"]["rawMessage"]
|
||||
else:
|
||||
# Couldn't find a station
|
||||
reply = "No Metar station found"
|
||||
|
||||
return reply
|
||||
|
||||
Loading…
Reference in New Issue
Block a user