Application.py

import os
import random
import time
from pathlib import Path
import tkinter as tk

from PIL import Image, ImageTk

import ModesEnum
import my_utils
import Report

from db import TinyDB
from GroundStation import GroundStation
from Satellite import Satellite

current_path = os.path.join(os.path.dirname(os.path.abspath(__file__)))


# const double ARIEL_LAT = 32.10544342149793;
# const double ARIEL_LNG = 35.211391207993;

# AOS: Acquisition of Signal (or Satellite). AOS is the time that a satellite rises above the horizon of an observer.
# TCA: Time of Closest Approach. This is the time when the satellite is closest to the observer and when Doppler shift is zero. This usually corresponds to the time that the satellite reaches maximum elevation above the horizon.
# LOS: Loss of Signal (or Satellite). The time that a satellite passes below the observer’s horizon.


class Application:
    """
    Satellite Ground Station System Application.

    Attributes:
    - gs_image: Image for ground stations.
    - sat_image: Image for satellites.

    Methods:
    - __init__: Initialize the Application.
    - execution: Execute the application with specified settings.
    - execution_init: Initialize the execution based on the mode.
    - get_sat_dict: Get the dictionary of satellites.
    - stop_and_exit: Stop and exit the application.
    - stop: Stop all ground stations and satellite transmissions.
    - handle_keypress: Handle keypress events.
    - my_time: Perform actions for each time increment.
    - my_time_gui: Perform GUI-related actions for each time increment.
    - run: Run the application.
    - sat_send_message: Send messages from satellites.
    - gs_receive_message: Receive messages by ground stations.
    """
    gs_image = None
    sat_image = None

    def __init__(self, time_increment=60, alg_mode=None, exec_period=1, gui=False, timestamp='', datestart='',
                 gs_exp=100, sat_exp=100):
        """
        Initialize the Application.

        Parameters:
        - time_increment (int): Time increment for each step (default is 60 seconds).
        - alg_mode (list): List of algorithm modes (default is None).
        - exec_period (int): Execution period (default is 1).
        - gui (bool): Enable GUI mode (default is False).
        - timestamp (str): Timestamp for the application (default is '').
        - datestart (str): Start date for the application (default is '').

        Returns:
        None
        """
        self.alg_mode = alg_mode
        self.root_tk = None
        self.map_widget = None
        # tk globals
        self.markers = {}
        self.time_lbl = None
        if len(timestamp) > 0:
            self.timestamp = timestamp
        else:
            self.timestamp = my_utils.date_timestamp()

        if len(datestart) > 0:
            self.datestart = datestart
        else:
            self.datestart = [my_utils.date_start]
        self.exec_period = exec_period
        self.time_increment = time_increment
        # Globals
        self.state = ()
        self.sat_dict = {}
        self.gs_dict = {}
        self.weight_dict = {}
        self.gui = gui
        self.messages = []
        self.sat_exp = sat_exp
        self.gs_exp = gs_exp

    def execution(self):
        """
        Execute the application with specified settings.

        Returns:
        None
        """
        mode = []
        if len(self.alg_mode) == 0:
            for m in ModesEnum.Mode:
                mode.append(m.value)
        else:
            mode = self.alg_mode

        for ds in self.datestart:
            for m in mode:
                print(f'*** Starting: Mode={m}, Date: {ds}')
                my_utils.set_date_period(exec_period=self.exec_period, increment=self.time_increment, ds=ds)
                my_utils.rest_random()
                self.execution_init(m)
                self.run()
                self.stop()
            df = Report.aggregate(report_folder=self.timestamp, col_names=Report.col_names,prefix='gs_')
            Report.save(df, report_folder=self.timestamp, report_name='gs_combined.csv')

            df = Report.aggregate(report_folder=self.timestamp, col_names=Report.sat_col_names, prefix='sat_')
            Report.save(df, report_folder=self.timestamp, report_name='sat_combined.csv')

            df = Report.aggregate(report_folder=self.timestamp, col_names=Report.sat_vis_col_names, prefix='vis_')
            Report.save(df, report_folder=self.timestamp, report_name='vis_combined.csv')
            self.timestamp = my_utils.date_timestamp()
        exit(0)

    def execution_init(self, mode):
        """
        Initialize the execution based on the mode.

        Parameters:
        - mode: Mode for the execution.

        Returns:
        None
        """
        # Start the DB
        absolute_path = Path().absolute()
        db_file_path = os.path.join(absolute_path, 'db/edgegs_db.json')
        tiny_db = TinyDB.SatDatabase(db_file_path)
        # Initiating the database creates a new one if none exists or updates TLE if two weeks have passed since the last update.
        TinyDB.init_db(tiny_db, absolute_path)

        cnt = 0
        # Get Sats
        sat_list = tiny_db.get_entities_by_type('satellite')
        for cnt, sat_doc in enumerate(sat_list):
            sat = Satellite.from_json(sat_doc, self.timestamp)
            if len(sat.tle) > 0:
                self.sat_dict[str(sat.sat_id)] = sat
                if len(self.sat_dict) > self.sat_exp:
                    break
        # logger.info(f'Sat loaded: {len(sat_list)}')
        print(f'Sat loaded: {cnt+1}')

        # Get GS
        gs_list = tiny_db.get_entities_by_type('ground_station')
        for cnt, gs_doc in enumerate(gs_list):
            gs = GroundStation.from_json(gs_doc, self.timestamp)
            self.gs_dict[gs.gs_id] = gs
            if len(self.gs_dict) > self.gs_exp:
                break
        print(f'GS loaded: {cnt+1}')

        sat_tx_prob = my_utils.get_sat_tx_prob(self.sat_dict)
        sat_rx_prob = my_utils.get_sat_rx_prob(self.sat_dict, self.gs_dict)
        self.weight_dict = my_utils.get_weight(self.sat_dict)
        tle_list = []
        for sat in self.sat_dict.values():
            tle = (sat.sat_id, sat.tle[0], sat.tle[1])
            if tle is None:
                continue
            tle_list.append(my_utils.parse_tle(tle))

        for gs in self.gs_dict.values():
            # print(f'{gs.to_string()} started')
            gs.set_tle_list(tle_list)
            gs.set_sat_dict(self.sat_dict)
            sat_rx_prob_dict = {}
            for sat in self.sat_dict.values():
                gs_prob = sat_rx_prob[gs.gs_id + '_' + str(sat.sat_id)]
                sat_rx_prob_dict[str(sat.sat_id)] = gs_prob
            gs.set_sat_rx_prob_dict(sat_rx_prob_dict)
            gs.set_sat_weight_dict(self.weight_dict)
            gs.mode = mode
            gs.turn_on()

        for sat in self.sat_dict.values():
            # print(f'Sat ID: {sat.sat_id}, Name: {sat.name} is active')
            sat.set_tx_prob = sat_tx_prob[str(sat.sat_id)]
            sat.start_transmit()

        if self.gui:
            import tkintermapview

            self.root_tk = tk.Tk()
            self.root_tk.geometry(f"{1024}x{768}")
            self.root_tk.title("SATLLA-GS")
            self.gs_image = ImageTk.PhotoImage(
                Image.open(os.path.join(current_path, "images", "gs-39.png")).resize((30, 30)))
            self.sat_image = ImageTk.PhotoImage(
                Image.open(os.path.join(current_path, "images", "sat-39.png")).resize((30, 30)))
            # create map widget
            self.map_widget = tkintermapview.TkinterMapView(self.root_tk, width=1024, height=768, corner_radius=0)
            self.map_widget.place(relx=0.5, rely=0.5, anchor=tk.CENTER)
            markers = dict()
            # set current widget position and zoom
            self.map_widget.set_position(0.0, 0.0)
            self.map_widget.set_zoom(1)

            self.root_tk.bind("<Key>", self.handle_keypress)

            self.time_lbl = tk.Label(
                self.root_tk,
                text=time.strftime("%d/%m/%Y %H:%M:%S"),
                font=(21),
                bg='#d9d8d7'
            )
            self.time_lbl.grid(row=0, column=0, padx=50, pady=10)
            self.time_lbl.pack()

    def get_sat_dict(self):
        return self.sat_dict

    def stop_and_exit(self):
        self.stop()
        exit()

    def stop(self):
        """
        Stop all ground stations and satellite transmissions.

        Returns:
        None
        """
        [gs.turn_off() for gs in self.gs_dict.values()]
        [sat.stop_transmit() for sat in self.sat_dict.values()]

        # If the GUI is enabled, quit the root_tk
        if self.gui:
            self.root_tk.quit()

    def handle_keypress(self, event):
        """Print the character associated to the key pressed"""
        # print(event.char)
        char_to_function = {
            'x': self.stop_and_exit,
            '+': my_utils.increment,
            '-': my_utils.decrease,
            's': self.my_time_gui
        }

        if event.char in char_to_function:
            char_to_function[event.char]()

    def my_time(self):
        """
        Perform actions for each time increment.

        Returns:
        None
        """
        my_utils.time_loop()
        # gs progress

        # Shuffle the values before iterating over them
        values = list(self.gs_dict.values())
        random.shuffle(values)

        # Execute the list comprehension with shuffled values
        [gs.gs_loop(self.gs_dict) for gs in values]
        # [gs.gs_loop(self.gs_dict) for gs in self.gs_dict.values()]

        # sat update location
        # Shuffle the values before iterating over them
        values = list(self.sat_dict.values())
        random.shuffle(values)
        [sat.get_location(self.gs_dict) for sat in values]
        # [sat.get_location(self.gs_dict) for sat in self.sat_dict.values()]
        self.sat_send_message()
        self.gs_receive_message()

        time_string = my_utils.get_now().strftime("%d/%m/%Y %H:%M:%S")  # time format
        print(f'{time_string}', end="\r")

    def my_time_gui(self):
        """
        Perform GUI-related actions for each time increment.

        Returns:
        None
        """
        my_utils.time_loop()
        now = my_utils.get_now()
        # my_utils.time_loop()
        if now > my_utils.time_end:
            self.stop()
        for gs in self.gs_dict.values():
            # markers[gs.gs_id].delete()
            # markers[gs.gs_id] = map_widget.set_marker(gs.location[0], gs.location[1], text=f'{gs.gs_id} ({len(gs.messages)})', icon=gs_image)
            gs.gs_loop(self.gs_dict)  # refresh
            if gs.gs_id not in self.markers:
                self.markers[gs.gs_id] = self.map_widget.set_marker(gs.location[0],
                                                                    gs.location[1],
                                                                    text=f'{gs.gs_id} ({gs.message_count})',
                                                                    icon=self.gs_image,
                                                                    font="Tahoma 12 bold")
            if len(gs.sat_id) > 0:
                self.markers[gs.gs_id].set_text(
                    text=f'{gs.gs_id} ({gs.message_count})\n({self.sat_dict[gs.sat_id].name})')

        for sat in self.sat_dict.values():
            sat_location = sat.get_location()
            if sat.sat_id in self.markers:
                self.markers[sat.sat_id].delete()
            if self.map_widget is not None and (sat.location[0] != 0 or sat.location[0] != 0):
                self.markers[sat.sat_id] = self.map_widget.set_marker(sat_location[0], sat_location[1],
                                                                      text=self.sat_dict[str(sat.sat_id)].name,
                                                                      icon=self.sat_image,
                                                                      font="Tahoma 12 bold")

        self.sat_send_message()
        self.gs_receive_message()

        time_string = my_utils.get_now()  # time format
        time_string = f'{time_string.strftime("%d/%m/%Y %H:%M:%S")}'
        self.time_lbl.config(text=f'{time_string}, {my_utils.time_increment}')
        # print(f'{time_string}')
        self.time_lbl.after(1, self.my_time_gui)  # time delay of 1000 milliseconds

    def run(self):
        """
        Run the application.

        Returns:
        None
        """
        if not self.gui:
            now = my_utils.get_now()
            while now < my_utils.time_end:
                self.my_time()
                # print(f'{now.strftime("%d/%m/%Y %H:%M:%S")}')
                now = my_utils.get_now()
        else:
            self.my_time_gui()
            self.root_tk.mainloop()

    def sat_send_message(self):
        for sat in self.sat_dict.values():
            message = sat.get_message()
            if message is not None:
                self.messages.append(message)

    def gs_receive_message(self):
        for message in self.messages:
            for gs in self.gs_dict.values():
                gs.received_message(message)
        self.messages.clear()