818 lines
34 KiB
Python
818 lines
34 KiB
Python
# -*- 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 spacepackets.ecss.tc import PusTelecommand
|
|
from tmtccmd.config.definitions import QueueCommands
|
|
from tmtccmd.tc.packer import TcQueueT
|
|
from tmtccmd.logging import get_console_logger
|
|
from tmtccmd.tc.service_200_mode import pack_mode_data, Modes
|
|
from utility.input_helper import InputHelper
|
|
|
|
LOGGER = get_console_logger()
|
|
|
|
latchup_id_dict = {
|
|
"0": "0.85V",
|
|
"1": "1.8V",
|
|
"2": "MISC",
|
|
"3": "3.3V",
|
|
"4": "NVM_4XO",
|
|
"5": "MISSION",
|
|
"6": "SAFECOTS",
|
|
}
|
|
|
|
MANUAL_INPUT = "1"
|
|
|
|
update_file_dict = {
|
|
MANUAL_INPUT: ["manual input", ""],
|
|
"2": ["/mnt/sd0/ploc/supervisor/update.bin", "/mnt/sd0/ploc/supervisor/update.bin"],
|
|
"3": ["/mnt/sd0/ploc/supervisor/update-large.bin", "/mnt/sd0/ploc/supervisor/update-large.bin"],
|
|
"4": ["/mnt/sd0/ploc/supervisor/update-small.bin", "/mnt/sd0/ploc/supervisor/update-small.bin"],
|
|
}
|
|
|
|
event_buffer_path_dict = {
|
|
MANUAL_INPUT: ["manual input", ""],
|
|
"2": ["/mnt/sd0/ploc/supervisor", "/mnt/sd0/ploc/supervisor"],
|
|
}
|
|
|
|
|
|
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_CLEAR_ALL = 37
|
|
LOGGING_REQUEST_COUNTERS = 38
|
|
UPDATE_IMAGE_DATA = 39
|
|
FACTORY_RESET_CLEAR_MIRROR = 40
|
|
FACTORY_RESET_CLEAR_CIRCULAR = 41
|
|
START_MPSOC_QUIET = 45
|
|
SET_SHUTDOWN_TIMEOUT = 46
|
|
FACTORY_FLASH = 47
|
|
PERFORM_UPDATE = 48
|
|
TERMINATE_SUPV_HELPER = 49
|
|
ENABLE_AUTO_TM = 50
|
|
DISABLE_AUTO_TM = 51
|
|
ENABLE_ADC_MONITOR_TASK = 52
|
|
DISABLE_ADC_MONITOR_TASK = 53
|
|
LOGGING_REQUEST_EVENT_BUFFERS = 54
|
|
LOGGING_CLEAR_COUNTERS = 55
|
|
LOGGING_SET_TOPIC = 56
|
|
REQUEST_ADC_REPORT = 57
|
|
|
|
|
|
class SupvHkIds:
|
|
HK_REPORT = 52
|
|
BOOT_STATUS_REPORT = 53
|
|
|
|
|
|
def pack_ploc_supv_commands(
|
|
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 == "0":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set mode off"))
|
|
command = pack_mode_data(object_id, Modes.OFF, 0)
|
|
command = PusTelecommand(service=200, subservice=1, ssc=9, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
if op_code == "1":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set mode on"))
|
|
command = pack_mode_data(object_id, Modes.ON, 0)
|
|
command = PusTelecommand(service=200, subservice=1, ssc=10, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
if op_code == "2":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "Star tracker: Mode Normal"))
|
|
command = pack_mode_data(object_id, Modes.NORMAL, 0)
|
|
command = PusTelecommand(service=200, subservice=1, ssc=11, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
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 == "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 == "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())
|
|
elif op_code == "35":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set GPIO command"))
|
|
command = pack_set_gpio_cmd(object_id)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=50, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "36":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Read GPIO command"))
|
|
command = pack_read_gpio_cmd(object_id)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=51, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "37":
|
|
tc_queue.appendleft(
|
|
(QueueCommands.PRINT, "PLOC Supervisor: Restart supervisor")
|
|
)
|
|
command = command = object_id + struct.pack(
|
|
"!I", SupvActionIds.RESTART_SUPERVISOR
|
|
)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=52, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "38":
|
|
tc_queue.appendleft(
|
|
(QueueCommands.PRINT, "PLOC Supervisor: Factory reset clear all")
|
|
)
|
|
command = command = object_id + struct.pack(
|
|
"!I", SupvActionIds.FACTORY_RESET_CLEAR_ALL
|
|
)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=53, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "39":
|
|
tc_queue.appendleft(
|
|
(QueueCommands.PRINT, "PLOC Supervisor: Factory reset clear mirror entries")
|
|
)
|
|
command = command = object_id + struct.pack(
|
|
"!I", SupvActionIds.FACTORY_RESET_CLEAR_MIRROR
|
|
)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=54, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "40":
|
|
tc_queue.appendleft(
|
|
(
|
|
QueueCommands.PRINT,
|
|
"PLOC Supervisor: Factory reset clear circular entries",
|
|
)
|
|
)
|
|
command = command = object_id + struct.pack(
|
|
"!I", SupvActionIds.FACTORY_RESET_CLEAR_CIRCULAR
|
|
)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=55, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "41":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: CAN loopback test"))
|
|
command = command = object_id + struct.pack(
|
|
"!I", SupvActionIds.CAN_LOOPBACK_TEST
|
|
)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=56, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "42":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Perform update"))
|
|
command = pack_update_command(object_id)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=57, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "43":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Terminate supervisor process"))
|
|
command = object_id + struct.pack(
|
|
"!I", SupvActionIds.TERMINATE_SUPV_HELPER
|
|
)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=58, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "44":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Start MPSoC quiet"))
|
|
command = object_id + struct.pack(
|
|
"!I", SupvActionIds.START_MPSOC_QUIET
|
|
)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=59, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "45":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set shutdown timeout"))
|
|
command = pack_set_shutdown_timeout_command(object_id)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=60, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "46":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Factory flash"))
|
|
command = object_id + struct.pack(
|
|
"!I", SupvActionIds.FACTORY_FLASH
|
|
)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=61, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "47":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Enable auto TM"))
|
|
command = object_id + struct.pack(
|
|
"!I", SupvActionIds.ENABLE_AUTO_TM
|
|
)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=62, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "48":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Disable auto TM"))
|
|
command = object_id + struct.pack(
|
|
"!I", SupvActionIds.DISABLE_AUTO_TM
|
|
)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=63, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "49":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Enable ADC monitor task"))
|
|
command = object_id + struct.pack(
|
|
"!I", SupvActionIds.ENABLE_ADC_MONITOR_TASK
|
|
)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=64, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "50":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Disable ADC monitor task"))
|
|
command = object_id + struct.pack(
|
|
"!I", SupvActionIds.DISABLE_ADC_MONITOR_TASK
|
|
)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=65, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "51":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Logging request event buffers"))
|
|
command = pack_logging_buffer_request(object_id)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=66, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "52":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Logging clear counters"))
|
|
command = object_id + struct.pack(
|
|
"!I", SupvActionIds.LOGGING_CLEAR_COUNTERS
|
|
)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=67, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "53":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Logging set topic"))
|
|
command = pack_logging_set_topic(object_id)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=68, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "54":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Logging request counters"))
|
|
command = object_id + struct.pack('!I', SupvActionIds.LOGGING_REQUEST_COUNTERS)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=69, app_data=command)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
elif op_code == "55":
|
|
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Request ADC report"))
|
|
command = object_id + struct.pack('!I', SupvActionIds.REQUEST_ADC_REPORT)
|
|
command = PusTelecommand(service=8, subservice=128, ssc=70, 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.SET_ALERT_LIMIT)
|
|
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 filter type (TP):"))
|
|
div = int(input("Specify clock divider (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: 0x"), 16)
|
|
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_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()
|
|
verbosity = get_debug_verbosity()
|
|
command = object_id + struct.pack("!I", SupvActionIds.SET_DBG_VERBOSITY)
|
|
command = command + struct.pack("!B", verbosity)
|
|
return command
|
|
|
|
|
|
def pack_update_command(object_id: bytearray) -> bytearray:
|
|
command = bytearray()
|
|
memory_id = int(input("Specify memory ID: "))
|
|
start_address = int(input("Specify start address: 0x"), 16)
|
|
update_file = get_update_file()
|
|
command += object_id
|
|
command += struct.pack('!I', SupvActionIds.PERFORM_UPDATE)
|
|
command += bytearray(update_file, 'utf-8')
|
|
# Adding null terminator
|
|
command += struct.pack('!B', 0)
|
|
command += struct.pack('!B', memory_id)
|
|
command += struct.pack('!I', start_address)
|
|
return command
|
|
|
|
|
|
def pack_set_shutdown_timeout_command(object_id: bytearray) -> bytearray:
|
|
command = bytearray()
|
|
command += object_id
|
|
command += struct.pack('!I', SupvActionIds.SET_SHUTDOWN_TIMEOUT)
|
|
timeout = int(input("Specify shutdown timeout (ms): "))
|
|
command += struct.pack('!I', timeout)
|
|
return command
|
|
|
|
|
|
def pack_logging_buffer_request(object_id: bytearray) -> bytearray:
|
|
command = bytearray()
|
|
command += object_id
|
|
command += struct.pack('!I', SupvActionIds.LOGGING_REQUEST_EVENT_BUFFERS)
|
|
path = get_event_buffer_path()
|
|
command += bytearray(path, 'utf-8')
|
|
return command
|
|
|
|
|
|
def get_debug_verbosity() -> int:
|
|
tries = 0
|
|
while tries < 3:
|
|
try:
|
|
print("Debug verbosity options")
|
|
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: "))
|
|
if verbosity > len(verbosity_options_dict) - 1:
|
|
raise ValueError
|
|
return verbosity
|
|
except ValueError:
|
|
LOGGER.warning("Invalid verbosity key specified")
|
|
tries = tries + 1
|
|
LOGGER.error("get_debug_verbosity: Exceeded max tries to input verbosity key")
|
|
quit()
|
|
|
|
|
|
def pack_set_gpio_cmd(object_id: bytearray) -> bytearray:
|
|
port = int(input("Specify port : 0x"), 16)
|
|
pin = int(input("Specify pin: 0x"), 16)
|
|
val = int(input("Specify val: 0x"), 16)
|
|
command = object_id + struct.pack("!I", SupvActionIds.SET_GPIO)
|
|
command = command + struct.pack("!B", port)
|
|
command = command + struct.pack("!B", pin)
|
|
command = command + struct.pack("!B", val)
|
|
return command
|
|
|
|
|
|
def pack_read_gpio_cmd(object_id: bytearray) -> bytearray:
|
|
port = int(input("Specify port: 0x"), 16)
|
|
pin = int(input("Specify pin: 0x"), 16)
|
|
command = object_id + struct.pack("!I", SupvActionIds.READ_GPIO)
|
|
command = command + struct.pack("!B", port)
|
|
command = command + struct.pack("!B", pin)
|
|
return command
|
|
|
|
|
|
def pack_logging_set_topic(objetc_id: bytearray) -> bytearray:
|
|
command = objetc_id + struct.pack('!I', SupvActionIds.LOGGING_SET_TOPIC)
|
|
tpc = int(input("Specify logging topic: "))
|
|
command += struct.pack('!B', tpc)
|
|
return command
|
|
|
|
|
|
def get_update_file() -> str:
|
|
LOGGER.info("Specify update file ")
|
|
input_helper = InputHelper(update_file_dict)
|
|
key = input_helper.get_key()
|
|
if key == MANUAL_INPUT:
|
|
file = input("Ploc Supervisor: Specify absolute name of update file: ")
|
|
else:
|
|
file = update_file_dict[key][1]
|
|
return file
|
|
|
|
|
|
def get_event_buffer_path() -> str:
|
|
LOGGER.info("Specify path where to store event buffer file ")
|
|
input_helper = InputHelper(event_buffer_path_dict)
|
|
key = input_helper.get_key()
|
|
if key == MANUAL_INPUT:
|
|
file = input("Ploc Supervisor: Specify path: ")
|
|
else:
|
|
file = event_buffer_path_dict[key][1]
|
|
return file
|