eive-tmtc/pus_tm/action_reply_handler.py

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 gomspace.gomspace_common import GomspaceDeviceActionIds
from tmtccmd.logging import get_console_logger
from tmtccmd.config.definitions import DataReplyUnpacked
from tmtccmd.tm import Service8FsfwTm
from tmtccmd.utility.tmtc_printer import FsfwTmTcPrinter

LOGGER = get_console_logger()


def handle_action_reply(
    raw_tm: bytes, printer: FsfwTmTcPrinter, obj_id_dict: ObjectIdDictT
):
    """Core Action reply handler
    :return:
    """
    tm_packet = Service8FsfwTm.unpack(raw_telemetry=raw_tm)
    printer.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
    object_id = obj_id_dict.get(tm_packet.source_object_id_as_bytes)
    custom_data = tm_packet.custom_data
    action_id = tm_packet.action_id
    generic_print_str = printer.generic_action_packet_tm_print(
        packet=tm_packet, obj_id=object_id
    )
    print(generic_print_str)
    printer.file_logger.info(generic_print_str)
    if object_id.as_bytes == IMTQ_HANDLER_ID:
        return handle_imtq_replies(action_id, printer, custom_data)
    elif object_id.as_bytes == PLOC_MPSOC_ID:
        return handle_ploc_replies(action_id, printer, custom_data)
    elif object_id.as_bytes == PLOC_SUPV_ID:
        return handle_supervisor_replies(action_id, printer, custom_data)
    elif object_id.as_bytes == STAR_TRACKER_ID:
        return handle_startracker_replies(action_id, printer, custom_data)
    elif object_id.as_bytes == ACU_HANDLER_ID:
        return handle_acu_replies(action_id, printer, custom_data)


def handle_imtq_replies(
    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
):
    if action_id == struct.unpack("!I", ImtqActionIds.get_commanded_dipole)[0]:
        header_list = [
            "Commanded X-Dipole",
            "Commanded Y-Dipole",
            "Commanded Z-Dipole",
        ]
        [x_dipole, y_dipole, z_dipole] = struct.unpack("!HHH", custom_data[0:6])
        content_list = [x_dipole, y_dipole, z_dipole]
        print(header_list)
        print(content_list)
        printer.file_logger.info(header_list)
        printer.file_logger.info(content_list)


def handle_ploc_replies(
    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
):
    if action_id == PlocReplyIds.TM_MEM_READ_RPT:
        header_list = [
            "PLOC Memory Address",
            "PLOC Mem Len",
            "PLOC Read Memory Data",
        ]
        content_list = [
            "0x" + custom_data[:4].hex(),
            struct.unpack("!H", custom_data[4:6])[0],
            "0x" + custom_data[6:10].hex(),
        ]
        print(header_list)
        print(content_list)
        printer.file_logger.info(header_list)
        printer.file_logger.info(content_list)
    elif action_id == PlocReplyIds.TM_CAM_CMD_RPT:
        header_list = ["Camera reply string", "ACK"]
        content_list = [
            custom_data[: len(custom_data) - 1].decode("utf-8"),
            hex(custom_data[-1]),
        ]
        print(header_list)
        print(content_list)
        printer.file_logger.info(header_list)
        printer.file_logger.info(content_list)


def handle_supervisor_replies(
    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
):
    reply = DataReplyUnpacked()
    if action_id == SupvActionIds.DUMP_MRAM:
        header_list = ["MRAM Dump"]
        content_list = [custom_data[: len(custom_data)]]
        print(header_list)
        print(content_list)
        printer.file_logger.info(header_list)
        printer.file_logger.info(content_list)
    elif action_id == SupvActionIds.READ_GPIO:
        header_list = ["GPIO state"]
        content_list = [struct.unpack("!H", custom_data[:2])[0]]
        print(header_list)
        print(content_list)
        printer.file_logger.info(header_list)
        printer.file_logger.info(content_list)


def handle_startracker_replies(
    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
):
    if action_id == StarTrackerActionIds.CHECKSUM:
        if len(custom_data) != 5:
            LOGGER.warning(
                "Star tracker reply has invalid length {0}".format(len(custom_data))
            )
            return
        header_list = ["Checksum", "Checksum valid"]
        print(custom_data[4])
        checksum_valid_flag = custom_data[4] >> 8
        content_list = ["0x" + custom_data[:4].hex(), checksum_valid_flag]
        print(header_list)
        print(content_list)
        printer.file_logger.info(header_list)
        printer.file_logger.info(content_list)


def handle_acu_replies(
    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
):
    if action_id == GomspaceDeviceActionIds.PARAM_GET:
        header_list = [
            "Gomspace action ID"
            "Table ID",
            "Memory Address",
            "Payload length"
            "Payload"
        ]
        fmt_str = "!BBHH"
        (action, table_id, address, payload_length) = struct.unpack(fmt_str, custom_data[:6])
        content_list = [
            action,
            table_id,
            "0x" + hex(address),
            payload_length,
            "0x" + custom_data[6:].hex(),
        ]
        print(header_list)
        print(content_list)