eive-tmtc/pus_tc/tc_packer_hook.py

"""
@brief      This file transfers control of TC packing to the user
@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
            it to your needs.
"""

import os
from collections import deque
from typing import Union

from tmtccmd.core.definitions import CoreServiceList
from tmtccmd.utility.tmtcc_logger import get_logger
from tmtccmd.pus_tc.definitions import TcQueueT
from tmtccmd.pus_tc.service_5_event import pack_generic_service5_test_into
from tmtccmd.pus_tc.service_17_test import pack_service17_ping_command
from tmtccmd.core.object_id_manager import get_object_id

from pus_tc.p60dock import pack_p60dock_test_into
from pus_tc.pdu2 import pack_pdu2_test_into
from pus_tc.pdu1 import pack_pdu1_test_into
from pus_tc.acu import pack_acu_test_into
from pus_tc.tmp1075 import pack_tmp1075_test_into
from pus_tc.heater import pack_heater_test_into
from config.definitions import CustomServiceList
from config.object_ids import ObjIdIds


LOGGER = get_logger()


def pack_service_queue_user(service: Union[str, int], op_code: str, service_queue: TcQueueT):
    if service == CoreServiceList.SERVICE_5:
        return pack_generic_service5_test_into(service_queue)
    if service == CoreServiceList.SERVICE_17:
        return service_queue.appendleft(pack_service17_ping_command(ssc=1700).pack_command_tuple())
    if service == CustomServiceList.P60DOCK.value:
        object_id = get_object_id(ObjIdIds.P60DOCK_HANDLER_ID)
        return pack_p60dock_test_into(object_id, service_queue)
    if service == CustomServiceList.PDU1.value:
        pdu1_object_id = get_object_id(ObjIdIds.PDU1_HANDLER_ID)
        p60dock_object_id = get_object_id(ObjIdIds.P60DOCK_HANDLER_ID)
        return pack_pdu1_test_into(pdu1_object_id, p60dock_object_id, service_queue)
    if service == CustomServiceList.PDU2.value:
        pdu2_object_id = get_object_id(ObjIdIds.PDU2_HANDLER_ID)
        p60dock_object_id = get_object_id(ObjIdIds.P60DOCK_HANDLER_ID)
        return pack_pdu2_test_into(pdu2_object_id, p60dock_object_id, service_queue)
    if service == CustomServiceList.ACU.value:
        object_id = get_object_id(ObjIdIds.ACU_HANDLER_ID)
        return pack_acu_test_into(object_id, service_queue)
    if service == CustomServiceList.TMP1075_1.value:
        object_id = get_object_id(ObjIdIds.TMP1075_1_HANDLER_ID)
        return pack_tmp1075_test_into(object_id, service_queue)
    if service == CustomServiceList.TMP1075_2.value:
        object_id = get_object_id(ObjIdIds.TMP1075_2_HANDLER_ID)
        return pack_tmp1075_test_into(object_id, service_queue)
    if service == CustomServiceList.HEATER.value:
        object_id = get_object_id(ObjIdIds.HEATER)
        return pack_heater_test_into(object_id, service_queue)
    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_service17_ping_command(ssc=1700).pack_command_tuple())
    return tc_queue