68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
import struct
|
|
|
|
from eive_tmtc.pus_tm.defs import PrintWrapper
|
|
from eive_tmtc.pus_tc.devs.syrlinks_hk_handler import SetIds
|
|
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
|
|
|
|
|
|
def handle_syrlinks_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
|
|
if set_id == SetIds.RX_REGISTERS_DATASET:
|
|
return handle_syrlinks_rx_registers_dataset(printer, hk_data)
|
|
elif set_id == SetIds.TX_REGISTERS_DATASET:
|
|
return handle_syrlinks_tx_registers_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_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",
|
|
"RX Demod Eb",
|
|
"RX Demod N0",
|
|
"RX Datarate",
|
|
]
|
|
rx_status = hk_data[0]
|
|
rx_sensitivity = struct.unpack("!I", hk_data[1:5])
|
|
rx_frequency_shift = struct.unpack("!I", hk_data[5:9])
|
|
rx_iq_power = struct.unpack("!H", hk_data[9:11])
|
|
rx_agc_value = struct.unpack("!H", hk_data[11:13])
|
|
rx_demod_eb = struct.unpack("!I", hk_data[13:17])
|
|
rx_demod_n0 = struct.unpack("!I", hk_data[17:21])
|
|
rx_data_rate = hk_data[21]
|
|
content_list = [
|
|
rx_status,
|
|
rx_sensitivity,
|
|
rx_frequency_shift,
|
|
rx_iq_power,
|
|
rx_agc_value,
|
|
rx_demod_eb,
|
|
rx_demod_n0,
|
|
rx_data_rate,
|
|
]
|
|
validity_buffer = hk_data[22:]
|
|
pw.dlog(str(header_list))
|
|
pw.dlog(str(content_list))
|
|
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=8)
|
|
|
|
|
|
def handle_syrlinks_tx_registers_dataset(
|
|
printer: FsfwTmTcPrinter,
|
|
hk_data: bytes,
|
|
):
|
|
pw = PrintWrapper(printer)
|
|
header_list = ["TX Status", "TX Waveform", "TX AGC value"]
|
|
tx_status = hk_data[0]
|
|
tx_waveform = hk_data[1]
|
|
tx_agc_value = struct.unpack("!H", hk_data[2:4])
|
|
content_list = [tx_status, tx_waveform, tx_agc_value]
|
|
validity_buffer = hk_data[4:]
|
|
pw.dlog(str(header_list))
|
|
pw.dlog(str(content_list))
|
|
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=3)
|