eive-tmtc/eive_tmtc/pus_tc/cmd_demux.py

295 lines
12 KiB
Python
Raw Normal View History

2021-09-13 18:06:25 +02:00
"""Hook function which packs telecommands based on service and operation code string
2020-12-17 17:50:00 +01:00
"""
2023-02-01 11:17:04 +01:00
import logging
2023-11-22 19:42:26 +01:00
from typing import List, cast
2021-03-19 17:39:52 +01:00
2023-11-22 11:21:26 +01:00
from tmtccmd import DefaultProcedureInfo
2023-11-10 19:23:06 +01:00
from tmtccmd.tmtc import DefaultPusQueueHelper
2023-11-22 19:42:26 +01:00
from tmtccmd.util import ObjectIdU32
2021-03-19 17:39:52 +01:00
2023-11-22 19:42:26 +01:00
import eive_tmtc.config.object_ids as oids
2022-11-29 16:53:29 +01:00
from eive_tmtc.config.object_ids import (
2022-01-18 14:03:56 +01:00
ACU_HANDLER_ID,
2023-11-22 19:42:26 +01:00
CCSDS_HANDLER_ID,
HEATER_CONTROLLER_ID,
2022-01-18 14:03:56 +01:00
IMTQ_HANDLER_ID,
2023-11-22 19:42:26 +01:00
P60_DOCK_HANDLER,
PDEC_HANDLER_ID,
PDU_1_HANDLER_ID,
PDU_2_HANDLER_ID,
RAD_SENSOR_ID,
2022-01-18 14:03:56 +01:00
RW1_ID,
RW2_ID,
RW3_ID,
RW4_ID,
2023-11-22 19:42:26 +01:00
RW_ASSEMBLY,
2022-01-18 14:03:56 +01:00
STAR_TRACKER_ID,
STR_IMG_HELPER_ID,
SYRLINKS_HANDLER_ID,
2023-11-22 19:42:26 +01:00
TMP1075_HANDLER_IF_BRD_ID,
TMP1075_HANDLER_PLPCDU_0_ID,
TMP1075_HANDLER_TCS_BRD_0_ID,
TMP1075_HANDLER_TCS_BRD_1_ID,
2022-05-23 18:25:25 +02:00
get_object_ids,
2022-01-18 14:03:56 +01:00
)
2023-11-22 19:42:26 +01:00
from eive_tmtc.tmtc.acs.acs_board import pack_acs_command
from eive_tmtc.tmtc.acs.acs_ctrl import pack_acs_ctrl_command
2023-02-01 16:25:17 +01:00
from eive_tmtc.tmtc.acs.gps import pack_gps_command
2023-11-22 19:42:26 +01:00
from eive_tmtc.tmtc.acs.gyros import handle_gyr_cmd
from eive_tmtc.tmtc.acs.imtq import create_imtq_command
from eive_tmtc.tmtc.acs.mgms import handle_mgm_cmd
from eive_tmtc.tmtc.acs.reaction_wheels import (
create_single_rw_cmd,
pack_rw_ass_cmds,
)
from eive_tmtc.tmtc.acs.star_tracker import pack_star_tracker_commands
2023-02-01 16:25:17 +01:00
from eive_tmtc.tmtc.acs.str_img_helper import pack_str_img_helper_command
2023-11-22 19:42:26 +01:00
from eive_tmtc.tmtc.com.ccsds_handler import pack_ccsds_handler_command
from eive_tmtc.tmtc.com.pdec_handler import pack_pdec_handler_commands
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.core import pack_core_commands
from eive_tmtc.tmtc.health import build_health_cmds
from eive_tmtc.tmtc.payload.plpcdu import pack_pl_pcdu_commands
from eive_tmtc.tmtc.payload.rad_sensor import create_rad_sensor_cmd
from eive_tmtc.tmtc.payload.scex import pack_scex_cmds
from eive_tmtc.tmtc.power.acu import pack_acu_commands
from eive_tmtc.tmtc.power.p60dock import pack_p60dock_cmds
from eive_tmtc.tmtc.power.pdu1 import pack_pdu1_commands
from eive_tmtc.tmtc.power.pdu2 import pack_pdu2_commands
from eive_tmtc.tmtc.power.power import pack_power_commands
from eive_tmtc.tmtc.system import build_system_cmds
from eive_tmtc.tmtc.tcs.ctrl import pack_tcs_ctrl_commands
from eive_tmtc.tmtc.tcs.heater import pack_heater_cmds
from eive_tmtc.tmtc.tcs.rtd import pack_rtd_commands
from eive_tmtc.tmtc.tcs.subsystem import pack_tcs_sys_commands
from eive_tmtc.tmtc.tcs.tmp1075 import pack_tmp1075_test_into
from eive_tmtc.tmtc.test import build_test_commands
from eive_tmtc.tmtc.time import pack_time_management_cmd
from eive_tmtc.tmtc.wdt import pack_wdt_commands
2023-02-06 14:27:00 +01:00
from eive_tmtc.utility.input_helper import InputHelper
2020-12-17 17:50:00 +01:00
2023-11-22 19:22:51 +01:00
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("/")
2023-11-22 19:25:04 +01:00
if cmd_path_list[0] == "":
2023-11-22 19:22:51 +01:00
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"
)
2023-11-22 19:42:26 +01:00
if cmd_path_list[0] == "system":
assert len(cmd_path_list) >= 1
return build_system_cmds(queue_helper, cmd_path_list[1])
if cmd_path_list[0] == "health":
assert len(cmd_path_list) >= 1
return build_health_cmds(queue_helper, cmd_path_list[1])
2023-11-22 19:22:51 +01:00
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
2023-11-22 19:42:26 +01:00
return build_test_commands(queue_helper, cmd_path_list[1])
2023-11-22 19:22:51 +01:00
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}"
)
2023-11-22 19:13:22 +01:00
def handle_eps_procedure(queue_helper: DefaultPusQueueHelper, cmd_path_list: List[str]):
2022-05-23 11:24:55 +02:00
obj_id_man = get_object_ids()
2023-11-22 19:38:04 +01:00
if len(cmd_path_list) == 1:
return pack_power_commands(queue_helper, cmd_path_list[0])
assert len(cmd_path_list) >= 2
if cmd_path_list[0] == "p60_dock":
object_id = cast(ObjectIdU32, obj_id_man.get(P60_DOCK_HANDLER))
2023-11-22 10:17:05 +01:00
assert len(cmd_path_list) >= 2
return pack_p60dock_cmds(
2023-11-22 19:38:04 +01:00
object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[1]
2023-11-22 10:17:05 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "pdu1":
object_id = cast(ObjectIdU32, obj_id_man.get(PDU_1_HANDLER_ID))
2023-11-22 10:17:05 +01:00
return pack_pdu1_commands(
2023-11-22 19:38:04 +01:00
object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[1]
2023-11-22 10:17:05 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "pdu2":
object_id = cast(ObjectIdU32, obj_id_man.get(PDU_2_HANDLER_ID))
2023-11-22 10:17:05 +01:00
return pack_pdu2_commands(
2023-11-22 19:38:04 +01:00
object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[1]
2023-11-22 10:17:05 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "acu":
object_id = cast(ObjectIdU32, obj_id_man.get(ACU_HANDLER_ID))
2023-11-22 10:17:05 +01:00
return pack_acu_commands(
2023-11-22 19:38:04 +01:00
object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[1]
2023-11-22 10:17:05 +01:00
)
2023-11-22 19:13:22 +01:00
def handle_tcs_procedure(queue_helper: DefaultPusQueueHelper, cmd_path_list: List[str]):
obj_id_man = get_object_ids()
2023-11-22 19:38:04 +01:00
if len(cmd_path_list) == 1:
return pack_tcs_sys_commands(q=queue_helper, cmd_str=cmd_path_list[0])
assert len(cmd_path_list) >= 2
if cmd_path_list[0] == "tcs_brd_assy":
return pack_tcs_sys_commands(q=queue_helper, cmd_str=cmd_path_list[1])
if cmd_path_list[0] == "rtd_devs":
2023-11-22 19:13:22 +01:00
assert len(cmd_path_list) >= 2
return pack_rtd_commands(
2023-11-22 19:38:04 +01:00
object_id=None, q=queue_helper, cmd_str=cmd_path_list[1]
2023-11-22 19:13:22 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "tcs_ctrl":
2023-11-22 10:17:05 +01:00
assert len(cmd_path_list) >= 2
return pack_tcs_ctrl_commands(q=queue_helper, cmd_str=cmd_path_list[2])
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "tmp1075_devs":
2023-02-06 14:27:00 +01:00
menu_dict = {
"0": ("TMP1075 TCS Board 0", TMP1075_HANDLER_TCS_BRD_0_ID),
"1": ("TMP1075 TCS Board 1", TMP1075_HANDLER_TCS_BRD_1_ID),
"2": ("TMP1075 PL PCDU 0", TMP1075_HANDLER_PLPCDU_0_ID),
2023-06-18 13:37:21 +02:00
"3": ("TMP1075 PL PCDU 1", oids.TMP1075_HANDLER_PLPCDU_1_ID),
2023-02-06 14:27:00 +01:00
"4": ("TMP1075 IF Board", TMP1075_HANDLER_IF_BRD_ID),
}
input_helper = InputHelper(menu_dict)
tmp_select = input_helper.get_key()
2023-11-22 10:17:05 +01:00
assert len(cmd_path_list) >= 2
2023-02-06 14:49:50 +01:00
object_id = obj_id_man.get(menu_dict[tmp_select][1])
2023-11-22 10:17:05 +01:00
assert object_id is not None
2022-01-18 14:03:56 +01:00
return pack_tmp1075_test_into(
2023-11-22 19:38:04 +01:00
object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[1]
2022-01-18 14:03:56 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "heaters":
object_id = HEATER_CONTROLLER_ID
2023-11-22 10:17:05 +01:00
return pack_heater_cmds(
2023-11-22 19:38:04 +01:00
object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[1]
2023-11-22 10:17:05 +01:00
)
2023-11-22 19:13:22 +01:00
def handle_acs_procedure(queue_helper: DefaultPusQueueHelper, cmd_path_list: List[str]):
obj_id_man = get_object_ids()
2023-11-22 19:38:04 +01:00
if len(cmd_path_list) == 1:
return pack_acs_command(q=queue_helper, cmd_str=cmd_path_list[0])
2023-11-22 19:13:22 +01:00
assert len(cmd_path_list) >= 2
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "mgt":
object_id = cast(ObjectIdU32, obj_id_man.get(IMTQ_HANDLER_ID))
2023-11-22 10:17:05 +01:00
return create_imtq_command(
2023-11-22 19:38:04 +01:00
object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[1]
2023-11-22 10:17:05 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "str":
object_id = cast(ObjectIdU32, obj_id_man.get(STAR_TRACKER_ID))
2021-12-02 08:01:18 +01:00
return pack_star_tracker_commands(
2023-11-22 19:38:04 +01:00
object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[1]
2021-12-02 08:01:18 +01:00
)
2023-11-22 19:13:22 +01:00
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "rws":
2023-11-22 19:13:22 +01:00
assert len(cmd_path_list) >= 3
2023-11-22 19:38:04 +01:00
if cmd_path_list[1] == "rw_assy":
2023-11-22 19:13:22 +01:00
assert len(cmd_path_list) >= 4
return pack_rw_ass_cmds(
2023-11-22 19:38:04 +01:00
q=queue_helper, object_id=RW_ASSEMBLY, cmd_str=cmd_path_list[2]
2023-11-22 19:13:22 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[1] == "rw_1":
2023-11-22 19:13:22 +01:00
assert len(cmd_path_list) >= 4
return create_single_rw_cmd(
2023-11-22 19:38:04 +01:00
object_id=RW1_ID, rw_idx=1, q=queue_helper, cmd_str=cmd_path_list[2]
2023-11-22 19:13:22 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[1] == "rw_2":
2023-11-22 19:13:22 +01:00
assert len(cmd_path_list) >= 4
return create_single_rw_cmd(
2023-11-22 19:38:04 +01:00
object_id=RW2_ID, rw_idx=2, q=queue_helper, cmd_str=cmd_path_list[2]
2023-11-22 19:13:22 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[1] == "rw_3":
2023-11-22 19:13:22 +01:00
assert len(cmd_path_list) >= 4
return create_single_rw_cmd(
2023-11-22 19:38:04 +01:00
object_id=RW3_ID, rw_idx=3, q=queue_helper, cmd_str=cmd_path_list[2]
2023-11-22 19:13:22 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[1] == "rw_4":
2023-11-22 19:13:22 +01:00
assert len(cmd_path_list) >= 4
return create_single_rw_cmd(
2023-11-22 19:38:04 +01:00
object_id=RW4_ID, rw_idx=4, q=queue_helper, cmd_str=cmd_path_list[2]
2023-11-22 19:13:22 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "gnss_devs":
2023-11-22 19:13:22 +01:00
return pack_gps_command(
2023-11-22 19:38:04 +01:00
object_id=oids.GPS_CONTROLLER, q=queue_helper, cmd_str=cmd_path_list[1]
2023-11-22 19:13:22 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "acs_brd_assy":
2023-11-22 10:17:05 +01:00
assert len(cmd_path_list) >= 3
2023-11-22 19:38:04 +01:00
if cmd_path_list[1] == "mgm_devs":
return handle_mgm_cmd(q=queue_helper, cmd_str=cmd_path_list[2])
if cmd_path_list[1] == "gyro_devs":
assert len(cmd_path_list) >= 3
return handle_gyr_cmd(q=queue_helper, cmd_str=cmd_path_list[2])
if cmd_path_list[0] == "str_img_helper":
object_id = cast(ObjectIdU32, obj_id_man.get(STR_IMG_HELPER_ID))
2021-12-02 08:01:18 +01:00
return pack_str_img_helper_command(
2023-11-22 19:38:04 +01:00
object_id=object_id, q=queue_helper, op_code=cmd_path_list[1]
2023-11-22 10:17:05 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "acs_ctrl":
return pack_acs_ctrl_command(queue_helper, cmd_path_list[1])
2023-11-22 19:13:22 +01:00
def handle_payload_procedure(
queue_helper: DefaultPusQueueHelper, cmd_path_list: List[str]
):
obj_id_man = get_object_ids()
assert len(cmd_path_list) >= 2
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "rad_sensor":
2023-11-22 10:17:05 +01:00
assert len(cmd_path_list) >= 3
2023-11-22 19:13:22 +01:00
object_id = cast(ObjectIdU32, obj_id_man.get(RAD_SENSOR_ID))
return create_rad_sensor_cmd(
object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[2]
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "pl_pcdu":
2023-11-22 19:13:22 +01:00
assert len(cmd_path_list) >= 3
return pack_pl_pcdu_commands(q=queue_helper, cmd_str=cmd_path_list[2])
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "scex":
2023-11-22 19:13:22 +01:00
assert len(cmd_path_list) >= 3
return pack_scex_cmds(
cmd_str=cmd_path_list[2],
q=queue_helper,
2022-01-18 14:03:56 +01:00
)
2023-11-22 19:13:22 +01:00
def handle_obdh_procedure(
queue_helper: DefaultPusQueueHelper, cmd_path_list: List[str]
):
assert len(cmd_path_list) >= 2
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "core":
return pack_core_commands(q=queue_helper, cmd_str=cmd_path_list[1])
if cmd_path_list[0] == "xiphos_wdt":
return pack_wdt_commands(queue_helper, cmd_path_list[1])
if cmd_path_list[0] == "time":
2023-11-22 19:13:22 +01:00
return pack_time_management_cmd(queue_helper, cmd_path_list[1])
def handle_com_procedure(queue_helper: DefaultPusQueueHelper, cmd_path_list: List[str]):
obj_id_man = get_object_ids()
2023-11-22 19:38:04 +01:00
if len(cmd_path_list) == 1:
return build_com_subsystem_procedure(queue_helper, cmd_path_list[0])
2023-11-22 19:13:22 +01:00
assert len(cmd_path_list) >= 2
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "ccsds":
2023-01-24 19:07:11 +01:00
object_id = cast(ObjectIdU32, obj_id_man.get(CCSDS_HANDLER_ID))
2023-06-19 17:16:00 +02:00
return pack_ccsds_handler_command(
2023-11-22 10:17:05 +01:00
object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[2]
2022-01-18 14:03:56 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "pdec":
2023-11-22 10:17:05 +01:00
return pack_pdec_handler_commands(
object_id=PDEC_HANDLER_ID, q=queue_helper, cmd_str=cmd_path_list[2]
2022-01-18 14:03:56 +01:00
)
2023-11-22 19:38:04 +01:00
if cmd_path_list[0] == "syrlinks":
2022-08-18 11:09:35 +02:00
object_id = cast(ObjectIdU32, obj_id_man.get(SYRLINKS_HANDLER_ID))
2022-01-18 14:03:56 +01:00
return pack_syrlinks_command(
2023-11-22 10:17:05 +01:00
object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[2]
)