diff --git a/.idea/runConfigurations/tmtcclient_Hamming_from_FRAM.xml b/.idea/runConfigurations/tmtcclient_Hamming_from_FRAM.xml
new file mode 100644
index 0000000000000000000000000000000000000000..86e6dbb681b37ef6a8873140acc9b2a2350e3f0e
--- /dev/null
+++ b/.idea/runConfigurations/tmtcclient_Hamming_from_FRAM.xml
@@ -0,0 +1,24 @@
+<component name="ProjectRunConfigurationManager">
+ <configuration default="false" name="tmtcclient Hamming from FRAM" type="PythonConfigurationType" factoryName="Python" folderName="Serial SD-Card and Image">
+ <module name="tmtc" />
+ <option name="INTERPRETER_OPTIONS" value="" />
+ <option name="PARENT_ENVS" value="true" />
+ <envs>
+ <env name="PYTHONUNBUFFERED" value="1" />
+ </envs>
+ <option name="SDK_HOME" value="" />
+ <option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
+ <option name="IS_MODULE_SDK" value="true" />
+ <option name="ADD_CONTENT_ROOTS" value="true" />
+ <option name="ADD_SOURCE_ROOTS" value="true" />
+ <EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
+ <option name="SCRIPT_NAME" value="$PROJECT_DIR$/obsw_tmtc_client.py" />
+ <option name="PARAMETERS" value="-m 3 -c 1 -s Img -o P1 -t 3" />
+ <option name="SHOW_COMMAND_LINE" value="false" />
+ <option name="EMULATE_TERMINAL" value="true" />
+ <option name="MODULE_MODE" value="false" />
+ <option name="REDIRECT_INPUT" value="false" />
+ <option name="INPUT_FILE" value="" />
+ <method v="2" />
+ </configuration>
+</component>
\ No newline at end of file
diff --git a/.idea/runConfigurations/tmtcclient_Hamming_from_SD.xml b/.idea/runConfigurations/tmtcclient_Hamming_from_SD.xml
new file mode 100644
index 0000000000000000000000000000000000000000..43c2bce795ed4979b42d4bca37d72c9163c3dfcc
--- /dev/null
+++ b/.idea/runConfigurations/tmtcclient_Hamming_from_SD.xml
@@ -0,0 +1,24 @@
+<component name="ProjectRunConfigurationManager">
+ <configuration default="false" name="tmtcclient Hamming from SD" type="PythonConfigurationType" factoryName="Python" folderName="Serial SD-Card and Image">
+ <module name="tmtc" />
+ <option name="INTERPRETER_OPTIONS" value="" />
+ <option name="PARENT_ENVS" value="true" />
+ <envs>
+ <env name="PYTHONUNBUFFERED" value="1" />
+ </envs>
+ <option name="SDK_HOME" value="" />
+ <option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
+ <option name="IS_MODULE_SDK" value="true" />
+ <option name="ADD_CONTENT_ROOTS" value="true" />
+ <option name="ADD_SOURCE_ROOTS" value="true" />
+ <EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
+ <option name="SCRIPT_NAME" value="$PROJECT_DIR$/obsw_tmtc_client.py" />
+ <option name="PARAMETERS" value="-m 3 -c 1 -s Img -o P0 -t 3" />
+ <option name="SHOW_COMMAND_LINE" value="false" />
+ <option name="EMULATE_TERMINAL" value="true" />
+ <option name="MODULE_MODE" value="false" />
+ <option name="REDIRECT_INPUT" value="false" />
+ <option name="INPUT_FILE" value="" />
+ <method v="2" />
+ </configuration>
+</component>
\ No newline at end of file
diff --git a/obsw_tmtc_client.py b/obsw_tmtc_client.py
index e502717fda8cd2ba2db5c963c394fec2eec3ea43..600db5aef0cd0e10ae9f4602259a7a36c595a72e 100755
--- a/obsw_tmtc_client.py
+++ b/obsw_tmtc_client.py
@@ -48,7 +48,7 @@ from tmtc_core.utility.obsw_logger import set_tmtc_logger, get_logger
from test.obsw_pus_service_test import run_selected_pus_tests
from tc.obsw_pus_tc_packer import create_total_tc_queue, ServiceQueuePacker
from utility.obsw_args_parser import parse_input_arguments
-from utility.obsw_binary_uploader import perform_file_upload
+from utility.obsw_binary_uploader import BinaryFileUploader
from gui.obsw_tmtc_gui import TmTcGUI
from gui.obsw_backend_test import TmTcBackend
@@ -205,7 +205,9 @@ class TmTcHandler:
elif self.mode == g.ModeList.BinaryUploadMode:
# Upload binary, prompt user for input, in the end prompt for new mode and enter that
# mode
- perform_file_upload(self.communication_interface, self.tmtc_printer, self.tm_listener)
+ file_uploader = BinaryFileUploader(self.communication_interface, self.tmtc_printer,
+ self.tm_listener)
+ file_uploader.perform_file_upload()
self.command_received = True
self.mode = g.ModeList.ListenerMode
diff --git a/tc/obsw_image_handler.py b/tc/obsw_image_handler.py
index 5f96ad68b051046aea733aa89bb2321f9c792f3d..be5fd0b174ac405a9acc625997dd6f74273cc038 100644
--- a/tc/obsw_image_handler.py
+++ b/tc/obsw_image_handler.py
@@ -1,5 +1,6 @@
from tmtc_core.tc.obsw_pus_tc_base import PusTelecommand, Deque
from tc.obsw_tc_service8 import make_action_id
+from tc.obsw_tc_service20 import pack_boolean_parameter_setting
from tmtc_core.utility.obsw_logger import get_logger
from config.obsw_config import SW_IMAGE_HANDLER_ID
@@ -23,15 +24,29 @@ def generate_copy_obsw_sdc_to_flash_packet(ssc: int, slot: int,
return PusTelecommand(service=8, subservice=128, ssc=ssc, app_data=app_data)
+def generate_param_packet_hamming_from_sdcard(enable: bool, ssc: int,
+ object_id: bytearray = SW_IMAGE_HANDLER_ID):
+ return pack_boolean_parameter_setting(object_id=object_id, domain_id=0, unique_id=0,
+ parameter=enable, ssc=ssc)
+
+
def generate_img_handler_packet(service_queue: Deque, op_code: int):
# Action ID 4 (Copy SDC to Flash, Software Update Image)
if op_code == "A4U":
service_queue.appendleft(("print", "Generating command to copy SDC OBSW to flash."))
command = generate_copy_obsw_sdc_to_flash_packet(ssc=0, slot=2)
service_queue.appendleft(command.pack_command_tuple())
-
# Action ID 11 (Copy bootloader from SDC to flash
elif op_code == "A11S":
service_queue.appendleft(("print", "Generating command to copy SDC bootloader to flash."))
command = generate_copy_bl_sdc_to_flash_packet(0)
service_queue.appendleft(command.pack_command_tuple())
+ elif op_code == "P0":
+ service_queue.appendleft(("print", "Configuring hamming code to be taken from SD card"))
+ command = generate_param_packet_hamming_from_sdcard(enable=True, ssc=0)
+ service_queue.appendleft(command.pack_command_tuple())
+ elif op_code == "P1":
+ service_queue.appendleft(("print", "Configuring hamming code to be taken from FRAM"))
+ command = generate_param_packet_hamming_from_sdcard(enable=False, ssc=0)
+ service_queue.appendleft(command.pack_command_tuple())
+
diff --git a/tc/obsw_tc_service20.py b/tc/obsw_tc_service20.py
index 24ce024e069c49f1b22d4b9c8dfb86ab6bbfa12e..34bdd32cdd17e2fedac76da8436c449e1a2c6623 100644
--- a/tc/obsw_tc_service20.py
+++ b/tc/obsw_tc_service20.py
@@ -9,9 +9,46 @@ import struct
from typing import Deque
import config.obsw_config as g
-from tmtc_core.tc.obsw_pus_tc_base import PusTelecommand
+from tmtc_core.tc.obsw_pus_tc_base import PusTelecommand, TcQueueT
+from tmtc_core.utility.obsw_logger import get_logger
from tc.obsw_tc_service200 import pack_mode_data
+LOGGER = get_logger()
+
+
+def pack_boolean_parameter_setting(object_id: bytearray, domain_id: int,
+ unique_id: int, parameter: bool, ssc: int):
+ """
+ Generic function to pack a telecommand to tweak a boolean parameter
+ @param object_id:
+ @param domain_id:
+ @param unique_id:
+ @param parameter:
+ @param ssc:
+ @return:
+ """
+ parameter_id = bytearray(4)
+ parameter_id[0] = domain_id
+ if unique_id > 255:
+ LOGGER.warning("Invalid unique ID, should be smaller than 255!")
+ return
+ parameter_id[1] = unique_id
+ parameter_id[2] = 0
+ parameter_id[3] = 0
+ data_to_pack = bytearray(object_id)
+ data_to_pack.extend(parameter_id)
+ # PTC and PFC for uint8_t according to CCSDS
+ ptc = 3
+ pfc = 4
+ rows = 1
+ columns = 1
+ data_to_pack.append(ptc)
+ data_to_pack.append(pfc)
+ data_to_pack.append(rows)
+ data_to_pack.append(columns)
+ data_to_pack.append(parameter)
+ return PusTelecommand(service=20, subservice=128, ssc=ssc, app_data=data_to_pack)
+
def pack_service20_test_into(tc_queue: Deque, called_externally: bool = False) -> Deque:
# parameter IDs
@@ -75,6 +112,11 @@ def pack_service20_test_into(tc_queue: Deque, called_externally: bool = False) -
return tc_queue
+def pack_service23_commands_into(tc_queue: TcQueueT, op_code: int):
+ if op_code == 0:
+ pack_service20_test_into(tc_queue=tc_queue)
+
+
"""
#test checking Load for int32_t
tc_queue.appendleft(("print", "Testing Service 20: Load int32_t"))
diff --git a/tc/obsw_tc_service23_sdcard.py b/tc/obsw_tc_service23_sdcard.py
index ccd07d808c1b253838c2d5e3e4b9322b88c0fdbf..b89ae1bfb3f10dc9cd95af6f9d92ed503f2e20a0 100644
--- a/tc/obsw_tc_service23_sdcard.py
+++ b/tc/obsw_tc_service23_sdcard.py
@@ -4,9 +4,6 @@ Created: 21.01.2020 07:48
@author: Jakob Meier
"""
-import math
-from enum import Enum
-
import config.obsw_config as g
from typing import Deque, Union
@@ -17,206 +14,6 @@ from tmtc_core.utility.obsw_logger import get_logger
LOGGER = get_logger()
-class FileTransferHelper:
- """
- This helper class fills the provided TC queue with appropriate PUS telecommands
- to transfer a large file.
- There are three modes which determine which telecommands will be generated:
- 1. NORMAL: Generate telecommand to create a new file and append data packets if
- the file data is too large. This will be the default mode.
- 2. DELETE_OLD: Generate telecommand to delete old file and then perform same steps as the
- normal mode
- 3. RENAME_OLD: Rename old file and then perform same steps as in normal mode.
-
- Please note that the setter functions set_data have to be used to assign data, otherwise
- an empty file will be created. The mode is set with setter commands as well.
- """
- class TransferMode(Enum):
- # Normal mode
- NORMAL = 1
- # Generate a command to delete the old file first
- DELETE_OLD = 2
- # Generate a command to rename the old file first.
- RENAME_OLD = 3
-
- def __init__(self, tc_queue: TcQueueT, max_size_of_app_data: int,
- target_repository: str, target_filename: str,
- object_id=g.SD_CARD_HANDLER_ID):
- """
- @param tc_queue: TC queue which will be filled
- @param max_size_of_app_data: Maximum allowed app data size. Number of generated packets
- will depend on this value
- @param target_repository: Repository path on target.
- @param target_filename: Filename on target
- @param object_id:
- """
- self.__transfer_mode = self.TransferMode.NORMAL
- self.max_size_of_app_data = max_size_of_app_data
- self.__max_file_data_size = 0
- self.allowed_file_data_size = calculate_allowed_file_data_size(
- max_size_of_app_data, target_filename, target_repository)
- self.target_filename = target_filename
- self.target_repository = target_repository
- self.__renamed_name = self.target_filename + "old"
- self.object_id = object_id
- self.tc_queue = tc_queue
-
- self.__number_of_packets = 0
- self.__number_of_append_packets = 0
- self.__number_of_create_packets = 1
- self.__number_of_delete_packets = 0
- self.__number_of_finish_packets = 1
-
- self.__lock_file = True
- self.__local_filename = ""
- self.__file_data = bytearray()
- # This will generate a telecommand to delete the old file, if it exists
- self.delete_old_file = False
- # This will generater a telecommand to rename the old file, if it exists
- self.rename_old_file = False
-
- def set_data_from_file(self, local_filename: str):
- with open(local_filename, 'rb') as file:
- self.__file_data = file.read()
-
- def set_data_raw(self, tc_data: bytearray):
- self.__file_data = tc_data
-
- def set_to_delete_old_file(self):
- self.__transfer_mode = self.TransferMode.DELETE_OLD
-
- def set_to_rename_old_file(self, renamed_name: str):
- self.__transfer_mode = self.TransferMode.RENAME_OLD
- self.__renamed_name = renamed_name
-
- def set_to_lock_file(self, lock_file: bool):
- self.__lock_file = lock_file
-
- def get_number_of_packets_generated(self):
- return self.__number_of_packets
-
- def set_max_file_data_size(self, max_file_data_size: int):
- """
- If this value is specified and the source file is large (larger than the maximum allowed
- app data!), the file data size will be set to this value.
- @param max_file_data_size:
- @return:
- """
- self.__max_file_data_size = max_file_data_size
-
- def file_size(self):
- return len(self.__file_data)
-
- def generate_packets(self, ssc: int):
- if self.__transfer_mode == self.TransferMode.DELETE_OLD:
- command = generate_rm_file_srv23_2_packet(self.target_filename, self.target_repository,
- ssc, self.object_id)
- self.__number_of_delete_packets = 1
- ssc += 1
- self.tc_queue.appendleft(command.pack_command_tuple())
- elif self.__transfer_mode == self.TransferMode.RENAME_OLD:
- # not implemented yet
- pass
- if len(self.__file_data) > self.allowed_file_data_size:
- # Large file, create file with init_data
- if self.__max_file_data_size > 0:
- init_data = self.__file_data[0:self.__max_file_data_size]
- else:
- init_data = self.__file_data[0:self.allowed_file_data_size]
- else:
- # Small file, one packet for file creation sufficient
- command = generate_create_file_srv23_1_packet(
- self.target_filename, self.target_repository, ssc, self.max_size_of_app_data,
- self.__file_data)
- ssc += 1
- self.tc_queue.appendleft(command.pack_command_tuple())
- return
-
- # Create large file.
- command = generate_create_file_srv23_1_packet(
- self.target_filename, self.target_repository, ssc, self.max_size_of_app_data,
- init_data)
- ssc += 1
- self.tc_queue.appendleft(command.pack_command_tuple())
- rest_of_data = self.__file_data[self.allowed_file_data_size:]
- # Generate the rest of the packets to write to large file
- if self.__max_file_data_size > 0:
- self.__generate_append_to_file_packets_automatically(
- data=rest_of_data, target_repository=self.target_repository,
- target_filename=self.target_filename, size_of_data_blocks=self.__max_file_data_size,
- init_ssc=ssc)
- else:
- self.__generate_append_to_file_packets_automatically(
- data=rest_of_data, target_repository=self.target_repository,
- target_filename=self.target_filename, size_of_data_blocks=self.max_size_of_app_data,
- init_ssc=ssc)
- ssc += 1
- last_command = generate_finish_append_to_file_srv23_131_packet(
- filename=self.target_filename, repository_path=self.target_repository,
- ssc=ssc, lock_file=self.__lock_file)
- self.tc_queue.appendleft(last_command.pack_command_tuple())
- self.__number_of_packets = \
- self.__number_of_append_packets + self.__number_of_create_packets + \
- self.__number_of_delete_packets + 1
-
- def __generate_append_to_file_packets_automatically(
- self, data: bytearray, target_repository: str, target_filename: str,
- size_of_data_blocks: int, init_ssc: int):
- """
- This function generates PUS packets which is used to write data in a file.
- A new file will be created if not already existing. If the file already exists, this might
- lead to
-
- If the file data is larger than the maximum allowed size of application data, this function
- will split the data into multiple packets and increment the initial SSC number by one for
- each packet.
- @param data: Data which will be split up.
- @param init_ssc: First SSC, which will be incremented for each packet.
- """
- header = bytearray(self.object_id)
- header += target_repository.encode('utf-8')
- # Add string terminator of repository path
- header.append(0)
- header += target_filename.encode('utf-8')
- # Add string terminator of filename
- header.append(0)
- self.__split_large_file(header, size_of_data_blocks, data, init_ssc)
-
- def __split_large_file(self, header: bytearray, size_of_data_blocks: int,
- data: bytearray, init_ssc: int):
- """
- This function splits a large file in multiple packets and packs the generated packets
- into the member deque. This is necessary because the packet size is limited.
- @param header: Repository and file name which will always stay the same
- @param size_of_data_blocks: The file data blocks will have this size
- @param data: The data to pack in multiple packets
- @param init_ssc: The ssc of the first command, will be incremented by one for each packet.
- """
- number_of_packets = math.floor(len(data) / size_of_data_blocks)
- packet_sequence_number = 0
-
- for i in range(number_of_packets):
- header.append(packet_sequence_number >> 8)
- header.append(0xFF & packet_sequence_number)
- header += data[i * size_of_data_blocks:(i + 1) * size_of_data_blocks]
-
- commands = PusTelecommand(service=23, subservice=130, ssc=init_ssc + i,
- app_data=header)
- self.tc_queue.appendleft(commands.pack_command_tuple())
-
- # Remove everything except the header
- header = header[:len(header) - size_of_data_blocks - 2]
- packet_sequence_number = packet_sequence_number + 1
- # Last packet will be subservice 131 to finish the append operation
- header.append(packet_sequence_number >> 8)
- header.append(0xFF & packet_sequence_number)
- self.__number_of_append_packets += number_of_packets
- header += data[number_of_packets * size_of_data_blocks:len(data)]
- commands = PusTelecommand(service=23, subservice=130, ssc=init_ssc + packet_sequence_number,
- app_data=header)
- self.tc_queue.appendleft(commands.pack_command_tuple())
-
-
def generate_print_sd_card_packet(
ssc: int, object_id: bytearray = g.SD_CARD_HANDLER_ID) -> PusTelecommand:
app_data = bytearray(object_id)
diff --git a/tmtc_core b/tmtc_core
index 233383842fdaca1480d25ce4568da54e8305ffaa..228d57103721942918f38b54e0b5e84f3acae427 160000
--- a/tmtc_core
+++ b/tmtc_core
@@ -1 +1 @@
-Subproject commit 233383842fdaca1480d25ce4568da54e8305ffaa
+Subproject commit 228d57103721942918f38b54e0b5e84f3acae427
diff --git a/utility/obsw_binary_uploader.py b/utility/obsw_binary_uploader.py
index 7f27a9e3f6255ef4cac7671785d2af0ee00f9e31..add1696883cd2dad5bd42d1069432505234b3205 100644
--- a/utility/obsw_binary_uploader.py
+++ b/utility/obsw_binary_uploader.py
@@ -12,183 +12,208 @@ import tkinter as tk
from tkinter import filedialog
from collections import deque
from glob import glob
+from typing import Deque
from tmtc_core.comIF.obsw_com_interface import CommunicationInterface
-from tc.obsw_tc_service23_sdcard import FileTransferHelper
+from utility.obsw_file_transfer_helper 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 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..")
+class BinaryFileUploader:
+ def __init__(self, com_if: CommunicationInterface, tmtc_printer: TmTcPrinter,
+ tm_listener: TmListener):
+ """
+ Initializes the binary file uploader with the required components.
+ @param com_if:
+ @param tmtc_printer:
+ @param tm_listener:
+ """
+ self.com_if = com_if
+ self.tmtc_printer = tmtc_printer
+ self.tm_listener = tm_listener
+
+ def perform_file_upload(self):
+ 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
+ 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
- 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:
+ 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(os.path.sep, 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")
- print_string = file_path.rsplit(os.path.sep, 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
-
- prompt_lock = input("Lock file with last packet? [y/n]: ")
- if prompt_lock in ['n', "no", 0]:
- prompt_lock = False
- else:
- prompt_lock = True
-
- if bootloader_prompt:
- file_name = "bl.bin"
- else:
- file_name = "obsw_up.bin"
-
- if iobc_prompt:
- if bootloader_prompt:
- repository_name = "BIN/IOBC/BL"
+ iobc_prompt = input("iOBC? [y/n]: ")
+ if iobc_prompt in ['y', 'yes', 1]:
+ iobc_prompt = True
else:
- repository_name = "BIN/IOBC/OBSW"
- 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
+
+ prompt_lock = input("Lock file with last packet? [y/n]: ")
+ if prompt_lock in ['n', "no", 0]:
+ prompt_lock = False
+ else:
+ prompt_lock = True
+
if bootloader_prompt:
- repository_name = "BIN/AT91/BL"
+ 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"
+
+ if calc_hamming_code:
+ pass
+
+ # 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
+
+ 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
+ self.tmtc_printer.set_display_mode(DisplayMode.SHORT)
+
+ # Configure file transfer helper
+ file_transfer_helper.set_data_from_file(file_path)
+ file_transfer_helper.set_to_delete_old_file()
+ if prompt_lock:
+ file_transfer_helper.set_to_lock_file(prompt_lock)
+ else:
+ file_transfer_helper.set_to_lock_file(prompt_lock)
+
+ # Generate the packets.
+ file_transfer_helper.generate_packets(init_ssc)
+
+ self.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.get_number_of_packets_generated()
+ print_string = "BinaryUploader: " + str(total_num_packets) + \
+ " packets generated."
+ if prompt_lock:
+ print_string += " File will be locked."
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
-
- 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)
-
- # Configure file transfer helper
- file_transfer_helper.set_data_from_file(file_path)
- file_transfer_helper.set_to_delete_old_file()
- if prompt_lock:
- file_transfer_helper.set_to_lock_file(prompt_lock)
- else:
- file_transfer_helper.set_to_lock_file(prompt_lock)
-
- # Generate the packets.
- 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.get_number_of_packets_generated()
- print_string = "BinaryUploader: " + str(total_num_packets) + \
- " packets generated."
- if prompt_lock:
- print_string += " File will be locked."
- else:
- print_string += " File will not be locked."
- 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(round(total_time - (idx - 2) * interval, 2)) + \
- " 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)
- pass
- tm_listener.clear_tm_packet_queue()
- LOGGER.info("Transitioning back to listener mode..")
+ print_string += " File will not be locked."
+ LOGGER.info(print_string)
+
+ reception_deque = deque()
+ self.__perform_send_algorithm(tc_queue, total_num_packets, reception_deque)
+
+ 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(self.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)
+ pass
+ self.tm_listener.clear_tm_packet_queue()
+ LOGGER.info("Transitioning back to listener mode..")
+
+ def __perform_send_algorithm(self, tc_queue: Deque, number_of_packets: int, reception_deque:
+ Deque):
+ interval = 0.6
+ last_check = time.time()
+ last_sent = time.time()
+ total_time = interval * number_of_packets
+ idx = 1
+ 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
+ self.com_if.send_telecommand(tc_packet, tc_info)
+ self.tmtc_printer.print_telecommand(tc_packet, tc_info)
+ elif tc_packet == "print":
+ LOGGER.info(tc_info)
+ remaining_time_string = "Remaining time: " + \
+ str(round(total_time - (idx - 2) * interval, 2)) + \
+ " seconds"
+ print_progress_bar(idx - 2, number_of_packets, print_end="\n",
+ suffix=remaining_time_string)
+ # sys.stdout.write("\033[F") # Cursor up one line
+ packets_received = self.tm_listener.retrieve_tm_packet_queue()
+ reception_deque.extend(packets_received)
+ # Every 5 seconds, check whether any reply has been received. If not, cancel operation.
+ if time.time() - last_check > 5.0 and len(packets_received) == 0:
+ LOGGER.warning("No replies are being received, cancelling upload operation..")
+ time_to_sleep = next_send - time.time()
+ last_sent = next_send
+ time.sleep(time_to_sleep)
# https://stackoverflow.com/questions/3173320/text-progress-bar-in-the-console
diff --git a/utility/obsw_file_transfer_helper.py b/utility/obsw_file_transfer_helper.py
new file mode 100644
index 0000000000000000000000000000000000000000..c6efb90e7e15cf98b494e9c895f9137b88107c6f
--- /dev/null
+++ b/utility/obsw_file_transfer_helper.py
@@ -0,0 +1,241 @@
+from enum import Enum
+import math
+
+from config.obsw_config import SD_CARD_HANDLER_ID
+from tmtc_core.tc.obsw_pus_tc_base import TcQueueT, PusTelecommand
+from tc.obsw_tc_service23_sdcard import \
+ calculate_allowed_file_data_size, generate_rm_file_srv23_2_packet, \
+ generate_create_file_srv23_1_packet, generate_finish_append_to_file_srv23_131_packet, \
+ generate_lock_file_srv23_5_6_packet
+
+
+class FileTransferHelper:
+ """
+ This helper class fills the provided TC queue with appropriate PUS telecommands
+ to transfer a file.
+ There are three modes which determine which telecommands will be generated:
+ 1. NORMAL: Generate telecommand to create a new file and append data packets if
+ the file data is too large. This will be the default mode.
+ 2. DELETE_OLD: Generate telecommand to delete old file and then perform same steps as the
+ normal mode
+ 3. RENAME_OLD: Rename old file and then perform same steps as in normal mode.
+
+ Please note that the setter functions set_data have to be used to assign data, otherwise
+ an empty file will be created. The mode is set with setter commands as well.
+ """
+ class TransferMode(Enum):
+ # Normal mode
+ NORMAL = 1
+ # Generate a command to delete the old file first
+ DELETE_OLD = 2
+ # Generate a command to rename the old file first.
+ RENAME_OLD = 3
+
+ def __init__(self, tc_queue: TcQueueT, max_size_of_app_data: int,
+ target_repository: str, target_filename: str,
+ object_id=SD_CARD_HANDLER_ID):
+ """
+ @param tc_queue: TC queue which will be filled
+ @param max_size_of_app_data: Maximum allowed app data size. Number of generated packets
+ will depend on this value
+ @param target_repository: Repository path on target.
+ @param target_filename: Filename on target
+ @param object_id:
+ """
+ self.object_id = object_id
+ self.max_size_of_app_data = max_size_of_app_data
+ self.allowed_file_data_size = calculate_allowed_file_data_size(
+ max_size_of_app_data, target_filename, target_repository)
+
+ self.target_filename = target_filename
+ self.target_repository = target_repository
+
+ self.tc_queue = tc_queue
+
+ self.__transfer_mode = self.TransferMode.NORMAL
+ self.__max_file_data_size = 0
+ self.__renamed_name = self.target_filename + "old"
+
+ self.__large_file = False
+ self.__number_of_packets = 0
+ self.__number_of_append_packets = 0
+ self.__number_of_create_packets = 1
+ self.__number_of_delete_packets = 0
+ self.__number_of_finish_packets = 1
+
+ self.__current_ssc = 0
+ self.__lock_file = True
+ self.__local_filename = ""
+ self.__file_data = bytearray()
+ # This will generate a telecommand to delete the old file, if it exists
+ self.delete_old_file = False
+ # This will generater a telecommand to rename the old file, if it exists
+ self.rename_old_file = False
+
+ def set_data_from_file(self, local_filename: str):
+ with open(local_filename, 'rb') as file:
+ self.__file_data = file.read()
+
+ def set_data_raw(self, tc_data: bytearray):
+ self.__file_data = tc_data
+
+ def set_to_delete_old_file(self):
+ self.__transfer_mode = self.TransferMode.DELETE_OLD
+
+ def set_to_rename_old_file(self, renamed_name: str):
+ self.__transfer_mode = self.TransferMode.RENAME_OLD
+ self.__renamed_name = renamed_name
+
+ def set_to_lock_file(self, lock_file: bool):
+ """
+ Command will be sent to lock file after succesfull transfer
+ @param lock_file:
+ @return:
+ """
+ self.__lock_file = lock_file
+
+ def get_number_of_packets_generated(self):
+ return self.__number_of_packets
+
+ def set_max_file_data_size(self, max_file_data_size: int):
+ """
+ If this value is specified and the source file is large (larger than the maximum allowed
+ app data!), the file data size will be set to this value.
+ @param max_file_data_size:
+ @return:
+ """
+ self.__max_file_data_size = max_file_data_size
+
+ def file_size(self):
+ return len(self.__file_data)
+
+ def generate_packets(self, ssc: int):
+ """
+ Main function to generate all packets and fill them into the provided deque.
+ @param ssc:
+ @return:
+ """
+ self.__current_ssc = ssc
+ self.__handle_delete_packet_generation()
+ if self.__transfer_mode == self.TransferMode.RENAME_OLD:
+ # not implemented yet
+ pass
+ self.__handle_create_file_packet_generation()
+ self.__handle_finish_and_lock_packet_generation()
+ self.__number_of_packets = \
+ self.__number_of_create_packets + self.__number_of_append_packets + \
+ self.__number_of_delete_packets + self.__number_of_finish_packets
+
+ def __handle_delete_packet_generation(self):
+ if self.__transfer_mode == self.TransferMode.DELETE_OLD:
+ command = generate_rm_file_srv23_2_packet(
+ filename=self.target_filename, repository_path=self.target_repository,
+ ssc=self.__current_ssc, object_id=self.object_id)
+ self.__number_of_delete_packets = 1
+ self.__current_ssc += 1
+ self.tc_queue.appendleft(command.pack_command_tuple())
+
+ def __handle_create_file_packet_generation(self):
+ if len(self.__file_data) > self.allowed_file_data_size:
+ # Large file, create file with init_data
+ if self.__max_file_data_size > 0:
+ init_data = self.__file_data[0:self.__max_file_data_size]
+ else:
+ init_data = self.__file_data[0:self.allowed_file_data_size]
+ self.__large_file = True
+ else:
+ init_data = self.__file_data
+
+ # Create file.
+ command = generate_create_file_srv23_1_packet(
+ self.target_filename, self.target_repository, ssc=self.__current_ssc,
+ max_size_of_app_data=self.max_size_of_app_data, initial_data=init_data)
+ self.__current_ssc += 1
+ self.tc_queue.appendleft(command.pack_command_tuple())
+ if not self.__large_file:
+ return
+ rest_of_data = self.__file_data[self.allowed_file_data_size:]
+ # Generate the rest of the packets to write to large file
+ if self.__max_file_data_size > 0:
+ self.__generate_append_to_file_packets_automatically(
+ data=rest_of_data, target_repository=self.target_repository,
+ target_filename=self.target_filename, size_of_data_blocks=self.__max_file_data_size,
+ init_ssc=self.__current_ssc)
+ else:
+ self.__generate_append_to_file_packets_automatically(
+ data=rest_of_data, target_repository=self.target_repository,
+ target_filename=self.target_filename, size_of_data_blocks=self.max_size_of_app_data,
+ init_ssc=self.__current_ssc)
+ self.__current_ssc += 1
+
+ def __generate_append_to_file_packets_automatically(
+ self, data: bytearray, target_repository: str, target_filename: str,
+ size_of_data_blocks: int, init_ssc: int):
+ """
+ This function generates PUS packets which is used to write data in a file.
+ A new file will be created if not already existing. If the file already exists, this might
+ lead to
+
+ If the file data is larger than the maximum allowed size of application data, this function
+ will split the data into multiple packets and increment the initial SSC number by one for
+ each packet.
+ @param data: Data which will be split up.
+ @param init_ssc: First SSC, which will be incremented for each packet.
+ """
+ header = bytearray(self.object_id)
+ header += target_repository.encode('utf-8')
+ # Add string terminator of repository path
+ header.append(0)
+ header += target_filename.encode('utf-8')
+ # Add string terminator of filename
+ header.append(0)
+ self.__split_large_file(header, size_of_data_blocks, data, init_ssc)
+
+ def __split_large_file(self, header: bytearray, size_of_data_blocks: int,
+ data: bytearray, init_ssc: int):
+ """
+ This function splits a large file in multiple packets and packs the generated packets
+ into the member deque. This is necessary because the packet size is limited.
+ @param header: Repository and file name which will always stay the same
+ @param size_of_data_blocks: The file data blocks will have this size
+ @param data: The data to pack in multiple packets
+ @param init_ssc: The ssc of the first command, will be incremented by one for each packet.
+ """
+ number_of_packets = math.floor(len(data) / size_of_data_blocks)
+ packet_sequence_number = 0
+
+ for i in range(number_of_packets):
+ header.append(packet_sequence_number >> 8)
+ header.append(0xFF & packet_sequence_number)
+ header += data[i * size_of_data_blocks:(i + 1) * size_of_data_blocks]
+
+ commands = PusTelecommand(service=23, subservice=130, ssc=init_ssc + i,
+ app_data=header)
+ self.tc_queue.appendleft(commands.pack_command_tuple())
+
+ # Remove everything except the header
+ header = header[:len(header) - size_of_data_blocks - 2]
+ packet_sequence_number = packet_sequence_number + 1
+ # Last packet will be subservice 131 to finish the append operation
+ header.append(packet_sequence_number >> 8)
+ header.append(0xFF & packet_sequence_number)
+ self.__number_of_append_packets += number_of_packets
+ header += data[number_of_packets * size_of_data_blocks:len(data)]
+ commands = PusTelecommand(service=23, subservice=130, ssc=init_ssc + packet_sequence_number,
+ app_data=header)
+ self.tc_queue.appendleft(commands.pack_command_tuple())
+
+ def __handle_finish_and_lock_packet_generation(self):
+ if self.__large_file:
+ last_command = generate_finish_append_to_file_srv23_131_packet(
+ filename=self.target_filename, repository_path=self.target_repository,
+ ssc=self.__current_ssc, lock_file=self.__lock_file)
+ else:
+ if self.__lock_file:
+ last_command = generate_lock_file_srv23_5_6_packet(
+ filename=self.target_filename, repository_path=self.target_repository,
+ object_id=self.object_id, lock=True, ssc=self.__current_ssc)
+ else:
+ self.__number_of_finish_packets = 0
+ return
+ self.tc_queue.appendleft(last_command.pack_command_tuple())
\ No newline at end of file