new updated location plugin

This commit is contained in:
Hemna 2021-06-17 13:25:00 -04:00
parent 488b83a157
commit fc15ffd6eb
2 changed files with 219 additions and 0 deletions

View File

@ -0,0 +1,73 @@
import logging
from aprsd import plugin
from aprsd import utils as aprsd_utils
from slack_sdk import WebClient
import aprsd_slack_plugin
LOG = logging.getLogger("APRSD")
class SlackPluginBase(plugin.APRSDPluginBase):
"""SlackCommandPlugin.
This APRSD plugin looks for the location command comming in
to aprsd, then fetches the caller's location, and then reports
that location string to the configured slack channel.
To use this:
Create a slack bot for your workspace at api.slack.com.
A good source of information on how to create the app
and the tokens and permissions and install the app in your
workspace is here:
https://api.slack.com/start/building/bolt-python
You will need the signing secret from the
Basic Information -> App Credentials form.
You will also need the Bot User OAuth Access Token from
OAuth & Permissions -> OAuth Tokens for Your Team ->
Bot User OAuth Access Token.
Install the app/bot into your workspace.
Edit your ~/.config/aprsd/aprsd.yml and add the section
slack:
signing_secret: <signing secret token here>
bot_token: <Bot User OAuth Access Token here>
channel: <channel name here>
"""
version = aprsd_slack_plugin.__version__
def setup_slack(self):
"""Create the slack require client from config."""
# signing_secret = self.config["slack"]["signing_secret"]
try:
aprsd_utils.check_config_option(self.config, ["services", "slack", "bot_token"])
except Exception as ex:
LOG.error("Failed to find config slack:bot_token {}".format(ex))
return "No slack bot_token found"
bot_token = self.config["services"]["slack"]["bot_token"]
if not bot_token:
LOG.error(
"APRSD config is missing slack: bot_token:<token>. "
"Please install the slack app and get the "
"Bot User OAth Access Token.",
)
return False
self.swc = WebClient(token=bot_token)
self.slack_channels = self.config["services"]["slack"].get("channels", None)
if not self.slack_channels:
LOG.error(
"APRSD config is missing slack: channels: <name> "
"Please add a slack channel name to send messages.",
)
return False
return True

View File

@ -0,0 +1,146 @@
import logging
import re
import time
from aprsd import messaging, plugin_utils
from aprsd import utils as aprsd_utils
import aprsd_slack_plugin
from aprsd_slack_plugin import base_plugin
LOG = logging.getLogger("APRSD")
class SlackLocationPlugin(base_plugin.SlackPluginBase):
"""SlackCommandPlugin.
This APRSD plugin looks for the location command comming in
to aprsd, then fetches the caller's location, and then reports
that location string to the configured slack channel.
To use this:
Create a slack bot for your workspace at api.slack.com.
A good source of information on how to create the app
and the tokens and permissions and install the app in your
workspace is here:
https://api.slack.com/start/building/bolt-python
You will need the signing secret from the
Basic Information -> App Credentials form.
You will also need the Bot User OAuth Access Token from
OAuth & Permissions -> OAuth Tokens for Your Team ->
Bot User OAuth Access Token.
Install the app/bot into your workspace.
Edit your ~/.config/aprsd/aprsd.yml and add the section
slack:
signing_secret: <signing secret token here>
bot_token: <Bot User OAuth Access Token here>
channel: <channel name here>
"""
version = aprsd_slack_plugin.__version__
# matches any string starting with h or H
command_regex = "^[lL]"
command_name = "location-slack"
def command(self, fromcall, message, ack):
LOG.info("SlackCommandPlugin")
is_setup = self.setup_slack()
if not is_setup:
return
# get last location of a callsign, get descriptive name from weather service
try:
aprsd_utils.check_config_option(self.config, ["services", "aprs.fi", "apiKey"])
except Exception as ex:
LOG.error("Failed to find config aprs.fi:apikey {}".format(ex))
return "No aprs.fi apikey found"
api_key = self.config["services"]["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("Failed to fetch aprs.fi '{}'".format(ex))
return "Failed to fetch aprs.fi location"
LOG.debug("LocationPlugin: aprs_data = {}".format(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 = aprs_data["entries"][0]["lat"]
lon = aprs_data["entries"][0]["lng"]
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
callsign_url = "<http://aprs.fi/info/a/{}|{}>".format(searchcall, searchcall)
aprs_url = "<http://aprs.fi/#!mt=roadmap&z=15&lat={}&lng={}|" " http://aprs.fi/>".format(
lat,
lon,
)
message = {}
message["username"] = "APRSD - Slack Location Plugin"
message["icon_emoji"] = ":satellite_antenna:"
message["attachments"] = [{}]
message["text"] = "{} - Location".format(callsign_url)
message["channel"] = "#random"
attachment = message["attachments"][0]
attachment["fallback"] = message["text"]
attachment["fields"] = []
attachment["fields"].append(
{"title": "Location", "value": aprs_url, "short": True},
)
attachment["fields"].append(
{
"title": "Altitude",
"value": altfeet,
"short": True,
"fallback": "Altitude - {}".format(altfeet),
},
)
attachment["fields"].append(
{
"title": "Time",
"value": "{} h ago".format(round(delta_hours, 1)),
"short": True,
"fallback": "Time {} h ago".format(round(delta_hours, 1)),
},
)
LOG.debug(message)
# self.swc.chat_postMessage(**message)
for channel in self.slack_channels:
message["channel"] = channel
self.swc.chat_postMessage(**message)
# Don't have aprsd try and send a reply
return messaging.NULL_MESSAGE