front-end improvements

README.md

@@ -12,9 +12,14 @@ pip install packetraven
 #### Usage:
 to start the client, run the following:
 ```bash
-packetraven <serial_port> [log_file] [output_file]
+packetraven
 ```
-or start the graphical interface:
+for usage, do
+```bash
+packetraven -h
+```
+
+there is also a Tkinter interface:
 ```bash
 packetraven_gui
 ```

client/packetraven_cli.py

@@ -1,13 +1,7 @@
-"""
-Balloon telemetry parsing, display, and logging.
-
-__authors__ = ['Quinn Kupec', 'Zachary Burnett']
-"""
-
+import argparse
 from datetime import datetime
 import os
 from pathlib import Path
-import sys
 import time
 
 from packetraven import BALLOON_CALLSIGNS
@@ -16,94 +10,102 @@
 from packetraven.utilities import get_logger
 from packetraven.writer import write_aprs_packet_tracks
 
-INTERVAL_SECONDS = 5
-DESKTOP_PATH = Path('~').resolve() / 'Desktop'
-
 LOGGER = get_logger('packetraven')
 
+DEFAULT_INTERVAL_SECONDS = 5
+DESKTOP_PATH = Path('~').expanduser() / 'Desktop'
 
-def main():
- if len(sys.argv) > 0 and '-h' not in sys.argv:
- serial_port = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else None
- log_filename = Path(sys.argv[2]).resolve() if len(sys.argv) > 2 else DESKTOP_PATH
- output_filename = Path(sys.argv[3]).resolve() if len(sys.argv) > 3 else None
 
+def main():
+ parser = argparse.ArgumentParser()
+ parser.add_argument('-k', '--key', help='API key from https://aprs.fi/page/api')
+ parser.add_argument('-p', '--port', help='name of serial port connected to APRS packet radio')
+ parser.add_argument('-l', '--log', help='path to log file to save log messages')
+ parser.add_argument('-o', '--output', help='path to output file to save packets')
+ parser.add_argument('-t', '--interval', help='seconds between each main loop')
+ args = parser.parse_args()
+
+ serial_port = args.port
+ aprs_fi_api_key = args.key
+
+ if args.log is not None:
+ log_filename = Path(args.log).expanduser()
  if log_filename.is_dir():
  if not log_filename.exists():
  os.makedirs(log_filename, exist_ok=True)
  log_filename = log_filename / f'{datetime.now():%Y%m%dT%H%M%S}_packetraven_log.txt'
-
- if output_filename is not None:
- output_directory = output_filename.parent
- if not output_directory.exists():
- os.makedirs(output_directory, exist_ok=True)
-
  get_logger(LOGGER.name, log_filename)
 
- aprs_connections = []
- if serial_port is not None and 'txt' in serial_port:
- try:
- text_file = PacketTextFile(serial_port)
- aprs_connections.append(text_file)
- except Exception as error:
- LOGGER.exception(f'{error.__class__.__name__} - {error}')
- else:
- try:
- radio = PacketRadio(serial_port)
- serial_port = radio.serial_port
- LOGGER.info(f'opened port {serial_port}')
- aprs_connections.append(radio)
- except Exception as error:
- LOGGER.exception(f'{error.__class__.__name__} - {error}')
+ if args.output is not None:
+ output_filename = Path(args.output).expanduser()
+ output_directory = output_filename.parent
+ if not output_directory.exists():
+ os.makedirs(output_directory, exist_ok=True)
+ else:
+ output_filename = None
 
+ aprs_connections = []
+ if serial_port is not None and 'txt' in serial_port:
  try:
- aprs_api = APRS_fi(BALLOON_CALLSIGNS)
- LOGGER.info(f'established connection to API')
- aprs_connections.append(aprs_api)
- except Exception as error:
- LOGGER.exception(f'{error.__class__.__name__} - {error}')
-
- packet_tracks = {}
- while True:
- LOGGER.debug(f'receiving packets from {len(aprs_connections)} source(s)')
-
- parsed_packets = []
- for aprs_connection in aprs_connections:
- parsed_packets.extend(aprs_connection.packets)
-
- LOGGER.debug(f'received {len(parsed_packets)} packets')
-
- if len(parsed_packets) > 0:
- for parsed_packet in parsed_packets:
- callsign = parsed_packet['callsign']
-
- if callsign in packet_tracks:
- if parsed_packet not in packet_tracks[callsign]:
- packet_tracks[callsign].append(parsed_packet)
- else:
- LOGGER.debug(f'{callsign:8} - received duplicate packet: {parsed_packet}')
- continue
- else:
- packet_tracks[callsign] = APRSTrack(callsign, [parsed_packet])
- LOGGER.debug(f'{callsign:8} - started tracking')
-
- LOGGER.info(f'{callsign:8} - received new packet: {parsed_packet}')
-
- if 'longitude' in parsed_packet and 'latitude' in parsed_packet:
- ascent_rate = packet_tracks[callsign].ascent_rate[-1]
- ground_speed = packet_tracks[callsign].ground_speed[-1]
- seconds_to_impact = packet_tracks[callsign].seconds_to_impact
- LOGGER.info(f'{callsign:8} - Ascent rate (m/s): {ascent_rate}')
- LOGGER.info(f'{callsign:8} - Ground speed (m/s): {ground_speed}')
- if seconds_to_impact >= 0:
- LOGGER.info(f'{callsign:8} - Estimated time until landing (s): {seconds_to_impact}')
-
- if output_filename is not None:
- write_aprs_packet_tracks(packet_tracks.values(), output_filename)
-
- time.sleep(INTERVAL_SECONDS)
+ text_file = PacketTextFile(serial_port)
+ aprs_connections.append(text_file)
+ except ConnectionError as error:
+ LOGGER.warning(f'{error.__class__.__name__} - {error}')
  else:
- print('usage: packetraven serial_port [log_path] [output_file]')
+ try:
+ radio = PacketRadio(serial_port)
+ serial_port = radio.location
+ LOGGER.info(f'opened port {serial_port}')
+ aprs_connections.append(radio)
+ except ConnectionError as error:
+ LOGGER.warning(f'{error.__class__.__name__} - {error}')
+
+ try:
+ aprs_api = APRS_fi(BALLOON_CALLSIGNS, api_key=aprs_fi_api_key)
+ LOGGER.info(f'established connection to {aprs_api.location}')
+ aprs_connections.append(aprs_api)
+ except ConnectionError as error:
+ LOGGER.warning(f'{error.__class__.__name__} - {error}')
+
+ packet_tracks = {}
+ while len(aprs_connections) > 0:
+ LOGGER.debug(f'receiving packets from {len(aprs_connections)} source(s)')
+
+ parsed_packets = []
+ for aprs_connection in aprs_connections:
+ parsed_packets.extend(aprs_connection.packets)
+
+ LOGGER.debug(f'received {len(parsed_packets)} packets')
+
+ if len(parsed_packets) > 0:
+ for parsed_packet in parsed_packets:
+ callsign = parsed_packet['callsign']
+
+ if callsign in packet_tracks:
+ if parsed_packet not in packet_tracks[callsign]:
+ packet_tracks[callsign].append(parsed_packet)
+ else:
+ LOGGER.debug(f'{callsign:8} - received duplicate packet: {parsed_packet}')
+ continue
+ else:
+ packet_tracks[callsign] = APRSTrack(callsign, [parsed_packet])
+ LOGGER.debug(f'{callsign:8} - started tracking')
+
+ LOGGER.info(f'{callsign:8} - received new packet: {parsed_packet}')
+
+ if 'longitude' in parsed_packet and 'latitude' in parsed_packet:
+ ascent_rate = packet_tracks[callsign].ascent_rate[-1]
+ ground_speed = packet_tracks[callsign].ground_speed[-1]
+ seconds_to_impact = packet_tracks[callsign].seconds_to_impact
+ LOGGER.info(f'{callsign:8} - Ascent rate (m/s): {ascent_rate}')
+ LOGGER.info(f'{callsign:8} - Ground speed (m/s): {ground_speed}')
+ if seconds_to_impact >= 0:
+ LOGGER.info(f'{callsign:8} - Estimated time until landing (s): {seconds_to_impact}')
+
+ if output_filename is not None:
+ write_aprs_packet_tracks(packet_tracks.values(), output_filename)
+
+ time.sleep(DEFAULT_INTERVAL_SECONDS)
 
 
 if __name__ == '__main__':

client/packetraven_gui.py

@@ -2,18 +2,18 @@
 import logging
 from pathlib import Path
 import tkinter
-from tkinter import filedialog, messagebox
+from tkinter import filedialog, messagebox, simpledialog
 
 from packetraven import BALLOON_CALLSIGNS
 from packetraven.connections import APRS_fi, PacketRadio, PacketTextFile, next_available_port
 from packetraven.tracks import APRSTrack
 from packetraven.utilities import get_logger
 from packetraven.writer import write_aprs_packet_tracks
 
-INTERVAL_SECONDS = 5
-DESKTOP_PATH = Path('~').resolve() / 'Desktop'
+LOGGER = get_logger('packetraven_gui')
 
-LOGGER = get_logger('packetraven')
+DEFAULT_INTERVAL_SECONDS = 5
+DESKTOP_PATH = Path('~').expanduser() / 'Desktop'
 
 
 class PacketRavenGUI:
@@ -129,7 +129,7 @@ def toggle(self):
  connection.close()
 
  if type(connection) is PacketRadio:
- LOGGER.info(f'closing port {connection.serial_port}')
+ LOGGER.info(f'closing port {connection.location}')
 
  for element in self.frames['bottom'].winfo_children():
  element.configure(state=tkinter.DISABLED)
@@ -167,14 +167,16 @@ def toggle(self):
  else:
  try:
  radio = PacketRadio(self.serial_port)
- self.serial_port = radio.serial_port
+ self.serial_port = radio.location
  LOGGER.info(f'opened port {self.serial_port}')
  self.connections.append(radio)
  except Exception as error:
  LOGGER.exception(f'{error.__class__.__name__} - {error}')
 
+ aprs_fi_api_key = simpledialog.askstring('APRS.fi API Key', 'enter API key for https://aprs.fi', parent=self.main_window)
+
  try:
- aprs_api = APRS_fi(BALLOON_CALLSIGNS)
+ aprs_api = APRS_fi(BALLOON_CALLSIGNS, api_key=aprs_fi_api_key)
  LOGGER.info(f'established connection to API')
  self.connections.append(aprs_api)
  except Exception as error:
@@ -244,7 +246,7 @@ def run(self):
  write_aprs_packet_tracks(self.packet_tracks.values(), output_filename)
 
  if self.active:
- self.main_window.after(INTERVAL_SECONDS * 1000, self.run)
+ self.main_window.after(DEFAULT_INTERVAL_SECONDS * 1000, self.run)
 
  @staticmethod
  def replace_text(element, value):

packetraven/connections.py

@@ -18,13 +18,13 @@
 
 from packetraven.database import DatabaseTable
 from packetraven.packets import APRSLocationPacket, LocationPacket
-from packetraven.utilities import CREDENTIALS_FILENAME, get_logger
+from packetraven.utilities import CREDENTIALS_FILENAME, get_logger, read_configuration
 
 LOGGER = get_logger('packetraven.connection')
 
 
 class PacketConnection(ABC):
- address: str
+ location: str
 
  @property
  @abstractmethod
@@ -60,13 +60,13 @@ def __init__(self, serial_port: str = None):
 
  if serial_port is None:
  try:
- self.serial_port = next_available_port()
+ self.location = next_available_port()
  except ConnectionError:
  raise ConnectionError('could not find radio over serial connection')
  else:
- self.serial_port = serial_port.strip('"')
+ self.location = serial_port.strip('"')
 
- self.connection = Serial(self.serial_port, baudrate=9600, timeout=1)
+ self.connection = Serial(self.location, baudrate=9600, timeout=1)
 
  @property
  def packets(self) -> [APRSLocationPacket]:
@@ -82,7 +82,7 @@ def close(self):
  self.connection.close()
 
  def __repr__(self):
- return f'{self.__class__.__name__}("{self.serial_port}")'
+ return f'{self.__class__.__name__}("{self.location}")'
 
 
 class PacketTextFile(PacketConnection):
@@ -98,10 +98,10 @@ def __init__(self, filename: PathLike = None):
  filename = filename.strip('"')
  filename = Path(filename)
 
- self.filename = filename
+ self.location = filename
 
  # open text file
- self.connection = open(self.filename)
+ self.connection = open(self.location)
 
  @property
  def packets(self) -> [APRSLocationPacket]:
@@ -118,11 +118,11 @@ def close(self):
  self.connection.close()
 
  def __repr__(self):
- return f'{self.__class__.__name__}("{self.filename}")'
+ return f'{self.__class__.__name__}("{self.location}")'
 
 
 class APRS_fi(PacketConnection):
- api_url = 'https://api.aprs.fi/api/get'
+ location = 'https://api.aprs.fi/api/get'
 
  def __init__(self, callsigns: [str], api_key: str = None):
  """
@@ -132,15 +132,19 @@ def __init__(self, callsigns: [str], api_key: str = None):
  :param api_key: API key for aprs.fi
  """
 
- if api_key is None:
- with open(CREDENTIALS_FILENAME) as secret_file:
- api_key = secret_file.read().strip()
+ if api_key is None or api_key == '':
+ configuration = read_configuration(CREDENTIALS_FILENAME)
+
+ if 'APRS_FI' in configuration:
+ api_key = configuration['APRS_FI']['api_key']
+ else:
+ raise ConnectionError(f'no APRS.fi API key specified')
 
- self.callsigns = callsigns
  self.api_key = api_key
+ self.callsigns = callsigns
 
  if not self.has_network_connection():
- raise ConnectionError(f'No network connection.')
+ raise ConnectionError(f'no network connection')
 
  @property
  def packets(self) -> [APRSLocationPacket]:
@@ -153,7 +157,7 @@ def packets(self) -> [APRSLocationPacket]:
 
  query = '&'.join(f'{key}={value}' for key, value in query.items())
 
- response = requests.get(f'{self.api_url}?{query}').json()
+ response = requests.get(f'{self.location}?{query}').json()
  if response['result'] != 'fail':
  packets = [parse_packet(packet_candidate) for packet_candidate in response['entries']]
  else:
@@ -170,7 +174,7 @@ def has_network_connection(self) -> bool:
  """
 
  try:
- requests.get(self.api_url, timeout=2)
+ requests.get(self.location, timeout=2)
  return True
  except ConnectionError:
  return False
@@ -204,6 +208,7 @@ def __init__(self, hostname: str, database: str, table: str, fields: {str: type}
  fields = {}
  fields = {**self.__default_fields, **fields}
  super().__init__(hostname, database, table, fields, 'time', **kwargs)
+ self.location = f'{self.hostname}:{self.port}/{self.database}/{self.table}'
 
  @property
  def packets(self) -> [LocationPacket]:

packetraven/packets.py

@@ -1,9 +1,3 @@
-"""
-APRS packet class for packet operations.
-
-__authors__ = ['Quinn Kupec', 'Zachary Burnett']
-"""
-
 from datetime import datetime, timedelta
 import math
 from typing import Union