eive/eive-tmtc
eive
/
eive-tmtc
13
0
Fork 0
Code Issues Pull Requests Projects Releases 97 Wiki Activity

last tests

This commit is contained in:
Robin Müller 2023-11-22 19:22:51 +01:00
parent 688602e486
commit f756df43a2
Signed by: muellerr
GPG Key ID: A649FB78196E3849
2 changed files with 36 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
eive_tmtc/pus_tc/cmd_demux.py
View File

@ -6,6 +6,7 @@ from typing import cast, List
from eive_tmtc.tmtc.acs.gyros import handle_gyr_cmd
from eive_tmtc.tmtc.acs.acs_ctrl import pack_acs_ctrl_command
from eive_tmtc.tmtc.test import pack_test_command
from eive_tmtc.tmtc.acs.mgms import handle_mgm_cmd
from eive_tmtc.tmtc.power.power import pack_power_commands
from eive_tmtc.tmtc.tcs.ctrl import pack_tcs_ctrl_commands
@ -70,6 +71,39 @@ from tmtccmd.util import ObjectIdU32
from eive_tmtc.utility.input_helper import InputHelper
def handle_pus_procedure( # noqa C901: Complexity okay here.
info: DefaultProcedureInfo,
queue_helper: DefaultPusQueueHelper,
):
cmd_path = info.cmd_path
assert cmd_path is not None
cmd_path_list = cmd_path.split("/")
if cmd_path_list[0] == "/":
cmd_path_list = cmd_path_list[1:]
if len(cmd_path_list) == 0:
raise ValueError(
"command path list empty. Full command path {cmd_path} might have invalid format"
)
if cmd_path_list[0] == "eps":
return handle_eps_procedure(queue_helper, cmd_path_list[1:])
if cmd_path_list[0] == "tcs":
return handle_tcs_procedure(queue_helper, cmd_path_list[1:])
if cmd_path_list[0] == "acs":
return handle_acs_procedure(queue_helper, cmd_path_list[1:])
if cmd_path_list[0] == "payload":
return handle_payload_procedure(queue_helper, cmd_path_list[1:])
if cmd_path_list[0] == "obdh":
return handle_obdh_procedure(queue_helper, cmd_path_list[1:])
if cmd_path_list[0] == "test":
assert len(cmd_path_list) >= 1
return pack_test_command(queue_helper, cmd_path_list[1])
if cmd_path_list[0] == "com":
return handle_com_procedure(queue_helper, cmd_path_list[1:])
logging.getLogger(__name__).warning(
f"invalid or unimplemented command path {cmd_path}"
)
def handle_eps_procedure(queue_helper: DefaultPusQueueHelper, cmd_path_list: List[str]):
obj_id_man = get_object_ids()
assert len(cmd_path_list) >= 1
@ -257,33 +291,3 @@ def handle_com_procedure(queue_helper: DefaultPusQueueHelper, cmd_path_list: Lis
return pack_syrlinks_command(
object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[2]
)
def handle_default_procedure( # noqa C901: Complexity okay here.
info: DefaultProcedureInfo,
queue_helper: DefaultPusQueueHelper,
):
cmd_path = info.cmd_path
assert cmd_path is not None
cmd_path_list = cmd_path.split("/")
if cmd_path_list[0] == "/":
cmd_path_list = cmd_path_list[1:]
if len(cmd_path_list) == 0:
raise ValueError(
"command path list empty. Full command path {cmd_path} might have invalid format"
)
if cmd_path_list[0] == "eps":
return handle_eps_procedure(queue_helper, cmd_path_list[1:])
if cmd_path_list[0] == "tcs":
return handle_tcs_procedure(queue_helper, cmd_path_list[1:])
if cmd_path_list[0] == "acs":
return handle_acs_procedure(queue_helper, cmd_path_list[1:])
if cmd_path_list[0] == "payload":
return handle_payload_procedure(queue_helper, cmd_path_list[1:])
if cmd_path_list[0] == "obdh":
return handle_obdh_procedure(queue_helper, cmd_path_list[1:])
if cmd_path_list[0] == "com":
return handle_com_procedure(queue_helper, cmd_path_list[1:])
logging.getLogger(__name__).warning(
f"invalid or unimplemented command path {cmd_path}"
)

4
eive_tmtc/pus_tc/tc_handler.py
View File

@ -6,7 +6,7 @@ from eive_tmtc.config.definitions import (
PUS_APID,
CFDP_LOCAL_ENTITY_ID,
)
from eive_tmtc.pus_tc.cmd_demux import handle_default_procedure
from eive_tmtc.pus_tc.cmd_demux import handle_pus_procedure
from eive_tmtc.cfdp.handler import CfdpInCcsdsHandler
from tmtccmd import TcHandlerBase, ProcedureWrapper
from tmtccmd.cfdp.defs import CfdpRequestType
@ -69,7 +69,7 @@ class TcHandler(TcHandlerBase):
def feed_cb(self, info: ProcedureWrapper, wrapper: FeedWrapper):
self.queue_helper.queue_wrapper = wrapper.queue_wrapper
if info.proc_type == TcProcedureType.DEFAULT:
handle_default_procedure(info.to_def_procedure(), self.queue_helper)
handle_pus_procedure(info.to_def_procedure(), self.queue_helper)
elif info.proc_type == TcProcedureType.CFDP:
self.handle_cfdp_procedure(info)