start fixing PLOC tree
All checks were successful
EIVE/-/pipeline/head This commit looks good

This commit is contained in:
Robin Müller 2024-04-09 11:32:57 +02:00
parent 33cf7b1613
commit 11d7ad0f8d
7 changed files with 29 additions and 27 deletions

View File

@ -3,6 +3,7 @@
@details Template configuration file. Copy this folder to the TMTC commander root and adapt @details Template configuration file. Copy this folder to the TMTC commander root and adapt
it to your needs. it to your needs.
""" """
import enum import enum
from tmtccmd import CcsdsTmtcBackend from tmtccmd import CcsdsTmtcBackend

View File

@ -3,6 +3,7 @@
@details Template configuration file. Copy this folder to the TMTC commander root and adapt @details Template configuration file. Copy this folder to the TMTC commander root and adapt
it to your needs. it to your needs.
""" """
import logging import logging
import os.path import os.path
from typing import Dict from typing import Dict

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

@ -1,5 +1,6 @@
"""Hook function which packs telecommands based on service and operation code string """Hook function which packs telecommands based on service and operation code string
""" """
import logging import logging
from typing import List, cast from typing import List, cast
@ -51,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
@ -251,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":

View File

@ -1,4 +1,5 @@
"""HK Handling for EIVE OBSW""" """HK Handling for EIVE OBSW"""
import dataclasses import dataclasses
import logging import logging
import base64 # noqa import base64 # noqa

View File

@ -1,5 +1,6 @@
"""Core EIVE TM handler module """Core EIVE TM handler module
""" """
import logging import logging
import sqlite3 import sqlite3
import uuid import uuid

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