eive-tmtc/pus_tm/service_8_hook.py

import struct
from typing import Tuple
from config.object_ids import *
from pus_tc.imtq import ImtqActionIds
from pus_tc.ploc_mpsoc import PlocReplyIds
from pus_tc.ploc_supervisor import SupvActionIds


def user_analyze_service_8_data(
        object_id: bytes, action_id: int, custom_data: bytearray) -> Tuple[list, list]:
    """
    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:
        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()
        content_list = [data_string]
    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)
    else:
        header_list = []
        content_list = []
    return header_list, content_list


def handle_imtq_replies(action_id: int, custom_data: bytearray) -> Tuple[list, list]:
    header_list = []
    content_list = []
    if action_id == struct.unpack('!I', ImtqActionIds.get_commanded_dipole)[0]:
        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])
        content_list = [x_dipole[0], y_dipole[0], z_dipole[0]]


def handle_ploc_replies(action_id: int, custom_data: bytearray) -> Tuple[list, list]:
    header_list = []
    content_list = []
    if action_id == PlocReplyIds.tm_mem_read_report:
        header_list = ['PLOC Memory Address', 'PLOC Mem Len', 'PLOC Read Memory Data']
        content_list = [custom_data[:4], custom_data[4:6], custom_data[6:10]]
    return header_list, content_list


def handle_supervisor_replies(action_id: int, custom_data: bytearray) -> Tuple[list, list]:
    header_list = []
    content_list = []
    if action_id == SupvActionIds.DUMP_MRAM:
        header_list = ['MRAM Dump']
        content_list = [custom_data[:len(custom_data)]]
    return header_list, content_list