113 lines
5.2 KiB
Python
113 lines
5.2 KiB
Python
"""
|
|
@brief This file transfers control of housekeeping handling (PUS service 3) to the
|
|
developer
|
|
@details Template configuration file. Copy this folder to the TMTC commander root and adapt
|
|
it to your needs.
|
|
"""
|
|
import struct
|
|
from typing import Tuple
|
|
|
|
from tmtccmd.pus_tm.service_3_housekeeping import Service3Base
|
|
from tmtccmd.utility.logger import get_logger
|
|
from pus_tc.syrlinks_hk_handler import SetIds
|
|
from config.object_ids import SYRLINKS_HANDLER, IMTQ_HANDLER_ID
|
|
LOGGER = get_logger()
|
|
|
|
|
|
def handle_user_hk_packet(object_id: bytes, set_id: int, hk_data: bytearray,
|
|
service3_packet: Service3Base) -> Tuple[list, list, bytearray, int]:
|
|
"""
|
|
This function is called when a Service 3 Housekeeping packet is received.
|
|
|
|
Please note that the object IDs should be compared by value because direct comparison of
|
|
enumerations does not work in Python. For example use:
|
|
|
|
if object_id.value == ObjectIds.TEST_OBJECT.value
|
|
|
|
to test equality based on the object ID list.
|
|
|
|
@param object_id:
|
|
@param set_id:
|
|
@param hk_data:
|
|
@param service3_packet:
|
|
@return: Expects a tuple, consisting of two lists, a bytearray and an integer
|
|
The first list contains the header columns, the second list the list with
|
|
the corresponding values. The bytearray is the validity buffer, which is usually appended
|
|
at the end of the housekeeping packet. The last value is the number of parameters.
|
|
"""
|
|
if object_id == SYRLINKS_HANDLER:
|
|
if set_id == SetIds.RX_REGISTERS_DATASET:
|
|
return handle_syrlinks_rx_registers_dataset(hk_data)
|
|
elif set_id == SetIds.TX_REGISTERS_DATASET:
|
|
return handle_syrlinks_tx_registers_dataset(hk_data)
|
|
else:
|
|
LOGGER.info("Serive 3 TM: Syrlinks handler reply with unknown set id")
|
|
return [], [], bytearray(), 0
|
|
if object_id == IMTQ_HANDLER_ID:
|
|
if set_id == SetIds.RX_REGISTERS_DATASET:
|
|
return imtq_positive_x_test(hk_data)
|
|
else:
|
|
LOGGER.info("Serive 3 TM: Syrlinks handler reply with unknown set id")
|
|
return [], [], bytearray(), 0
|
|
else:
|
|
LOGGER.info("Service 3 TM: Parsing for this SID has not been implemented.")
|
|
return [], [], bytearray(), 0
|
|
|
|
|
|
def handle_syrlinks_rx_registers_dataset(hk_data: bytearray) -> Tuple[list, list, bytearray, int]:
|
|
hk_header = []
|
|
hk_content = []
|
|
validity_buffer = bytearray()
|
|
hk_header = ["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]
|
|
hk_content = [rx_status, rx_sensitivity, rx_frequency_shift, rx_iq_power, rx_agc_value, rx_demod_eb, rx_demod_n0,
|
|
rx_data_rate]
|
|
return hk_header, hk_content, validity_buffer, 8
|
|
|
|
|
|
def handle_syrlinks_tx_registers_dataset(hk_data: bytearray) -> Tuple[list, list, bytearray, int]:
|
|
hk_header = []
|
|
hk_content = []
|
|
validity_buffer = bytearray()
|
|
hk_header = ["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])
|
|
hk_content = [tx_status, tx_waveform, tx_agc_value]
|
|
return hk_header, hk_content, validity_buffer, 3
|
|
|
|
|
|
def imtq_positive_x_test(hk_data: bytearray) -> Tuple[list, list, bytearray, int]:
|
|
hk_header = []
|
|
hk_content = []
|
|
validity_buffer = bytearray()
|
|
hk_header = ["Init Err", "Init Raw Mag X [T]", "Init Raw Mag Y [T]", "Init Raw Mag Z [T]", "Init Cal Mag X [T]",
|
|
"Init Cal Mag Y [T]", "Init Cal Mag Z [T]", "Init Coil X Current [A]", "Init Coil Y Current [A]",
|
|
"Init Coil Z Current [A]", "Init Coil X Temperature [°C]", "Init Coil Y Temperature [°C]",
|
|
"Init Coil Z Temperature [°C]"]
|
|
init_err = hk_data[0]
|
|
init_raw_mag_x = struct.unpack('!f', hk_data[1:5])
|
|
init_raw_mag_y = struct.unpack('!f', hk_data[5:9])
|
|
init_raw_mag_z = struct.unpack('!f', hk_data[9:13])
|
|
init_cal_mag_x = struct.unpack('!f', hk_data[13:17])
|
|
init_cal_mag_y = struct.unpack('!f', hk_data[17:21])
|
|
init_cal_mag_z = struct.unpack('!f', hk_data[21:25])
|
|
init_coil_x_current = struct.unpack('!f', hk_data[25:29])
|
|
init_coil_y_current = struct.unpack('!f', hk_data[29:33])
|
|
init_coil_z_current = struct.unpack('!f', hk_data[33:37])
|
|
init_coil_x_temperature = struct.unpack('!H', hk_data[37:39])
|
|
init_coil_y_temperature = struct.unpack('!H', hk_data[39:41])
|
|
init_coil_z_temperature = struct.unpack('!H', hk_data[41:43])
|
|
hk_content = [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]
|
|
return hk_header, hk_content, len(hk_header)
|