direwatch.py

#!/usr/bin/python3

# direwatch

"""
Craig Lamparter KM6LYW,  2025, MIT Licnese

Code derived from Adafruit PIL python example

Usage:
  direwatch.py --log="/run/direwolf.log" --fontsize="30" --title_text="DigiPi"  -o
  (-o option displays one station at a time, full screen)

This will tail a direwolf log file and display callsigns on an
adafruit ST7789, ILI9341, ILI9486 tft display 

  https://www.adafruit.com/product/4484  
  https://www.adafruit.com/product/2423
  https://www.amazon.com/Resistive-Screen-IPS-Resolution-Controller/dp/B07V9WW96D

More information on the ST7789 display is here:

https://learn.adafruit.com/adafruit-mini-pitft-135x240-color-tft-add-on-for-raspberry-pi/python-setup

One of three display types should be specified on the command line
   st7789
   ili9341
   ili9486

SPI interface must be enabled in /boot/firmware/config.txt or in raspi-config

Do not install the kernel module/framebuffer.

GPIO pins 12 (PTT) and 16 (DCD) are monitored and light green/red icons respectively.
Configure these gpio pins in direwolf.

Much code taken from ladyada for her great work driving these devices,

Included ILI9486 python library taken from https://github.com/SirLefti/Python_ILI9486

# SPDX-FileCopyrightText: 2021 ladyada for Adafruit Industries
# SPDX-License-Identifier: MIT

"""

import argparse
import time
import subprocess
import digitalio
import board
from PIL import Image, ImageDraw, ImageFont
import re
import adafruit_rgb_display.st7789 as st7789  
import adafruit_rgb_display.ili9341 as ili9341 
#import ILI9486_gpiod as ili9486
import ILI9486 as ili9486
import pyinotify
import gpiod
import threading
import signal
import os
import aprslib
import math
import numpy

# Config for display baudrate (default max is 24mhz):
BAUDRATE = 64000000

def parse_arguments():
    ap = argparse.ArgumentParser()
    ap.add_argument("-l", "--log", required=True, help="Direwolf log file location")
    ap.add_argument("-f", "--fontsize", required=False, help="Font size for callsigns")
    ap.add_argument("-t", "--title_text", required=False, help="Text displayed in title bar")
    ap.add_argument("-o", "--one", action='store_true', required=False, help="Show one station at a time full screen")
    ap.add_argument("-y", "--lat", required=False, help="Your Latitude -123.4567")
    ap.add_argument("-x", "--lon", required=False, help="Your Longitude  23.4567")
    ap.add_argument("-s", "--savefile", required=False, help="Save screen updates to png file")
    ap.add_argument("-d", "--display", required=False, help="st7789, ili9341, or ili9486")
    args = vars(ap.parse_args())
    return args

args = parse_arguments()
logfile = args["log"]
if args["fontsize"]:
   # 30 puts 6 lines
   # 33 puts 5 lines, max width
   fontsize = int(args["fontsize"])
else:
   fontsize = 30   # default 30
if args["title_text"]:
   title_text = args["title_text"]
else:
   title_text = "Direwatch"

if args["lat"]:
   lat1 = float(args["lat"])
else:
   lat1 = 0

if args["lon"]:
   lon1 = float(args["lon"])
else:
   lon1 = 0

if args["savefile"]:
   savefile = args["savefile"]
else:
   savefile = None

if args["display"]:
   displaytype = args["display"]
else:
   displaytype = st7789

if displaytype == 'ili9341':
   ## Large wide-screen adafruit screen 2.8" (320x240), two buttons
   spi = board.SPI()
   dc_pin = digitalio.DigitalInOut(board.D25)
   cs_pin = digitalio.DigitalInOut(board.D4)
   disp = ili9341.ILI9341(
       spi,
       cs=cs_pin,
       dc=dc_pin,
       baudrate=BAUDRATE,
       width=240,
       height=320,
       rotation=270
   )
elif displaytype == 'ili9486':
   ## Largest ili9486 display, no buttons
   from spidev import SpiDev
   spi = SpiDev(0,0)
   spi.mode = 0b10  
   spi.max_speed_hz = 48000000
   disp = ili9486.ILI9486(
       spi=spi,
       rst=25,
       dc=24,
       origin=ili9486.Origin.LOWER_RIGHT
   ).begin()
   disp.invert()
else:
   ## square adafruit screen 1.3" (240x240), two buttons
   dc_pin = digitalio.DigitalInOut(board.D25)
   cs_pin = digitalio.DigitalInOut(board.D4)
   spi = board.SPI()
   disp = st7789.ST7789(
       spi,
       cs=cs_pin,
       dc=dc_pin,
       baudrate=BAUDRATE,
       height=240,
       width=240,
       y_offset=80,
       rotation=180
   )

# don't write to display concurrently with thread
display_lock = threading.Lock()

# Create image and drawing object
if displaytype == 'st7789' or displaytype == 'ili9341':
   if disp.rotation % 180 == 90:
       height = disp.width    
       width = disp.height
   else:
       width = disp.width     
       height = disp.height
elif displaytype == 'ili9486': 
   width=480
   height=320
else:                         # sane default
   width=240
   height=240

#ili9486 must be in RGB format, no Alpha channel
#image = Image.new("RGBA", (width, height))
image = Image.new("RGB", (width, height))
draw = ImageDraw.Draw(image)

# define some constants to help with graphics layout
padding = 4 
title_bar_height = 34

if os.path.exists('/dev/gpiochip4'):
   gpiochip = gpiod.Chip('gpiochip4')
else:
   gpiochip = gpiod.Chip('gpiochip0')
blue_line = gpiochip.get_line(5)
blue_line.request(consumer="blue", type=gpiod.LINE_REQ_DIR_OUT)
red_line = gpiochip.get_line(26)
red_line.request(consumer="red", type=gpiod.LINE_REQ_DIR_OUT)

def get_direction(origin, destination):
   lat1, lon1 = origin 
   lat2, lon2 = destination
   dLon = (lon2 - lon1)
   x = math.cos(math.radians(lat2)) * math.sin(math.radians(dLon))
   y = math.cos(math.radians(lat1)) * math.sin(math.radians(lat2)) - math.sin(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.cos(math.radians(dLon))
   bearing = numpy.arctan2(x,y)
   bearing = numpy.degrees(bearing)
   bearing = (bearing + 360) % 360  # make positive
   dirs = ['N', 'NNE', 'NE', 'ENE', 'E', 'ESE', 'SE', 'SSE', 'S', 'SSW', 'SW', 'WSW', 'W', 'WNW', 'NW', 'NNW']
   ix = int(round(bearing / (360. / len(dirs))))  
   return dirs[ix % len(dirs)]
   #return bearing

def get_distance(origin, destination):
   lat1, lon1 = origin
   lat2, lon2 = destination
   # radius = 6371  # km
   radius = 3959    # miles
   dlat = math.radians(lat2 - lat1)
   dlon = math.radians(lon2 - lon1)
   a = (math.sin(dlat / 2) * math.sin(dlat / 2) +
        math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) *
        math.sin(dlon / 2) * math.sin(dlon / 2))
   c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
   d = radius * c
   return d

def signal_handler(signal, frame):
   print("Got ", signal, " exiting.")
   draw.rectangle((0, 0, width, height), outline=0, fill=(30,30,30))
   with display_lock:
       disp.image(image)
   red_line.release()
   blue_line.release()
   os._exit(0)

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

# LED threads, bluetooth, RED, GREEN
def bluetooth_connection_poll_thread():  #FIXME convert to libgpiod, not gpiozero
    bt_status = 0
    time.sleep(2) # so screen initialization doesn't overdraw bluetooth as off

    while True:
        cmd = "hcitool con | wc -l"
        connection_count = subprocess.check_output(cmd, shell=True).decode("utf-8")
        if int(connection_count) > 1:
            if bt_status == 0:
                bt_status = 1
                bticon = Image.open('bt.small.on.png')   
                blue_line.set_value(1) 
                image.paste(bticon, (width - title_bar_height * 3 + 12  , padding + 2 ), bticon)
                with display_lock:
                    disp.image(image)
        else:
            if bt_status == 1:
                bt_status = 0  
                bticon = Image.open('bt.small.off.png')   
                blue_line.set_value(0) 
                image.paste(bticon, (width - title_bar_height * 3 + 12  , padding + 2 ), bticon)
                with display_lock:
                    disp.image(image)
        time.sleep(2)

bluetooth_thread = threading.Thread(target=bluetooth_connection_poll_thread, name="btwatch")
bluetooth_thread.start()

def redgreen_thread():                      ## change red or green status indicators and red diode
   f = subprocess.Popen(['tail','-F','-n','10',logfile], stdout=subprocess.PIPE,stderr=subprocess.PIPE)
   while True:
      line = f.stdout.readline().decode("utf-8", errors="ignore")
      search = re.match("^((?:DCD |PTT ).*)", line)
      if search is not None:
         status = search.group(1)
         if status == 'DCD 0 = 1':
            draw.ellipse(( width - title_bar_height     , padding,    width - padding * 2,                     title_bar_height - padding), fill=(0,200,0,255))
         elif status == 'DCD 0 = 0':
            draw.ellipse(( width - title_bar_height     , padding,    width - padding * 2,                     title_bar_height - padding), fill=(0,80,0,255))
         elif status == 'PTT 0 = 1':
            draw.ellipse(( width - title_bar_height * 2 , padding,    width - title_bar_height - padding * 2 , title_bar_height - padding), fill=(200,0,0,255))
            #red_led.on()
            red_line.set_value(1)
         elif status == 'PTT 0 = 0':
            draw.ellipse(( width - title_bar_height * 2 , padding,    width - title_bar_height - padding * 2 , title_bar_height - padding), fill=(80,0,0,255))
            #red_led.off()
            red_line.set_value(0)
         else:
            print("Unknown DCD/PTT event\n")
         with display_lock:
            disp.image(image)

redgreen_thread = threading.Thread(target=redgreen_thread, name="rgwatch")
redgreen_thread.start()


# Load a TTF font. 
fontname = "DejaVuSans.ttf"
fontname_bold = "DejaVuSans-Bold.ttf"
if os.path.exists("/usr/share/fonts/truetype/dejavu/" + fontname):
   fontpath = "/usr/share/fonts/truetype/dejavu/" + fontname
elif os.path.exists("./" + fontname):
   fontpath = "./" + fontname
else:
   print("Couldn't find font " +  fontname + " in working dir or /usr/share/fonts/truetype/dejavu/")
   exit(1)
if os.path.exists("/usr/share/fonts/truetype/dejavu/" + fontname_bold):
   fontpath_bold = "/usr/share/fonts/truetype/dejavu/" + fontname_bold
elif os.path.exists("./" + fontname_bold):
   fontpath_bold = "./" + fontname_bold
else:
   print("Couldn't find font " +  fontname_bold + " in working dir or /usr/share/fonts/truetype/dejavu/")
   exit(1)

if displaytype == 'ili9486':
   bump = 10
else:
   bump = 0

#font(no suffix) is defined on command line, used in list mode
font = ImageFont.truetype(fontpath, fontsize)
font_small = ImageFont.truetype(fontpath_bold, 18 + bump)
font_big = ImageFont.truetype(fontpath_bold, 24)            # title bar font
font_huge = ImageFont.truetype(fontpath_bold, 34 + bump)
font_epic = ImageFont.truetype(fontpath_bold, 38 + bump)

# load symbol chart based on font height
symbol_chart0x128 = Image.open("aprs-symbols-128-0.png")
symbol_chart1x128 = Image.open("aprs-symbols-128-1.png")
symbol_dimension = 128

# Draw a black filled box to clear the image.
draw.rectangle((0, 0, width, height), outline=0, fill="#000000")

# Draw our logo
h = font.getbbox(title_text)[3]
draw.text(   (padding * 3  ,  height // 2 - h) ,   title_text, font=font_huge,   fill="#999999")
with display_lock:
    disp.image(image)
time.sleep(1)

# erase the screen
draw.rectangle((0, 0, width, height), outline=(0,0,0), fill="#000000")

# draw the header bar
draw.rectangle((0, 0, width, title_bar_height), fill=(30, 30, 30))
draw.text((padding, padding), title_text, font=font_big, fill="#999999")

# draw the bluetooth icon
bticon = Image.open('bt.small.off.png')
image.paste(bticon, (width - title_bar_height * 3 + 12  , padding + 2 ), bticon)

# draw Green LED
draw.ellipse(( width - title_bar_height               , padding,       width - padding * 2,                  title_bar_height - padding), fill=(0,80,0,255))

# draw Red LED
draw.ellipse(( width - title_bar_height * 2           , padding,    width - title_bar_height - padding * 2 , title_bar_height - padding), fill=(80,0,0,255))

with display_lock:
    disp.image(image)
    if savefile: image.save(savefile, compress_level=1) 

# fire up green/red led threads
#watch_threadG.start()
#watch_threadR.start()

# setup geometry defaults
call = "null"
x = padding
line_count = 0
col_count = 0 

# tail and block on the log file
f = subprocess.Popen(['tail','-F','-n','10',logfile], stdout=subprocess.PIPE,stderr=subprocess.PIPE)

# single loop, display one station at a time full screen, when using "-o" option on command line
def single_loop():

   # contstants
   infotopmargin = title_bar_height + ( padding * 4 )
   infolinespacing = font_small.getbbox("ABCJQ")[3] + padding

   # we try to get callsign, symbol and four relevant info lines from every packet
   while True:
      info1 = info2 = info3 = info4 = ''      # sane defaults

      line = f.stdout.readline().decode("utf-8", errors="ignore")
      
      search = re.search("^\[\d\.*\d*\] (.*)", line)

      if search is not None:
         packetstring = search.group(1)
         packetstring = packetstring.replace('<0x0d>','\x0d').replace('<0x1c>','\x1c').replace('<0x1e>','\x1e').replace('<0x1f>','\0x1f').replace('<0x0a>','\0x0a')
      else:
         continue
  
      try:  
         packet = aprslib.parse(packetstring)                           # parse packet
         call = packet['from']
         supported_packet = True 
      except Exception as e:                                            # aprslib doesn't support all packet types
         supported_packet = False
         packet = {}   
         search = re.search("^\[\d\.*\d*\] ([a-zA-Z0-9-]*)", line)      # snag callsign from unsupported packet
         if search is not None:
            call = search.group(1) 
            symbol = '/'                                                # unsupported packet symbol set to bronze ball
            symbol_table = '/' 
         else:
            continue

      try:          
         if supported_packet:
            if 'symbol' in packet:                                      # get symbol from valid packet or use red ball
               symbol = packet['symbol']
               symbol_table = packet['symbol_table']
            else:
               symbol = '/'
               symbol_table = '/'
                                                                        # extract relevant info lines
         if not supported_packet:                                      
               info1 = info2 = info3 = info4 = ''                       # no info in unsupported packet
         elif 'weather' in packet:                                      # weather (often contained in compressed/uncompressed type packets)
            lat2 = packet['latitude']
            lon2 = packet['longitude']
            if lat1:
               distance = get_distance((lat1,lon1),(lat2,lon2))
               direction = get_direction((lat1,lon1),(lat2,lon2))   
               info1 = str(round(distance)) + "mi " + direction
            else:
               info1 = ""
            info2 = round(packet['weather']['temperature'])
            info2 = str(round(int(info2) * 1.8 + 32)) + 'F'
            info3 = ""
            info4 = str(packet['comment'])
         elif packet['format'] == 'mic-e' or packet['format'] == 'compressed'  or packet['format'] == 'uncompressed' or packet['format'] == 'object':  
            info4 = re.sub('^[^0-9a-zA-Z]*', '', packet['comment'])     # get rid of leading punctuation
            lat2 = packet['latitude']
            lon2 = packet['longitude']
            if lat1:
               distance = get_distance((lat1,lon1),(lat2,lon2))
               direction = get_direction((lat1,lon1),(lat2,lon2))
               info1 = str(round(distance)) + "mi " + direction
            else:
               info1 = ""
         elif 'status' in packet:                                       # status packet
            info4 = re.sub('^[^0-9a-zA-Z]+', '', packet['status'])      # get rid of leading punctuation
      except Exception as e:
         print("Malformed/missing data: ", str(e), ": ", packetstring)

      offset = ord(symbol) - 33
      row = offset // 16 
      col = offset % 16 
      draw.rectangle((0, title_bar_height, width, height), fill="#000000")  # erase most of screen
      crop_area = (col*symbol_dimension, row*symbol_dimension, col*symbol_dimension+symbol_dimension, row*symbol_dimension+symbol_dimension)
      if symbol_table == '/':
         symbolimage = symbol_chart0x128.crop(crop_area)
      else:
         symbolimage = symbol_chart1x128.crop(crop_area)
      #symbolimage = symbolimage.resize((height // 2, height // 2), Image.NEAREST)
      if height >= 320:
         symbolimage = symbolimage.resize((180, 180), Image.LANCZOS)
      image.paste(symbolimage, (0, title_bar_height), symbolimage)
      infoleftmargin = symbolimage.width + padding
      draw.text((infoleftmargin, infotopmargin),                         str(info1), font=font_small, fill="#AAAAAA")
      draw.text((infoleftmargin, infotopmargin + infolinespacing),       str(info2), font=font_small, fill="#AAAAAA")
      draw.text((infoleftmargin, infotopmargin + (infolinespacing * 2)), str(info3), font=font_small, fill="#AAAAAA")
      statustopmargin = symbolimage.height + title_bar_height 
      draw.text((5, statustopmargin), str(info4), font=font_small, fill="#AAAAAA")
      draw.text((5, height - font_epic.getbbox("J")[3]), call, font=font_epic, fill="#AAAAAA") # text up from bottom edge
  
      with display_lock:
          disp.image(image)
          if savefile: image.save(savefile, compress_level=1) 

      time.sleep(1)


# list of stations on screen (no -o option on command line)
def list_loop():
  call = "null"

  # scale symbols based on font height
  fontvertical = font.getbbox("ABCJQ")[3]       # tallest callsign, with dangling J/Q tails
  symbol_chart0x128.thumbnail(((fontvertical + fontvertical // 8) * 16, (fontvertical + fontvertical // 8) * 6)) # nudge larger than font, into space between lines
  symbol_chart1x128.thumbnail(((fontvertical + fontvertical // 8) * 16, (fontvertical + fontvertical // 8) * 6)) # nudge larger than font, into space between lines
  symbol_dimension = symbol_chart0x128.width//16
  max_line_width = font.getbbox("KN6MUC-15")[2] + symbol_dimension + (symbol_dimension // 8)   # longest callsign i can think of in pixels, plus symbo width + space
  max_cols = width // max_line_width

  # position cursor in -1 slot, as the first thing the loop does is increment slot
  y = padding + title_bar_height - font.getbbox("ABCJQ")[3]  
  x = padding
  line_height = font.getbbox("ABCJQ")[3] - 1          # tallest callsign, with dangling J/Q tails
  max_lines  = ( height - title_bar_height - padding )  //   line_height 
  max_cols = ( width // max_line_width )
  line_count = 0
  col_count = 0 

  while True:
    line = f.stdout.readline().decode("utf-8", errors="ignore")

    # watch for regular packet
    #search = re.search("^\[\d\.\d\] (.*)", line)
    #Direwolf lines start with [#.#], or just [#], observed on HF
    search = re.search("^\[\d\.*\d*\] (.*)", line)

    if search is not None:
       packetstring = search.group(1)
       packetstring = packetstring.replace('<0x0d>','\x0d').replace('<0x1c>','\x1c').replace('<0x1e>','\x1e').replace('<0x1f>','\0x1f').replace('<0x0a>','\0x0a')
    else:
       continue

    lastcall = call

    try:                                        #   aprslib has trouble parsing all packets
       packet = aprslib.parse(packetstring) 
       call = packet['from']
       if 'symbol' in packet:
          symbol = packet['symbol']
          symbol_table = packet['symbol_table']
       else:
          symbol = '/'
          symbol_table = '/'
    except:                                     #   if it fails, let's just snag the callsign
       #print("aprslib failed to parse.")
       #search = re.search("^\[\d\.\d\] ([a-zA-Z0-9-]*)", line)
       search = re.search("^\[\d\.*\d*\] ([a-zA-Z0-9-]*)", line)
       if search is not None:
          call = search.group(1) 
          symbol = '/'
          symbol_table = '/'
       else:
          continue

    offset = ord(symbol) - 33
    row = offset // 16 
    col = offset % 16 

    if call == lastcall:   # blink duplicates
       time.sleep(0.5)
       draw.text((x + symbol_dimension + (symbol_dimension // 8) , y), call, font=font, fill="#000000") # start text after symbol, relative padding
       with display_lock:
           disp.image(image)
       time.sleep(0.1)
       draw.text((x + symbol_dimension + (symbol_dimension // 8) , y), call, font=font, fill="#AAAAAA") # start text after symbol, relative padding
       with display_lock:
           disp.image(image)
    else:
       y += line_height
       if line_count == max_lines:       # about to write off bottom edge of screen
           col_count += 1
           x = col_count * max_line_width  
           y = padding + title_bar_height
           line_count = 0
       if col_count == max_cols:         # about to write off right edge of screen
           x = padding 
           y = padding + title_bar_height
           draw.rectangle((0, title_bar_height + 1, width, height), outline=0, fill="#000000") # erase lines
           line_count = 0
           col_count = 0
           time.sleep(2.0)
       crop_area = (col*symbol_dimension, row*symbol_dimension, col*symbol_dimension+symbol_dimension, row*symbol_dimension+symbol_dimension)
       if symbol_table == '/':
          symbolimage = symbol_chart0x128.crop(crop_area)
       else:
          symbolimage = symbol_chart1x128.crop(crop_area)
       image.paste(symbolimage, (x, y), symbolimage)
       draw.text((x + symbol_dimension + (symbol_dimension // 8) , y), call, font=font, fill="#AAAAAA") # start text after symbol, relative padding
       line_count += 1
       with display_lock:
           disp.image(image)
           if savefile: image.save(savefile, compress_level=1) 

if __name__ == '__main__':
   if args["one"]:
      single_loop()
   else:
      list_loop()
   os._exit(0)