eive/eive-tmtc
14
0
Fork 0
Code Issues Pull Requests 1 Projects Releases 99 Wiki Activity

Merge remote-tracking branch 'origin/main' into ploc-supv-tm-handling
Some checks failed
EIVE/-/pipeline/head Build queued...
Details
EIVE/-/pipeline/pr-main There was a failure building this commit
Details

Browse Source
This commit is contained in:
Robin Müller
2024-04-17 15:04:57 +02:00
parent 342a3bbcc9 a8d0143b1e
commit 3700769e46
72 changed files with 3038 additions and 2014 deletions
Download Patch File Download Diff File Expand all files Collapse all files

339
eive_tmtc/tmtc/payload/ploc_supervisor.py
View File

@ -6,23 +6,20 @@
@author J. Meier
@date 10.07.2021
"""
from datetime import datetime
import enum
import logging
import struct
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.config import TmtcDefinitionWrapper
from tmtccmd.config.tmtc import OpCodeEntry, tmtc_definitions_provider
from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter
from tmtccmd.pus.s8_fsfw_action import create_action_cmd
from tmtccmd.pus.s200_fsfw_mode import Mode, pack_mode_data
from tmtccmd.pus.tc.s3_fsfw_hk import generate_one_hk_command, make_sid
from tmtccmd.tmtc import service_provider
from tmtccmd.tmtc.decorator import ServiceProviderParams
from eive_tmtc.config.definitions import CustomServiceList
from eive_tmtc.config.object_ids import PLOC_SUPV_ID, get_object_ids
from eive_tmtc.pus_tm.defs import PrintWrapper
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.pus.tc.s3_fsfw_hk import generate_one_hk_command, make_sid
from tmtccmd.config.tmtc import CmdTreeNode
from tmtccmd.tmtc import DefaultPusQueueHelper
from tmtccmd.pus.s200_fsfw_mode import pack_mode_data, Mode
from tmtccmd.pus.s8_fsfw_action import create_action_cmd
from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter
from eive_tmtc.utility.input_helper import InputHelper
_LOGGER = logging.getLogger(__name__)
@ -42,14 +39,14 @@ MANUAL_INPUT = "1"
HARDCODED_FILE = "/home/rmueller/EIVE/mpsoc_boot.bin"
UPDATE_FILE_DICT = {
HARDCODED: ["hardcoded", ""],
MANUAL_INPUT: ["manual input", ""],
"2": ["/mnt/sd0/ploc/mpsoc/image.bin", "/mnt/sd0/ploc/mpsoc/image.bin"],
HARDCODED: ("hardcoded", ""),
MANUAL_INPUT: ("manual input", ""),
"2": ("/mnt/sd0/ploc/mpsoc/image.bin", "/mnt/sd0/ploc/mpsoc/image.bin"),
}
EVENT_BUFFER_PATH_DICT = {
MANUAL_INPUT: ["manual input", ""],
"2": ["/mnt/sd0/ploc/supervisor", "/mnt/sd0/ploc/supervisor"],
MANUAL_INPUT: ("manual input", ""),
"2": ("/mnt/sd0/ploc/supervisor", "/mnt/sd0/ploc/supervisor"),
}
@ -133,126 +130,90 @@ class OpCode:
ON = "on"
NORMAL = "nml"
HK_TO_OBC = "hk_to_obc"
REQUEST_HK_SET_FROM_DEV = "req_hk_from_dev"
REQUEST_HK_SET = "req_hk"
REQUEST_GENERIC_HK_SET = "req_generic_hk"
START_MPSOC = "start_mpsoc"
SHUTDOWN_MPSOC = "stop_mpsoc"
SEL_NVM = "sel_nvm"
SET_TIME_REF = "set_time_ref"
FACTORY_FLASH = "factory_flash"
REQ_BOOT_STATUS_REPORT = "boot_report"
START_UPDATE = "start_update"
PERFORM_UPDATE = "update"
FACTORY_RESET = "factory_reset"
MEM_CHECK = "mem_check"
RESET_MPSOC = "reset_mpsoc"
SET_GPIO = "set_gpio"
READ_GPIO = "read_gpio"
READ_STATUS_REPORT = "read_status_report"
class Info(str, enum.Enum):
value: str
OFF = "Switch Off"
ON = "Switch On"
NML = "Switch Normal"
NORMAL = "Switch Normal"
HK_TO_OBC = "Request HK from PLOC SUPV"
REQUEST_HK_SET_FROM_DEV = "Request HK set from the device to the PLOC Handler"
REQUEST_HK_SET = "Request HK set from PLOC Handler"
REQUEST_GENERIC_HK_SET = "Request prompted HK set from PLOC Handler"
START_MPSOC = "Start MPSoC"
SHUTDOWN_MPSOC = "Shutdown MPSoC"
SET_TIME_REF = "Set time reference"
FACTORY_FLASH = "Factory Flash Mode"
PERFORM_UPDATE = "Start or continue MPSoC SW update at starting bytes"
START_UPDATE = "Start new MPSoC SW update"
FACTORY_RESET = "Factory Reset of loggers"
REQ_BOOT_STATUS_REPORT = "Request boot status report and HK"
MEM_CHECK = "Memory Check"
SEL_NVM = "Select NVM"
RESET_MPSOC = "Reset MPSoC"
SET_GPIO = "Set GPIO"
READ_GPIO = "Read GPIO"
READ_STATUS_REPORT = "Read HK status report"
@tmtc_definitions_provider
def add_ploc_supv_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
oce.add(OpCode.OFF, Info.OFF)
oce.add(OpCode.ON, Info.ON)
oce.add(OpCode.NORMAL, Info.NML)
oce.add(OpCode.HK_TO_OBC, Info.HK_TO_OBC)
oce.add(OpCode.REQUEST_HK_SET, Info.REQUEST_HK_SET)
oce.add(OpCode.REQUEST_HK_SET_FROM_DEV, Info.REQUEST_HK_SET_FROM_DEV)
oce.add(OpCode.START_MPSOC, "PLOC Supervisor: Start MPSoC")
oce.add(OpCode.SHUTDOWN_MPSOC, "PLOC Supervisor: Shutdown MPSoC")
oce.add(OpCode.SEL_NVM, Info.SEL_NVM)
oce.add(OpCode.SET_TIME_REF, Info.SET_TIME_REF)
oce.add(OpCode.FACTORY_RESET, Info.FACTORY_RESET)
oce.add(OpCode.RESET_MPSOC, Info.RESET_MPSOC)
oce.add("8", "PLOC Supervisor: Set max restart tries")
oce.add("11", "PLOC Supervisor: Set boot timeout")
oce.add("12", "PLOC Supervisor: Disable Hk")
oce.add(OpCode.REQ_BOOT_STATUS_REPORT, Info.REQ_BOOT_STATUS_REPORT)
oce.add("17", "PLOC Supervisor: Enable latchup alert")
oce.add("18", "PLOC Supervisor: Disable latchup alert")
oce.add("20", "PLOC Supervisor: Set alert limit")
oce.add("23", "PLOC Supervisor: Set ADC enabled channels")
oce.add("24", "PLOC Supervisor: Set ADC window and stride")
oce.add("25", "PLOC Supervisor: Set ADC threshold")
oce.add("26", "PLOC Supervisor: Request latchup status report")
oce.add("27", "PLOC Supervisor: Copy ADC data to MRAM")
oce.add("30", "PLOC Supervisor: Run auto EM tests")
oce.add("31", "PLOC Supervisor: MRAM Wipe")
oce.add("35", "PLOC Supervisor: Set GPIO")
oce.add("36", "PLOC Supervisor: Read GPIO")
oce.add("37", "PLOC Supervisor: Restart supervisor")
oce.add(OpCode.PERFORM_UPDATE, Info.PERFORM_UPDATE)
oce.add(OpCode.START_UPDATE, Info.START_UPDATE)
oce.add("43", "PLOC Supervisor: Terminate supervisor process")
oce.add("44", "PLOC Supervisor: Start MPSoC quiet")
oce.add("45", "PLOC Supervisor: Set shutdown timeout")
oce.add(OpCode.FACTORY_FLASH, Info.FACTORY_FLASH)
oce.add("47", "PLOC Supervisor: Enable auto TM")
oce.add("48", "PLOC Supervisor: Disable auto TM")
oce.add("51", "PLOC Supervisor: Logging request event buffers")
oce.add("52", "PLOC Supervisor: Logging clear counters")
oce.add("53", "PLOC Supervisor: Logging set topic")
oce.add("54", "PLOC Supervisor: Logging request counters")
oce.add("55", "PLOC Supervisor: Request ADC Report")
oce.add("56", "PLOC Supervisor: Reset PL")
oce.add("57", "PLOC Supervisor: Enable NVMs")
oce.add("58", "PLOC Supervisor: Continue update")
oce.add(OpCode.MEM_CHECK, Info.MEM_CHECK)
defs.add_service(CustomServiceList.PLOC_SUPV.value, "PLOC Supervisor", oce)
def create_ploc_supv_node() -> CmdTreeNode:
op_code_strs = [
getattr(OpCode, key) for key in dir(OpCode) if not key.startswith("__")
]
info_strs = [getattr(Info, key) for key in dir(OpCode) if not key.startswith("__")]
combined_dict = dict(zip(op_code_strs, info_strs))
node = CmdTreeNode("ploc_supv", "PLOC Supervisor", hide_children_for_print=True)
for op_code, info in combined_dict.items():
node.add_child(CmdTreeNode(op_code, info))
return node
@service_provider(CustomServiceList.PLOC_SUPV)
def pack_ploc_supv_commands(p: ServiceProviderParams): # noqa C901
q = p.queue_helper
op_code = p.op_code
def pack_ploc_supv_commands(q: DefaultPusQueueHelper, cmd_str: str): # noqa C901
object_id = get_object_ids().get(PLOC_SUPV_ID)
assert object_id is not None
q.add_log_cmd(f"Testing PLOC Supervisor with object id: {object_id.as_hex_string}")
obyt = object_id.as_bytes
prefix = "PLOC Supervisor"
if op_code == OpCode.OFF:
if cmd_str == OpCode.OFF:
q.add_log_cmd(f"{prefix}: {Info.OFF}")
command = pack_mode_data(object_id.as_bytes, Mode.OFF, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code == OpCode.ON:
if cmd_str == OpCode.ON:
q.add_log_cmd(f"{prefix}: {Info.ON}")
command = pack_mode_data(object_id.as_bytes, Mode.ON, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code == OpCode.NORMAL:
q.add_log_cmd(f"{prefix}: {Info.NML}")
if cmd_str == OpCode.NORMAL:
q.add_log_cmd(f"{prefix}: {Info.NORMAL}")
command = pack_mode_data(object_id.as_bytes, Mode.NORMAL, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code == OpCode.HK_TO_OBC:
if cmd_str == OpCode.HK_TO_OBC:
q.add_log_cmd(f"{prefix}: {Info.HK_TO_OBC}")
command = obyt + struct.pack("!I", SupvActionId.REQUEST_HK_REPORT)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == OpCode.REQUEST_HK_SET:
q.add_log_cmd(f"{prefix}: {Info.REQUEST_HK_SET}")
sid = make_sid(object_id.as_bytes, prompt_set_id())
cmd = generate_one_hk_command(sid)
q.add_pus_tc(cmd)
if op_code == OpCode.REQUEST_HK_SET_FROM_DEV:
q.add_log_cmd(f"{prefix}: {Info.REQUEST_HK_SET_FROM_DEV}")
if cmd_str == OpCode.START_MPSOC:
q.add_log_cmd("PLOC Supervisor: Start MPSoC")
command = obyt + struct.pack("!I", SupvActionId.START_MPSOC)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if cmd_str == OpCode.SHUTDOWN_MPSOC:
q.add_log_cmd("PLOC Supervisor: Shutdown MPSoC")
command = object_id.as_bytes + struct.pack("!I", SupvActionId.SHUTWOWN_MPSOC)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if cmd_str == OpCode.READ_STATUS_REPORT:
q.add_log_cmd(f"{prefix}: {Info.READ_STATUS_REPORT}")
set_id = prompt_set_id()
action_cmd = None
# First read the set from the device.
if set_id == SetId.HK_REPORT:
action_cmd = create_action_cmd(PLOC_SUPV_ID, SupvActionId.REQUEST_HK_REPORT)
if set_id == SetId.ADC_REPORT:
@ -263,17 +224,33 @@ def pack_ploc_supv_commands(p: ServiceProviderParams): # noqa C901
action_cmd = create_action_cmd(
PLOC_SUPV_ID, SupvActionId.REQUEST_LOGGING_COUNTERS
)
assert action_cmd is not None
if set_id == SetId.LATCHUP_REPORT:
action_cmd = create_action_cmd(
PLOC_SUPV_ID, SupvActionId.GET_LATCHUP_STATUS_REPORT
)
if set_id == SetId.BOOT_STATUS_REPORT:
action_cmd = create_action_cmd(
PLOC_SUPV_ID, SupvActionId.GET_BOOT_STATUS_REPORT
)
if action_cmd is None:
_LOGGER.warning(f"invalid set ID {set_id!r} for PLOC SUPV")
return
# Now dump the HK set.
sid = make_sid(object_id.as_bytes, set_id)
req_hk = generate_one_hk_command(sid)
q.add_pus_tc(action_cmd)
elif op_code == OpCode.START_MPSOC:
q.add_wait_seconds(2.0)
q.add_pus_tc(req_hk)
assert action_cmd is not None
elif cmd_str == OpCode.START_MPSOC:
q.add_log_cmd("PLOC Supervisor: Start MPSoC")
command = obyt + struct.pack("!I", SupvActionId.START_MPSOC)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == OpCode.SHUTDOWN_MPSOC:
if cmd_str == OpCode.SHUTDOWN_MPSOC:
q.add_log_cmd("PLOC Supervisor: Shutdown MPSoC")
command = object_id.as_bytes + struct.pack("!I", SupvActionId.SHUTWOWN_MPSOC)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == OpCode.SEL_NVM:
if cmd_str == OpCode.SEL_NVM:
q.add_log_cmd("PLOC Supervisor: Select MPSoC boot image")
mem = int(input("MEM (NVM0 - 0 or NVM1 - 1): "))
bp0 = int(input("BP0 (0 or 1): "))
@ -281,7 +258,7 @@ def pack_ploc_supv_commands(p: ServiceProviderParams): # noqa C901
bp2 = int(input("BP2 (0 or 1): "))
command = pack_sel_boot_image_cmd(object_id.as_bytes, mem, bp0, bp1, bp2)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == OpCode.FACTORY_RESET:
if cmd_str == OpCode.FACTORY_RESET:
q.add_log_cmd(f"{prefix}: {Info.FACTORY_RESET}")
while True:
print("Please select the key for a factory reset operation")
@ -298,7 +275,7 @@ def pack_ploc_supv_commands(p: ServiceProviderParams): # noqa C901
user_data=bytes([key]),
)
)
if op_code == "8":
if cmd_str == "8":
q.add_log_cmd("PLOC Supervisor: Set max restart tries")
restart_tries = int(input("Specify maximum restart tries: "))
command = (
@ -307,15 +284,15 @@ def pack_ploc_supv_commands(p: ServiceProviderParams): # noqa C901
+ struct.pack("!B", restart_tries)
)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == OpCode.RESET_MPSOC:
if cmd_str == OpCode.RESET_MPSOC:
q.add_log_cmd(Info.RESET_MPSOC)
command = object_id.as_bytes + struct.pack("!I", SupvActionId.RESET_MPSOC)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == OpCode.SET_TIME_REF:
if cmd_str == OpCode.SET_TIME_REF:
q.add_log_cmd("PLOC Supervisor: Set time reference")
command = object_id.as_bytes + struct.pack("!I", SupvActionId.SET_TIME_REF)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "11":
if cmd_str == "11":
q.add_log_cmd("PLOC Supervisor: Set boot timeout")
boot_timeout = int(input("Specify boot timeout [ms]: "))
command = (
@ -324,143 +301,63 @@ def pack_ploc_supv_commands(p: ServiceProviderParams): # noqa C901
+ struct.pack("!I", boot_timeout)
)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "12":
q.add_log_cmd("PLOC Supervisor: Disable HK")
command = object_id.as_bytes + struct.pack("!I", SupvActionId.DISABLE_HK)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code in OpCode.REQ_BOOT_STATUS_REPORT:
q.add_log_cmd(f"{prefix}: {Info.REQ_BOOT_STATUS_REPORT}")
command = object_id.as_bytes + struct.pack(
"!I", SupvActionId.GET_BOOT_STATUS_REPORT
)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_wait_seconds(2.0)
sid = make_sid(object_id.as_bytes, SetId.BOOT_STATUS_REPORT)
req_hk = generate_one_hk_command(sid)
q.add_pus_tc(req_hk)
if op_code == "17":
q.add_log_cmd("PLOC Supervisor: Enable latchup alert")
command = pack_lachtup_alert_cmd(object_id.as_bytes, True)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "18":
q.add_log_cmd("PLOC Supervisor: Disable latchup alert")
command = pack_lachtup_alert_cmd(object_id.as_bytes, False)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "20":
q.add_log_cmd("PLOC Supervisor: Set alert limit")
command = pack_set_alert_limit_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "23":
if cmd_str == "23":
q.add_log_cmd("PLOC Supervisor: Set ADC enabled channels")
command = pack_set_adc_enabled_channels_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "24":
if cmd_str == "24":
q.add_log_cmd("PLOC Supervisor: Set ADC window and stride")
command = pack_set_adc_window_and_stride_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "25":
if cmd_str == "25":
q.add_log_cmd("PLOC Supervisor: Set ADC threshold")
command = pack_set_adc_threshold_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "26":
q.add_log_cmd("PLOC Supervisor: Request latchup status report")
command = object_id.as_bytes + struct.pack(
"!I", SupvActionId.GET_LATCHUP_STATUS_REPORT
)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "27":
q.add_log_cmd("PLOC Supervisor: Copy ADC data to MRAM")
command = object_id.as_bytes + struct.pack(
"!I", SupvActionId.COPY_ADC_DATA_TO_MRAM
)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "30":
q.add_log_cmd("PLOC Supervisor: Run auto EM tests")
command = pack_auto_em_tests_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "31":
q.add_log_cmd("PLOC Supervisor: Wipe MRAM")
command = pack_mram_wipe_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "35":
if cmd_str == OpCode.SET_GPIO:
q.add_log_cmd("PLOC Supervisor: Set GPIO command")
command = pack_set_gpio_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "36":
if cmd_str == OpCode.READ_GPIO:
q.add_log_cmd("PLOC Supervisor: Read GPIO command")
command = pack_read_gpio_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "37":
if cmd_str == "37":
q.add_log_cmd("PLOC Supervisor: Restart supervisor")
command = object_id.as_bytes + struct.pack(
"!I", SupvActionId.RESTART_SUPERVISOR
)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code in OpCode.START_UPDATE:
if cmd_str in OpCode.START_UPDATE:
q.add_log_cmd("PLOC Supversior: Start new MPSoC SW update")
command = pack_update_command(object_id.as_bytes, True)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code in OpCode.PERFORM_UPDATE:
if cmd_str in OpCode.PERFORM_UPDATE:
q.add_log_cmd("PLOC Supervisor: Perform MPSoC SW update")
command = pack_update_command(object_id.as_bytes, False)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "43":
if cmd_str == "43":
q.add_log_cmd("PLOC Supervisor: Terminate supervisor process")
command = object_id.as_bytes + struct.pack(
"!I", SupvActionId.TERMINATE_SUPV_HELPER
)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "44":
if cmd_str == "44":
q.add_log_cmd("PLOC Supervisor: Start MPSoC quiet")
command = object_id.as_bytes + struct.pack("!I", SupvActionId.START_MPSOC_QUIET)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "45":
if cmd_str == "45":
q.add_log_cmd("PLOC Supervisor: Set shutdown timeout")
command = pack_set_shutdown_timeout_command(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code in OpCode.FACTORY_FLASH:
if cmd_str in OpCode.FACTORY_FLASH:
q.add_log_cmd(f"{prefix}: {Info.FACTORY_FLASH}")
command = object_id.as_bytes + struct.pack("!I", SupvActionId.FACTORY_FLASH)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "47":
q.add_log_cmd("PLOC Supervisor: Enable auto TM")
command = object_id.as_bytes + struct.pack("!I", SupvActionId.ENABLE_AUTO_TM)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "48":
q.add_log_cmd("PLOC Supervisor: Disable auto TM")
command = object_id.as_bytes + struct.pack("!I", SupvActionId.DISABLE_AUTO_TM)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "51":
q.add_log_cmd("PLOC Supervisor: Logging request event buffers")
command = pack_logging_buffer_request(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "52":
q.add_log_cmd("PLOC Supervisor: Logging clear counters")
command = object_id.as_bytes + struct.pack(
"!I", SupvActionId.LOGGING_CLEAR_COUNTERS
)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "53":
q.add_log_cmd("PLOC Supervisor: Logging set topic")
command = pack_logging_set_topic(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "54":
q.add_log_cmd("PLOC Supervisor: Logging request counters")
command = object_id.as_bytes + struct.pack(
"!I", SupvActionId.REQUEST_LOGGING_COUNTERS
)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "55":
q.add_log_cmd("PLOC Supervisor: Request ADC report")
command = object_id.as_bytes + struct.pack(
"!I", SupvActionId.REQUEST_ADC_REPORT
)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "56":
if cmd_str == "56":
q.add_log_cmd("PLOC Supervisor: Reset PL")
command = object_id.as_bytes + struct.pack("!I", SupvActionId.RESET_PL)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "57":
if cmd_str == "57":
q.add_log_cmd("PLOC Supervisor: Enable NVMs")
nvm01 = int(input("Enable (1) or disable(0) NVM 0 and 1: "))
nvm3 = int(input("Enable (1) or disable(0) NVM 3: "))
@ -471,11 +368,11 @@ def pack_ploc_supv_commands(p: ServiceProviderParams): # noqa C901
+ struct.pack("B", nvm3)
)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "58":
if cmd_str == "58":
q.add_log_cmd("PLOC Supervisor: Continue update")
command = object_id.as_bytes + struct.pack("!I", SupvActionId.CONTINUE_UPDATE)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == OpCode.MEM_CHECK:
if cmd_str == OpCode.MEM_CHECK:
custom_data = bytearray()
update_file = get_update_file()
memory_id = int(input("Specify memory ID: "))
@ -768,6 +665,8 @@ def handle_supv_hk_data(set_id: int, hk_data: bytes, pw: PrintWrapper):
handle_adc_report(hk_data)
elif set_id == SetId.COUNTERS_REPORT:
handle_counters_report(hk_data)
elif set_id == SetId.LATCHUP_REPORT:
handle_latchup_status_report(hk_data)
else:
pw.dlog(f"PLOC SUPV: HK handling not implemented for set ID {set_id}")
pw.dlog(f"Raw Data: 0x[{hk_data.hex(sep=',')}]")
@ -912,3 +811,53 @@ def handle_counters_report(hk_data: bytes):
print(f"MM task lost: {mm_task_lost}")
print(f"HK task lost: {hk_task_lost}")
print(f"DL task lost: {dl_task_lost}")
def handle_latchup_status_report(hk_data: bytes):
# 1 byte ID, 7 times 2 bytes of counts, and 8 bytes of time and 1 byte sync status.
if len(hk_data) < 24:
raise ValueError("Latchup status report smaller than expected")
current_idx = 0
id = hk_data[current_idx]
current_idx += 1
counts_of_alerts = []
for _ in range(7):
counts_of_alerts.append(
struct.unpack("!H", hk_data[current_idx : current_idx + 2])[0]
)
current_idx += 2
print(f"ID: {id}")
print(f"Counts of alerts: {counts_of_alerts}")
time_ms = struct.unpack("!H", hk_data[current_idx : current_idx + 2])[0]
current_idx += 2
time_seconds = hk_data[current_idx]
current_idx += 1
time_minutes = hk_data[current_idx]
current_idx += 1
time_hour = hk_data[current_idx]
current_idx += 1
time_day = hk_data[current_idx]
current_idx += 1
time_month = hk_data[current_idx]
current_idx += 1
# Is stored as years since 1900.
time_year = 1900 + hk_data[current_idx]
current_idx += 1
is_synced = hk_data[current_idx]
print(f"Time Sync Status: {is_synced}")
try:
dt = datetime(
year=time_year,
month=time_month,
day=time_day,
hour=time_hour,
minute=time_minutes,
second=time_seconds,
microsecond=time_ms * 1000,
)
print(f"Time Now: {dt}")
except ValueError:
print(
f"Time: {time_day}.{time_month}.{time_year}T"
f"{time_hour}:{time_minutes}:{time_seconds}.{time_ms}"
)