eive-tmtc/eive_tmtc/config/hook.py

from typing import Optional

from prompt_toolkit.history import FileHistory
from tmtccmd import CcsdsTmtcBackend, HookBase
from tmtccmd.com import ComInterface
from tmtccmd.config import CmdTreeNode
from tmtccmd.config.hook import History
from tmtccmd.util import ObjectIdDictT, RetvalDictT

from eive_tmtc.config.definitions import SPACE_PACKET_IDS
from eive_tmtc.config.retvals import get_retval_dict
from eive_tmtc.tmtc.acs.acs_ctrl import create_acs_ctrl_node
from eive_tmtc.tmtc.acs.gps import create_gnss_node
from eive_tmtc.tmtc.acs.imtq import create_mgt_node
from eive_tmtc.tmtc.acs.star_tracker import create_str_node
from eive_tmtc.tmtc.acs.subsystem import create_acs_subsystem_node
from eive_tmtc.tmtc.com.ccsds_handler import create_ccsds_node
from eive_tmtc.tmtc.com.subsystem import create_com_subsystem_node
from eive_tmtc.tmtc.com.syrlinks_handler import create_syrlinks_node
from eive_tmtc.tmtc.core import create_core_node
from eive_tmtc.tmtc.health import create_global_health_node
from eive_tmtc.tmtc.payload.ploc_mpsoc import create_ploc_mpsoc_node
from eive_tmtc.tmtc.payload.ploc_supervisor import create_ploc_supv_node
from eive_tmtc.tmtc.payload.plpcdu import create_pl_pcdu_node
from eive_tmtc.tmtc.payload.scex import create_scex_node
from eive_tmtc.tmtc.payload.subsystem import create_payload_subsystem_node
from eive_tmtc.tmtc.power.bpx_batt import create_bpx_batt_node
from eive_tmtc.tmtc.power.pwr_ctrl import create_pwr_ctrl_node
from eive_tmtc.tmtc.power.subsystem import create_eps_subsystem_node
from eive_tmtc.tmtc.system import create_system_node
from eive_tmtc.tmtc.tcs.ctrl import create_tcs_ctrl_node
from eive_tmtc.tmtc.tcs.heater import create_heater_node
from eive_tmtc.tmtc.tcs.rtd import create_rtd_node
from eive_tmtc.tmtc.tcs.tmp1075 import create_tmp_sens_node
from eive_tmtc.tmtc.test import create_test_node
from eive_tmtc.tmtc.time import create_time_node
from eive_tmtc.tmtc.tm_store import create_persistent_tm_store_node
from eive_tmtc.tmtc.wdt import create_wdt_node


class EiveHookObject(HookBase):
    def __init__(self, json_cfg_path: str):
        super().__init__(json_cfg_path=json_cfg_path)

    def get_command_definitions(self) -> CmdTreeNode:
        root_node = CmdTreeNode.root_node()
        mode_node = CmdTreeNode("mode", "Mode Commands")
        hk_node = CmdTreeNode("hk", "HK Commands")
        action_node = CmdTreeNode("action", "Action Commands")
        param_node = CmdTreeNode("params", "Parameter Commands")
        health_node = CmdTreeNode("health", "Health Commands")
        dev_node = CmdTreeNode("dev", "Device Commands")
        dev_node.add_child(hk_node)
        dev_node.add_child(mode_node)
        dev_node.add_child(health_node)
        dev_node.add_child(param_node)
        # Not strictly necessary, but I do not know a single device which does not have actions..
        dev_node.add_child(action_node)
        assy_node = CmdTreeNode("assy", "Assembly Commands")
        assy_node.add_child(mode_node)

        acs_node = create_acs_subsystem_node()
        acs_brd_assy_node = CmdTreeNode("acs_brd_assy", "ACS Board Assembly")
        acs_brd_assy_node.add_child(mode_node)
        mgm_devs = CmdTreeNode("mgm_devs", "MGM Devices")
        gyro_devs = CmdTreeNode("gyro_devs", "Gyro Devices")
        acs_brd_assy_node.add_child(mgm_devs)
        acs_brd_assy_node.add_child(gyro_devs)
        acs_ctrl = create_acs_ctrl_node()
        rws = CmdTreeNode("rws", "Reaction Wheel Devices")
        rw_assy = CmdTreeNode("rw_assy", "Reaction Wheel Assembly")
        rw_1 = CmdTreeNode("rw_1", "Reaction Wheel 1")
        rw_2 = CmdTreeNode("rw_2", "Reaction Wheel 2")
        rw_3 = CmdTreeNode("rw_3", "Reaction Wheel 3")
        rw_4 = CmdTreeNode("rw_4", "Reaction Wheel 4")
        rws.add_child(rw_assy)
        rws.add_child(rw_1)
        rws.add_child(rw_2)
        rws.add_child(rw_3)
        rws.add_child(rw_4)
        star_tracker = create_str_node()
        star_tracker_img_helper = CmdTreeNode(
            "str_img_helper", "Star Tracker Image Helper"
        )
        star_tracker.add_child(star_tracker_img_helper)
        gnss_devs = create_gnss_node()

        acs_node.add_child(acs_brd_assy_node)
        acs_node.add_child(acs_ctrl)
        acs_node.add_child(mode_node)
        acs_node.add_child(rws)
        acs_node.add_child(create_mgt_node())
        acs_node.add_child(star_tracker)
        acs_node.add_child(gnss_devs)

        tcs_node = CmdTreeNode("tcs", "TCS Subsystem")
        tmp_1075_node = create_tmp_sens_node()
        rtds_node = create_rtd_node()
        heaters_node = create_heater_node()
        tcs_ctrl = create_tcs_ctrl_node()
        tcs_node.add_child(rtds_node)
        tcs_node.add_child(tmp_1075_node)
        tcs_node.add_child(tcs_ctrl)
        tcs_node.add_child(heaters_node)
        tcs_brd_assy = CmdTreeNode("tcs_brd_assy", "TCS Board Assembly")
        tcs_node.add_child(tcs_brd_assy)

        com_node = create_com_subsystem_node()
        com_node.add_child(create_syrlinks_node())
        com_node.add_child(create_ccsds_node())

        eps_node = create_eps_subsystem_node()
        acu_node = CmdTreeNode("acu", "PCDU ACU component")
        pdu_1_node = CmdTreeNode("pdu1", "PCDU PDU 1 component")
        pdu_2_node = CmdTreeNode("pdu2", "PCDU PDU 2 component")
        p60_dock_node = CmdTreeNode("p60_dock", "PCDU P60 Dock component")
        eps_node.add_child(create_pwr_ctrl_node())
        eps_node.add_child(acu_node)
        eps_node.add_child(pdu_1_node)
        eps_node.add_child(pdu_2_node)
        eps_node.add_child(p60_dock_node)
        eps_node.add_child(create_bpx_batt_node())

        payload_node = create_payload_subsystem_node()
        payload_node.add_child(create_pl_pcdu_node())
        payload_node.add_child(create_scex_node())
        payload_node.add_child(create_ploc_mpsoc_node())
        payload_node.add_child(create_ploc_supv_node())

        obdh_node = CmdTreeNode("obdh", "OBDH Subsystem")
        obdh_node.add_child(create_wdt_node())
        obdh_node.add_child(create_core_node())
        obdh_node.add_child(create_time_node())
        obdh_node.add_child(create_persistent_tm_store_node())

        root_node.add_child(create_test_node())
        root_node.add_child(create_system_node())
        root_node.add_child(create_global_health_node())
        root_node.add_child(acs_node)
        root_node.add_child(tcs_node)
        root_node.add_child(com_node)
        root_node.add_child(eps_node)
        root_node.add_child(payload_node)
        root_node.add_child(obdh_node)
        return root_node

    def get_communication_interface(self, com_if_key: str) -> Optional[ComInterface]:
        from tmtccmd.config.com import (
            create_com_interface_cfg_default,
            create_com_interface_default,
        )

        assert self.cfg_path is not None

        cfg = create_com_interface_cfg_default(
            com_if_key=com_if_key,
            json_cfg_path=self.cfg_path,
            space_packet_ids=SPACE_PACKET_IDS,
        )
        assert cfg is not None
        return create_com_interface_default(cfg)

    def perform_mode_operation(self, tmtc_backend: CcsdsTmtcBackend, mode: int):
        from eive_tmtc.config.custom_mode_op import custom_mode_operation

        custom_mode_operation(tmtc_backend, mode)

    def get_object_ids(self) -> ObjectIdDictT:
        from eive_tmtc.config.object_ids import get_object_ids

        return get_object_ids()

    def get_retval_dict(self) -> RetvalDictT:
        return get_retval_dict()

    def get_cmd_history(self) -> Optional[History]:
        return FileHistory(".tmtc-history.txt")