import logging import re import time from aprsd import packets, plugin, plugin_utils from oslo_config import cfg import aprsd_slack_plugin from aprsd_slack_plugin import base_plugin CONF = cfg.CONF LOG = logging.getLogger("APRSD") class SlackLocationPlugin( base_plugin.SlackPluginBase, plugin.APRSDRegexCommandPluginBase, plugin.APRSFIKEYMixin, ): """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.conf and add the section slack: signing_secret: bot_token: channel: """ version = aprsd_slack_plugin.__version__ # matches any string starting with h or H command_regex = "^[lL]" command_name = "location-slack" def setup(self): self.ensure_aprs_fi_key() if self.enabled: config_set = self.setup_slack() if not config_set: self.enabled = False def process(self, packet): LOG.info("SlackCommandPlugin") fromcall = packet.from_call message = packet.message_text # get last location of a callsign, get descriptive name from weather service 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 = 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"] delta_seconds = time.time() - int(aprs_lasttime_seconds) delta_hours = delta_seconds / 60 / 60 wx_data = None try: wx_data = plugin_utils.get_weather_gov_for_gps(lat, lon) except Exception: LOG.warning("Couldn't fetch forecast.weather.gov") callsign_url = f"" aprs_url = "".format( lat, lon, ) message = {} message["username"] = "APRSD - Slack Location Plugin" message["icon_emoji"] = ":satellite_antenna:" message["attachments"] = [{}] message["text"] = f"{callsign_url} - Location" message["channel"] = "#random" attachment = message["attachments"][0] attachment["fallback"] = message["text"] attachment["fields"] = [] # if the coordinates are outside of the US, we don't get this # aread description if wx_data and "location" in wx_data and "areaDescription" in wx_data["location"]: attachment["fields"].append( { "title": "Location", "value": wx_data["location"]["areaDescription"], "short": True, }, ) attachment["fields"].append( {"title": "Map Location", "value": aprs_url, "short": True}, ) attachment["fields"].append( { "title": "Altitude", "value": altfeet, "short": True, "fallback": f"Altitude - {altfeet}", }, ) attachment["fields"].append( { "title": "Time", "value": f"{round(delta_hours, 1)} h ago", "short": True, "fallback": f"Time {round(delta_hours, 1)} h ago", }, ) 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 packets.NULL_MESSAGE