eive-tmtc/eive_tmtc/tmtc/acs/imtq.py

# -*- coding: utf-8 -*-
"""
@file   imtq.py
@brief  Tests for the ISIS IMTQ (Magnettorquer) device handler
@author J. Meier
@date   25.03.2021
"""
import enum
import logging
import struct
from typing import List

from tmtccmd.config import CmdTreeNode

from eive_tmtc.config.definitions import CustomServiceList
from eive_tmtc.pus_tm.defs import PrintWrapper
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.config.tmtc import (
    tmtc_definitions_provider,
    OpCodeEntry,
    TmtcDefinitionWrapper,
)
from tmtccmd.tmtc import DefaultPusQueueHelper
from tmtccmd.pus.tc.s3_fsfw_hk import (
    make_sid,
    generate_one_diag_command,
    generate_one_hk_command,
    create_request_one_diag_command,
    create_enable_periodic_hk_command_with_interval_with_diag,
    create_disable_periodic_hk_command_with_diag,
)
from tmtccmd.pus.s200_fsfw_mode import pack_mode_data, Mode
from tmtccmd.util import ObjectIdU32
from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter


_LOGGER = logging.getLogger(__name__)


class OpCode:
    ON = "on"
    NORMAL = "normal"
    OFF = "off"
    SET_DIPOLE = "set_dipole"
    REQUEST_ENG_HK_NO_TORQUE = "hk_os_eng_hk_no_torque"
    REQUEST_ENG_HK_WITH_TORQUE = "hk_os_eng_hk_with_torque"
    REQUEST_MGM_RAW_NO_TORQUE = "hk_os_mgm_raw_no_torque"
    ENABLE_MGM_RAW_NO_TORQUE = "enb_mgm_raw_no_torque"
    DISABLE_MGM_RAW_NO_TORQUE = "dis_mgm_raw_no_torque"
    REQUEST_MGM_RAW_WITH_TORQUE = "hk_os_mgm_raw_with_torque"
    ENABLE_MGM_RAW_WITH_TORQUE = "enb_mgm_raw_with_torque"
    DISABLE_MGM_RAW_WITH_TORQUE = "dis_mgm_raw_with_torque"
    ENABLE_ENG_HK_NO_TORQUE = "enb_eng_hk_no_torque"
    DISABLE_ENG_HK_NO_TORQUE = "dis_eng_hk_no_torque"
    ENABLE_ENG_HK_WITH_TORQUE = "enb_eng_hk_with_torque"
    DISABLE_ENG_HK_WITH_TORQUE = "dis_eng_hk_with_torque"
    POS_X_SELF_TEST = "self_test_pos_x"
    NEG_X_SELF_TEST = "self_test_neg_x"
    POS_Y_SELF_TEST = "self_test_pos_y"
    NEG_Y_SELF_TEST = "self_test_neg_y"
    POS_Z_SELF_TEST = "self_test_pos_z"
    NEG_Z_SELF_TEST = "self_test_neg_z"


class ImtqSetId(enum.IntEnum):
    ENG_HK_NO_TORQUE = 1
    RAW_MTM_NO_TORQUE = 2
    ENG_HK_SET_WITH_TORQUE = 3
    RAW_MTM_WITH_TORQUE = 4
    STATUS_SET = 5
    DIPOLES = 6

    CAL_MTM_SET = 9
    POSITIVE_X_TEST = 10
    NEGATIVE_X_TEST = 11
    POSITIVE_Y_TEST = 12
    NEGATIVE_Y_TEST = 13
    POSITIVE_Z_TEST = 14
    NEGATIVE_Z_TEST = 15


class ImtqActionId:
    start_actuation_dipole = bytearray([0x0, 0x0, 0x0, 0x02])
    get_commanded_dipole = bytearray([0x0, 0x0, 0x0, 0x03])
    perform_positive_x_test = bytearray([0x0, 0x0, 0x0, 0x07])
    perform_negative_x_test = bytearray([0x0, 0x0, 0x0, 0x08])
    perform_positive_y_test = bytearray([0x0, 0x0, 0x0, 0x09])
    perform_negative_y_test = bytearray([0x0, 0x0, 0x0, 0x0A])
    perform_positive_z_test = bytearray([0x0, 0x0, 0x0, 0x0B])
    perform_negative_z_test = bytearray([0x0, 0x0, 0x0, 0x0C])
    # Initiates the reading of the last performed self test. After sending this command the results
    # can be downlinked via the housekeeping service by using the appropriate set ids listed above.
    read_self_test_results = bytearray([0x0, 0x0, 0x0, 0x0D])


CTN = CmdTreeNode


def create_mgt_node() -> CmdTreeNode:
    node = CmdTreeNode(
        "mgt", "iMTQ MGT device node", hide_children_which_are_leaves=True
    )
    node.add_child(CmdTreeNode(OpCode.OFF, "Mode OFF"))
    node.add_child(CmdTreeNode(OpCode.ON, "Mode ON"))
    node.add_child(CmdTreeNode(OpCode.NORMAL, "Mode Normal"))
    node.add_child(CmdTreeNode(OpCode.OFF, "Mode OFF"))
    node.add_child(
        CmdTreeNode(OpCode.REQUEST_ENG_HK_NO_TORQUE, "Request Engineering HK One Shot")
    )
    node.add_child(
        CmdTreeNode(
            OpCode.REQUEST_ENG_HK_WITH_TORQUE,
            "Request Engineering HK One Shot during Torque",
        )
    )
    node.add_child(
        CmdTreeNode(OpCode.ENABLE_ENG_HK_NO_TORQUE, "Enable ENG HK not torque")
    )
    node.add_child(
        CmdTreeNode(OpCode.ENABLE_ENG_HK_WITH_TORQUE, "Enable ENG HK with torque")
    )
    node.add_child(
        CmdTreeNode(OpCode.DISABLE_ENG_HK_NO_TORQUE, "Disable ENG HK not torque")
    )
    node.add_child(
        CmdTreeNode(OpCode.DISABLE_ENG_HK_WITH_TORQUE, "Disable ENG HK with torque")
    )
    node.add_child(
        CmdTreeNode(
            OpCode.REQUEST_MGM_RAW_NO_TORQUE,
            "Request MGM Raw Without Torque HK One Shot",
        )
    )
    node.add_child(
        CmdTreeNode(OpCode.ENABLE_MGM_RAW_NO_TORQUE, "Enable MGM Raw Without Torque HK")
    )
    node.add_child(
        CmdTreeNode(
            OpCode.DISABLE_MGM_RAW_NO_TORQUE, "Disable MGM Raw Without Torque HK"
        )
    )
    node.add_child(
        CmdTreeNode(
            OpCode.REQUEST_MGM_RAW_WITH_TORQUE,
            "Request MGM Raw With Torque HK One Shot",
        )
    )
    node.add_child(
        CmdTreeNode(OpCode.ENABLE_MGM_RAW_WITH_TORQUE, "Enable MGM Raw With Torque HK")
    )
    node.add_child(
        CTN(OpCode.DISABLE_MGM_RAW_WITH_TORQUE, "Disable MGM Raw With Torque HK")
    )
    node.add_child(CTN(OpCode.POS_X_SELF_TEST, "IMTQ perform pos X self test"))
    node.add_child(CTN(OpCode.NEG_X_SELF_TEST, "IMTQ perform neg X self test"))
    node.add_child(CTN(OpCode.POS_Y_SELF_TEST, "IMTQ perform pos Y self test"))
    node.add_child(CTN(OpCode.NEG_Y_SELF_TEST, "IMTQ perform neg Y self test"))
    node.add_child(CTN(OpCode.POS_Z_SELF_TEST, "IMTQ perform pos Z self test"))
    node.add_child(CTN(OpCode.NEG_Z_SELF_TEST, "IMTQ perform neg Z self test"))
    node.add_child(CTN(OpCode.SET_DIPOLE, "IMTQ command dipole"))
    return node


@tmtc_definitions_provider
def add_imtq_cmds(defs: TmtcDefinitionWrapper):
    oce = OpCodeEntry()
    oce.add(OpCode.OFF, "Mode Off")
    oce.add(OpCode.ON, "Mode On")
    oce.add(OpCode.NORMAL, "Mode Normal")
    oce.add(OpCode.REQUEST_ENG_HK_NO_TORQUE, "Request Engineering HK One Shot")
    oce.add(
        OpCode.REQUEST_ENG_HK_WITH_TORQUE,
        "Request Engineering HK One Shot during Torque",
    )
    oce.add(OpCode.ENABLE_ENG_HK_NO_TORQUE, "Enable ENG HK not torque")
    oce.add(OpCode.ENABLE_ENG_HK_WITH_TORQUE, "Enable ENG HK with torque")
    oce.add(OpCode.DISABLE_ENG_HK_NO_TORQUE, "Disable ENG HK not torque")
    oce.add(OpCode.DISABLE_ENG_HK_WITH_TORQUE, "Disable ENG HK with torque")
    oce.add(
        OpCode.REQUEST_MGM_RAW_NO_TORQUE, "Request MGM Raw Without Torque HK One Shot"
    )
    oce.add(OpCode.ENABLE_MGM_RAW_NO_TORQUE, "Enable MGM Raw Without Torque HK")
    oce.add(OpCode.DISABLE_MGM_RAW_NO_TORQUE, "Disable MGM Raw Without Torque HK")
    oce.add(
        OpCode.REQUEST_MGM_RAW_WITH_TORQUE, "Request MGM Raw With Torque HK One Shot"
    )
    oce.add(OpCode.ENABLE_MGM_RAW_WITH_TORQUE, "Enable MGM Raw With Torque HK")
    oce.add(OpCode.DISABLE_MGM_RAW_WITH_TORQUE, "Disable MGM Raw With Torque HK")
    oce.add(OpCode.POS_X_SELF_TEST, "IMTQ perform pos X self test")
    oce.add(OpCode.NEG_X_SELF_TEST, "IMTQ perform neg X self test")
    oce.add(OpCode.POS_Y_SELF_TEST, "IMTQ perform pos Y self test")
    oce.add(OpCode.NEG_Y_SELF_TEST, "IMTQ perform neg Y self test")
    oce.add(OpCode.POS_Z_SELF_TEST, "IMTQ perform pos Z self test")
    oce.add(OpCode.NEG_Z_SELF_TEST, "IMTQ perform neg Z self test")
    oce.add(OpCode.SET_DIPOLE, "IMTQ command dipole")
    oce.add("10", "IMTQ get commanded dipole")
    oce.add("12", "IMTQ get calibrated MTM measurement one shot")
    defs.add_service(CustomServiceList.IMTQ.value, "IMQT Device", oce)


def create_imtq_command(  # noqa C901
    object_id: ObjectIdU32, q: DefaultPusQueueHelper, cmd_str: str
):
    q.add_log_cmd(
        f"Testing ISIS IMTQ handler with object id: {object_id.as_hex_string}"
    )

    if cmd_str == OpCode.OFF:
        q.add_log_cmd("IMTQ: Set mode off")
        command = pack_mode_data(object_id.as_bytes, Mode.OFF, 0)
        q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
    if cmd_str == OpCode.ON:
        q.add_log_cmd("IMTQ: Set mode on")
        command = pack_mode_data(object_id.as_bytes, Mode.ON, 0)
        q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
    if cmd_str == OpCode.NORMAL:
        q.add_log_cmd("IMTQ: Mode Normal")
        command = pack_mode_data(object_id.as_bytes, Mode.NORMAL, 0)
        q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
    if cmd_str == OpCode.POS_X_SELF_TEST:
        q.add_log_cmd("IMTQ: Perform positive x self test")
        command = object_id.as_bytes + ImtqActionId.perform_positive_x_test
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))

        q.add_log_cmd("IMTQ: Initiate reading of positive x self test results")
        command = object_id.as_bytes + ImtqActionId.read_self_test_results
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))

        q.add_log_cmd("IMTQ: Request dataset with positive x self test results")
        sid = make_sid(object_id.as_bytes, ImtqSetId.POSITIVE_X_TEST)
        q.add_pus_tc(generate_one_hk_command(sid))

    if cmd_str == OpCode.NEG_X_SELF_TEST:
        q.add_log_cmd("IMTQ: Perform negative x self test")
        command = object_id.as_bytes + ImtqActionId.perform_negative_x_test
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
        q.add_log_cmd("IMTQ: Initiate reading of negative x self test results")
        command = object_id.as_bytes + ImtqActionId.read_self_test_results
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
        q.add_log_cmd("IMTQ: Request dataset with negative x self test results")
        sid = make_sid(object_id.as_bytes, ImtqSetId.NEGATIVE_X_TEST)
        q.add_pus_tc(generate_one_hk_command(sid))

    if cmd_str == OpCode.POS_Y_SELF_TEST:
        q.add_log_cmd("IMTQ: Perform positive y self test")
        command = object_id.as_bytes + ImtqActionId.perform_positive_y_test
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
        q.add_log_cmd("IMTQ: Initiate reading of positive y self test results")
        command = object_id.as_bytes + ImtqActionId.read_self_test_results
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))

        q.add_log_cmd("IMTQ: Request dataset with positive y self test results")
        sid = make_sid(object_id.as_bytes, ImtqSetId.POSITIVE_Y_TEST)
        q.add_pus_tc(generate_one_hk_command(sid))

    if cmd_str == OpCode.NEG_Y_SELF_TEST:
        q.add_log_cmd("IMTQ: Perform negative y self test")
        command = object_id.as_bytes + ImtqActionId.perform_negative_y_test
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))

        q.add_log_cmd("IMTQ: Initiate reading of negative y self test results")
        command = object_id.as_bytes + ImtqActionId.read_self_test_results
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))

        q.add_log_cmd("IMTQ: Request dataset with negative y self test results")
        sid = make_sid(object_id.as_bytes, ImtqSetId.NEGATIVE_Y_TEST)
        q.add_pus_tc(generate_one_hk_command(sid))

    if cmd_str == OpCode.POS_Z_SELF_TEST:
        q.add_log_cmd("IMTQ: Perform positive z self test")
        command = object_id.as_bytes + ImtqActionId.perform_positive_z_test
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))

        q.add_log_cmd("IMTQ: Initiate reading of positive z self test results")
        command = object_id.as_bytes + ImtqActionId.read_self_test_results
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))

        q.add_log_cmd("IMTQ: Request dataset with positive z self test results")
        sid = make_sid(object_id.as_bytes, ImtqSetId.POSITIVE_Y_TEST)
        q.add_pus_tc(generate_one_hk_command(sid))

    if cmd_str == OpCode.NEG_Z_SELF_TEST:
        q.add_log_cmd("IMTQ: Perform negative z self test")
        command = object_id.as_bytes + ImtqActionId.perform_negative_z_test
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
        q.add_log_cmd("IMTQ: Initiate reading of negative z self test results")
        command = object_id.as_bytes + ImtqActionId.read_self_test_results
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
        q.add_log_cmd("IMTQ: Request dataset with negative z self test results")
        sid = make_sid(object_id.as_bytes, ImtqSetId.NEGATIVE_Z_TEST)
        q.add_pus_tc(generate_one_hk_command(sid))

    if cmd_str in OpCode.SET_DIPOLE:
        x_dipole = int(input("Specify X dipole [range [0, 2000] * 10^-4 * Am^2]: "))
        y_dipole = int(input("Specify Y dipole [range [0, 2000] * 10^-4 * Am^2]: "))
        z_dipole = int(input("Specify Z dipole [range [0, 2000] * 10^-4 * Am^2]: "))
        duration = int(
            input(
                f"Specify torque duration [range [0, {pow(2, 16) - 1}, "
                "0 for continuous generation until update]: "
            )
        )
        dur_str = "infinite" if duration == 0 else str(duration)
        q.add_log_cmd(
            f"IMTQ: Commanding dipole X={x_dipole}, Y={y_dipole}, Z={y_dipole},"
            f" duration={dur_str}"
        )
        q.add_pus_tc(
            pack_dipole_command(
                object_id.as_bytes, x_dipole, y_dipole, z_dipole, duration
            )
        )

    if cmd_str == "10":  # doesnt seem to work anymore
        q.add_log_cmd("IMTQ: Get commanded dipole")
        command = object_id.as_bytes + ImtqActionId.get_commanded_dipole
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))

    if cmd_str == OpCode.ENABLE_ENG_HK_NO_TORQUE:
        q.add_log_cmd("IMTQ: Enable ENG HK")
        interval = float(input("Please enter collection interval in seconds: "))
        cmds = create_enable_periodic_hk_command_with_interval_with_diag(
            diag=True,
            sid=make_sid(object_id.as_bytes, ImtqSetId.ENG_HK_NO_TORQUE),
            interval_seconds=interval,
        )
        for cmd in cmds:
            q.add_pus_tc(cmd)
    if cmd_str == OpCode.DISABLE_ENG_HK_NO_TORQUE:
        q.add_log_cmd("IMTQ: Disable ENG HK (No Torque)")
        q.add_pus_tc(
            create_disable_periodic_hk_command_with_diag(
                True, make_sid(object_id.as_bytes, ImtqSetId.ENG_HK_NO_TORQUE)
            )
        )
    if cmd_str == OpCode.REQUEST_ENG_HK_WITH_TORQUE:
        q.add_log_cmd("IMTQ: Get engineering hk set with torque")
        q.add_pus_tc(
            create_request_one_diag_command(
                sid=make_sid(
                    object_id=object_id.as_bytes,
                    set_id=ImtqSetId.ENG_HK_SET_WITH_TORQUE,
                )
            )
        )
    if cmd_str == OpCode.ENABLE_ENG_HK_WITH_TORQUE:
        q.add_log_cmd("IMTQ: Enable ENG HK with torque")
        interval = float(input("Please enter collection interval in seconds: "))
        cmds = create_enable_periodic_hk_command_with_interval_with_diag(
            diag=True,
            sid=make_sid(object_id.as_bytes, ImtqSetId.ENG_HK_SET_WITH_TORQUE),
            interval_seconds=interval,
        )
        for cmd in cmds:
            q.add_pus_tc(cmd)
    if cmd_str == OpCode.DISABLE_ENG_HK_WITH_TORQUE:
        q.add_log_cmd("IMTQ: Disable ENG HK with Torque")
        q.add_pus_tc(
            create_disable_periodic_hk_command_with_diag(
                True, make_sid(object_id.as_bytes, ImtqSetId.ENG_HK_SET_WITH_TORQUE)
            )
        )
    if cmd_str == OpCode.REQUEST_ENG_HK_NO_TORQUE:
        q.add_log_cmd("IMTQ: Get engineering hk set (no torque)")
        q.add_pus_tc(
            generate_one_diag_command(
                sid=make_sid(
                    object_id=object_id.as_bytes, set_id=ImtqSetId.ENG_HK_NO_TORQUE
                )
            )
        )

    if cmd_str == "12":
        q.add_log_cmd("IMTQ: Get calibrated MTM hk set")
        q.add_pus_tc(
            create_request_one_diag_command(
                sid=make_sid(object_id=object_id.as_bytes, set_id=ImtqSetId.CAL_MTM_SET)
            )
        )

    if cmd_str == OpCode.REQUEST_MGM_RAW_NO_TORQUE:
        q.add_log_cmd("IMTQ: Get raw MTM hk set")
        q.add_pus_tc(
            create_request_one_diag_command(
                sid=make_sid(
                    object_id=object_id.as_bytes, set_id=ImtqSetId.RAW_MTM_NO_TORQUE
                )
            )
        )

    if cmd_str == OpCode.DISABLE_MGM_RAW_NO_TORQUE:
        q.add_log_cmd("IMTQ: Disable MGM RAW HK (No Torque)")
        q.add_pus_tc(
            create_disable_periodic_hk_command_with_diag(
                True, make_sid(object_id.as_bytes, ImtqSetId.RAW_MTM_NO_TORQUE)
            )
        )
    if cmd_str == OpCode.ENABLE_MGM_RAW_NO_TORQUE:
        q.add_log_cmd("IMTQ: Enable MGM RAW HK (No Torque)")
        interval = float(input("Please enter collection interval in seconds: "))
        cmds = create_enable_periodic_hk_command_with_interval_with_diag(
            diag=True,
            sid=make_sid(object_id.as_bytes, ImtqSetId.RAW_MTM_NO_TORQUE),
            interval_seconds=interval,
        )
        for cmd in cmds:
            q.add_pus_tc(cmd)
    if cmd_str == OpCode.REQUEST_MGM_RAW_WITH_TORQUE:
        q.add_log_cmd("IMTQ: Get raw MTM hk set")
        q.add_pus_tc(
            create_request_one_diag_command(
                sid=make_sid(
                    object_id=object_id.as_bytes, set_id=ImtqSetId.RAW_MTM_WITH_TORQUE
                )
            )
        )
    if cmd_str == OpCode.ENABLE_MGM_RAW_WITH_TORQUE:
        q.add_log_cmd("IMTQ: Enable MGM RAW HK (No Torque)")
        interval = float(input("Please enter collection interval in seconds: "))
        cmds = create_enable_periodic_hk_command_with_interval_with_diag(
            diag=True,
            sid=make_sid(object_id.as_bytes, ImtqSetId.RAW_MTM_WITH_TORQUE),
            interval_seconds=interval,
        )
        for cmd in cmds:
            q.add_pus_tc(cmd)
    if cmd_str == OpCode.DISABLE_MGM_RAW_WITH_TORQUE:
        q.add_log_cmd("IMTQ: Disable MGM RAW HK (No Torque)")
        q.add_pus_tc(
            create_disable_periodic_hk_command_with_diag(
                True, make_sid(object_id.as_bytes, ImtqSetId.RAW_MTM_WITH_TORQUE)
            )
        )


def pack_dipole_command(
    object_id: bytes, x_dipole: int, y_dipole: int, z_dipole: int, duration: int
) -> PusTelecommand:
    """This function packs the command causing the ISIS IMTQ to generate a dipole.
    :param object_id:    The object id of the IMTQ handler.
    :param x_dipole:  The dipole of the x coil in 10^-4*Am^2 (max. 2000)
    :param y_dipole:   The dipole of the y coil in 10^-4*Am^2 (max. 2000)
    :param z_dipole:   The dipole of the z coil in 10^-4*Am^2 (max. 2000)
    :param duration:  The duration in milliseconds the dipole will be generated by the coils.
                     When set to 0, the dipole will be generated until a new dipole actuation
                     command is sent.
    """
    action_id = ImtqActionId.start_actuation_dipole
    command = object_id + action_id
    x_dipole = int(round(x_dipole))
    y_dipole = int(round(y_dipole))
    z_dipole = int(round(z_dipole))
    if x_dipole < -2000 or x_dipole > 2000:
        raise_dipole_error("X dipole", x_dipole)
    if y_dipole < -2000 or y_dipole > 2000:
        raise_dipole_error("Y dipole", y_dipole)
    if z_dipole < -2000 or z_dipole > 2000:
        raise_dipole_error("Z dipole", z_dipole)
    duration = int(round(duration))
    if duration < 0 or duration > pow(2, 16) - 1:
        raise ValueError(
            f"Duration in ms of {duration} smaller than 0 or larger than allowed"
            f" {pow(2, 16) - 1}"
        )
    command += struct.pack("!h", x_dipole)
    command += struct.pack("!h", y_dipole)
    command += struct.pack("!h", z_dipole)
    command += struct.pack("!H", duration)
    command = PusTelecommand(service=8, subservice=128, app_data=command)
    return command


def raise_dipole_error(dipole_str: str, value: int):
    raise ValueError(
        f"{dipole_str} {value} negative or larger than maximum allowed 2000 *"
        " 10^-4*Am^2"
    )


STATUS_HEADERS = [
    "Status Byte Mode",
    "Status Byte Error",
    "Status Byte Config",
    "Status Byte Uptime [s]",
]


ENG_HK_HEADERS = [
    "Digital Voltage [mV]",
    "Analog Voltage [mV]",
    "Digital Current [mA]",
    "Analog Current [mA]",
    "Coil Current X [mA]",
    "Coil Current Y [mA]",
    "Coil Current Z [mA]",
    "Coil X Temperature [°C]",
    "Coil Y Temperature [°C]",
    "Coil Z Temperature [°C]",
    "MCU Temperature [°C]",
]


def handle_imtq_hk(pw: PrintWrapper, hk_data: bytes, set_id: int):
    if (set_id >= ImtqSetId.POSITIVE_X_TEST) and (set_id <= ImtqSetId.NEGATIVE_Z_TEST):
        return handle_self_test_data(pw, hk_data)
    elif set_id == ImtqSetId.ENG_HK_NO_TORQUE:
        return handle_eng_set(pw, hk_data, False)
    elif set_id == ImtqSetId.ENG_HK_SET_WITH_TORQUE:
        return handle_eng_set(pw, hk_data, True)
    elif set_id == ImtqSetId.CAL_MTM_SET:
        return handle_calibrated_mtm_measurement(pw, hk_data)
    elif set_id == ImtqSetId.RAW_MTM_NO_TORQUE:
        return handle_raw_mtm_measurement(pw, hk_data, False)
    elif set_id == ImtqSetId.RAW_MTM_WITH_TORQUE:
        return handle_raw_mtm_measurement(pw, hk_data, True)
    elif set_id == ImtqSetId.DIPOLES:
        return handle_dipole_set(pw, hk_data)
    elif set_id == ImtqSetId.STATUS_SET:
        return handle_status_set(pw, hk_data)
    else:
        pw.wlog(
            _LOGGER,
            f"IMTQ handler HK reply with unknown or unimplemented set id {set_id}",
        )


def unpack_status_set(hk_data: bytes) -> List:
    status_mode = hk_data[0]
    status_error = hk_data[1]
    status_conf = hk_data[2]
    status_uptime = struct.unpack("!I", hk_data[3:7])[0]
    return [status_mode, status_error, status_conf, status_uptime]


def handle_dipole_set(pw: PrintWrapper, hk_data: bytes):
    pw.dlog("Received iMTQ dipole set")
    fmt_str = "!hhhH"
    fmt_len = struct.calcsize(fmt_str)
    (dipole_x, dipole_y, dipole_z, current_torque_duration) = struct.unpack(
        fmt_str, hk_data[0:8]
    )
    pw.dlog(f"Dipole X: {dipole_x}")
    pw.dlog(f"Dipole Y: {dipole_y}")
    pw.dlog(f"Dipole Z: {dipole_z}")
    pw.dlog(f"Current torque duration: {current_torque_duration}")
    FsfwTmTcPrinter.get_validity_buffer(hk_data[fmt_len:], 2)


def unpack_eng_hk(hk_data: bytes) -> List:
    digital_voltage = struct.unpack("!H", hk_data[0:2])[0]
    analog_voltage = struct.unpack("!H", hk_data[2:4])[0]
    digital_current = struct.unpack("!f", hk_data[4:8])[0]
    analog_current = struct.unpack("!f", hk_data[8:12])[0]
    coil_x_current = struct.unpack("!f", hk_data[12:16])[0]
    coil_y_current = struct.unpack("!f", hk_data[16:20])[0]
    coil_z_current = struct.unpack("!f", hk_data[20:24])[0]
    coil_x_temperature = struct.unpack("!h", hk_data[24:26])[0]
    coil_y_temperature = struct.unpack("!h", hk_data[26:28])[0]
    coil_z_temperature = struct.unpack("!h", hk_data[28:30])[0]
    mcu_temperature = struct.unpack("!h", hk_data[30:32])[0]
    content_list = [
        digital_voltage,
        analog_voltage,
        digital_current,
        analog_current,
        coil_x_current,
        coil_y_current,
        coil_z_current,
        coil_x_temperature,
        coil_y_temperature,
        coil_z_temperature,
        mcu_temperature,
    ]
    return content_list


def handle_eng_set(pw: PrintWrapper, hk_data: bytes, torque_on: bool):
    pw.dlog(f"Found engineering HK. Torque Status: {torque_on}")
    content_list = unpack_eng_hk(hk_data)
    validity_buffer = hk_data[32:]

    num_of_vars = len(ENG_HK_HEADERS)
    for k, v in zip(ENG_HK_HEADERS, content_list):
        pw.dlog(f"{k.ljust(30)}: {v}")
    pw.dlog(
        FsfwTmTcPrinter.get_validity_buffer(
            validity_buffer=validity_buffer, num_vars=num_of_vars
        )
    )


def handle_status_set(pw: PrintWrapper, hk_data: bytes):
    content_list = unpack_status_set(hk_data)
    validity_buffer = hk_data[7:]

    num_of_vars = 4
    for k, v in zip(STATUS_HEADERS, content_list):
        pw.dlog(f"{k.ljust(30)}: {v}")
    pw.dlog(
        FsfwTmTcPrinter.get_validity_buffer(
            validity_buffer=validity_buffer, num_vars=num_of_vars
        )
    )


def handle_calibrated_mtm_measurement(pw: PrintWrapper, hk_data: bytes):
    header_list = [
        "Calibrated MTM X [nT]",
        "Calibrated MTM Y [nT]",
        "Calibrated MTM Z [nT]",
        "Coil actuation status",
    ]
    mtm_x = struct.unpack("!I", hk_data[0:4])[0]
    mtm_y = struct.unpack("!I", hk_data[4:8])[0]
    mtm_z = struct.unpack("!I", hk_data[8:12])[0]
    coil_actuation_status = hk_data[12]
    validity_buffer = hk_data[12:]
    content_list = [mtm_x, mtm_y, mtm_z, coil_actuation_status]
    num_of_vars = len(header_list)
    pw.dlog(str(header_list))
    pw.dlog(str(content_list))
    pw.dlog(
        FsfwTmTcPrinter.get_validity_buffer(
            validity_buffer=validity_buffer, num_vars=num_of_vars
        )
    )


def handle_raw_mtm_measurement(pw: PrintWrapper, hk_data: bytes, torque_status: bool):
    pw.dlog(f"Found raw MTM measurement. Torque Status: {torque_status}")
    header_list = [
        "Raw MTM X [nT]",
        "Raw MTM Y [nT]",
        "Raw MTM Z [nT]",
        "Coil actuation status",
    ]
    mtm_x = struct.unpack("!f", hk_data[0:4])[0]
    mtm_y = struct.unpack("!f", hk_data[4:8])[0]
    mtm_z = struct.unpack("!f", hk_data[8:12])[0]
    coil_actuation_status = hk_data[12]
    validity_buffer = hk_data[13:]
    content_list = [mtm_x, mtm_y, mtm_z, coil_actuation_status]
    num_of_vars = 2
    pw.dlog(str(header_list))
    pw.dlog(str(content_list))
    pw.dlog(
        FsfwTmTcPrinter.get_validity_buffer(
            validity_buffer=validity_buffer, num_vars=num_of_vars
        )
    )


def handle_self_test_data(pw: PrintWrapper, hk_data: bytes):
    header_list = [
        "Init Err",
        "Init Raw Mag X [nT]",
        "Init Raw Mag Y [nT]",
        "Init Raw Mag Z [nT]",
        "Init Cal Mag X [nT]",
        "Init Cal Mag Y [nT]",
        "Init Cal Mag Z [nT]",
        "Init Coil X Current [mA]",
        "Init Coil Y Current [mA]",
        "Init Coil Z Current [mA]",
        "Init Coil X Temperature [°C]",
        "Init Coil Y Temperature [°C]",
        "Init Coil Z Temperature [°C]",
        "Err",
        "Raw Mag X [nT]",
        "Raw Mag Y [nT]",
        "Raw Mag Z [nT]",
        "Cal Mag X [nT]",
        "Cal Mag Y [nT]",
        "Cal Mag Z [nT]",
        "Coil X Current [mA]",
        "Coil Y Current [mA]",
        "Coil Z Current [mA]",
        "Coil X Temperature [°C]",
        "Coil Y Temperature [°C]",
        "Coil Z Temperature [°C]",
        "Fina Err",
        "Fina Raw Mag X [nT]",
        "Fina Raw Mag Y [nT]",
        "Fina Raw Mag Z [nT]",
        "Fina Cal Mag X [nT]",
        "Fina Cal Mag Y [nT]",
        "Fina Cal Mag Z [nT]",
        "Fina Coil X Current [mA]",
        "Fina Coil Y Current [mA]",
        "Fina Coil Z Current [mA]",
        "Fina Coil X Temperature [°C]",
        "Fina Coil Y Temperature [°C]",
        "Fina Coil Z Temperature [°C]",
    ]
    # INIT step (no coil actuation)
    init_err = hk_data[0]
    init_raw_mag_x = struct.unpack("!f", hk_data[1:5])[0]
    init_raw_mag_y = struct.unpack("!f", hk_data[5:9])[0]
    init_raw_mag_z = struct.unpack("!f", hk_data[9:13])[0]
    init_cal_mag_x = struct.unpack("!f", hk_data[13:17])[0]
    init_cal_mag_y = struct.unpack("!f", hk_data[17:21])[0]
    init_cal_mag_z = struct.unpack("!f", hk_data[21:25])[0]
    init_coil_x_current = struct.unpack("!f", hk_data[25:29])[0]
    init_coil_y_current = struct.unpack("!f", hk_data[29:33])[0]
    init_coil_z_current = struct.unpack("!f", hk_data[33:37])[0]
    init_coil_x_temperature = struct.unpack("!H", hk_data[37:39])[0]
    init_coil_y_temperature = struct.unpack("!H", hk_data[39:41])[0]
    init_coil_z_temperature = struct.unpack("!H", hk_data[41:43])[0]

    # Actuation step
    err = hk_data[43]
    raw_mag_x = struct.unpack("!f", hk_data[44:48])[0]
    raw_mag_y = struct.unpack("!f", hk_data[48:52])[0]
    raw_mag_z = struct.unpack("!f", hk_data[52:56])[0]
    cal_mag_x = struct.unpack("!f", hk_data[56:60])[0]
    cal_mag_y = struct.unpack("!f", hk_data[60:64])[0]
    cal_mag_z = struct.unpack("!f", hk_data[64:68])[0]
    coil_x_current = struct.unpack("!f", hk_data[68:72])[0]
    coil_y_current = struct.unpack("!f", hk_data[72:76])[0]
    coil_z_current = struct.unpack("!f", hk_data[76:80])[0]
    coil_x_temperature = struct.unpack("!H", hk_data[80:82])[0]
    coil_y_temperature = struct.unpack("!H", hk_data[82:84])[0]
    coil_z_temperature = struct.unpack("!H", hk_data[84:86])[0]

    # FINA step (no coil actuation)
    fina_err = hk_data[86]
    fina_raw_mag_x = struct.unpack("!f", hk_data[87:91])[0]
    fina_raw_mag_y = struct.unpack("!f", hk_data[91:95])[0]
    fina_raw_mag_z = struct.unpack("!f", hk_data[95:99])[0]
    fina_cal_mag_x = struct.unpack("!f", hk_data[99:103])[0]
    fina_cal_mag_y = struct.unpack("!f", hk_data[103:107])[0]
    fina_cal_mag_z = struct.unpack("!f", hk_data[107:111])[0]
    fina_coil_x_current = struct.unpack("!f", hk_data[111:115])[0]
    fina_coil_y_current = struct.unpack("!f", hk_data[115:119])[0]
    fina_coil_z_current = struct.unpack("!f", hk_data[119:123])[0]
    fina_coil_x_temperature = struct.unpack("!H", hk_data[123:125])[0]
    fina_coil_y_temperature = struct.unpack("!H", hk_data[125:127])[0]
    fina_coil_z_temperature = struct.unpack("!H", hk_data[127:129])[0]

    validity_buffer = hk_data[129:]
    content_list = [
        init_err,
        init_raw_mag_x,
        init_raw_mag_y,
        init_raw_mag_z,
        init_cal_mag_x,
        init_cal_mag_y,
        init_cal_mag_z,
        init_coil_x_current,
        init_coil_y_current,
        init_coil_z_current,
        init_coil_x_temperature,
        init_coil_y_temperature,
        init_coil_z_temperature,
        err,
        raw_mag_x,
        raw_mag_y,
        raw_mag_z,
        cal_mag_x,
        cal_mag_y,
        cal_mag_z,
        coil_x_current,
        coil_y_current,
        coil_z_current,
        coil_x_temperature,
        coil_y_temperature,
        coil_z_temperature,
        fina_err,
        fina_raw_mag_x,
        fina_raw_mag_y,
        fina_raw_mag_z,
        fina_cal_mag_x,
        fina_cal_mag_y,
        fina_cal_mag_z,
        fina_coil_x_current,
        fina_coil_y_current,
        fina_coil_z_current,
        fina_coil_x_temperature,
        fina_coil_y_temperature,
        fina_coil_z_temperature,
    ]
    num_of_vars = len(header_list)
    pw.dlog(str(header_list))
    pw.dlog(str(content_list))
    pw.dlog(
        FsfwTmTcPrinter.get_validity_buffer(
            validity_buffer=validity_buffer, num_vars=num_of_vars
        )
    )