eive-tmtc/pus_tc/tmtcc_tc_packer_hook.py

73 lines
3.5 KiB
Python

"""
@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 config.tmtcc_definitions import ServiceList
from tmtc_core.utility.tmtcc_logger import get_logger
from tmtc_core.pus_tc.tmtcc_pus_tc_base import TcQueueT
from tmtc_core.pus_tc.tmtcc_tc_service_5_event import pack_generic_service5_test_into
from tmtc_core.pus_tc.tmtcc_tc_service_17_test import pack_service17_ping_command
from pus_tc.tmtcc_tc_p60dock import pack_p60dock_test_into
from pus_tc.tmtcc_tc_pdu2 import pack_pdu2_test_into
from pus_tc.tmtcc_tc_pdu1 import pack_pdu1_test_into
from pus_tc.tmtcc_tc_acu import pack_acu_test_into
from tmtc_core.core.tmtcc_object_id_manager import get_object_id
from config.tmtcc_object_ids import ObjectIds
from pus_tc.tmtcc_tc_tmp1075 import pack_tmp1075_test_into
from pus_tc.tmtcc_tc_heater import pack_heater_test_into
from pus_tc.tmtcc_tc_solar_array_deployment import pack_solar_array_deployment_test_into
from pus_tc.tmtcc_tc_syrlinks_hk_handler import pack_syrlinks_hk_handler_test_into
LOGGER = get_logger()
def pack_service_queue_user(service: ServiceList, op_code: str, service_queue: TcQueueT):
if service == ServiceList.SERVICE_5:
return pack_generic_service5_test_into(service_queue)
if service == ServiceList.SERVICE_17:
return service_queue.appendleft(pack_service17_ping_command(ssc=1700).pack_command_tuple())
if service == ServiceList.P60DOCK:
object_id = get_object_id(ObjectIds.P60DOCK_HANDLER_ID)
return pack_p60dock_test_into(object_id, service_queue)
if service == ServiceList.PDU1:
pdu1_object_id = get_object_id(ObjectIds.PDU1_HANDLER_ID)
p60dock_object_id = get_object_id(ObjectIds.P60DOCK_HANDLER_ID)
return pack_pdu1_test_into(pdu1_object_id, p60dock_object_id, service_queue)
if service == ServiceList.PDU2:
pdu2_object_id = get_object_id(ObjectIds.PDU2_HANDLER_ID)
p60dock_object_id = get_object_id(ObjectIds.P60DOCK_HANDLER_ID)
return pack_pdu2_test_into(pdu2_object_id, p60dock_object_id, service_queue)
if service == ServiceList.ACU:
object_id = get_object_id(ObjectIds.ACU_HANDLER_ID)
return pack_acu_test_into(object_id, service_queue)
if service == ServiceList.TMP1075_1:
object_id = get_object_id(ObjectIds.TMP1075_1_HANDLER_ID)
return pack_tmp1075_test_into(object_id, service_queue)
if service == ServiceList.TMP1075_2:
object_id = get_object_id(ObjectIds.TMP1075_2_HANDLER_ID)
return pack_tmp1075_test_into(object_id, service_queue)
if service == ServiceList.HEATER:
object_id = get_object_id(ObjectIds.HEATER)
return pack_heater_test_into(object_id, service_queue)
if service == ServiceList.SA_DEPL:
object_id = get_object_id(ObjectIds.SOLAR_ARRAY_DEPLOYMENT)
return pack_solar_array_deployment_test_into(object_id, service_queue)
if service == ServiceList.SYRLINKS:
object_id = get_object_id(ObjectIds.SYRLINKS_HK_HANDLER)
return pack_syrlinks_hk_handler_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