initiated tmtc commander

README.md

@@ -1,2 +1,8 @@
-EIVE TMTC Commander
-======
+### How to use this folder
+
+This folder contains template files to set up the TMTC commander 
+for a new mission or project. These files are the adaption
+point to customize the TMTC commander.
+
+To do so, simply copy all folder inside the TMTC commander root. This
+step is also required because the TMTC commander core will load some modules.

config/tmtcc_com_config.py

@@ -0,0 +1,18 @@
+"""
+@brief      This file transfers control of communication interface setup to the user.
+@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
+            it to your needs.
+"""
+from typing import Union
+
+from tmtc_core.com_if.tmtcc_com_interface_base import CommunicationInterface
+from tmtc_core.core.tmtc_core_definitions import ComInterfaces
+from tmtc_core.core.tmtcc_com_if_setup import create_communication_interface_default
+from tmtc_core.utility.tmtcc_tmtc_printer import TmTcPrinter
+
+
+def create_communication_interface_user(com_if: ComInterfaces, tmtc_printer: TmTcPrinter) -> \
+        Union[CommunicationInterface, None]:
+    return create_communication_interface_default(com_if, tmtc_printer)
+
+

config/tmtcc_definitions.py

@@ -0,0 +1,41 @@
+"""
+@brief      This file transfers control of the custom definitions like modes to the user.
+@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
+            it to your needs.
+"""
+
+import enum
+from enum import auto
+
+
+# Mode options, set by args parser
+class ModeList(enum.Enum):
+    Idle = 0
+    ListenerMode = 1
+    SingleCommandMode = 2
+    ServiceTestMode = 3
+    SoftwareTestMode = 4
+    PromptMode = 32
+
+
+class ServiceList(enum.Enum):
+    SERVICE_2 = 0
+    SERVICE_3 = auto()
+    SERVICE_5 = auto()
+    SERVICE_8 = auto()
+    SERVICE_9 = auto()
+    SERVICE_17 = auto()
+    SERVICE_20 = auto()
+    SERVICE_200 = auto()
+
+
+class SerialConfig(enum.Enum):
+    SERIAL_PORT = auto()
+    SERIAL_BAUD_RATE = auto()
+    SERIAL_TIMEOUT = auto()
+    SERIAL_COMM_TYPE = auto()
+
+
+class EthernetConfig(enum.Enum):
+    SEND_ADDRESS = auto()
+    RECV_ADDRESS = auto()

config/tmtcc_globals.py

@@ -0,0 +1,223 @@
+"""
+@brief      This file transfers definitions of global variables to the user.
+@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
+            it to your needs.
+"""
+
+import enum
+import argparse
+
+
+# All globals can be added here and will be part of a globals dictionary.
+from config.tmtcc_definitions import ServiceList, ModeList
+
+from tmtc_core.com_if.tmtcc_serial_com_if import SerialCommunicationType
+from tmtc_core.com_if.tmtcc_serial_utilities import determine_com_port
+from tmtc_core.core.tmtc_core_definitions import ComInterfaces
+from tmtc_core.utility.tmtcc_logger import get_logger
+
+
+class GlobalIds(enum.Enum):
+    from enum import auto
+    # Generic parameters
+    APID = auto()
+    MODE = auto()
+    SERVICE = auto()
+    SERVICELIST = auto()
+    COM_IF = auto()
+    OP_CODE = auto()
+    TM_TIMEOUT = auto()
+
+    # Miscellaneous
+    DISPLAY_MODE = auto()
+    USE_LISTENER_AFTER_OP = auto()
+    PRINT_HK = auto()
+    PRINT_TM = auto()
+    PRINT_RAW_TM = auto()
+    PRINT_TO_FILE = auto()
+    RESEND_TC = auto()
+    TC_SEND_TIMEOUT_FACTOR = auto()
+
+    # Config dictionaries
+    USE_SERIAL = auto()
+    SERIAL_CONFIG = auto()
+    USE_ETHERNET = auto()
+    ETHERNET_CONFIG = auto()
+
+    # Object handles
+    PRETTY_PRINTER = auto()
+    TM_LISTENER_HANDLE = auto()
+    COM_INTERFACE_HANDLE = auto()
+    TMTC_PRINTER_HANDLE = auto()
+
+
+def add_globals_pre_args_parsing(gui: bool = False):
+    from tmtc_core.core.tmtcc_globals_manager import update_global
+    import pprint
+
+    update_global(GlobalIds.APID, 0xef)
+    update_global(GlobalIds.COM_IF, ComInterfaces.EthernetUDP)
+    update_global(GlobalIds.TC_SEND_TIMEOUT_FACTOR, 2)
+    update_global(GlobalIds.TM_TIMEOUT, 4)
+    update_global(GlobalIds.DISPLAY_MODE, "long")
+    update_global(GlobalIds.PRINT_TO_FILE, True)
+
+    update_global(GlobalIds.SERIAL_CONFIG, dict())
+    update_global(GlobalIds.ETHERNET_CONFIG, dict())
+    pp = pprint.PrettyPrinter()
+    update_global(GlobalIds.PRETTY_PRINTER, pp)
+    update_global(GlobalIds.TM_LISTENER_HANDLE, None)
+    update_global(GlobalIds.COM_INTERFACE_HANDLE, None)
+    update_global(GlobalIds.TMTC_PRINTER_HANDLE, None)
+    update_global(GlobalIds.PRINT_RAW_TM, False)
+    update_global(GlobalIds.RESEND_TC, False)
+
+    update_global(GlobalIds.OP_CODE, "0")
+    update_global(GlobalIds.MODE, ModeList.ListenerMode)
+
+    if gui:
+        set_up_ethernet_cfg()
+
+    servicelist = dict()
+    servicelist[ServiceList.SERVICE_2] = ["Service 2 Raw Commanding"]
+    servicelist[ServiceList.SERVICE_3] = ["Service 3 Housekeeping"]
+    servicelist[ServiceList.SERVICE_5] = ["Service 5 Event"]
+    servicelist[ServiceList.SERVICE_8] = ["Service 8 Functional Commanding"]
+    servicelist[ServiceList.SERVICE_9] = ["Service 9 Time"]
+    servicelist[ServiceList.SERVICE_17] = ["Service 17 Test"]
+    servicelist[ServiceList.SERVICE_20] = ["Service 20 Parameters"]
+    servicelist[ServiceList.SERVICE_23] = ["Service 23 File Management"]
+    servicelist[ServiceList.SERVICE_200] = ["Service 200 Mode Management"]
+    update_global(GlobalIds.SERVICE, ServiceList.SERVICE_17)
+    update_global(GlobalIds.SERVICELIST, servicelist)
+
+
+def add_globals_post_args_parsing(args: argparse.Namespace):
+    from tmtc_core.core.tmtcc_globals_manager import update_global
+    from config.tmtcc_definitions import ModeList
+    logger = get_logger()
+
+    mode_param = ModeList.ListenerMode
+    if 0 <= args.mode <= 6:
+        if args.mode == 0:
+            mode_param = ModeList.GUIMode
+        elif args.mode == 1:
+            mode_param = ModeList.ListenerMode
+        elif args.mode == 2:
+            mode_param = ModeList.SingleCommandMode
+        elif args.mode == 3:
+            mode_param = ModeList.ServiceTestMode
+        elif args.mode == 4:
+            mode_param = ModeList.SoftwareTestMode
+    update_global(GlobalIds.MODE, mode_param)
+
+    if args.com_if == ComInterfaces.EthernetUDP.value:
+        com_if = ComInterfaces.EthernetUDP
+    elif args.com_if == ComInterfaces.Serial.value:
+        com_if = ComInterfaces.Serial
+    elif args.com_if == ComInterfaces.Dummy.value:
+        com_if = ComInterfaces.Dummy
+    elif args.com_if == ComInterfaces.QEMU.value:
+        com_if = ComInterfaces.QEMU
+    else:
+        com_if = ComInterfaces.Serial
+    update_global(GlobalIds.COM_IF, com_if)
+
+    if args.short_display_mode:
+        display_mode_param = "short"
+    else:
+        display_mode_param = "long"
+    update_global(GlobalIds.DISPLAY_MODE, display_mode_param)
+
+    service = str(args.service).lower()
+    if service == "2":
+        service = ServiceList.SERVICE_2
+    elif service == "3":
+        service = ServiceList.SERVICE_3
+    elif service == "5":
+        service = ServiceList.SERVICE_5
+    elif service == "8":
+        service = ServiceList.SERVICE_8
+    elif service == "9":
+        service = ServiceList.SERVICE_9
+    elif service == "17":
+        service = ServiceList.SERVICE_17
+    elif service == "20":
+        service = ServiceList.SERVICE_20
+    elif service == "23":
+        service = ServiceList.SERVICE_23
+    else:
+        logger.warning("Service not known! Setting standard service 17")
+        service = ServiceList.SERVICE_17
+
+    update_global(GlobalIds.SERVICE, service)
+
+    if args.op_code is None:
+        op_code = 0
+    else:
+        op_code = str(args.op_code).lower()
+    update_global(GlobalIds.OP_CODE, op_code)
+
+    update_global(GlobalIds.USE_LISTENER_AFTER_OP, args.listener)
+    update_global(GlobalIds.TM_TIMEOUT, args.tm_timeout)
+    update_global(GlobalIds.PRINT_HK, args.print_hk)
+    update_global(GlobalIds.PRINT_TM, args.print_tm)
+    update_global(GlobalIds.PRINT_RAW_TM, args.raw_data_print)
+    update_global(GlobalIds.PRINT_TO_FILE, args.print_log)
+    update_global(GlobalIds.RESEND_TC, args.resend_tc)
+    update_global(GlobalIds.TC_SEND_TIMEOUT_FACTOR, 3)
+
+    use_serial_cfg = False
+    if com_if == ComInterfaces.Serial or com_if == ComInterfaces.QEMU:
+        use_serial_cfg = True
+    if use_serial_cfg:
+        set_up_serial_cfg(com_if)
+
+    use_ethernet_cfg = False
+    if com_if == ComInterfaces.EthernetUDP:
+        use_ethernet_cfg = True
+    if use_ethernet_cfg:
+        # TODO: Port and IP address can also be passed as CLI parameters. Use them here if applicable
+        set_up_ethernet_cfg()
+
+
+def set_up_serial_cfg(com_if: ComInterfaces):
+    from tmtc_core.core.tmtcc_globals_manager import update_global
+    update_global(GlobalIds.USE_SERIAL, True)
+    from tmtc_core.core.tmtcc_globals_manager import get_global
+    from config.tmtcc_definitions import SerialConfig
+    serial_cfg_dict = get_global(GlobalIds.SERIAL_CONFIG)
+    if com_if == ComInterfaces.Serial:
+        com_port = determine_com_port()
+    else:
+        com_port = ""
+    serial_cfg_dict.update({SerialConfig.SERIAL_PORT: com_port})
+    serial_cfg_dict.update({SerialConfig.SERIAL_BAUD_RATE: 115200})
+    serial_cfg_dict.update({SerialConfig.SERIAL_TIMEOUT: 0.01})
+    serial_cfg_dict.update({SerialConfig.SERIAL_COMM_TYPE: SerialCommunicationType.DLE_ENCODING})
+    serial_cfg_dict.update({SerialConfig.SERIAL_FRAME_SIZE: 256})
+    serial_cfg_dict.update({SerialConfig.SERIAL_DLE_QUEUE_LEN: 25})
+    serial_cfg_dict.update({SerialConfig.SERIAL_DLE_MAX_FRAME_SIZE: 1024})
+    update_global(GlobalIds.SERIAL_CONFIG, serial_cfg_dict)
+
+
+def set_up_ethernet_cfg():
+    from tmtc_core.core.tmtcc_globals_manager import update_global
+    update_global(GlobalIds.USE_ETHERNET, True)
+    from tmtc_core.core.tmtcc_globals_manager import get_global
+    from config.tmtcc_definitions import EthernetConfig
+    ethernet_cfg_dict = get_global(GlobalIds.ETHERNET_CONFIG)
+    # Local host and unused port as default config
+    default_send_ip = "127.0.0.1"
+    default_send_port = 7301
+    send_address = (default_send_ip, default_send_port)
+    ethernet_cfg_dict.update({EthernetConfig.SEND_ADDRESS: send_address})
+    # Bind to all interfaces (might be insecure!)
+    default_rcv_ip = ''
+    default_rcv_port = 7302
+    recv_address = (default_rcv_ip, default_rcv_port)
+    ethernet_cfg_dict.update({EthernetConfig.RECV_ADDRESS: recv_address})
+    update_global(GlobalIds.ETHERNET_CONFIG, ethernet_cfg_dict)
+
+
+

config/tmtcc_hooks.py

@@ -0,0 +1,15 @@
+"""
+@brief      This file exposes hook functions to the user.
+@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
+            it to your needs.
+"""
+from tmtc_core.pus_tc.tmtcc_pus_tc_base import PusTelecommand
+
+
+def command_preparation_hook() -> PusTelecommand:
+    """
+    Can be used to pack user-defined commands by generating and returning a PusTelecommand
+    class instance
+    """
+    return PusTelecommand(service=17, subservice=1, ssc=20)
+

config/tmtcc_object_ids.py

@@ -0,0 +1,23 @@
+"""
+@brief      This file transfers control of the object IDs to the user.
+@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
+            it to your needs.
+"""
+
+import enum
+from typing import Dict
+
+
+class ObjectIds(enum.Enum):
+    from enum import auto
+    PUS_SERVICE_17 = auto()
+    TEST_DEVICE = auto()
+
+
+def set_object_ids(object_id_dict: Dict[ObjectIds, bytearray]):
+    o_ids = ObjectIds
+    object_id_dict.update(
+        {o_ids.PUS_SERVICE_17: bytearray([0x53, 0x00, 0x00, 0x17]),
+         o_ids.TEST_DEVICE: bytearray([0x44, 0x00, 0xAF, 0xFE]),
+         }
+    )

config/tmtcc_user_mode_op.py

@@ -0,0 +1,22 @@
+"""
+@brief      This file transfers control of custom mode handling to the user.
+@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
+            it to your needs.
+"""
+import sys
+
+from config.tmtcc_definitions import ModeList
+from core.tmtc_backend import TmTcHandler
+from test.obsw_pus_service_test import run_selected_pus_tests
+from tmtc_core.utility.tmtcc_logger import get_logger
+from utility.tmtcc_binary_uploader import BinaryFileUploader
+
+LOGGER = get_logger()
+
+
+def perform_mode_operation_user(tmtc_backend: TmTcHandler, mode: ModeList):
+    """
+    Custom modes can be implemented here
+    """
+    LOGGER.error(f"Unknown mode {mode}, Configuration error !")
+    sys.exit()

config/tmtcc_version.py

@@ -0,0 +1,10 @@
+"""
+@brief      This file transfers control of versioning to the user.
+@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
+            it to your needs.
+"""
+import tmtc_core.core.tmtcc_version as core
+
+SW_NAME = core.SW_NAME
+SW_VERSION = core.SW_VERSION
+SW_SUBVERSION = core.SW_SUBVERSION

pus_tc/tmtcc_tc_packer_hook.py

@@ -0,0 +1,34 @@
+"""
+@brief      This file transfers control of TC packing to the user
+@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
+            it to your needs.
+"""
+
+import os
+from collections import deque
+from typing import Union
+
+from config.tmtcc_definitions import ServiceList
+from tmtc_core.utility.tmtcc_logger import get_logger
+from tmtc_core.pus_tc.tmtcc_pus_tc_base import TcQueueT
+from tmtc_core.pus_tc.tmtcc_tc_service5_event import pack_service5_test_into
+from tmtc_core.pus_tc.tmtcc_tc_service17_test import pack_service17_ping_command
+
+LOGGER = get_logger()
+
+
+def pack_service_queue_user(service: Union[int, str], op_code: int, service_queue: TcQueueT):
+    if service == ServiceList.SERVICE_5:
+        return pack_service5_test_into(service_queue)
+    if service == ServiceList.SERVICE_17:
+        return service_queue.appendleft(pack_service17_ping_command(ssc=1700).pack_command_tuple())
+    LOGGER.warning("Invalid Service !")
+
+
+def create_total_tc_queue_user() -> TcQueueT:
+    if not os.path.exists("log"):
+        os.mkdir("log")
+    tc_queue = deque()
+    pack_service5_test_into(tc_queue)
+    tc_queue.appendleft(pack_service17_ping_command(ssc=1700).pack_command_tuple())
+    return tc_queue

pus_tm/tmtcc_pus_hk_handling.py

@@ -0,0 +1,26 @@
+"""
+@brief      This file transfers control of housekeeping handling (PUS service 3) to the
+            developer
+@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
+            it to your needs.
+"""
+from typing import Tuple
+
+from tmtc_core.pus_tm.obsw_tm_service_3 import Service3Base
+from tmtc_core.utility.tmtcc_logger import get_logger
+
+LOGGER = get_logger()
+
+
+def handle_user_hk_packet(
+        object_id: bytearray, hk_data: bytearray,
+        service3_packet: Service3Base) -> Tuple[list, list, bytearray]:
+    """
+    This function is called when a Service 3 Housekeeping packet is received.
+    @param object_id:
+    @param hk_data:
+    @param service3_packet:
+    @return:
+    """
+    LOGGER.info("Service3TM: Parsing for this SID has not been implemented.")
+    return [], [], bytearray()

pus_tm/tmtcc_pus_tm_factory_hook.py

@@ -0,0 +1,26 @@
+"""
+@brief      This file transfers control of TM parsing to the user
+@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
+            it to your needs.
+"""
+
+from tmtc_core.pus_tm.tmtcc_pus_tm_base import PusTelemetry
+from tmtc_core.utility.tmtcc_logger import get_logger
+
+from tmtc_core.pus_tm.tmtcc_tm_service1 import Service1TM
+from tmtc_core.pus_tm.tmtcc_tm_service5 import Service5TM
+from tmtc_core.pus_tm.tmtcc_tm_service17 import Service17TM
+
+LOGGER = get_logger()
+
+
+def tm_user_factory_hook(raw_tm_packet: bytearray) -> PusTelemetry:
+    service_type = raw_tm_packet[7]
+    if service_type == 1:
+        return Service1TM(raw_tm_packet)
+    if service_type == 5:
+        return Service5TM(raw_tm_packet)
+    if service_type == 17:
+        return Service17TM(raw_tm_packet)
+    LOGGER.info("The service " + str(service_type) + " is not implemented in Telemetry Factory")
+    return PusTelemetry(raw_tm_packet)

requirements.txt

@@ -0,0 +1,4 @@
+crcmod>=1.7
+PyQt5>=5.15.1
+PyQt5-stubs>=5.14.2.2
+pyserial>=3.4

tmtc_client_cli.py

@@ -0,0 +1,37 @@
+#!/usr/bin/python3
+"""
+@brief      TMTC Commander entry point for command line mode.
+@details
+This client was developed by KSat for the SOURCE project to test the on-board software but
+has evolved into a more generic tool for satellite developers to perform TMTC (Telemetry and Telecommand)
+handling and testing via different communication interfaces. Currently, only the PUS standard is
+implemented as a packet standard.
+
+Run this file with the -h flag to display options.
+
+@license
+Copyright 2020 KSat e.V. Stuttgart
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+@author     R. Mueller
+"""
+from tmtc_core.tmtcc_runner import run_tmtc_client
+
+
+def main():
+    run_tmtc_client(False)
+
+
+if __name__ == "__main__":
+    main()

tmtc_client_gui.py

@@ -0,0 +1,37 @@
+#!/usr/bin/python3
+"""
+@brief      TMTC Commander entry point for GUI mode.
+@details
+This client was developed by KSat for the SOURCE project to test the on-board software but
+has evolved into a more generic tool for satellite developers to perform TMTC (Telemetry and Telecommand)
+handling and testing via different communication interfaces. Currently, only the PUS standard is
+implemented as a packet standard.
+
+Run this file with the -h flag to display options.
+
+@license
+Copyright 2020 KSat e.V. Stuttgart
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+@author     R. Mueller
+"""
+from tmtc_core.tmtcc_runner import run_tmtc_client
+
+
+def main():
+    run_tmtc_client(True)
+
+
+if __name__ == "__main__":
+    main()

utility/tmtcc_user_args_parser.py

@@ -0,0 +1,15 @@
+"""
+@brief      This file transfers control of the command line argument parsing to the user.
+@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
+            it to your needs.
+"""
+
+
+def parse_input_arguments_user(print_known_args: bool = False, print_unknown_args: bool = False):
+    """
+    This function by default will build the default argument parser. A custom CLI parser can be
+    built in this function if required.
+    """
+    from tmtc_core.utility.tmtcc_core_args_parser import parse_default_input_arguments
+    parse_default_input_arguments(print_known_args=print_known_args,
+                                  print_unknown_args=print_unknown_args)