780 lines
26 KiB
Python
780 lines
26 KiB
Python
from config.definitions import CustomServiceList
|
|
from tmtccmd.config import (
|
|
QueueCommands,
|
|
ServiceOpCodeDictT,
|
|
add_op_code_entry,
|
|
add_service_op_code_entry,
|
|
)
|
|
from tmtccmd.tc.definitions import TcQueueT
|
|
from tmtccmd.tc.pus_3_fsfw_hk import *
|
|
|
|
"""
|
|
from config.object_ids import (
|
|
BPX_HANDLER_ID,
|
|
P60_DOCK_HANDLER,
|
|
PDU_1_HANDLER_ID,
|
|
PDU_2_HANDLER_ID,
|
|
ACU_HANDLER_ID,
|
|
CORE_CONTROLLER_ID,
|
|
)
|
|
"""
|
|
import config.object_ids as oids
|
|
from pus_tc.devs.bpx_batt import BpxSetIds
|
|
from pus_tc.system.core import SetIds as CoreSetIds
|
|
from gomspace.gomspace_common import SetIds as GsSetIds
|
|
from pus_tc.devs.rad_sensor import SetIds as RadSetIds
|
|
from pus_tc.devs.mgms import MgmLis3SetIds as MgmSetIds_0_2
|
|
from pus_tc.devs.mgms import MgmRm3100SetIds as MgmSetIds_1_3
|
|
from pus_tc.devs.gyros import AdisGyroSetIds as GyroSetIds_0_2
|
|
from pus_tc.devs.gyros import L3gGyroSetIds as GyroSetIds_1_3
|
|
from pus_tc.devs.gps import SetIds as GpsSetIds
|
|
from pus_tc.devs.imtq import ImtqSetIds
|
|
from pus_tc.devs.sus import SetIds
|
|
from pus_tc.devs.star_tracker import SetIds as StrSetIds
|
|
from pus_tc.devs.reaction_wheels import RwSetIds
|
|
|
|
from pus_tc.system.tcs import pack_tcs_sys_commands
|
|
from pus_tc.system.controllers import pack_controller_commands
|
|
from pus_tc.system.acs import pack_acs_command, pack_sus_cmds
|
|
from pus_tc.devs.imtq import pack_imtq_test_into, pack_dipole_command
|
|
from pus_tc.devs.star_tracker import pack_star_tracker_commands
|
|
from pus_tc.devs.reaction_wheels import pack_rw_ass_cmds, pack_set_speed_command
|
|
|
|
|
|
class OpCodes:
|
|
HEATER = ["0", "heater"]
|
|
BAT_FT = ["bat-ft"]
|
|
CORE_FT = ["core-ft"]
|
|
PCDU_FT = ["pcdu-ft"]
|
|
RAD_SEN_FT = ["rad-sen-ft"]
|
|
TCS_FT_ON = ["tcs-ft-on"]
|
|
TCS_FT_OFF = ["tcs-ft-off"]
|
|
ACS_FT = ["acs-ft"]
|
|
MGT_FT = ["mgt-ft"]
|
|
MGT_FT_DP = ["mgt-ft-dp"]
|
|
SUS_FT = ["sus-ft"]
|
|
STR_FT = ["str-ft"]
|
|
RW_FT_ONE_RW = ["rw-ft-one-rw"]
|
|
RW_FT_TWO_RWS = ["rw-ft-two-rws"]
|
|
|
|
|
|
class KeyAndInfo:
|
|
HEATER = ["Heater", "heater procedure"]
|
|
BAT_FT = ["BPX Battery", "battery functional test"]
|
|
CORE_FT = ["OBC", "OBC functional test"]
|
|
PCDU_FT = ["PCDU", "PCDU functional test"]
|
|
RAD_SEN_FT = ["Radiation Sensor", "Radiation Sensor functional test"]
|
|
TCS_FT_ON = ["TCS Act.", "TCS functional test activation"]
|
|
TCS_FT_OFF = ["TCS Deact.", "TCS functional test deactivation"]
|
|
ACS_FT = ["ACS", "ACS functional test"]
|
|
MGT_FT = ["MGT", "MGT functional test"]
|
|
MGT_FT_DP = ["MGT dipole", "MGT functional test with dipole"]
|
|
SUS_FT = ["SUS", "SUS functional test"]
|
|
STR_FT = ["STR", "STR functional test"]
|
|
RW_FT_ONE_RW = ["RW ONE RW", "RW functional test with one RW"]
|
|
RW_FT_TWO_RWS = ["RW TWO RWS", "RW functional test with two RWs"]
|
|
|
|
|
|
KAI = KeyAndInfo
|
|
|
|
PROC_INFO_DICT = {
|
|
KAI.BAT_FT[0]: [OpCodes.BAT_FT, KAI.BAT_FT[1], 120.0, 10.0],
|
|
KAI.CORE_FT[0]: [OpCodes.CORE_FT, KAI.CORE_FT[1], 120.0, 10.0],
|
|
KAI.PCDU_FT[0]: [OpCodes.PCDU_FT, KAI.PCDU_FT[1], 120.0, 10.0],
|
|
KAI.RAD_SEN_FT[0]: [OpCodes.RAD_SEN_FT, KAI.RAD_SEN_FT[1], 120.0, 10.0],
|
|
KAI.TCS_FT_ON[0]: [OpCodes.TCS_FT_ON, KAI.TCS_FT_ON[1], 120.0, 10.0],
|
|
KAI.TCS_FT_OFF[0]: [OpCodes.TCS_FT_OFF, KAI.TCS_FT_OFF[1], 120.0, 10.0],
|
|
KAI.ACS_FT[0]: [OpCodes.ACS_FT, KAI.ACS_FT[1], 120.0, 10.0],
|
|
KAI.MGT_FT[0]: [OpCodes.MGT_FT, KAI.MGT_FT[1], 120.0, 10.0],
|
|
# collection_time for KAI.MGT_FT_DP maybe be reduced as a full 120seconds is not needed after MGTs are tested
|
|
KAI.MGT_FT_DP[0]: [OpCodes.MGT_FT_DP, KAI.MGT_FT_DP[1], 10.0, 10.0],
|
|
KAI.SUS_FT[0]: [OpCodes.SUS_FT, KAI.SUS_FT[1], 120.0, 10.0],
|
|
KAI.STR_FT[0]: [OpCodes.STR_FT, KAI.STR_FT[1], 120.0, 10.0],
|
|
# collection_time for KAI.RW_FT_ONE_RW maybe be reduced as a full 120seconds is not needed after RWs are tested
|
|
KAI.RW_FT_ONE_RW[0]: [OpCodes.RW_FT_ONE_RW, KAI.RW_FT_ONE_RW[1], 10.0, 1.0],
|
|
# collection_time for KAI.RW_FT_ONE_RW maybe be reduced as a full 120seconds is not needed after RWs are tested
|
|
KAI.RW_FT_TWO_RWS[0]: [OpCodes.RW_FT_TWO_RWS, KAI.RW_FT_TWO_RWS[1], 10.0, 1.0],
|
|
}
|
|
|
|
|
|
def generic_print(tc_queue: TcQueueT, info: dict):
|
|
tc_queue.appendleft(
|
|
(QueueCommands.PRINT, f"Executing {info[1]} Procedure (OpCodes: {info[0]})")
|
|
)
|
|
|
|
|
|
def add_proc_cmds(cmd_dict: ServiceOpCodeDictT):
|
|
|
|
op_code_dict = dict()
|
|
for proc_entry in PROC_INFO_DICT.values():
|
|
add_op_code_entry(
|
|
op_code_dict=op_code_dict, keys=proc_entry[0], info=proc_entry[1]
|
|
)
|
|
add_service_op_code_entry(
|
|
srv_op_code_dict=cmd_dict,
|
|
name=CustomServiceList.PROCEDURE.value,
|
|
info="TV Test Procedures",
|
|
op_code_entry=op_code_dict,
|
|
)
|
|
|
|
|
|
def pack_generic_hk_listening_cmds(
|
|
tc_queue: TcQueueT,
|
|
proc_key: str,
|
|
sid_list: list[bytearray],
|
|
diag: bool,
|
|
mgt: bool,
|
|
one_rw: bool,
|
|
two_rws: bool,
|
|
):
|
|
info = PROC_INFO_DICT[proc_key]
|
|
collection_time = info[2]
|
|
generic_print(tc_queue=tc_queue, info=info)
|
|
|
|
for sid in sid_list:
|
|
enable_listen_to_hk_for_x_seconds(
|
|
diag=diag,
|
|
tc_queue=tc_queue,
|
|
device=proc_key,
|
|
sid=sid,
|
|
interval_seconds=info[3],
|
|
)
|
|
|
|
if mgt is True:
|
|
activate_mgts_alternately(
|
|
tc_queue=tc_queue,
|
|
)
|
|
|
|
elif one_rw is True:
|
|
activate_one_rw_in_sequence(
|
|
tc_queue=tc_queue,
|
|
)
|
|
|
|
elif two_rws is True:
|
|
activate_two_rws_in_sequence(
|
|
tc_queue=tc_queue,
|
|
)
|
|
|
|
else:
|
|
pass
|
|
|
|
tc_queue.appendleft((QueueCommands.WAIT, collection_time))
|
|
|
|
for sid in sid_list:
|
|
disable_listen_to_hk_for_x_seconds(
|
|
diag=diag,
|
|
tc_queue=tc_queue,
|
|
device=proc_key,
|
|
sid=sid,
|
|
)
|
|
|
|
sid_list.clear()
|
|
|
|
|
|
def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
|
|
sid_list = []
|
|
if op_code in OpCodes.BAT_FT:
|
|
key = KAI.BAT_FT[0]
|
|
sid_list.append(make_sid(oids.BPX_HANDLER_ID, BpxSetIds.GET_HK_SET))
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=False,
|
|
one_rw=False,
|
|
two_rws=False,
|
|
)
|
|
|
|
if op_code in OpCodes.CORE_FT:
|
|
key = KAI.CORE_FT[0]
|
|
sid_list.append(make_sid(oids.CORE_CONTROLLER_ID, CoreSetIds.HK))
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=False,
|
|
one_rw=False,
|
|
two_rws=False,
|
|
)
|
|
|
|
if op_code in OpCodes.PCDU_FT:
|
|
key = KAI.PCDU_FT[0]
|
|
sid_list.append(make_sid(oids.P60_DOCK_HANDLER, GsSetIds.P60_CORE))
|
|
sid_list.append(make_sid(oids.PDU_1_HANDLER_ID, GsSetIds.PDU_1_CORE))
|
|
sid_list.append(make_sid(oids.PDU_2_HANDLER_ID, GsSetIds.PDU_2_CORE))
|
|
sid_list.append(make_sid(oids.ACU_HANDLER_ID, GsSetIds.ACU_CORE))
|
|
sid_list.append(make_sid(oids.P60_DOCK_HANDLER, GsSetIds.P60_AUX))
|
|
sid_list.append(make_sid(oids.PDU_1_HANDLER_ID, GsSetIds.PDU_1_AUX))
|
|
sid_list.append(make_sid(oids.PDU_2_HANDLER_ID, GsSetIds.PDU_2_AUX))
|
|
sid_list.append(make_sid(oids.ACU_HANDLER_ID, GsSetIds.ACU_AUX))
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=False,
|
|
one_rw=False,
|
|
two_rws=False,
|
|
)
|
|
|
|
if op_code in OpCodes.RAD_SEN_FT:
|
|
key = KAI.RAD_SEN_FT[0]
|
|
sid_list.append(make_sid(oids.RAD_SENSOR_ID, RadSetIds.RAD_SEN_CORE))
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=False,
|
|
one_rw=False,
|
|
two_rws=False,
|
|
)
|
|
|
|
if op_code in OpCodes.TCS_FT_ON:
|
|
# check whether tcs_assembly also has to be commanded to NORMAL Mode
|
|
# pack_tcs_sys_commands(tc_queue=tc_queue, op_code="tcs-normal")
|
|
pack_controller_commands(tc_queue=tc_queue, op_code="thermal_controller")
|
|
|
|
if op_code in OpCodes.TCS_FT_OFF:
|
|
# check whether tcs_assembly also has to be commanded to OFF Mode
|
|
# pack_tcs_sys_commands(tc_queue=tc_queue, op_code="tcs-off")
|
|
pack_controller_commands(tc_queue=tc_queue, op_code="thermal_controller")
|
|
|
|
if op_code in OpCodes.ACS_FT:
|
|
key = KAI.ACS_FT[0]
|
|
|
|
pack_acs_command(tc_queue=tc_queue, op_code="acs-a")
|
|
|
|
# MGMs
|
|
sid_list.append(make_sid(oids.MGM_0_LIS3_HANDLER_ID, MgmSetIds_0_2.CORE_HK))
|
|
sid_list.append(make_sid(oids.MGM_1_RM3100_HANDLER_ID, MgmSetIds_1_3.CORE_HK))
|
|
# Gyros
|
|
sid_list.append(make_sid(oids.GYRO_0_ADIS_HANDLER_ID, GyroSetIds_0_2.CORE_HK))
|
|
sid_list.append(make_sid(oids.GYRO_0_ADIS_HANDLER_ID, GyroSetIds_0_2.CFG_HK))
|
|
sid_list.append(make_sid(oids.GYRO_1_L3G_HANDLER_ID, GyroSetIds_1_3.CORE_HK))
|
|
# GNSS0
|
|
sid_list.append(make_sid(oids.GPS_HANDLER_0_ID, GpsSetIds.HK))
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=False,
|
|
one_rw=False,
|
|
two_rws=False,
|
|
)
|
|
|
|
pack_acs_command(tc_queue=tc_queue, op_code="acs-off")
|
|
tc_queue.appendleft((QueueCommands.WAIT, 5.0))
|
|
pack_acs_command(tc_queue=tc_queue, op_code="acs-b")
|
|
|
|
# MGMs
|
|
sid_list.append(make_sid(oids.MGM_2_LIS3_HANDLER_ID, MgmSetIds_0_2.CORE_HK))
|
|
sid_list.append(make_sid(oids.MGM_3_RM3100_HANDLER_ID, MgmSetIds_1_3.CORE_HK))
|
|
# Gyros
|
|
sid_list.append(make_sid(oids.GYRO_2_ADIS_HANDLER_ID, GyroSetIds_0_2.CORE_HK))
|
|
sid_list.append(make_sid(oids.GYRO_2_ADIS_HANDLER_ID, GyroSetIds_0_2.CFG_HK))
|
|
sid_list.append(make_sid(oids.GYRO_3_L3G_HANDLER_ID, GyroSetIds_1_3.CORE_HK))
|
|
# GNSS1
|
|
sid_list.append(make_sid(oids.GPS_HANDLER_1_ID, GpsSetIds.HK))
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=False,
|
|
one_rw=False,
|
|
two_rws=False,
|
|
)
|
|
|
|
pack_acs_command(tc_queue=tc_queue, op_code="acs-off")
|
|
tc_queue.appendleft((QueueCommands.WAIT, 5.0))
|
|
pack_acs_command(tc_queue=tc_queue, op_code="acs-d")
|
|
|
|
# MGMs
|
|
sid_list.append(make_sid(oids.MGM_0_LIS3_HANDLER_ID, MgmSetIds_0_2.CORE_HK))
|
|
sid_list.append(make_sid(oids.MGM_1_RM3100_HANDLER_ID, MgmSetIds_1_3.CORE_HK))
|
|
sid_list.append(make_sid(oids.MGM_2_LIS3_HANDLER_ID, MgmSetIds_0_2.CORE_HK))
|
|
sid_list.append(make_sid(oids.MGM_3_RM3100_HANDLER_ID, MgmSetIds_1_3.CORE_HK))
|
|
# Gyros
|
|
sid_list.append(make_sid(oids.GYRO_0_ADIS_HANDLER_ID, GyroSetIds_0_2.CORE_HK))
|
|
sid_list.append(make_sid(oids.GYRO_0_ADIS_HANDLER_ID, GyroSetIds_0_2.CFG_HK))
|
|
sid_list.append(make_sid(oids.GYRO_1_L3G_HANDLER_ID, GyroSetIds_1_3.CORE_HK))
|
|
sid_list.append(make_sid(oids.GYRO_2_ADIS_HANDLER_ID, GyroSetIds_0_2.CORE_HK))
|
|
sid_list.append(make_sid(oids.GYRO_2_ADIS_HANDLER_ID, GyroSetIds_0_2.CFG_HK))
|
|
sid_list.append(make_sid(oids.GYRO_3_L3G_HANDLER_ID, GyroSetIds_1_3.CORE_HK))
|
|
# GNSS0+1
|
|
sid_list.append(make_sid(oids.GPS_HANDLER_0_ID, GpsSetIds.HK))
|
|
sid_list.append(make_sid(oids.GPS_HANDLER_1_ID, GpsSetIds.HK))
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=False,
|
|
one_rw=False,
|
|
two_rws=False,
|
|
)
|
|
|
|
pack_acs_command(tc_queue=tc_queue, op_code="acs-off")
|
|
|
|
if op_code in OpCodes.MGT_FT:
|
|
key = KAI.MGT_FT[0]
|
|
|
|
# mgt 1: imtq und hk
|
|
pack_imtq_test_into(oids.IMTQ_HANDLER_ID, tc_queue=tc_queue, op_code="2")
|
|
|
|
sid_list.append(make_sid(oids.IMTQ_HANDLER_ID, ImtqSetIds.ENG_HK_SET))
|
|
sid_list.append(make_sid(oids.IMTQ_HANDLER_ID, ImtqSetIds.CAL_MTM_SET))
|
|
sid_list.append(make_sid(oids.IMTQ_HANDLER_ID, ImtqSetIds.RAW_MTM_SET))
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=False,
|
|
one_rw=False,
|
|
two_rws=False,
|
|
)
|
|
|
|
pack_imtq_test_into(oids.IMTQ_HANDLER_ID, tc_queue=tc_queue, op_code="0")
|
|
|
|
if op_code in OpCodes.MGT_FT_DP:
|
|
key = KAI.MGT_FT_DP[0]
|
|
|
|
pack_acs_command(tc_queue=tc_queue, op_code="acs-d")
|
|
# mgt 2.: imtq + dual side + dipole
|
|
pack_imtq_test_into(
|
|
object_id=oids.IMTQ_HANDLER_ID, tc_queue=tc_queue, op_code="2"
|
|
)
|
|
|
|
sid_list.append(make_sid(oids.IMTQ_HANDLER_ID, ImtqSetIds.ENG_HK_SET))
|
|
sid_list.append(make_sid(oids.IMTQ_HANDLER_ID, ImtqSetIds.CAL_MTM_SET))
|
|
sid_list.append(make_sid(oids.IMTQ_HANDLER_ID, ImtqSetIds.RAW_MTM_SET))
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=True,
|
|
one_rw=False,
|
|
two_rws=False,
|
|
)
|
|
|
|
pack_imtq_test_into(oids.IMTQ_HANDLER_ID, tc_queue=tc_queue, op_code="0")
|
|
pack_acs_command(tc_queue=tc_queue, op_code="acs-off")
|
|
|
|
if op_code in OpCodes.SUS_FT:
|
|
key = KAI.SUS_FT[0]
|
|
|
|
pack_sus_cmds(tc_queue=tc_queue, op_code="sus-nom")
|
|
|
|
sus_n_ids = [
|
|
oids.SUS_0_N_LOC_XFYFZM_PT_XF,
|
|
oids.SUS_1_N_LOC_XBYFZM_PT_XB,
|
|
oids.SUS_2_N_LOC_XFYBZB_PT_YB,
|
|
oids.SUS_3_N_LOC_XFYBZF_PT_YF,
|
|
oids.SUS_4_N_LOC_XMYFZF_PT_ZF,
|
|
oids.SUS_5_N_LOC_XFYMZB_PT_ZB,
|
|
]
|
|
sus_r_ids = [
|
|
oids.SUS_6_R_LOC_XFYBZM_PT_XF,
|
|
oids.SUS_7_R_LOC_XBYBZM_PT_XB,
|
|
oids.SUS_8_R_LOC_XBYBZB_PT_YB,
|
|
oids.SUS_9_R_LOC_XBYBZB_PT_YF,
|
|
oids.SUS_10_R_LOC_XMYBZF_PT_ZF,
|
|
oids.SUS_11_R_LOC_XBYMZB_PT_ZB,
|
|
]
|
|
# SUSs
|
|
for nom_sus in sus_n_ids:
|
|
sid_list.append(make_sid(nom_sus, SetIds.HK))
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=False,
|
|
one_rw=False,
|
|
two_rws=False,
|
|
)
|
|
|
|
pack_acs_command(tc_queue=tc_queue, op_code="sus-off")
|
|
tc_queue.appendleft((QueueCommands.WAIT, 5.0))
|
|
pack_sus_cmds(tc_queue=tc_queue, op_code="sus-red")
|
|
|
|
# SUSs
|
|
for red_sus in sus_r_ids:
|
|
sid_list.append(make_sid(red_sus, SetIds.HK))
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=False,
|
|
one_rw=False,
|
|
two_rws=False,
|
|
)
|
|
|
|
pack_acs_command(tc_queue=tc_queue, op_code="sus-off")
|
|
tc_queue.appendleft((QueueCommands.WAIT, 5.0))
|
|
pack_sus_cmds(tc_queue=tc_queue, op_code="sus-d")
|
|
|
|
# SUSs
|
|
for nom_sus in sus_n_ids:
|
|
sid_list.append(make_sid(nom_sus, SetIds.HK))
|
|
for red_sus in sus_r_ids:
|
|
sid_list.append(make_sid(red_sus, SetIds.HK))
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=False,
|
|
one_rw=False,
|
|
two_rws=False,
|
|
)
|
|
|
|
pack_acs_command(tc_queue=tc_queue, op_code="sus-off")
|
|
|
|
if op_code in OpCodes.STR_FT:
|
|
key = KAI.STR_FT[0]
|
|
|
|
pack_star_tracker_commands(
|
|
object_id=oids.STAR_TRACKER_ID, tc_queue=tc_queue, op_code="2"
|
|
)
|
|
|
|
# STR
|
|
sid_list.append(make_sid(oids.STAR_TRACKER_ID, StrSetIds.TEMPERATURE))
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=False,
|
|
one_rw=False,
|
|
two_rws=False,
|
|
)
|
|
|
|
pack_star_tracker_commands(
|
|
object_id=oids.STAR_TRACKER_ID, tc_queue=tc_queue, op_code="3"
|
|
)
|
|
|
|
if op_code in OpCodes.RW_FT_ONE_RW:
|
|
key = KAI.RW_FT_ONE_RW[0]
|
|
# RW NORMAL
|
|
pack_rw_ass_cmds(object_id=oids.RW_ASSEMBLY, tc_queue=tc_queue, op_code="nml")
|
|
|
|
# RW HK für alle RWs nur einzeln
|
|
sid_list.append(make_sid(oids.RW1_ID, RwSetIds.STATUS_SET_ID))
|
|
sid_list.append(make_sid(oids.RW1_ID, RwSetIds.TEMPERATURE_SET_ID))
|
|
sid_list.append(make_sid(oids.RW1_ID, RwSetIds.LAST_RESET))
|
|
sid_list.append(make_sid(oids.RW1_ID, RwSetIds.TM_SET))
|
|
sid_list.append(make_sid(oids.RW2_ID, RwSetIds.STATUS_SET_ID))
|
|
sid_list.append(make_sid(oids.RW2_ID, RwSetIds.TEMPERATURE_SET_ID))
|
|
sid_list.append(make_sid(oids.RW2_ID, RwSetIds.LAST_RESET))
|
|
sid_list.append(make_sid(oids.RW2_ID, RwSetIds.TM_SET))
|
|
sid_list.append(make_sid(oids.RW3_ID, RwSetIds.STATUS_SET_ID))
|
|
sid_list.append(make_sid(oids.RW3_ID, RwSetIds.TEMPERATURE_SET_ID))
|
|
sid_list.append(make_sid(oids.RW3_ID, RwSetIds.LAST_RESET))
|
|
sid_list.append(make_sid(oids.RW3_ID, RwSetIds.TM_SET))
|
|
sid_list.append(make_sid(oids.RW4_ID, RwSetIds.STATUS_SET_ID))
|
|
sid_list.append(make_sid(oids.RW4_ID, RwSetIds.TEMPERATURE_SET_ID))
|
|
sid_list.append(make_sid(oids.RW4_ID, RwSetIds.LAST_RESET))
|
|
sid_list.append(make_sid(oids.RW4_ID, RwSetIds.TM_SET))
|
|
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=False,
|
|
one_rw=True,
|
|
two_rws=False,
|
|
)
|
|
# RW OFF
|
|
pack_rw_ass_cmds(object_id=oids.RW_ASSEMBLY, tc_queue=tc_queue, op_code="off")
|
|
|
|
# ass command with 2 rws to speed
|
|
if op_code in OpCodes.RW_FT_TWO_RWS:
|
|
key = KAI.RW_FT_TWO_RWS[0]
|
|
# RW NORMAL
|
|
pack_rw_ass_cmds(object_id=oids.RW_ASSEMBLY, tc_queue=tc_queue, op_code="nml")
|
|
|
|
# RW
|
|
sid_list.append(make_sid(oids.RW1_ID, RwSetIds.STATUS_SET_ID))
|
|
sid_list.append(make_sid(oids.RW1_ID, RwSetIds.TEMPERATURE_SET_ID))
|
|
sid_list.append(make_sid(oids.RW1_ID, RwSetIds.LAST_RESET))
|
|
sid_list.append(make_sid(oids.RW1_ID, RwSetIds.TM_SET))
|
|
sid_list.append(make_sid(oids.RW2_ID, RwSetIds.STATUS_SET_ID))
|
|
sid_list.append(make_sid(oids.RW2_ID, RwSetIds.TEMPERATURE_SET_ID))
|
|
sid_list.append(make_sid(oids.RW2_ID, RwSetIds.LAST_RESET))
|
|
sid_list.append(make_sid(oids.RW2_ID, RwSetIds.TM_SET))
|
|
sid_list.append(make_sid(oids.RW3_ID, RwSetIds.STATUS_SET_ID))
|
|
sid_list.append(make_sid(oids.RW3_ID, RwSetIds.TEMPERATURE_SET_ID))
|
|
sid_list.append(make_sid(oids.RW3_ID, RwSetIds.LAST_RESET))
|
|
sid_list.append(make_sid(oids.RW3_ID, RwSetIds.TM_SET))
|
|
sid_list.append(make_sid(oids.RW4_ID, RwSetIds.STATUS_SET_ID))
|
|
sid_list.append(make_sid(oids.RW4_ID, RwSetIds.TEMPERATURE_SET_ID))
|
|
sid_list.append(make_sid(oids.RW4_ID, RwSetIds.LAST_RESET))
|
|
sid_list.append(make_sid(oids.RW4_ID, RwSetIds.TM_SET))
|
|
pack_generic_hk_listening_cmds(
|
|
tc_queue=tc_queue,
|
|
proc_key=key,
|
|
sid_list=sid_list,
|
|
diag=False,
|
|
mgt=False,
|
|
one_rw=False,
|
|
two_rws=True,
|
|
)
|
|
# RW OFF
|
|
pack_rw_ass_cmds(object_id=oids.RW_ASSEMBLY, tc_queue=tc_queue, op_code="off")
|
|
|
|
|
|
"""
|
|
def listen_to_hk_for_x_seconds(
|
|
tc_queue: TcQueueT,
|
|
diag: bool,
|
|
device: str,
|
|
sid: bytes,
|
|
interval_seconds: float,
|
|
collection_time: float,
|
|
):
|
|
|
|
tc_queue.appendleft((QueueCommands.PRINT, f"Enabling periodic HK for {device}"))
|
|
cmd_tuple = enable_periodic_hk_command_with_interval(
|
|
diag=diag, sid=sid, interval_seconds=interval_seconds, ssc=0
|
|
)
|
|
for cmd in cmd_tuple:
|
|
tc_queue.appendleft(cmd.pack_command_tuple())
|
|
|
|
tc_queue.appendleft((QueueCommands.WAIT, collection_time))
|
|
|
|
tc_queue.appendleft((QueueCommands.PRINT, f"Disabling periodic HK for {device}"))
|
|
tc_queue.appendleft(
|
|
disable_periodic_hk_command(diag=diag, sid=sid, ssc=0).pack_command_tuple()
|
|
)
|
|
"""
|
|
|
|
|
|
def enable_listen_to_hk_for_x_seconds(
|
|
tc_queue: TcQueueT,
|
|
diag: bool,
|
|
device: str,
|
|
sid: bytes,
|
|
interval_seconds: float,
|
|
):
|
|
tc_queue.appendleft((QueueCommands.PRINT, f"Enabling periodic HK for {device}"))
|
|
cmd_tuple = enable_periodic_hk_command_with_interval(
|
|
diag=diag, sid=sid, interval_seconds=interval_seconds, ssc=0
|
|
)
|
|
for cmd in cmd_tuple:
|
|
tc_queue.appendleft(cmd.pack_command_tuple())
|
|
|
|
tc_queue.appendleft((QueueCommands.WAIT, 2.0))
|
|
|
|
|
|
def disable_listen_to_hk_for_x_seconds(
|
|
tc_queue: TcQueueT,
|
|
diag: bool,
|
|
device: str,
|
|
sid: bytes,
|
|
):
|
|
tc_queue.appendleft((QueueCommands.PRINT, f"Disabling periodic HK for {device}"))
|
|
tc_queue.appendleft(
|
|
disable_periodic_hk_command(diag=diag, sid=sid, ssc=0).pack_command_tuple()
|
|
)
|
|
|
|
|
|
def activate_mgts_alternately(
|
|
tc_queue: TcQueueT,
|
|
):
|
|
|
|
command = pack_dipole_command(
|
|
object_id=oids.IMTQ_HANDLER_ID,
|
|
x_dipole=2000,
|
|
y_dipole=0,
|
|
z_dipole=0,
|
|
duration=30000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 40.0))
|
|
|
|
command = pack_dipole_command(
|
|
object_id=oids.IMTQ_HANDLER_ID,
|
|
x_dipole=-2000,
|
|
y_dipole=0,
|
|
z_dipole=0,
|
|
duration=30000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 40.0))
|
|
|
|
command = pack_dipole_command(
|
|
object_id=oids.IMTQ_HANDLER_ID,
|
|
x_dipole=0,
|
|
y_dipole=2000,
|
|
z_dipole=0,
|
|
duration=30000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 40.0))
|
|
|
|
command = pack_dipole_command(
|
|
object_id=oids.IMTQ_HANDLER_ID,
|
|
x_dipole=0,
|
|
y_dipole=-2000,
|
|
z_dipole=0,
|
|
duration=30000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 40.0))
|
|
|
|
command = pack_dipole_command(
|
|
object_id=oids.IMTQ_HANDLER_ID,
|
|
x_dipole=0,
|
|
y_dipole=0,
|
|
z_dipole=2000,
|
|
duration=30000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 40.0))
|
|
|
|
command = pack_dipole_command(
|
|
object_id=oids.IMTQ_HANDLER_ID,
|
|
x_dipole=0,
|
|
y_dipole=0,
|
|
z_dipole=-2000,
|
|
duration=30000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 40.0))
|
|
|
|
|
|
def activate_one_rw_in_sequence(
|
|
tc_queue: TcQueueT,
|
|
):
|
|
# RW1 speed cmd
|
|
tc_queue.appendleft((QueueCommands.WAIT, 5.0))
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW1_ID,
|
|
speed=20000,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 70.0))
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW1_ID,
|
|
speed=0,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 10.0))
|
|
|
|
# RW2 speed cmd
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW2_ID,
|
|
speed=20000,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 70.0))
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW2_ID,
|
|
speed=0,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 10.0))
|
|
|
|
# RW3 speed cmd
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW3_ID,
|
|
speed=20000,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 70.0))
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW3_ID,
|
|
speed=0,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 10.0))
|
|
|
|
# RW4 speed cmd
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW4_ID,
|
|
speed=20000,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 70.0))
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW4_ID,
|
|
speed=0,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 5.0))
|
|
|
|
|
|
def activate_two_rws_in_sequence(tc_queue: TcQueueT):
|
|
# RW1+3 speed cmd
|
|
tc_queue.appendleft((QueueCommands.WAIT, 5.0))
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW1_ID,
|
|
speed=-20000,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW3_ID,
|
|
speed=-20000,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 70.0))
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW1_ID,
|
|
speed=0,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW3_ID,
|
|
speed=0,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 10.0))
|
|
# RW2+4 speed cmd
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW2_ID,
|
|
speed=-20000,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW4_ID,
|
|
speed=-20000,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 70.0))
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW2_ID,
|
|
speed=0,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
command = pack_set_speed_command(
|
|
object_id=oids.RW4_ID,
|
|
speed=0,
|
|
ramp_time=10000,
|
|
)
|
|
tc_queue.appendleft(command.pack_command_tuple())
|
|
tc_queue.appendleft((QueueCommands.WAIT, 5.0))
|