eive-tmtc/pus_tc/tc_packer_hook.py

"""Hook function which packs telecommands based on service and operation code string
"""
import os
from collections import deque
from typing import Union

from tmtccmd.config.definitions import CoreServiceList
from tmtccmd.utility.logger import get_console_logger
from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.tc.service_5_event import pack_generic_service5_test_into
from tmtccmd.pus.service_17_test import pack_service_17_ping_command

from pus_tc.service_200_mode import pack_service200_test_into
from pus_tc.p60dock import pack_p60dock_test_into
from pus_tc.pdu2 import pack_pdu2_commands
from pus_tc.pdu1 import pack_pdu1_commands
from pus_tc.acu import pack_acu_test_into
from pus_tc.imtq import pack_imtq_test_into
from pus_tc.tmp1075 import pack_tmp1075_test_into
from pus_tc.ploc_mpsoc import pack_ploc_mpsoc_test_into
from pus_tc.ploc_supervisor import pack_ploc_supv_test_into
from pus_tc.heater import pack_heater_test_into
from pus_tc.reaction_wheels import pack_single_rw_test_into
from pus_tc.rad_sensor import pack_rad_sensor_test_into
from pus_tc.ploc_upater import pack_ploc_updater_test_into
from pus_tc.ploc_memory_dumper import pack_ploc_memory_dumper_cmd
from pus_tc.ccsds_handler import pack_ccsds_handler_test
from pus_tc.core import pack_core_commands
from pus_tc.star_tracker import pack_star_tracker_commands
from pus_tc.syrlinks_hk_handler import pack_syrlinks_command
from pus_tc.gps import pack_gps_command
from pus_tc.acs import pack_acs_command
from pus_tc.str_img_helper import pack_str_img_helper_command
from config.definitions import CustomServiceList
from config.object_ids import (
    P60_DOCK_HANDLER,
    PDU_1_HANDLER_ID,
    PDU_2_HANDLER_ID,
    ACU_HANDLER_ID,
    TMP_1075_1_HANDLER_ID,
    TMP_1075_2_HANDLER_ID,
    HEATER_ID,
    IMTQ_HANDLER_ID,
    PLOC_MPSOC_ID,
    RW1_ID,
    RW2_ID,
    RW3_ID,
    RW4_ID,
    RAD_SENSOR_ID,
    PLOC_SUPV_ID,
    PLOC_UPDATER_ID,
    STAR_TRACKER_ID,
    PLOC_MEMORY_DUMPER_ID,
    GPS_HANDLER_0_ID,
    GPS_HANDLER_1_ID,
    CCSDS_HANDLER_ID,
    PDEC_HANDLER_ID,
    STR_IMG_HELPER_ID,
    SYRLINKS_HANDLER_ID,
)


LOGGER = get_console_logger()


def pack_service_queue_user(
    service: Union[str, int], op_code: str, service_queue: TcQueueT
):
    if service == CoreServiceList.SERVICE_5.value:
        return pack_generic_service5_test_into(tc_queue=service_queue)
    if service == CoreServiceList.SERVICE_17.value:
        return service_queue.appendleft(
            pack_service_17_ping_command(ssc=1700).pack_command_tuple()
        )
    if service == CoreServiceList.SERVICE_200.value:
        return pack_service200_test_into(tc_queue=service_queue)
    if service == CustomServiceList.P60DOCK.value:
        object_id = P60_DOCK_HANDLER
        return pack_p60dock_test_into(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.PDU1.value:
        object_id = PDU_1_HANDLER_ID
        return pack_pdu1_commands(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.PDU2.value:
        object_id = PDU_2_HANDLER_ID
        return pack_pdu2_commands(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.ACU.value:
        object_id = ACU_HANDLER_ID
        return pack_acu_test_into(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.TMP1075_1.value:
        object_id = TMP_1075_1_HANDLER_ID
        return pack_tmp1075_test_into(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.TMP1075_2.value:
        object_id = TMP_1075_2_HANDLER_ID
        return pack_tmp1075_test_into(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.HEATER.value:
        object_id = HEATER_ID
        return pack_heater_test_into(object_id=object_id, tc_queue=service_queue)
    if service == CustomServiceList.IMTQ.value:
        object_id = IMTQ_HANDLER_ID
        return pack_imtq_test_into(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.PLOC.value:
        object_id = PLOC_MPSOC_ID
        return pack_ploc_mpsoc_test_into(object_id=object_id, tc_queue=service_queue)
    if service == CustomServiceList.REACTION_WHEEL_1.value:
        object_id = RW1_ID
        return pack_single_rw_test_into(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.REACTION_WHEEL_2.value:
        object_id = RW2_ID
        return pack_single_rw_test_into(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.REACTION_WHEEL_3.value:
        object_id = RW3_ID
        return pack_single_rw_test_into(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.REACTION_WHEEL_4.value:
        object_id = RW4_ID
        return pack_single_rw_test_into(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.RAD_SENSOR.value:
        object_id = RAD_SENSOR_ID
        return pack_rad_sensor_test_into(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.PLOC_SUPV.value:
        object_id = PLOC_SUPV_ID
        return pack_ploc_supv_test_into(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.PLOC_UPDATER.value:
        object_id = PLOC_UPDATER_ID
        return pack_ploc_updater_test_into(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.STAR_TRACKER.value:
        object_id = STAR_TRACKER_ID
        return pack_star_tracker_commands(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.STR_IMG_HELPER.value:
        object_id = STR_IMG_HELPER_ID
        return pack_str_img_helper_command(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.CORE.value:
        return pack_core_commands(tc_queue=service_queue, op_code=op_code)
    if service == CustomServiceList.PLOC_MEMORY_DUMPER.value:
        object_id = PLOC_MEMORY_DUMPER_ID
        return pack_ploc_memory_dumper_cmd(
            object_id=object_id, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.ACS.value:
        return pack_acs_command(tc_queue=service_queue, op_code=op_code)
    if service == CustomServiceList.GPS_0.value:
        return pack_gps_command(
            object_id=GPS_HANDLER_0_ID, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.GPS_1.value:
        return pack_gps_command(
            object_id=GPS_HANDLER_1_ID, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.CCSDS_HANDLER.value:
        return pack_ccsds_handler_test(
            object_id=CCSDS_HANDLER_ID, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.PDEC_HANDLER.value:
        return pack_ccsds_handler_test(
            object_id=PDEC_HANDLER_ID, tc_queue=service_queue, op_code=op_code
        )
    if service == CustomServiceList.SYRLINKS.value:
        return pack_syrlinks_command(
            object_id=SYRLINKS_HANDLER_ID, tc_queue=service_queue, op_code=op_code
        )
    LOGGER.warning("Invalid Service !")


def create_total_tc_queue_user() -> TcQueueT:
    if not os.path.exists("log"):
        os.mkdir("log")
    tc_queue = deque()
    pack_generic_service5_test_into(tc_queue)
    tc_queue.appendleft(pack_service_17_ping_command(ssc=1700).pack_command_tuple())
    return tc_queue