eive-tmtc/pus_tc/ploc_supervisor.py

# -*- coding: utf-8 -*-
"""
@file   ploc_supervisor.py
@brief  Tests for commanding the supervisor of the PLOC.
        The supervisor is programmed by Thales.
@author J. Meier
@date   10.07.2021
"""
import struct

from tmtccmd.config.definitions import QueueCommands

from tmtccmd.tc.packer import TcQueueT
from tmtccmd.ecss.tc import PusTelecommand

latchup_id_dict = {
        "0": "0.85V",
        "1": "1.8V",
        "2": "MISC",
        "3": "3.3V",
        "4": "NVM_4XO",
        "5": "MISSION",
        "6": "SAFECOTS"
    }

class SupvActionIds:
    HK_REPORT = 1
    RESTART_MPSOC = 2
    START_MPSOC = 3
    SHUTWOWN_MPSOC = 4
    SEL_MPSOC_BOOT_IMAGE = 5
    SET_BOOT_TIMEOUT = 6
    SET_MAX_RESTART_TRIES = 7
    RESET_MPSOC = 8
    SET_TIME_REF = 9
    DISABLE_HK = 10
    GET_BOOT_STATUS_REPORT = 11
    UPDATE_AVAILABLE = 12
    WATCHDOGS_ENABLE = 13
    WATCHDOGS_CONFIG_TIMEOUT = 14
    ENABLE_LATCHUP_ALERT = 15
    DISABLE_LATCHUP_ALERT = 16
    AUTO_CALIBRATE_ALERT = 17
    SET_ALERT_LIMIT = 18
    SET_ALERT_IRQ_FILTER = 19
    SET_ADC_SWEEP_PERIOD = 20
    SET_ADC_ENABLED_CHANNELS = 21
    SET_ADC_WINDOW_AND_STRIDE = 22
    SET_ADC_THRESHOLD = 23
    GET_LATCHUP_STATUS_REPORT = 24
    COPY_ADC_DATA_TO_MRAM = 25
    ENABLE_NVMS = 26
    SELECT_NVM = 27
    RUN_AUTO_EM_TESTS = 28
    WIPE_MRAM = 29
    DUMP_MRAM = 30
    SET_DBG_VERBOSITY = 31
    CAN_LOOPBACK_TEST = 32
    PRINT_CPU_STATS = 33
    SET_GPIO = 34
    READ_GPIO = 35
    RESTART_SUPERVISOR = 36
    FACTORY_RESET = 37
    REQUEST_LOGGING_DATA = 38
    UPDATE_IMAGE_DATA = 39


class SupvHkIds:
    HK_REPORT = 52
    BOOT_STATUS_REPORT = 53


def pack_ploc_supv_test_into(object_id: bytearray, tc_queue: TcQueueT, op_code: str) -> TcQueueT:
    tc_queue.appendleft(
        (QueueCommands.PRINT,
         "Testing PLOC Supervisor with object id: 0x" + object_id.hex())
    )

    if op_code == "3":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: TC Get Hk Report"))
        command = object_id + struct.pack('!I', SupvActionIds.HK_REPORT)
        command = PusTelecommand(service=8, subservice=128, ssc=20, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "4":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Restart MPSoC"))
        command = object_id + struct.pack('!I', SupvActionIds.RESTART_MPSOC)
        command = PusTelecommand(service=8, subservice=128, ssc=21, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "5":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Start MPSoC"))
        command = object_id + struct.pack('!I', SupvActionIds.START_MPSOC)
        command = PusTelecommand(service=8, subservice=128, ssc=22, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "6":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Shutdown MPSoC"))
        command = object_id + struct.pack('!I', SupvActionIds.SHUTWOWN_MPSOC)
        command = PusTelecommand(service=8, subservice=128, ssc=23, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "7":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Select MPSoC boot image"))
        mem = int(input("MEM (NVM0 - 0 or NVM1 - 1):"))
        bp0 = int(input("BP0 (0 or 1):"))
        bp1 = int(input("BP1 (0 or 1):"))
        bp2 = int(input("BP2 (0 or 1):"))
        command = pack_sel_boot_image_cmd(object_id, mem, bp0, bp1, bp2)
        command = PusTelecommand(service=8, subservice=128, ssc=24, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "8":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set max restart tries"))
        restart_tries = int(input("Set maximum restart tries:"))
        command = object_id + struct.pack('!I', SupvActionIds.SET_MAX_RESTART_TRIES) + struct.pack('!B', restart_tries)
        command = PusTelecommand(service=8, subservice=128, ssc=25, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "9":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Reset MPSoC"))
        command = object_id + struct.pack('!I', SupvActionIds.RESET_MPSOC)
        command = PusTelecommand(service=8, subservice=128, ssc=26, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "10":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set time reference"))
        command = object_id + struct.pack('!I', SupvActionIds.SET_TIME_REF)
        command = PusTelecommand(service=8, subservice=128, ssc=27, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "11":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set boot timeout"))
        boot_timeout = int(input("Specify boot timeout [ms]:"))
        command = object_id + struct.pack('!I', SupvActionIds.SET_BOOT_TIMEOUT) + struct.pack('!I', boot_timeout)
        command = PusTelecommand(service=8, subservice=128, ssc=28, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "12":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Disable HK"))
        command = object_id + struct.pack('!I', SupvActionIds.DISABLE_HK)
        command = PusTelecommand(service=8, subservice=128, ssc=29, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "13":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Request boot status report"))
        command = object_id + struct.pack('!I', SupvActionIds.GET_BOOT_STATUS_REPORT)
        command = PusTelecommand(service=8, subservice=128, ssc=30, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "14":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Update available"))
        command = pack_update_available_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=31, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "15":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Watchdogs Enable"))
        command = pack_watchdogs_enable_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=32, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "16":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Watchdog configure timeout"))
        command = pack_watchdog_config_timeout_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=33, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "17":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Enable latchup alert"))
        command = pack_lachtup_alert_cmd(object_id, True)
        command = PusTelecommand(service=8, subservice=128, ssc=34, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "18":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Disable latchup alert"))
        command = pack_lachtup_alert_cmd(object_id, False)
        command = PusTelecommand(service=8, subservice=128, ssc=35, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "19":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Auto calibrate alert"))
        command = pack_auto_calibrate_alert_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=36, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "20":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set alert limit"))
        command = pack_set_alert_limit_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=37, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "21":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set alert irq filter"))
        command = pack_set_alert_irq_filter_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=38, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "22":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set ADC sweep period"))
        command = pack_set_adc_sweep_period_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=39, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "23":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set ADC enabled channels"))
        command = pack_set_adc_enabled_channels_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=40, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "24":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set ADC window and stride"))
        command = pack_set_adc_window_and_stride_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=41, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "25":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set ADC threshold"))
        command = pack_set_adc_threshold_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=42, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "26":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Request latchup status report"))
        command = object_id + struct.pack('!I', SupvActionIds.GET_LATCHUP_STATUS_REPORT)
        command = PusTelecommand(service=8, subservice=128, ssc=43, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "27":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Copy ADC data to MRAM"))
        command = object_id + struct.pack('!I', SupvActionIds.COPY_ADC_DATA_TO_MRAM)
        command = PusTelecommand(service=8, subservice=128, ssc=44, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "28":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Enalbe/Disable NVMs"))
        command = pack_enable_nvms_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=45, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "29":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Select NVM"))
        command = pack_select_nvm_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=45, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "30":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Run auto EM tests"))
        command = pack_auto_em_tests_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=45, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "31":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Wipe MRAM"))
        command = pack_mram_wipe_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=46, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "32":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Dump MRAM"))
        command = pack_mram_dump_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=47, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "33":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Print CPU stats"))
        command = pack_print_cpu_stats_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=48, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "34":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set debug verbosity"))
        command = pack_set_debug_verbosity_cmd(object_id)
        command = PusTelecommand(service=8, subservice=128, ssc=49, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())

    return tc_queue


def pack_sel_boot_image_cmd(object_id: bytearray, mem: int, bp0: int, bp1: int, bp2: int) -> bytearray:
    """ This function can be used to generate the command to select the image from which the MPSoC will boot
    @param object_id    The object id of the PLOC supervisor handler.
    @param mem  The memory from which the MPSoC shall boot (NVM0 - 0, NVM1 - 1)
    @param bp0  Partition pin 0
    @param bp1  Partition pin 1
    @param bp2  Partition pin 2
    """
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.SEL_MPSOC_BOOT_IMAGE)
    command = command + struct.pack('!B', mem)
    command = command + struct.pack('!B', bp0)
    command = command + struct.pack('!B', bp1)
    command = command + struct.pack('!B', bp2)
    return command


def pack_update_available_cmd(object_id: bytearray) -> bytearray:
    """
    @brief  This function packs the udpate availabe command.
    @param object_id    The object id of the PLOC supervisor handler.
    """
    image_select = 1
    image_partition = 0
    image_size = 222
    image_crc = 0x0
    number_of_packets = 150
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.UPDATE_AVAILABLE)
    command = command + struct.pack('!B', image_select)
    command = command + struct.pack('!B', image_partition)
    command = command + struct.pack('!I', image_size)
    command = command + struct.pack('!I', image_crc)
    command = command + struct.pack('!I', number_of_packets)
    return command


def pack_watchdogs_enable_cmd(object_id: bytearray) -> bytearray:
    """
    @brief  This function packs the command to enable or disable watchdogs on the PLOC.
    @param object_id    The object id of the PLOC supervisor handler.
    @note   Enable = 1, Disable = 0
    """
    watchdog_ps = 1
    watchdog_pl = 1
    watchdog_int = 0
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.WATCHDOGS_ENABLE)
    command = command + struct.pack('!B', watchdog_ps)
    command = command + struct.pack('!B', watchdog_pl)
    command = command + struct.pack('!B', watchdog_int)
    return command


def pack_watchdog_config_timeout_cmd(object_id: bytearray) -> bytearray:
    """
    @brief  This function packs the command set the timeout of one of the three watchdogs of the PLOC.
    @param object_id    The object id of the PLOC supervisor handler.
    """
    watchdog = int(input("Specify watchdog (0 - PS, 1 - PL, 2 - INT):"))
    timeout = int(input("Specify timeout (1000 ms - 360000 ms):"))
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.WATCHDOGS_CONFIG_TIMEOUT)
    command = command + struct.pack('!B', watchdog)
    command = command + struct.pack('!I', timeout)
    return command


def pack_lachtup_alert_cmd(object_id: bytearray, state: bool) -> bytearray:
    """
    @brief  This function packs the command to enable or disable a certain latchup alerts.
    @param object_id    The object id of the PLOC supervisor handler.
    @param state    True - enable latchup alert, False - disable latchup alert
    """
    latchup_id = get_latchup_id()
    command = bytearray()
    if state:
        command = object_id + struct.pack('!I', SupvActionIds.ENABLE_LATCHUP_ALERT)
    else:
        command = object_id + struct.pack('!I', SupvActionIds.DISABLE_LATCHUP_ALERT)
    command = command + struct.pack('!B', latchup_id)
    return command


def pack_auto_calibrate_alert_cmd(object_id: bytearray) -> bytearray:
    """
    @brief  This function packs the command to auto calibrate a latchup alert.
    @param object_id    The object id of the PLOC supervisor handler.
    """
    latchup_id = get_latchup_id()
    mg = int(input("Specify MG:"))
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.AUTO_CALIBRATE_ALERT)
    command = command + struct.pack('!B', latchup_id)
    command = command + struct.pack('!I', mg)
    return command


def get_latchup_id() -> int:
    key_column_width = 10
    description_column_width = 50
    separator_width = key_column_width + description_column_width + 3
    separator_string = separator_width * "-"
    key_string = "Latchup ID".ljust(key_column_width)
    description_string = "Description".ljust(description_column_width)
    print(f"{key_string} | {description_string}")
    print(separator_string)
    for key in latchup_id_dict:
        key_string = key.ljust(key_column_width)
        description_string = latchup_id_dict[key].ljust(description_column_width)
        print(f"{key_string} | {description_string}")
    return int(input("Specify latchup ID:"))


def pack_set_alert_limit_cmd(object_id: bytearray) -> bytearray:
    """
    @brief  This function packs the command to set the limit of a latchup alert.
    @param object_id    The object id of the PLOC supervisor handler.
    """
    latchup_id = get_latchup_id()
    dutycycle = int(input("Specify dutycycle:"))
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.AUTO_CALIBRATE_ALERT)
    command = command + struct.pack('!B', latchup_id)
    command = command + struct.pack('!I', dutycycle)
    return command


def pack_set_alert_irq_filter_cmd(object_id: bytearray) -> bytearray:
    """
    @brief  This function packs the command to configure the latchup alert irq filter.
    @param object_id    The object id of the PLOC supervisor handler.
    """
    latchup_id = get_latchup_id()
    tp = int(input("Specify TP:"))
    div = int(input("Specify DIV:"))
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.SET_ALERT_IRQ_FILTER)
    command = command + struct.pack('!B', latchup_id)
    command = command + struct.pack('!B', tp)
    command = command + struct.pack('!B', div)
    return command


def pack_set_adc_sweep_period_cmd(object_id: bytearray) -> bytearray:
    """
    @brief  This function packs the command to set the limit of a latchup alert.
    @param object_id    The object id of the PLOC supervisor handler.
    """
    sweep_period = int(input("Specify sweep period (min 21 us):"))
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.SET_ADC_SWEEP_PERIOD)
    command = command + struct.pack('!I', sweep_period)
    return command


def pack_set_adc_enabled_channels_cmd(object_id: bytearray) -> bytearray:
    """
    @brief  This function packs the command to enable or disable channels of the ADC.
    @param object_id    The object id of the PLOC supervisor handler.
    """
    ch = int(input("Specify ch:"))
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.SET_ADC_ENABLED_CHANNELS)
    command = command + struct.pack('!H', ch)
    return command


def pack_set_adc_window_and_stride_cmd(object_id: bytearray) -> bytearray:
    window_size = int(input("Specify window size:"))
    striding_step_size = int(input("Specify striding step size:"))
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.SET_ADC_WINDOW_AND_STRIDE)
    command = command + struct.pack('!H', window_size)
    command = command + struct.pack('!H', striding_step_size)
    return command


def pack_set_adc_threshold_cmd(object_id: bytearray) -> bytearray:
    threshold = int(input("Specify threshold:"))
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.SET_ADC_THRESHOLD)
    command = command + struct.pack('!I', threshold)
    return command


def pack_enable_nvms_cmd(object_id: bytearray) -> bytearray:
    n01 = int(input("NVM0/1 (0 - off, 1 - on):"))
    n3 = int(input("NVM3 (0 - off, 1 - on):"))
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.ENABLE_NVMS)
    command = command + struct.pack('!B', n01)
    command = command + struct.pack('!B', n3)
    return command


def pack_select_nvm_cmd(object_id: bytearray) -> bytearray:
    mem = int(input("Specify NVM (0 - NVM0, 1 - MVM1):"))
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.SELECT_NVM)
    command = command + struct.pack('!B', mem)
    return command


def pack_auto_em_tests_cmd(object_id: bytearray) -> bytearray:
    test = int(input("Specify test (1 - complete, 2 - short):"))
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.RUN_AUTO_EM_TESTS)
    command = command + struct.pack('!B', test)
    return command


def pack_mram_wipe_cmd(object_id: bytearray) -> bytearray:
    start = int(input("Start address: 0x"), 16)
    stop = int(input("Stop address: 0x"), 16)
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.WIPE_MRAM)
    command = command + struct.pack('!I', start)
    command = command + struct.pack('!I', stop)
    return command


def pack_mram_dump_cmd(object_id: bytearray) -> bytearray:
    start = int(input("Start address: 0x"), 16)
    stop = int(input("Stop address: 0x"), 16)
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.DUMP_MRAM)
    command = command + struct.pack('!I', start)
    command = command + struct.pack('!I', stop)
    return command


def pack_print_cpu_stats_cmd(object_id: bytearray) -> bytearray:
    en = 1
    command = bytearray()
    command = object_id + struct.pack('!I', SupvActionIds.PRINT_CPU_STATS)
    command = command + struct.pack('!B', en)
    return command


def pack_set_debug_verbosity_cmd(object_id: bytearray) -> bytearray:
    command = bytearray()
    print("Specify debug verbosity")
    verbosity_options_dict = {
        0: "None",
        1: "Error",
        2: "Warn",
        3: "Info",
    }
    print("{:<6} | {}".format('Key', 'Description'))
    for entry in verbosity_options_dict.items():
        print("{:<6} | {}".format(entry[0], entry[1]))
    verbosity = int(input("Specify verbosity key: "))
    command = object_id + struct.pack('!I', SupvActionIds.SET_DBG_VERBOSITY)
    command = command + struct.pack('!B', verbosity)
    return command