eive-tmtc/eive_tmtc/tmtc/com/syrlinks_handler.py

# -*- coding: utf-8 -*-
"""
@file   syrlinks_hk_handler.py
@brief  Syrlinks Hk Handler tests
@author J. Meier
@date   13.12.2020
"""
import enum
import logging
import math

from eive_tmtc.pus_tm.defs import PrintWrapper
from eive_tmtc.tmtc.com.subsystem import Submode
from eive_tmtc.config.definitions import CustomServiceList
from tmtccmd.config.tmtc import (
    tmtc_definitions_provider,
    TmtcDefinitionWrapper,
    OpCodeEntry,
)
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import (
    make_sid,
    create_request_one_diag_command,
    create_enable_periodic_hk_command_with_interval,
    create_disable_periodic_hk_command,
    create_request_one_hk_command,
)
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc.pus_200_fsfw_mode import Mode, create_mode_command
from eive_tmtc.config.object_ids import SYRLINKS_HANDLER_ID
import struct

from tmtccmd.util import ObjectIdU32
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter


class SetId(enum.IntEnum):
    RX_REGISTERS_DATASET = 1
    TX_REGISTERS_DATASET = 2
    TEMPERATURE_SET_ID = 3


class OpCode:
    OFF = "off"
    ON = "on"
    NORMAL_RX_ONLY = "nml_rx_only"
    NORMAL_RX_AND_TX_CW = "nml_carrier_wave"
    NORMAL_RX_AND_TX_DEF_DATARATE = "nml_default_datarate"
    NORMAL_RX_AND_TX_LOW_DATARATE = "nml_low_datarate"
    NORMAL_RX_AND_TX_HIGH_DATARATE = "nml_high_datarate"
    HK_RX_REGS = "hk_rx_regs"
    HK_TEMPS = "hk_temps"
    ENABLE_HK_RX_REGS = "enable_hk_rx"
    DISABLE_HK_RX_REGS = "disable_hk_rx"
    ENABLE_HK_TX_REGS = "enable_hk_tx"
    DISABLE_HK_TX_REGS = "disable_hk_tx"
    HK_TX_REGS = "hk_tx_regs"
    TX_STATUS = "tx_status"
    RX_STATUS = "rx_status"


class Info:
    OFF = "Switch OFF"
    ON = "Switch ON"
    NORMAL_RX_ONLY = "NORMAL RX Only, set TX to standby"
    NORMAL_RX_AND_TX_CW = "NORMAL RX and TX, TX Carrier Wave"
    NORMAL_RX_AND_TX_DEF_DATARATE = "NORMAL RX and TX, TX with default datarate"
    NORMAL_RX_AND_TX_LOW_DATARATE = "NORMAL RX and TX, TX with low datarate"
    NORMAL_RX_AND_TX_HIGH_DATARATE = "NORMAL RX and TX, TX with high datarate"
    HK_RX_REGS = "Request RX register set"
    HK_TX_REGS = "Request TX register set"
    HK_TEMPS = "Request Temperatures HK"
    ENABLE_HK_RX_REGS = "Enable periodic RX register HK"
    DISABLE_HK_RX_REGS = "Disable periodic RX register HK"
    ENABLE_HK_TX_REGS = "Enable periodic TX register HK"
    DISABLE_HK_TX_REGS = "Disable periodic TX register HK"
    TX_STATUS = "Read TX status (always read in normal mode)"
    RX_STATUS = "Read RX status (always read in normal mode)"
    SET_CW = "Set TX carrier wave"


class CommandId(enum.IntEnum):
    READ_RX_STATUS_REGISTERS = 2
    SET_TX_MODE_STANDBY = 3
    SET_TX_MODE_MODULATION = 4
    SET_TX_MODE_CW = 5
    READ_TX_STATUS = 7
    READ_TX_WAVEFORM = 8
    READ_TX_AGC_VALUE_HIGH_BYTE = 9
    READ_TX_AGC_VALUE_LOW_BYTE = 10
    WRITE_LCL_CONFIG = 11
    READ_LCL_CONFIG_REGISTER = 12
    SET_WAVEFORM_OQPSK = 17
    SET_WAVEFORM_BPSK = 18
    SET_SECOND_CONFIG = 19
    ENABLE_DEBUG = 20
    DISABLE_DEBUG = 21


class Datarate(enum.IntEnum):
    LOW_RATE_MODULATION_BPSK = 0
    HIGH_RATE_MODULATION_0QPSK = 1


@tmtc_definitions_provider
def add_syrlinks_cmds(defs: TmtcDefinitionWrapper):
    oce = OpCodeEntry()
    oce.add(OpCode.OFF, Info.OFF)
    oce.add(OpCode.ON, Info.ON)
    oce.add(OpCode.NORMAL_RX_ONLY, Info.NORMAL_RX_ONLY)
    oce.add(OpCode.NORMAL_RX_AND_TX_CW, Info.NORMAL_RX_AND_TX_CW)
    oce.add(OpCode.NORMAL_RX_AND_TX_DEF_DATARATE, Info.NORMAL_RX_AND_TX_DEF_DATARATE)
    oce.add(OpCode.NORMAL_RX_AND_TX_LOW_DATARATE, Info.NORMAL_RX_AND_TX_LOW_DATARATE)
    oce.add(OpCode.NORMAL_RX_AND_TX_HIGH_DATARATE, Info.NORMAL_RX_AND_TX_HIGH_DATARATE)
    oce.add(OpCode.HK_RX_REGS, Info.HK_RX_REGS)
    oce.add(OpCode.HK_TX_REGS, Info.HK_TX_REGS)
    oce.add(OpCode.TX_STATUS, Info.TX_STATUS)
    oce.add(OpCode.RX_STATUS, Info.RX_STATUS)
    oce.add(OpCode.ENABLE_HK_RX_REGS, Info.ENABLE_HK_RX_REGS)
    oce.add(OpCode.DISABLE_HK_RX_REGS, Info.DISABLE_HK_RX_REGS)
    oce.add(OpCode.ENABLE_HK_TX_REGS, Info.ENABLE_HK_TX_REGS)
    oce.add(OpCode.DISABLE_HK_TX_REGS, Info.DISABLE_HK_TX_REGS)
    oce.add(OpCode.HK_TEMPS, Info.HK_TEMPS)
    oce.add("7", "Syrlinks Handler: Read TX waveform")
    oce.add("8", "Syrlinks Handler: Read TX AGC value high byte")
    oce.add("9", "Syrlinks Handler: Read TX AGC value low byte")
    oce.add("12", "Syrlinks Handler: Write LCL config")
    oce.add("14", "Syrlinks Handler: Read LCL config register")
    oce.add("15", "Syrlinks Handler: Set waveform OQPSK")
    oce.add("16", "Syrlinks Handler: Set waveform BPSK")
    oce.add("17", "Syrlinks Handler: Set second config")
    oce.add("18", "Syrlinks Handler: Enable debug output")
    oce.add("19", "Syrlinks Handler: Disable debug output")
    defs.add_service(CustomServiceList.SYRLINKS.value, "Syrlinks Handler", oce)


_PREFIX = "Syrlinks"


def normal_mode_cmd(q: DefaultPusQueueHelper, info: str, submode: int):
    q.add_log_cmd(f"{_PREFIX}: {info}")
    q.add_pus_tc(create_mode_command(SYRLINKS_HANDLER_ID, Mode.NORMAL, submode))


def pack_syrlinks_command(
    object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code: str
):
    obyt = object_id.as_bytes
    prefix = "Syrlinks"
    q.add_log_cmd(f"Testing Syrlinks with object id: {object_id.as_hex_string}")
    if op_code == OpCode.OFF:
        q.add_log_cmd(f"{prefix}: {Info.OFF}")
        q.add_pus_tc(create_mode_command(obyt, Mode.OFF, 0))
    if op_code == OpCode.ON:
        q.add_log_cmd(f"{prefix}: {Info.ON}")
        q.add_pus_tc(create_mode_command(obyt, Mode.ON, 0))
    if op_code == OpCode.NORMAL_RX_ONLY:
        normal_mode_cmd(q, Info.NORMAL_RX_ONLY, Submode.RX_ONLY)
    if op_code == OpCode.NORMAL_RX_AND_TX_LOW_DATARATE:
        normal_mode_cmd(
            q, Info.NORMAL_RX_AND_TX_LOW_DATARATE, Submode.RX_AND_TX_LOW_DATARATE
        )
    if op_code == OpCode.NORMAL_RX_AND_TX_DEF_DATARATE:
        normal_mode_cmd(
            q, Info.NORMAL_RX_AND_TX_DEF_DATARATE, Submode.RX_AND_TX_DEFAULT_DATARATE
        )
    if op_code == OpCode.NORMAL_RX_AND_TX_HIGH_DATARATE:
        normal_mode_cmd(
            q, Info.NORMAL_RX_AND_TX_HIGH_DATARATE, Submode.RX_AND_TX_HIGH_DATARATE
        )
    if op_code in OpCode.NORMAL_RX_AND_TX_CW:
        normal_mode_cmd(q, Info.NORMAL_RX_AND_TX_CW, Submode.RX_AND_TX_CW)
    if op_code in OpCode.HK_RX_REGS:
        q.add_log_cmd(f"{prefix}: {Info.HK_RX_REGS}")
        sid = make_sid(obyt, SetId.RX_REGISTERS_DATASET)
        q.add_pus_tc(create_request_one_diag_command(sid))
    if op_code in OpCode.HK_TEMPS:
        q.add_log_cmd(f"{prefix}: {Info.HK_TEMPS}")
        sid = make_sid(obyt, SetId.TEMPERATURE_SET_ID)
        q.add_pus_tc(create_request_one_hk_command(sid))
    if op_code in OpCode.ENABLE_HK_RX_REGS:
        q.add_log_cmd(f"{prefix}: {Info.ENABLE_HK_RX_REGS}")
        sid = make_sid(obyt, SetId.RX_REGISTERS_DATASET)
        interval = float(input("HK interval in floating point seconds"))
        cmds = create_enable_periodic_hk_command_with_interval(True, sid, interval)
        for cmd in cmds:
            q.add_pus_tc(cmd)
    if op_code in OpCode.DISABLE_HK_RX_REGS:
        q.add_log_cmd(f"{prefix}: {Info.DISABLE_HK_RX_REGS}")
        sid = make_sid(obyt, SetId.RX_REGISTERS_DATASET)
        q.add_pus_tc(create_disable_periodic_hk_command(True, sid))
    if op_code in OpCode.ENABLE_HK_TX_REGS:
        q.add_log_cmd(f"{prefix}: {Info.ENABLE_HK_TX_REGS}")
        sid = make_sid(obyt, SetId.TX_REGISTERS_DATASET)
        interval = float(input("HK interval in floating point seconds"))
        cmds = create_enable_periodic_hk_command_with_interval(True, sid, interval)
        for cmd in cmds:
            q.add_pus_tc(cmd)
    if op_code in OpCode.DISABLE_HK_TX_REGS:
        q.add_log_cmd(f"{prefix}: {Info.DISABLE_HK_TX_REGS}")
        sid = make_sid(obyt, SetId.TX_REGISTERS_DATASET)
        q.add_pus_tc(create_disable_periodic_hk_command(True, sid))
    if op_code in OpCode.HK_TX_REGS:
        q.add_log_cmd(f"{prefix}: {Info.HK_TX_REGS}")
        sid = make_sid(obyt, SetId.TX_REGISTERS_DATASET)
        q.add_pus_tc(create_request_one_diag_command(sid))
    if op_code in OpCode.TX_STATUS:
        q.add_log_cmd(f"{prefix}: {Info.TX_STATUS}")
        command = obyt + struct.pack("!I", CommandId.READ_TX_STATUS)
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
    if op_code == "9":
        q.add_log_cmd("Syrlinks: Read TX waveform")
        command = obyt + struct.pack("!I", CommandId.READ_TX_WAVEFORM)
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
    if op_code == "10":
        q.add_log_cmd("Syrlinks: Read TX AGC value high byte")
        command = obyt + struct.pack("!I", CommandId.READ_TX_AGC_VALUE_HIGH_BYTE)
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
    if op_code == "11":
        q.add_log_cmd("Syrlinks: Read TX AGC value low byte")
        command = obyt + struct.pack("!I", CommandId.READ_TX_AGC_VALUE_LOW_BYTE)
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
    if op_code == "12":
        q.add_log_cmd("Syrlinks: Write LCL config")
        command = obyt + struct.pack("!I", CommandId.WRITE_LCL_CONFIG)
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
    if op_code == "13":
        q.add_log_cmd("Syrlinks: Read RX status registers")
        command = obyt + struct.pack("!I", CommandId.READ_RX_STATUS_REGISTERS)
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
    if op_code == "14":
        q.add_log_cmd("Syrlinks: Read LCL config register")
        command = obyt + struct.pack("!I", CommandId.READ_LCL_CONFIG_REGISTER)
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
    if op_code == "15":
        q.add_log_cmd("Syrlinks: Set waveform OQPSK")
        command = obyt + struct.pack("!I", CommandId.SET_WAVEFORM_OQPSK)
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
    if op_code == "16":
        q.add_log_cmd("Syrlinks: Set waveform BPSK")
        command = obyt + struct.pack("!I", CommandId.SET_WAVEFORM_BPSK)
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
    if op_code == "17":
        q.add_log_cmd("Syrlinks: Set second config")
        command = obyt + struct.pack("!I", CommandId.SET_SECOND_CONFIG)
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
    if op_code == "18":
        q.add_log_cmd("Syrlinks: Enable debug printout")
        command = obyt + struct.pack("!I", CommandId.ENABLE_DEBUG)
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
    if op_code == "19":
        q.add_log_cmd("Syrlinks: Disable debug printout")
        command = obyt + struct.pack("!I", CommandId.DISABLE_DEBUG)
        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))


def handle_syrlinks_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
    if set_id == SetId.RX_REGISTERS_DATASET:
        return handle_syrlinks_rx_registers_dataset(printer, hk_data)
    elif set_id == SetId.TX_REGISTERS_DATASET:
        return handle_syrlinks_tx_registers_dataset(printer, hk_data)
    elif set_id == SetId.TEMPERATURE_SET_ID:
        return handle_syrlinks_temp_dataset(printer, hk_data)
    else:
        pw = PrintWrapper(printer)
        pw.dlog(f"Service 3 TM: Syrlinks handler reply with unknown set ID {set_id}")


def handle_syrlinks_temp_dataset(printer: FsfwTmTcPrinter, hk_data: bytes):
    pw = PrintWrapper(printer)
    if len(hk_data) < 8:
        raise ValueError("expected  at least 8 bytes of HK data")
    temp_power_amplifier = struct.unpack("!f", hk_data[0:4])[0]
    temp_baseband_board = struct.unpack("!f", hk_data[4:8])[0]
    pw.dlog(f"Temperatur Power Amplifier [C]: {temp_power_amplifier}")
    pw.dlog(f"Temperatur Baseband Board [C]: {temp_baseband_board}")
    printer.print_validity_buffer(hk_data[8:], 2)


def handle_syrlinks_rx_registers_dataset(printer: FsfwTmTcPrinter, hk_data: bytes):
    pw = PrintWrapper(printer)
    header_list = [
        "RX Status",
        "RX Sensitivity",
        "RX Frequency Shift",
        "RX IQ Power",
        "RX AGC Value (Raw)",
        "RX Demod Eb",
        "RX Demod N0",
        "RX Datarate [kbps]",
    ]
    rx_status = hk_data[0]
    carrier_detect = rx_status & 0b1
    carrier_lock = (rx_status >> 1) & 0b1
    data_lock = (rx_status >> 2) & 0b1
    data_valid = (rx_status >> 3) & 0b1
    rx_sensitivity = struct.unpack("!I", hk_data[1:5])[0]
    rx_frequency_shift = struct.unpack("!i", hk_data[5:9])[0]
    freq_shift_hz = rx_frequency_shift / 8.0
    freq_shift_printout = f"Raw: {rx_frequency_shift}, Eng: {freq_shift_hz} Hz"
    rx_iq_power = struct.unpack("!H", hk_data[9:11])[0]
    rx_agc_value = struct.unpack("!H", hk_data[11:13])[0]
    rx_agc_inhibit = (rx_agc_value >> 15) & 0b1
    rx_agc = rx_agc_value & 0xFFF
    rx_demod_eb = struct.unpack("!I", hk_data[13:17])[0] & 0xFFFFFF
    rx_demod_n0 = struct.unpack("!I", hk_data[17:21])[0] & 0xFFFFFF
    eb_to_n0 = 20 * math.log10(rx_demod_eb / rx_demod_n0) - 3
    rx_data_rate_raw = hk_data[21]
    rx_data_rate = -1
    if rx_data_rate_raw == 0:
        rx_data_rate = 256
    elif rx_data_rate_raw == 1:
        rx_data_rate = 128
    elif rx_data_rate_raw == 3:
        rx_data_rate = 64
    elif rx_data_rate_raw == 7:
        rx_data_rate = 32
    elif rx_data_rate_raw == 15:
        rx_data_rate = 16
    elif rx_data_rate_raw == 31:
        rx_data_rate = 8
    content_list = [
        rx_status,
        rx_sensitivity,
        freq_shift_printout,
        rx_iq_power,
        rx_agc_value,
        rx_demod_eb,
        rx_demod_n0,
        rx_data_rate,
    ]
    validity_buffer = hk_data[22:]
    for header, content in zip(header_list, content_list):
        pw.dlog(f"{header}: {content}")
    printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=8)
    pw.dlog(f"Carrier Detect: {carrier_detect}")
    pw.dlog(f"Carrier Lock: {carrier_lock}")
    pw.dlog(f"Data Lock (data clock recovery loop lock status): {data_lock}")
    pw.dlog(f"Data Valid (valid if TEB < 10e-5): {data_valid}")
    pw.dlog(f"Data Lock (data clock recovery loop lock status): {data_lock}")
    pw.dlog(f"RX AGC Inhibit: {rx_agc_inhibit}")
    pw.dlog(f"RX AGC: {rx_agc}")
    pw.dlog(f"Eb / E0RX [dB]: {eb_to_n0}")


class TxConv(enum.IntEnum):
    NO_CODING = 0b000
    VITERBI_HALF_G1G2INV = 0b010
    VITERBI_HALF = 0b111


class TxStatus(enum.IntEnum):
    NOT_AVAILABLE = 0b00
    MODULATION = 0b01
    CW = 0b10
    STANDBY = 0b11


class TxCfgSet(enum.IntEnum):
    START_WITH_CURRENT_CFG = 0b00
    START_WITH_CONF_0 = 0b01
    START_WITH_CONF_1 = 0b10


WAVEFORM_STRINGS = ["OFF", "CW", "QPSK", "0QPSK", "PCM/PM", "PSK/PM", "BPSK"]


def handle_syrlinks_tx_registers_dataset(
    printer: FsfwTmTcPrinter,
    hk_data: bytes,
):
    pw = PrintWrapper(printer)
    header_list = ["TX Status Raw", "TX Waveform", "TX AGC value"]
    tx_status = hk_data[0]
    """
    try:
        tx_conv = TxConv(tx_status & 0b111)
    except ValueError:
        logging.getLogger(__name__).warning(
            f"invalid TX conv value {tx_status & 0b111}"
        )
        tx_conv = -1
    tx_diff_encoder_enable = (tx_status >> 3) & 0b1
    """
    tx_status_status = TxStatus(tx_status & 0b11)
    try:
        tx_conf_set = TxCfgSet((tx_status >> 2) & 0b11)
    except ValueError:
        logging.getLogger(__name__).warning(
            f"invalid TX conf set {(tx_status >> 2) & 0b11}"
        )
        tx_conf_set = -1
    tx_clock_detect = (tx_status >> 4) & 0b1
    tx_waveform = hk_data[1]
    waveform = tx_waveform & 0b1111
    try:
        waveform_str = WAVEFORM_STRINGS[waveform]
    except IndexError:
        logging.getLogger(__name__).warning(f"Unknown waveform value {waveform}")
        waveform_str = "Unknown"
    pcm_mode = (tx_waveform >> 4) & 0b1
    tx_agc_value = struct.unpack("!H", hk_data[2:4])[0]
    tx_agc_inhibit = (tx_agc_value >> 15) & 0b1
    tx_agc = tx_agc_value & 0xFFF
    content_list = [tx_status, tx_waveform, tx_agc_value]
    validity_buffer = hk_data[4:]
    for header, content in zip(header_list, content_list):
        pw.dlog(f"{header}: {content}")
    printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=3)
    # pw.dlog(f"TX CONV: {tx_conv!r}")
    # pw.dlog(f"TX DIFF (differential encoder enable): {tx_diff_encoder_enable}")
    pw.dlog(f"TX Status: {tx_status_status!r}")
    pw.dlog(f"TX Config Set: {tx_conf_set!r}")
    pw.dlog(f"TX Clock Detect: {tx_clock_detect}")
    pw.dlog(f"Waveform: {waveform_str}")
    pw.dlog(f"PCM Mode: {pcm_mode}")
    pw.dlog(f"TX AGC Inhibit: {tx_agc_inhibit}")
    pw.dlog(f"TX AGC: {tx_agc}")