UNIHIKER: ADS-B Flight Tracker

# update package list
$ apt update

# install required packages
$ apt install -y rtl-sdr dump1090-mutability

# install optional package
$ apt install -y ncat

# SSH Connection (tty000)

# execute dump1090-mutability
$ dump1090-mutability --interactive --net

# SSH Connection (tty001)

# execute ncat
$ nc localhost 30003

from socket import socket, AF_INET, SOCK_STREAM
from re import match
from queue import Queue
from time import sleep

flight_data = {}


def should_remove(key: str) -> bool:
    """
    Check if a key should be removed.

    :param key: The key to check.
    :type key: str
    :return: True if the key should be removed, False otherwise.
    """
    if len(key) == 1:
        return True

    if match(r'^\d{4}/\d{2}/\d{2}$', key):
        return True

    return False


def update_flight_data(icao: str, flight: str) -> list:
    """
    Update flight data for the given ICAO code.

    :param icao: The ICAO code of the flight.
    :type icao: str
    :param flight: The flight information to update.
    :type flight: str
    :return: A list of tuples containing the ICAO code and flight information for all flights that should not be removed.
    """
    global flight_data

    if icao in flight_data:
        if not flight:
            flight = flight_data[icao]['Flight']
    else:
        flight_data[icao] = {'Flight': ''}

    flight_data[icao] = {
        'Flight': flight
    }

    return  [(key, value['Flight']) for key, value in flight_data.items() if not should_remove(key)]

def process_message(message: str, data_queue: Queue) -> None:
    """
    Process the given message and update the flight data.

    :param message: A string containing the message to process.
    :type message: str
    :param data_queue: A Queue object to store the updated flight data.
    :type data_queue: Queue
    :return: None
    """
    parts = message.split(',')

    if len(parts) >= 20:
        hex_code = parts[4]
        flight_code = parts[10].strip()

        updated_data = update_flight_data(icao=hex_code, flight=flight_code)
        data_queue.put(updated_data)

def receive_flight_data(host: str, port: int, delay: float, data_queue: Queue) -> None:
    """
    Receives flight data from a specified host and port, and processes the data by placing it into a specified Queue.

    :param host: The hostname or IP address to connect to.
    :type host: str
    :param port: The port number to connect to.
    :type port: int
    :param delay: The time delay (in seconds) to wait between receiving data.
    :type delay: float
    :param data_queue: The Queue object to store the received flight data.
    :type data_queue: Queue
    :return: None
    """
    with socket(AF_INET, SOCK_STREAM) as s:
        s.connect((str(host), int(port)))

        while True:
            data = s.recv(4096)

            if not data:
                break

            try:
                messages = data.decode('utf-8').strip().split('\n')
                for msg in messages:
                    process_message(msg, data_queue)
            except UnicodeDecodeError:
                print('Error')

            sleep(float(delay))

from tkinter import Tk, Frame, Label, Canvas, Scrollbar


class TrackerApp(Tk):

    __HEADER: tuple = ('HEX', 'Flight')

    def __init__(self, width: int, height: int):
        """
        Class constructor.

        :param width: screen width
        :type width: int
        :param height: screen height
        :type height: int
        """
        super().__init__()
        self._screen_width = int(width)
        self._screen_height = int(height)

        self.resizable(width=False, height=False)
        self.geometry(f'{self._screen_width}x{self._screen_height}+0+0')

        self.__add_header()

        self.canvas = Canvas(self)
        self.scroll_y = Scrollbar(self, orient='vertical', command=self.canvas.yview)
        self.canvas.config(yscrollcommand=self.scroll_y.set)

        self.canvas_frame = Frame(self.canvas)
        self.canvas.create_window((0, 0), window=self.canvas_frame, anchor='nw')

        self.canvas_frame.bind('<Configure>', lambda e: self.__on_frame_configure(self.canvas))

        self.scroll_y.pack(side='right', fill='y')
        self.canvas.pack(side='left', fill='both', expand=True)

        self.labels = {}

    @staticmethod
    def __on_frame_configure(canvas: Canvas) -> None:
        """
        Static method updates the scroll region of a canvas when size of frame is changed.

        :param canvas: The canvas widget whose scroll region needs to be updated.
        :type canvas: Canvas
        :return: None
        """
        canvas.config(scrollregion=canvas.bbox('all'))

    def __add_header(self) -> None:
        """
        Private method to add a header frame to the top of the window.

        :return: None
        """
        header_frame = Frame(self)
        header_frame.pack(fill='x', side='top')

        header_frame.grid_columnconfigure(index=0, weight=1)
        header_frame.grid_columnconfigure(index=1, weight=1)

        for number in range(2):
            text = self.__HEADER[number]
            header_label = Label(header_frame, text=text, borderwidth=1, relief='solid', height=2)
            header_label.grid(row=0, column=number, sticky='nsew')

        header_frame.grid_rowconfigure(index=0, weight=1)

    def update_flight_data(self, flight_data: list) -> None:
        """
        Update the flight data displayed on the canvas.

        :param flight_data: A list of lists containing the flight data to update.
        :return: None
        """
        for i, (hex_code, flight_code) in enumerate(flight_data):
            for j, text in enumerate((hex_code, flight_code)):
                if (i, j) in self.labels:
                    self.labels[(i, j)].config(text=text)
                else:
                    label = Label(self.canvas_frame, text=text, borderwidth=1, relief='solid', width=12, height=2)
                    label.grid(row=i, column=j, sticky='nsew')
                    self.labels[(i, j)] = label

        current_labels = set((i, j) for i in range(len(flight_data)) for j in range(2))

        for pos in list(self.labels):
            if pos not in current_labels:
                self.labels[pos].destroy()
                del self.labels[pos]

        self.canvas_frame.update_idletasks()
        self.__on_frame_configure(self.canvas)

from threading import Thread
from queue import Queue
from lib.gui import TrackerApp
from lib.flight_data import receive_flight_data

SCREEN_WIDTH: int = 240
SCREEN_HEIGHT: int = 320
HOST: str = '127.0.0.1'
PORT: int = 30003
DELAY: float = 2.5

def flight_data_thread(data_queue: Queue) -> None:
    """
    Receive flight data in a separate thread.

    :param data_queue: A queue to store the received flight data.
    :return: None
    """
    receive_flight_data(HOST, PORT, DELAY, data_queue)

def main() -> None:
    """
    This function is the entry point of the program.

    :return: None
    """
    app = TrackerApp(width=SCREEN_WIDTH, height=SCREEN_HEIGHT)

    flight_data_queue = Queue()

    thread = Thread(target=flight_data_thread, args=(flight_data_queue,), daemon=True)
    thread.start()

    def update_gui():
        """
        Update the GUI with flight data from the flight_data_queue.

        :return: None
        """
        while not flight_data_queue.empty():
            flight_data = flight_data_queue.get()
            app.update_flight_data(flight_data)
        app.after(100, update_gui)

    app.after(100, update_gui)

    app.mainloop()


if __name__ == '__main__':
    main()

# SSH Connection (tty000)

# execute dump1090-mutability
$ dump1090-mutability --interactive --net

# SSH Connection (tty001)

# change directory
$ cd /root/ADS-B/

# run Python application
$ python3 main.py