eive-tmtc/pus_tm/service_8_hook.py

101 lines
3.6 KiB
Python
Raw Normal View History

import struct
2021-06-11 13:51:56 +02:00
from config.object_ids import *
from pus_tc.imtq import ImtqActionIds
2021-07-11 14:29:11 +02:00
from pus_tc.ploc_mpsoc import PlocReplyIds
from pus_tc.ploc_supervisor import SupvActionIds
2021-12-24 07:32:35 +01:00
from pus_tc.star_tracker import StarTrackerActionIds
from tmtccmd.utility.logger import get_console_logger
from tmtccmd.config.definitions import DataReplyUnpacked
2021-12-24 07:32:35 +01:00
LOGGER = get_console_logger()
2021-02-09 12:35:10 +01:00
def user_analyze_service_8_data(
2022-01-18 14:03:56 +01:00
object_id: bytes, action_id: int, custom_data: bytearray
) -> DataReplyUnpacked:
2021-02-09 12:35:10 +01:00
"""
This function is called by the TMTC core if a Service 8 data reply (subservice 130)
is received. The user can return a tuple of two lists, where the first list
is a list of header strings to print and the second list is a list of values to print.
The TMTC core will take care of printing both lists and logging them.
@param object_id:
@param action_id:
@param custom_data:
@return:
"""
2021-05-17 17:42:04 +02:00
if object_id == PDU_2_HANDLER_ID:
reply = DataReplyUnpacked()
reply.header_list = ["PDU2 Service 8 Reply"]
2021-02-11 08:18:42 +01:00
data_string = str()
for index in range(len(custom_data)):
data_string += str(hex(custom_data[index])) + " , "
data_string = data_string.rstrip()
2022-01-18 14:03:56 +01:00
data_string = data_string.rstrip(",")
2021-02-11 08:18:42 +01:00
data_string = data_string.rstrip()
reply.content_list = [data_string]
return reply
2021-06-11 13:51:56 +02:00
elif object_id == IMTQ_HANDLER_ID:
return handle_imtq_replies(action_id, custom_data)
2021-07-09 12:57:39 +02:00
elif object_id == PLOC_MPSOC_ID:
2021-04-24 13:27:57 +02:00
return handle_ploc_replies(action_id, custom_data)
elif object_id == PLOC_SUPV_ID:
return handle_supervisor_replies(action_id, custom_data)
2021-12-24 07:32:35 +01:00
elif object_id == STAR_TRACKER_ID:
return handle_startracker_replies(action_id, custom_data)
return DataReplyUnpacked()
def handle_imtq_replies(action_id: int, custom_data: bytearray) -> DataReplyUnpacked:
reply = DataReplyUnpacked()
2022-01-18 14:03:56 +01:00
if action_id == struct.unpack("!I", ImtqActionIds.get_commanded_dipole)[0]:
2022-02-03 16:02:55 +01:00
reply.header_list = [
"Commanded X-Dipole",
"Commanded Y-Dipole",
"Commanded Z-Dipole",
]
2022-01-18 14:03:56 +01:00
x_dipole = struct.unpack("!H", custom_data[:2])
y_dipole = struct.unpack("!H", custom_data[2:4])
z_dipole = struct.unpack("!H", custom_data[4:6])
reply.content_list = [x_dipole[0], y_dipole[0], z_dipole[0]]
return reply
2021-04-26 15:01:22 +02:00
def handle_ploc_replies(action_id: int, custom_data: bytearray) -> DataReplyUnpacked:
reply = DataReplyUnpacked()
2021-04-24 13:27:57 +02:00
if action_id == PlocReplyIds.tm_mem_read_report:
2022-02-03 16:02:55 +01:00
reply.header_list = [
"PLOC Memory Address",
"PLOC Mem Len",
"PLOC Read Memory Data",
]
reply.content_list = [custom_data[:4], custom_data[4:6], custom_data[6:10]]
return reply
2022-01-18 14:03:56 +01:00
def handle_supervisor_replies(
action_id: int, custom_data: bytearray
) -> DataReplyUnpacked:
reply = DataReplyUnpacked()
if action_id == SupvActionIds.DUMP_MRAM:
reply.header_list = ["MRAM Dump"]
reply.content_list = [custom_data[: len(custom_data)]]
return reply
2021-12-24 07:32:35 +01:00
2022-01-18 14:03:56 +01:00
def handle_startracker_replies(
action_id: int, custom_data: bytearray
) -> DataReplyUnpacked:
reply = DataReplyUnpacked()
2021-12-24 07:32:35 +01:00
if action_id == StarTrackerActionIds.CHECKSUM:
if len(custom_data) != 5:
2022-01-18 14:03:56 +01:00
LOGGER.warning(
"Star tracker reply has invalid length {0}".format(len(custom_data))
)
return reply
reply.header_list = ["Checksum", "Checksum valid"]
2021-12-24 07:32:35 +01:00
print(custom_data[4])
checksum_valid_flag = custom_data[4] >> 8
reply.content_list = ["0x" + custom_data[:4].hex(), checksum_valid_flag]
return reply