243 lines
9.7 KiB
Python
243 lines
9.7 KiB
Python
"""Hook function which packs telecommands based on service and operation code string
|
|
"""
|
|
import logging
|
|
import os
|
|
from collections import deque
|
|
from typing import Union
|
|
|
|
from spacepackets.ecss import PusTelecommand
|
|
from tmtccmd.com_if.com_interface_base import CommunicationInterface
|
|
from tmtccmd.config.definitions import CoreServiceList
|
|
from tmtccmd.logging import get_console_logger
|
|
from tmtccmd.logging.pus import log_raw_pus_tc
|
|
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 tmtccmd.logging import get_current_time_string
|
|
|
|
from pus_tc.service_200_mode import pack_service200_test_into
|
|
from pus_tc.devs.p60dock import pack_p60dock_test_into
|
|
from pus_tc.devs.pdu2 import pack_pdu2_commands
|
|
from pus_tc.devs.pdu1 import pack_pdu1_commands
|
|
from pus_tc.devs.bpx_batt import pack_bpx_commands
|
|
from pus_tc.devs.acu import pack_acu_test_into
|
|
from pus_tc.devs.solar_array_deployment import pack_solar_array_deployment_test_into
|
|
from pus_tc.devs.imtq import pack_imtq_test_into
|
|
from pus_tc.devs.tmp1075 import pack_tmp1075_test_into
|
|
from pus_tc.devs.ploc_mpsoc import pack_ploc_mpsoc_commands
|
|
from pus_tc.devs.ploc_supervisor import pack_ploc_supv_commands
|
|
from pus_tc.devs.heater import pack_heater_test_into
|
|
from pus_tc.devs.reaction_wheels import pack_single_rw_test_into
|
|
from pus_tc.devs.rad_sensor import pack_rad_sensor_test_into
|
|
from pus_tc.devs.ploc_upater import pack_ploc_updater_test_into
|
|
from pus_tc.devs.ploc_memory_dumper import pack_ploc_memory_dumper_cmd
|
|
from pus_tc.devs.ccsds_handler import pack_ccsds_handler_test
|
|
from pus_tc.system.core import pack_core_commands
|
|
from pus_tc.devs.star_tracker import pack_star_tracker_commands
|
|
from pus_tc.devs.syrlinks_hk_handler import pack_syrlinks_command
|
|
from pus_tc.devs.gps import pack_gps_command
|
|
from pus_tc.system.acs import pack_acs_command, pack_sus_cmds
|
|
from pus_tc.devs.plpcdu import pack_pl_pcdu_commands
|
|
from pus_tc.devs.str_img_helper import pack_str_img_helper_command
|
|
from pus_tc.system.tcs import pack_tcs_sys_commands
|
|
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,
|
|
SOLAR_ARRAY_DEPLOYMENT_ID,
|
|
)
|
|
|
|
|
|
LOGGER = get_console_logger()
|
|
|
|
|
|
def pre_tc_send_cb(
|
|
packet: bytes,
|
|
com_if: CommunicationInterface,
|
|
pus_info: Union[PusTelecommand, any],
|
|
file_logger: logging.Logger,
|
|
):
|
|
log_raw_pus_tc(
|
|
packet=packet, srv_subservice=(pus_info.service, pus_info.subservice)
|
|
)
|
|
tc_info_string = f"Sent {pus_info}"
|
|
LOGGER.info(tc_info_string)
|
|
file_logger.info(f"{get_current_time_string(True)}: {tc_info_string}")
|
|
com_if.send(data=packet)
|
|
|
|
|
|
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.BPX_BATTERY.value:
|
|
return pack_bpx_commands(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_MPSOC.value:
|
|
object_id = PLOC_MPSOC_ID
|
|
return pack_ploc_mpsoc_commands(
|
|
object_id=object_id, tc_queue=service_queue, op_code=op_code
|
|
)
|
|
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_commands(
|
|
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
|
|
)
|
|
if service == CustomServiceList.SA_DEPLYOMENT.value:
|
|
return pack_solar_array_deployment_test_into(
|
|
object_id=SOLAR_ARRAY_DEPLOYMENT_ID, tc_queue=service_queue
|
|
)
|
|
if service == CustomServiceList.SUS_ASS.value:
|
|
return pack_sus_cmds(tc_queue=service_queue, op_code=op_code)
|
|
if service == CustomServiceList.PL_PCDU.value:
|
|
return pack_pl_pcdu_commands(tc_queue=service_queue, op_code=op_code)
|
|
if service == CustomServiceList.ACS_ASS.value:
|
|
return pack_acs_command(tc_queue=service_queue, op_code=op_code)
|
|
if service == CustomServiceList.TCS_ASS.value:
|
|
return pack_tcs_sys_commands(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
|