Add new PDEC commands

This commit is contained in:
Robin Müller 2023-04-14 19:21:51 +02:00
parent 183cd85907
commit 63c584e061
No known key found for this signature in database
GPG Key ID: 11D4952C8CCEF814
2 changed files with 56 additions and 20 deletions

View File

@ -5,11 +5,13 @@
@author J. Meier @author J. Meier
@date 22.11.2021 @date 22.11.2021
""" """
import enum
from spacepackets.ecss.tc import PusTelecommand from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc import DefaultPusQueueHelper from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_20_fsfw_param import create_load_param_cmd from tmtccmd.tc.pus_20_fsfw_param import create_load_param_cmd
from tmtccmd.tc.pus_8_fsfw_funccmd import create_action_cmd
from tmtccmd.pus.s20_fsfw_param_defs import create_scalar_u8_parameter from tmtccmd.pus.s20_fsfw_param_defs import create_scalar_u8_parameter
from tmtccmd.config.tmtc import ( from tmtccmd.config.tmtc import (
@ -28,16 +30,23 @@ class CommandId:
PRINT_PDEC_MON = bytearray([0x0, 0x0, 0x0, 0x1]) PRINT_PDEC_MON = bytearray([0x0, 0x0, 0x0, 0x1])
class ParameterId: class ParameterId(enum.IntEnum):
POSITIVE_WINDOW = 0 POSITIVE_WINDOW = 0
NEGATIVE_WINDOW = 1 NEGATIVE_WINDOW = 1
class ActionId(enum.IntEnum):
RESET_NO_INIT = 2
RESET_WITH_INIT = 3
class OpCode: class OpCode:
PRINT_CLCW = "print_clcw" PRINT_CLCW = "print_clcw"
PRINT_MON_REG = "print_mon_reg" PRINT_MON_REG = "print_mon_reg"
POSITIVE_WINDOW = "positive_window" POSITIVE_WINDOW = "positive_window"
NEGATIVE_WINDOW = "negative_window" NEGATIVE_WINDOW = "negative_window"
RESET_WITH_INIT = "reset_with_init"
RESET_NO_INIT = "reset_no_init"
class Info: class Info:
@ -47,6 +56,8 @@ class Info:
) )
POSITIVE_WINDOW = "Change positive window parameter for AD frames" POSITIVE_WINDOW = "Change positive window parameter for AD frames"
NEGATIVE_WINDOW = "Change negative window parameter for AD frames" NEGATIVE_WINDOW = "Change negative window parameter for AD frames"
RESET_WITH_INIT = "Reset with full initialization"
RESET_NO_INIT = "Reset with mandatory initialization"
def pack_pdec_handler_test( def pack_pdec_handler_test(
@ -88,6 +99,16 @@ def pack_pdec_handler_test(
).pack() ).pack()
) )
) )
if op_code == OpCode.RESET_NO_INIT:
q.add_log_cmd(f"{prefix}: {Info.RESET_NO_INIT}")
q.add_pus_tc(
create_action_cmd(object_id=object_id, action_id=ActionId.RESET_NO_INIT)
)
if op_code == OpCode.RESET_WITH_INIT:
q.add_log_cmd(f"{prefix}: {Info.RESET_WITH_INIT}")
q.add_pus_tc(
create_action_cmd(object_id=object_id, action_id=ActionId.RESET_WITH_INIT)
)
@tmtc_definitions_provider @tmtc_definitions_provider
@ -97,4 +118,6 @@ def add_pdec_cmds(defs: TmtcDefinitionWrapper):
oce.add(OpCode.PRINT_MON_REG, Info.PRINT_MON_REG) oce.add(OpCode.PRINT_MON_REG, Info.PRINT_MON_REG)
oce.add(OpCode.POSITIVE_WINDOW, Info.POSITIVE_WINDOW) oce.add(OpCode.POSITIVE_WINDOW, Info.POSITIVE_WINDOW)
oce.add(OpCode.NEGATIVE_WINDOW, Info.NEGATIVE_WINDOW) oce.add(OpCode.NEGATIVE_WINDOW, Info.NEGATIVE_WINDOW)
oce.add(OpCode.RESET_WITH_INIT, Info.RESET_WITH_INIT)
oce.add(OpCode.RESET_NO_INIT, Info.RESET_NO_INIT)
defs.add_service(CustomServiceList.PDEC_HANDLER.value, "PDEC Handler", oce) defs.add_service(CustomServiceList.PDEC_HANDLER.value, "PDEC Handler", oce)

View File

@ -139,8 +139,13 @@ def add_core_controller_definitions(defs: TmtcDefinitionWrapper):
oce.add(keys=OpCode.OBSW_UPDATE_FROM_SD_0, info=Info.OBSW_UPDATE_FROM_SD_0) oce.add(keys=OpCode.OBSW_UPDATE_FROM_SD_0, info=Info.OBSW_UPDATE_FROM_SD_0)
oce.add(keys=OpCode.OBSW_UPDATE_FROM_SD_1, info=Info.OBSW_UPDATE_FROM_SD_1) oce.add(keys=OpCode.OBSW_UPDATE_FROM_SD_1, info=Info.OBSW_UPDATE_FROM_SD_1)
oce.add(keys=OpCode.SYSTEMCTL_CMD_EXECUTOR, info=Info.SYSTEMCTL_CMD_EXECUTOR) oce.add(keys=OpCode.SYSTEMCTL_CMD_EXECUTOR, info=Info.SYSTEMCTL_CMD_EXECUTOR)
oce.add(keys=OpCode.EXECUTE_SHELL_CMD_BLOCKING, info=Info.EXECUTE_SHELL_CMD_BLOCKING) oce.add(
oce.add(keys=OpCode.EXECUTE_SHELL_CMD_NON_BLOCKING, info=Info.EXECUTE_SHELL_CMD_NON_BLOCKING) keys=OpCode.EXECUTE_SHELL_CMD_BLOCKING, info=Info.EXECUTE_SHELL_CMD_BLOCKING
)
oce.add(
keys=OpCode.EXECUTE_SHELL_CMD_NON_BLOCKING,
info=Info.EXECUTE_SHELL_CMD_NON_BLOCKING,
)
oce.add( oce.add(
keys=OpCode.GET_HK, keys=OpCode.GET_HK,
info="Request housekeeping set", info="Request housekeeping set",
@ -217,29 +222,37 @@ def pack_core_commands(q: DefaultPusQueueHelper, op_code: str):
print("systemctl command types: ") print("systemctl command types: ")
for entry in SystemctlCmd: for entry in SystemctlCmd:
print(f"{entry}: {entry.name}") print(f"{entry}: {entry.name}")
systemctl_cmd = SystemctlCmd(int(input("Specify systemctl command type by key: "))) systemctl_cmd = SystemctlCmd(
int(input("Specify systemctl command type by key: "))
)
unit_name = input("Specify unit name: ") unit_name = input("Specify unit name: ")
cmd_data = bytearray([systemctl_cmd]) cmd_data = bytearray([systemctl_cmd])
cmd_data.extend(unit_name.encode()) cmd_data.extend(unit_name.encode())
q.add_pus_tc(create_action_cmd( q.add_pus_tc(
object_id=CORE_CONTROLLER_ID, create_action_cmd(
action_id=ActionId.SYSTEMCTL_CMD_EXECUTOR, object_id=CORE_CONTROLLER_ID,
user_data=cmd_data action_id=ActionId.SYSTEMCTL_CMD_EXECUTOR,
)) user_data=cmd_data,
)
)
elif op_code == OpCode.EXECUTE_SHELL_CMD_BLOCKING: elif op_code == OpCode.EXECUTE_SHELL_CMD_BLOCKING:
custom_cmd = input("Please specify command to execute: ") custom_cmd = input("Please specify command to execute: ")
q.add_pus_tc(create_action_cmd( q.add_pus_tc(
object_id=CORE_CONTROLLER_ID, create_action_cmd(
action_id=ActionId.EXECUTE_SHELL_CMD_BLOCKING, object_id=CORE_CONTROLLER_ID,
user_data=custom_cmd.encode() action_id=ActionId.EXECUTE_SHELL_CMD_BLOCKING,
)) user_data=custom_cmd.encode(),
)
)
elif op_code == OpCode.EXECUTE_SHELL_CMD_NON_BLOCKING: elif op_code == OpCode.EXECUTE_SHELL_CMD_NON_BLOCKING:
custom_cmd = input("Please specify command to execute: ") custom_cmd = input("Please specify command to execute: ")
q.add_pus_tc(create_action_cmd( q.add_pus_tc(
object_id=CORE_CONTROLLER_ID, create_action_cmd(
action_id=ActionId.EXECUTE_SHELL_CMD_NON_BLOCKING, object_id=CORE_CONTROLLER_ID,
user_data=custom_cmd.encode() action_id=ActionId.EXECUTE_SHELL_CMD_NON_BLOCKING,
)) user_data=custom_cmd.encode(),
)
)
elif op_code in OpCode.XSC_REBOOT_0_0: elif op_code in OpCode.XSC_REBOOT_0_0:
add_xsc_reboot_cmd( add_xsc_reboot_cmd(
q=q, reboot_self=False, chip=Chip.CHIP_0, copy=Copy.COPY_0_NOM q=q, reboot_self=False, chip=Chip.CHIP_0, copy=Copy.COPY_0_NOM