obsw_binary_uploader.py

#!/usr/bin/python3.8
"""
@brief      Binary Uploader Module
@details
This module will be used to upload binaries to the OBC via a communication port, given
a supplied binary. The binary will be sent via the specified communication interface.
It will be possible to encode the data (for example using DLE encoding)
"""
import os
import time
import tkinter as tk
from tkinter import filedialog
from collections import deque
from glob import glob

from tmtc_core.comIF.obsw_com_interface import CommunicationInterface
from tc.obsw_tc_service23_sdcard import FileTransferHelper
import config.obsw_config as g
from tmtc_core.utility.obsw_tmtc_printer import TmTcPrinter, DisplayMode
from tmtc_core.utility.obsw_logger import get_logger
from tmtc_core.sendreceive.obsw_tm_listener import TmListener
from tmtc_core.tm.obsw_pus_tm_factory import PusTelemetryFactory, PusTmQueueT

LOGGER = get_logger()


def perform_file_upload(com_if: CommunicationInterface, tmtc_printer: TmTcPrinter,
                        tm_listener: TmListener):
    gui_cl_prompt = input("GUI(0) or command line version (1)? [0/1]: ")
    if gui_cl_prompt == 0:
        gui_cl_prompt = True
    else:
        gui_cl_prompt = False
    # TODO: prompt whether this is a binary upload or a normal file upload. Or use op code
    #       two different commands to achieve the same.
    print("Please select file to upload: ")
    file_path = ""
    if gui_cl_prompt:
        root = tk.Tk()
        root.withdraw()
        root.wm_attributes('-topmost', 1)
        file_path = filedialog.askopenfilename(parent=root)
        print("File select: " + str(file_path))
        if file_path == ():
            LOGGER.warning("Invalid file path, exiting binary upload mode.")
            return
    else:
        result = [y for x in os.walk("../_bin") for y in glob(os.path.join(x[0], '*.bin'))]
        print("Files found in _bin folder: ")
        for idx, path in enumerate(result):
            print("Selection " + str(idx) + ": " + str(path))
        select_valid = False
        selection = input("Please enter desired selection [c to cancel]: ")
        while not select_valid:
            if selection == 'c':
                print("Exiting binary upload mode..")
                return
            if not selection.isdigit():
                selection = input("Invalid input, try again [c to cancel]: ")
            if selection.isdigit():
                if int(selection) < len(result):
                    file_path = result[int(selection)]
                    select_valid = True
                else:
                    selection = input("Invalid input, try again [c to cancel]: ")

    print_string = file_path.rsplit('/', 1)[-1] + " was selected."
    LOGGER.info(print_string)
    calc_hamming_code = input("Calculate and send hamming code? [y/n]: ")
    if calc_hamming_code in ['y', 'yes', 1]:
        calc_hamming_code = True
        print("Hamming code will be calculated and sent in tail packet")
    else:
        calc_hamming_code = False
        print("Hamming code will not be calculated")

    iobc_prompt = input("iOBC? [y/n]: ")
    if iobc_prompt in ['y', 'yes', 1]:
        iobc_prompt = True
    else:
        iobc_prompt = False

    bootloader_prompt = input("Bootloader (0) or Software Image (1)? [0/1]: ")
    if str(bootloader_prompt) == "0":
        bootloader_prompt = True
    else:
        bootloader_prompt = False

    if bootloader_prompt:
        file_name = "bl.bin"
    else:
        file_name = "obsw_up.bin"

    if iobc_prompt:
        if bootloader_prompt:
            repository_name = "BIN/IOBC/BL"
        else:
            repository_name = "BIN/IOBC/OBSW"
    else:
        if bootloader_prompt:
            repository_name = "BIN/AT91/BL"
        else:
            repository_name = "BIN/AT91/OBSW"

    # Right now, the size of PUS packets is limited to 1500 bytes which also limits the app
    # data length.
    frame_length = g.G_MAX_APP_DATA_LENGTH

    if calc_hamming_code:
        # now we calculate the hamming code
        pass

    # Read file as binary file
    if file_path != ():
        with open(file_path, 'rb') as file:
            data_to_read = bytearray(file.read())
    else:
        LOGGER.info("Specified path invalid..")

    print("Generating folder structure..")
    tc_queue = deque()

    # Delete existing binary file first, otherwise data might be appended to otherwise
    # valid file which already exists.
    file_transfer_helper = FileTransferHelper(
        tc_queue=tc_queue, max_size_of_app_data=frame_length, target_repository=repository_name,
        target_filename=file_name)

    init_ssc = 0
    tmtc_printer.set_display_mode(DisplayMode.SHORT)

    file_transfer_helper.set_data_from_file(file_path)
    file_transfer_helper.set_to_delete_old_file()
    file_transfer_helper.generate_packets(init_ssc)
    tm_listener.set_listener_mode(TmListener.ListenerModes.MANUAL)
    print_string = "BinaryUploader: Detected file size: " + str(file_transfer_helper.file_size())
    LOGGER.info(print_string)
    total_num_packets = file_transfer_helper.number_of_packets
    print_string = "BinaryUploader: " + str(total_num_packets) + \
                   " packets generated with " + \
                   str(file_transfer_helper.number_of_append_packets) + " append packets."
    LOGGER.info(print_string)
    interval = 0.6
    last_sent = time.time()
    total_time = interval * total_num_packets
    idx = 1
    reception_deque: PusTmQueueT = deque()
    while tc_queue:
        next_send = last_sent + interval
        (tc_packet, tc_info) = tc_queue.pop()
        if not isinstance(tc_packet, str):
            # print_string = "Sending packet " + str(idx) + ".."
            # LOGGER.info(print_string)
            idx += 1
            com_if.send_telecommand(tc_packet, tc_info)
            tmtc_printer.print_telecommand(tc_packet, tc_info)
        elif tc_packet == "print":
            LOGGER.info(tc_info)
        remaining_time_string = "Remaining time: " + str(total_time - (idx - 2) * interval) + \
                                " seconds"
        print_progress_bar(idx - 2, total_num_packets, print_end="\n", suffix=remaining_time_string)
        # sys.stdout.write("\033[F")  # Cursor up one line
        reception_deque.extend(tm_listener.retrieve_tm_packet_queue())
        time_to_sleep = next_send - time.time()
        last_sent = next_send
        time.sleep(time_to_sleep)

    print_string = "BinaryUploader: All binary packets were sent!"
    LOGGER.info(print_string)
    print_string = str(reception_deque.__len__()) + " replies received."

    LOGGER.info(print_string)
    time.sleep(15)
    reception_deque.extend(tm_listener.retrieve_tm_packet_queue())
    for tm_list in reception_deque:
        for tm_packet in tm_list:
            if tm_packet.get_service() == 23 and tm_packet.get_subservice() == 132:
                tmtc_printer.print_telemetry(tm_packet)
    tm_listener.clear_tm_packet_queue()
    LOGGER.info("Transitioning back to listener mode..")


# https://stackoverflow.com/questions/3173320/text-progress-bar-in-the-console
# Thank you Greensticks :-)
def print_progress_bar(iteration, total, prefix='', suffix='', decimals=1, length=100,
                       fill='█', print_end="\r"):
    """
    Call in a loop to create terminal progress bar
    @params:
        iteration   - Required  : current iteration (Int)
        total       - Required  : total iterations (Int)
        prefix      - Optional  : prefix string (Str)
        suffix      - Optional  : suffix string (Str)
        decimals    - Optional  : positive number of decimals in percent complete (Int)
        length      - Optional  : character length of bar (Int)
        fill        - Optional  : bar fill character (Str)
        print_end    - Optional  : end character (e.g. "\r", "\r\n") (Str)
    """
    percent = ("{0:." + str(decimals) + "f}").format(100 * (iteration / float(total)))
    filled_length = int(length * iteration // total)
    bar = fill * filled_length + '-' * (length - filled_length)
    print(f'\r{prefix} |{bar}| {percent}% {suffix}', end=print_end)
    # Print New Line on Complete
    if iteration == total:
        print()