105 lines
3.8 KiB
Python
105 lines
3.8 KiB
Python
import struct
|
|
from config.object_ids import *
|
|
from pus_tc.devs.imtq import ImtqActionIds
|
|
from pus_tc.devs.ploc_mpsoc import PlocReplyIds
|
|
from pus_tc.devs.ploc_supervisor import SupvActionIds
|
|
from pus_tc.devs.star_tracker import StarTrackerActionIds
|
|
from tmtccmd.logging import get_console_logger
|
|
from tmtccmd.config.definitions import DataReplyUnpacked
|
|
|
|
LOGGER = get_console_logger()
|
|
|
|
|
|
def user_analyze_service_8_data(
|
|
object_id: bytes, action_id: int, custom_data: bytearray
|
|
) -> DataReplyUnpacked:
|
|
"""
|
|
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:
|
|
"""
|
|
if object_id == PDU_2_HANDLER_ID:
|
|
reply = DataReplyUnpacked()
|
|
reply.header_list = ["PDU2 Service 8 Reply"]
|
|
data_string = str()
|
|
for index in range(len(custom_data)):
|
|
data_string += str(hex(custom_data[index])) + " , "
|
|
data_string = data_string.rstrip()
|
|
data_string = data_string.rstrip(",")
|
|
data_string = data_string.rstrip()
|
|
reply.content_list = [data_string]
|
|
return reply
|
|
elif object_id == IMTQ_HANDLER_ID:
|
|
return handle_imtq_replies(action_id, custom_data)
|
|
elif object_id == PLOC_MPSOC_ID:
|
|
return handle_ploc_replies(action_id, custom_data)
|
|
elif object_id == PLOC_SUPV_ID:
|
|
return handle_supervisor_replies(action_id, custom_data)
|
|
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()
|
|
if action_id == struct.unpack("!I", ImtqActionIds.get_commanded_dipole)[0]:
|
|
reply.header_list = [
|
|
"Commanded X-Dipole",
|
|
"Commanded Y-Dipole",
|
|
"Commanded Z-Dipole",
|
|
]
|
|
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
|
|
|
|
|
|
def handle_ploc_replies(action_id: int, custom_data: bytearray) -> DataReplyUnpacked:
|
|
reply = DataReplyUnpacked()
|
|
if action_id == PlocReplyIds.tm_mem_read_report:
|
|
reply.header_list = [
|
|
"PLOC Memory Address",
|
|
"PLOC Mem Len",
|
|
"PLOC Read Memory Data",
|
|
]
|
|
reply.content_list = [
|
|
"0x" + custom_data[:4].hex(),
|
|
struct.unpack("!H", custom_data[4:6])[0],
|
|
"0x" + custom_data[6:10].hex(),
|
|
]
|
|
return reply
|
|
|
|
|
|
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
|
|
|
|
|
|
def handle_startracker_replies(
|
|
action_id: int, custom_data: bytearray
|
|
) -> DataReplyUnpacked:
|
|
reply = DataReplyUnpacked()
|
|
if action_id == StarTrackerActionIds.CHECKSUM:
|
|
if len(custom_data) != 5:
|
|
LOGGER.warning(
|
|
"Star tracker reply has invalid length {0}".format(len(custom_data))
|
|
)
|
|
return reply
|
|
reply.header_list = ["Checksum", "Checksum valid"]
|
|
print(custom_data[4])
|
|
checksum_valid_flag = custom_data[4] >> 8
|
|
reply.content_list = ["0x" + custom_data[:4].hex(), checksum_valid_flag]
|
|
return reply
|