521 lines
19 KiB
Python
521 lines
19 KiB
Python
# -*- 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
|
|
import struct
|
|
|
|
from spacepackets.ecss.tc import PusTelecommand
|
|
from tmtccmd.config.tmtc import (
|
|
CmdTreeNode,
|
|
OpCodeEntry,
|
|
TmtcDefinitionWrapper,
|
|
tmtc_definitions_provider,
|
|
)
|
|
from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter
|
|
from tmtccmd.pus.s200_fsfw_mode import Mode, create_mode_command
|
|
from tmtccmd.pus.tc.s3_fsfw_hk import (
|
|
create_disable_periodic_hk_command_with_diag,
|
|
create_enable_periodic_hk_command_with_interval_with_diag,
|
|
create_request_one_diag_command,
|
|
create_request_one_hk_command,
|
|
make_sid,
|
|
)
|
|
from tmtccmd.tmtc import DefaultPusQueueHelper
|
|
from tmtccmd.util import ObjectIdU32
|
|
|
|
from eive_tmtc.config.definitions import CustomServiceList
|
|
from eive_tmtc.config.object_ids import SYRLINKS_HANDLER_ID
|
|
from eive_tmtc.pus_tm.defs import PrintWrapper
|
|
from eive_tmtc.pus_tm.hk import HkTmInfo
|
|
from eive_tmtc.tmtc.com.defs import Mode as ComMode
|
|
|
|
|
|
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"
|
|
SET_CW = "tx_cw"
|
|
|
|
|
|
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
|
|
|
|
|
|
def create_syrlinks_node() -> CmdTreeNode:
|
|
op_code_strs = [
|
|
getattr(OpCode, key) for key in dir(OpCode) if not key.startswith("__")
|
|
]
|
|
info_strs = [getattr(Info, key) for key in dir(OpCode) if not key.startswith("__")]
|
|
combined_dict = dict(zip(op_code_strs, info_strs))
|
|
node = CmdTreeNode("syrlinks", "Syrlinks Device", hide_children_for_print=True)
|
|
for op_code, info in combined_dict.items():
|
|
node.add_child(CmdTreeNode(op_code, info))
|
|
return node
|
|
|
|
|
|
@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( # noqa C901: Complexity okay here.
|
|
object_id: ObjectIdU32, q: DefaultPusQueueHelper, cmd_str: str
|
|
):
|
|
obyt = object_id.as_bytes
|
|
prefix = "Syrlinks"
|
|
q.add_log_cmd(f"Testing Syrlinks with object id: {object_id.as_hex_string}")
|
|
if cmd_str == OpCode.OFF:
|
|
q.add_log_cmd(f"{prefix}: {Info.OFF}")
|
|
q.add_pus_tc(create_mode_command(obyt, Mode.OFF, 0))
|
|
if cmd_str == OpCode.ON:
|
|
q.add_log_cmd(f"{prefix}: {Info.ON}")
|
|
q.add_pus_tc(create_mode_command(obyt, Mode.ON, ComMode.RX_ONLY))
|
|
if cmd_str == OpCode.NORMAL_RX_ONLY:
|
|
normal_mode_cmd(q, Info.NORMAL_RX_ONLY, ComMode.RX_ONLY)
|
|
if cmd_str == OpCode.NORMAL_RX_AND_TX_LOW_DATARATE:
|
|
normal_mode_cmd(
|
|
q, Info.NORMAL_RX_AND_TX_LOW_DATARATE, ComMode.RX_AND_TX_LOW_DATARATE
|
|
)
|
|
if cmd_str == OpCode.NORMAL_RX_AND_TX_DEF_DATARATE:
|
|
normal_mode_cmd(
|
|
q, Info.NORMAL_RX_AND_TX_DEF_DATARATE, ComMode.RX_AND_TX_DEF_DATARATE
|
|
)
|
|
if cmd_str == OpCode.NORMAL_RX_AND_TX_HIGH_DATARATE:
|
|
normal_mode_cmd(
|
|
q, Info.NORMAL_RX_AND_TX_HIGH_DATARATE, ComMode.RX_AND_TX_HIGH_DATARATE
|
|
)
|
|
if cmd_str in OpCode.NORMAL_RX_AND_TX_CW:
|
|
normal_mode_cmd(q, Info.NORMAL_RX_AND_TX_CW, ComMode.RX_AND_TX_CARRIER_WAVE)
|
|
if cmd_str 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 cmd_str 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 cmd_str 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_with_diag(
|
|
True, sid, interval
|
|
)
|
|
for cmd in cmds:
|
|
q.add_pus_tc(cmd)
|
|
if cmd_str 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_with_diag(True, sid))
|
|
if cmd_str 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_with_diag(
|
|
True, sid, interval
|
|
)
|
|
for cmd in cmds:
|
|
q.add_pus_tc(cmd)
|
|
if cmd_str 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_with_diag(True, sid))
|
|
if cmd_str 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 cmd_str 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 cmd_str == "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 cmd_str == "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 cmd_str == "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 cmd_str == "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 cmd_str == "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 cmd_str == "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 cmd_str == "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 cmd_str == "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 cmd_str == "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 cmd_str == "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 cmd_str == "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(
|
|
hk_info: HkTmInfo,
|
|
pw: PrintWrapper,
|
|
):
|
|
if hk_info.set_id == SetId.RX_REGISTERS_DATASET:
|
|
return handle_syrlinks_rx_registers_dataset(hk_info, pw)
|
|
elif hk_info.set_id == SetId.TX_REGISTERS_DATASET:
|
|
return handle_syrlinks_tx_registers_dataset(hk_info, pw)
|
|
elif hk_info.set_id == SetId.TEMPERATURE_SET_ID:
|
|
return handle_syrlinks_temp_dataset(hk_info, pw)
|
|
else:
|
|
pw.dlog(
|
|
f"Service 3 TM: Syrlinks handler reply with unknown set ID {hk_info.set_id}"
|
|
)
|
|
|
|
|
|
def handle_syrlinks_temp_dataset(hk_info: HkTmInfo, pw: PrintWrapper):
|
|
hk_data = hk_info.hk_data
|
|
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}")
|
|
pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[8:], 2))
|
|
|
|
|
|
def handle_syrlinks_rx_registers_dataset(
|
|
hk_info: HkTmInfo,
|
|
pw: PrintWrapper,
|
|
):
|
|
hk_data = hk_info.hk_data
|
|
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
|
|
if rx_demod_n0 > 0:
|
|
eb_to_n0 = 20 * math.log10(rx_demod_eb / rx_demod_n0) - 3
|
|
else:
|
|
logging.getLogger(__name__).warning(
|
|
"RX Demod N0 is 0, can not calculate Eb to N0"
|
|
)
|
|
eb_to_n0 = 0
|
|
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}")
|
|
pw.dlog(
|
|
FsfwTmTcPrinter.get_validity_buffer(validity_buffer=validity_buffer, num_vars=8)
|
|
)
|
|
print(f"Carrier Detect: {carrier_detect}")
|
|
print(f"Carrier Lock: {carrier_lock}")
|
|
print(f"Data Lock (data clock recovery loop lock status): {data_lock}")
|
|
print(f"Data Valid (valid if TEB < 10e-5): {data_valid}")
|
|
print(f"Data Lock (data clock recovery loop lock status): {data_lock}")
|
|
print(f"RX AGC Inhibit: {rx_agc_inhibit}")
|
|
print(f"RX AGC: {rx_agc}")
|
|
print(f"Eb / E0RX [dB]: {eb_to_n0}")
|
|
cursor = hk_info.db_con.cursor()
|
|
cursor.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS syrlinks_rx_regs(
|
|
packet_uuid TEXT PRIMARY KEY,
|
|
generation_time TEXT,
|
|
carrier_detect NUM,
|
|
carrier_lock NUM,
|
|
data_lock NUM,
|
|
data_valid NUM,
|
|
rx_agc_inhibit NUM,
|
|
rx_agc NUM,
|
|
eb_to_e0_rx NUM
|
|
)"""
|
|
)
|
|
cursor.execute(
|
|
"INSERT INTO syrlinks_rx_regs VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(
|
|
str(hk_info.packet_uuid),
|
|
hk_info.hk_packet.pus_tm.time_provider.as_datetime(), # type: ignore
|
|
carrier_detect,
|
|
carrier_lock,
|
|
data_lock,
|
|
data_valid,
|
|
rx_agc_inhibit,
|
|
rx_agc,
|
|
eb_to_n0,
|
|
),
|
|
)
|
|
hk_info.db_con.commit()
|
|
|
|
|
|
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(
|
|
hk_info: HkTmInfo,
|
|
pw: PrintWrapper,
|
|
):
|
|
header_list = ["TX Status Raw", "TX Waveform", "TX AGC value"]
|
|
tx_status = hk_info.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}"
|
|
)
|
|
# Hack to make DB insertion work.
|
|
tx_conf_set = TxCfgSet.START_WITH_CURRENT_CFG
|
|
tx_clock_detect = (tx_status >> 4) & 0b1
|
|
tx_waveform = hk_info.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_info.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_info.hk_data[4:]
|
|
for header, content in zip(header_list, content_list):
|
|
pw.dlog(f"{header}: {content}")
|
|
pw.dlog(
|
|
FsfwTmTcPrinter.get_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}")
|
|
print(f"TX Status: {tx_status_status!r}")
|
|
print(f"TX Config Set: {tx_conf_set!r}")
|
|
print(f"TX Clock Detect: {tx_clock_detect}")
|
|
print(f"Waveform: {waveform_str}")
|
|
print(f"PCM Mode: {pcm_mode}")
|
|
print(f"TX AGC Inhibit: {tx_agc_inhibit}")
|
|
print(f"TX AGC: {tx_agc}")
|
|
cursor = hk_info.db_con.cursor()
|
|
cursor.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS syrlinks_tx_regs(
|
|
packet_uuid TEXT PRIMARY KEY,
|
|
generation_time TEXT,
|
|
tx_status NUM,
|
|
tx_status_str TEXT,
|
|
tx_cfg_set NUM,
|
|
tx_cfg_set_str TEXT,
|
|
tx_clock_detect NUM,
|
|
waveform NUM,
|
|
waveform_str TEXT,
|
|
pcm_mode NUM,
|
|
tx_agc_inhibut NUM,
|
|
tx_agc NUM
|
|
)"""
|
|
)
|
|
cursor.execute(
|
|
"INSERT INTO syrlinks_tx_regs VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(
|
|
str(hk_info.packet_uuid),
|
|
hk_info.hk_packet.pus_tm.time_provider.as_datetime(), # type: ignore
|
|
tx_status_status,
|
|
tx_status_status.name,
|
|
tx_conf_set,
|
|
tx_conf_set.name,
|
|
tx_clock_detect,
|
|
waveform,
|
|
waveform_str,
|
|
pcm_mode,
|
|
tx_agc_inhibit,
|
|
tx_agc,
|
|
),
|
|
)
|
|
hk_info.db_con.commit()
|