Some PLOC commanding fixes #288

Merged
meggert merged 4 commits from ploc-tree-test into main 2024-04-09 14:41:11 +02:00
4 changed files with 28 additions and 28 deletions

View File

@ -19,6 +19,10 @@ list yields a list of all related PRs for each release.
- EPS power commands working again. - EPS power commands working again.
## Changed
- Fixed PLOC commanding
# [v6.1.1] 2024-03-06 # [v6.1.1] 2024-03-06
## Added ## Added

View File

@ -1,21 +0,0 @@
from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry, CoreServiceList
from tmtccmd.config.tmtc import (
call_all_definitions_providers,
)
from tmtccmd.config.globals import get_default_tmtc_defs
def get_eive_service_op_code_dict() -> TmtcDefinitionWrapper:
"""Call all registered TMTC definition providers. They were registered using
the :py:func:`tmtc_definitions_provider` decorator.
"""
def_wrapper = get_default_tmtc_defs()
srv_5 = OpCodeEntry()
srv_5.add("0", "Event Test")
def_wrapper.add_service(
name=CoreServiceList.SERVICE_5.value,
info="PUS Service 5 Event",
op_code_entry=srv_5,
)
call_all_definitions_providers(def_wrapper)
return def_wrapper

View File

@ -52,6 +52,7 @@ from eive_tmtc.tmtc.com.subsystem import build_com_subsystem_procedure
from eive_tmtc.tmtc.com.syrlinks_handler import pack_syrlinks_command from eive_tmtc.tmtc.com.syrlinks_handler import pack_syrlinks_command
from eive_tmtc.tmtc.core import pack_core_commands from eive_tmtc.tmtc.core import pack_core_commands
from eive_tmtc.tmtc.health import build_health_cmds from eive_tmtc.tmtc.health import build_health_cmds
from eive_tmtc.tmtc.payload.subsystem import create_payload_subsystem_cmd
from eive_tmtc.tmtc.payload.ploc_mpsoc import pack_ploc_mpsoc_commands from eive_tmtc.tmtc.payload.ploc_mpsoc import pack_ploc_mpsoc_commands
from eive_tmtc.tmtc.payload.ploc_supervisor import pack_ploc_supv_commands from eive_tmtc.tmtc.payload.ploc_supervisor import pack_ploc_supv_commands
from eive_tmtc.tmtc.payload.plpcdu import pack_pl_pcdu_commands from eive_tmtc.tmtc.payload.plpcdu import pack_pl_pcdu_commands
@ -252,7 +253,8 @@ def handle_payload_procedure(
queue_helper: DefaultPusQueueHelper, cmd_path_list: List[str] queue_helper: DefaultPusQueueHelper, cmd_path_list: List[str]
): ):
obj_id_man = get_object_ids() obj_id_man = get_object_ids()
assert len(cmd_path_list) >= 2 if cmd_path_list[0] == "subsystem":
return create_payload_subsystem_cmd(queue_helper, cmd_path_list[1])
if cmd_path_list[0] == "ploc_mpsoc": if cmd_path_list[0] == "ploc_mpsoc":
return pack_ploc_mpsoc_commands(queue_helper, cmd_path_list[1]) return pack_ploc_mpsoc_commands(queue_helper, cmd_path_list[1])
if cmd_path_list[0] == "ploc_supv": if cmd_path_list[0] == "ploc_supv":
@ -264,7 +266,6 @@ def handle_payload_procedure(
object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[1] object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[1]
) )
if cmd_path_list[0] == "pl_pcdu": if cmd_path_list[0] == "pl_pcdu":
assert len(cmd_path_list) >= 3
return pack_pl_pcdu_commands(q=queue_helper, cmd_str=cmd_path_list[1]) return pack_pl_pcdu_commands(q=queue_helper, cmd_str=cmd_path_list[1])
if cmd_path_list[0] == "scex": if cmd_path_list[0] == "scex":
return pack_scex_cmds( return pack_scex_cmds(

View File

@ -21,20 +21,27 @@ class ModeId(enum.IntEnum):
class OpCode(str, enum.Enum): class OpCode(str, enum.Enum):
OFF = "off" OFF = "off"
REPORT_ALL_MODES = "report_modes" REPORT_ALL_MODES = "report_modes"
MPSOC_STREAM = "mode_mpsoc_stream"
CAM_STREAM = "mode_cam_stream"
EARTH_OBSV = "mode_eart_obsv"
class Info(str, enum.Enum): class Info(str, enum.Enum):
OFF = "Off Command" OFF = "Off Command"
REPORT_ALL_MODES = "Report all modes" REPORT_ALL_MODES = "Report all modes"
MPSOC_STREAM = "MPSoC Stream Mode"
CAM_STREAM = "Camera Stream Mode"
EARTH_OBSV = "Earth Observation Mode"
HANDLER_LIST: Dict[str, Tuple[int, str]] = { HANDLER_LIST: Dict[str, Tuple[int, str]] = {
OpCode.OFF: (ModeId.OFF, Info.OFF), OpCode.OFF: (ModeId.OFF, Info.OFF),
OpCode.MPSOC_STREAM: (ModeId.MPSOC_STREAM, Info.MPSOC_STREAM),
} }
def build_acs_subsystem_cmd(q: DefaultPusQueueHelper, cmd_str: str): def create_payload_subsystem_cmd(q: DefaultPusQueueHelper, cmd_str: str):
info_prefix = "ACS Subsystem" info_prefix = "Payload Subsystem"
if cmd_str == OpCode.REPORT_ALL_MODES: if cmd_str == OpCode.REPORT_ALL_MODES:
q.add_log_cmd(f"{info_prefix}: {Info.REPORT_ALL_MODES}") q.add_log_cmd(f"{info_prefix}: {Info.REPORT_ALL_MODES}")
q.add_pus_tc( q.add_pus_tc(
@ -45,19 +52,28 @@ def build_acs_subsystem_cmd(q: DefaultPusQueueHelper, cmd_str: str):
) )
) )
mode_info_tup = HANDLER_LIST.get(cmd_str) mode_info_tup = HANDLER_LIST.get(cmd_str)
assert mode_info_tup is not None
if mode_info_tup is None: if mode_info_tup is None:
return return
pack_mode_cmd_with_info( pack_mode_cmd_with_info(
object_id=PL_SUBSYSTEM_ID, object_id=PL_SUBSYSTEM_ID,
info=f"{info_prefix}: {mode_info_tup[1]}", info=f"{info_prefix}: {mode_info_tup[1]}",
submode=0,
mode=mode_info_tup[0], mode=mode_info_tup[0],
submode=0,
q=q, q=q,
) )
def create_payload_subsystem_node() -> CmdTreeNode: def create_payload_subsystem_node() -> CmdTreeNode:
payload_node = CmdTreeNode("payload", "Payload Subsystem") payload_node = CmdTreeNode("payload", "Payload Subsystem")
payload_node.add_child(CmdTreeNode(OpCode.OFF, Info.OFF)) subsystem_node = CmdTreeNode("subsystem", "Subsystem Commands")
payload_node.add_child(CmdTreeNode(OpCode.REPORT_ALL_MODES, Info.REPORT_ALL_MODES)) subsystem_node.add_child(CmdTreeNode(OpCode.OFF, Info.OFF))
subsystem_node.add_child(CmdTreeNode(OpCode.MPSOC_STREAM, Info.MPSOC_STREAM))
subsystem_node.add_child(CmdTreeNode(OpCode.EARTH_OBSV, Info.EARTH_OBSV))
subsystem_node.add_child(CmdTreeNode(OpCode.CAM_STREAM, Info.CAM_STREAM))
subsystem_node.add_child(
CmdTreeNode(OpCode.REPORT_ALL_MODES, Info.REPORT_ALL_MODES)
)
payload_node.add_child(subsystem_node)
return payload_node return payload_node