eive/eive-tmtc
eive/eive-tmtc
14
0
Fork 0
Code Issues Pull Requests 2 Projects Releases 97 Wiki Activity

Merge remote-tracking branch 'origin/develop' into mueller/gps-hk-parsing

Browse Source
This commit is contained in:
Robin Müller 2022-05-27 14:32:58 +02:00
commit 0795e7c279
No known key found for this signature in database
GPG Key ID: 11D4952C8CCEF814
7 changed files with 211 additions and 165 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
pus_tc/devs/gps.py
View File

@ -12,6 +12,7 @@ from tmtccmd.tc.pus_8_funccmd import generate_action_command
LOGGER = get_console_logger()
class OpCodes:
REQ_OS_HK = ["0", "hk-os"]
RESET_GNSS = ["5", "reset"]

8
pus_tc/system/common.py
View File

@ -1,10 +1,16 @@
from typing import Union
from tmtccmd.tc.definitions import TcQueueT, QueueCommands
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes, Subservices
def command_mode(
object_id: bytes, mode: Modes, submode: int, tc_queue: TcQueueT, info: str
object_id: bytes,
mode: Union[int, Modes],
submode: int,
tc_queue: TcQueueT,
info: str,
):
tc_queue.appendleft((QueueCommands.PRINT, info))
mode_data = pack_mode_data(

40
pus_tc/system/controllers.py
View File

@ -1,5 +1,7 @@
from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.config import QueueCommands
from tmtccmd.tc.pus_200_fsfw_modes import Modes
from tmtccmd.utility import ObjectId
from .common import command_mode
import config.object_ids as obj_ids
@ -8,8 +10,8 @@ from pus_tc.prompt_parameters import prompt_parameters
class OpCodes:
THERMAL_CONTROLLER = [obj_ids.THERMAL_CONTROLLER_ID.hex(), "thermal_controller"]
CORE_CONTROLLER = [obj_ids.CORE_CONTROLLER_ID.hex(), "core_controller"]
THERMAL_CONTROLLER = [obj_ids.THERMAL_CONTROLLER_ID.hex(), "ctrl-th"]
CORE_CONTROLLER = [obj_ids.CORE_CONTROLLER_ID.hex(), "ctrl-core"]
class Info:
@ -17,7 +19,7 @@ class Info:
CORE_CONTROLLER = "ACS controller"
def pack_controller_commands(tc_queue: TcQueueT, op_code: str):
def pack_cmd_ctrl_to_prompted_mode(tc_queue: TcQueueT, object_id: ObjectId):
parameters = prompt_parameters(
[
{"name": "Mode", "defaultValue": "2"},
@ -30,21 +32,41 @@ def pack_controller_commands(tc_queue: TcQueueT, op_code: str):
mode = 0
submode = int(parameters["Submode"])
command_mode(
object_id=get_object_from_op_code(op_code),
object_id=object_id.as_bytes,
mode=mode,
submode=submode,
tc_queue=tc_queue,
info=op_code + " to " + str(mode) + "," + str(submode),
info=f"Commanding {object_id} to {mode}, {submode}",
)
tc_queue.appendleft((QueueCommands.WAIT, 20))
def pack_cmd_ctrl_to_off(tc_queue: TcQueueT, object_id: ObjectId):
command_mode(
object_id=get_object_from_op_code(op_code),
mode=0,
object_id=object_id.as_bytes,
mode=Modes.OFF,
submode=0,
tc_queue=tc_queue,
info=op_code + " to 0,0",
info=f"Commanding {object_id} OFF",
)
def pack_cmd_ctrl_to_on(tc_queue: TcQueueT, object_id: ObjectId):
command_mode(
object_id=object_id.as_bytes,
mode=Modes.ON,
submode=0,
tc_queue=tc_queue,
info=f"Commanding {object_id} ON",
)
def pack_cmd_ctrl_to_nml(tc_queue: TcQueueT, object_id: ObjectId):
command_mode(
object_id=object_id.as_bytes,
mode=Modes.NORMAL,
submode=0,
tc_queue=tc_queue,
info=f"Commanding {object_id} NORMAL",
)

301
pus_tc/system/proc.py
View File

@ -1,6 +1,11 @@
from __future__ import annotations
import time
from typing import List
from config.definitions import CustomServiceList
from config.object_ids import get_object_ids
from pus_tc.system.tcs import pack_tcs_sys_commands
from tmtccmd.config import (
QueueCommands,
ServiceOpCodeDictT,
@ -8,25 +13,29 @@ from tmtccmd.config import (
add_service_op_code_entry,
)
from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.tc.pus_11_tc_sched import (
generate_time_tagged_cmd,
generate_enable_tc_sched_cmd,
generate_reset_tc_sched_cmd,
)
from tmtccmd.tc.pus_3_fsfw_hk import *
import config.object_ids as oids
from pus_tc.system.tcs import OpCodes as TcsOpCodes
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.mgms import MgmLis3SetIds as MgmLis3SetIds_0_2
from pus_tc.devs.mgms import MgmRm3100SetIds as MgmRm3100SetIds_1_3
from pus_tc.devs.gyros import AdisGyroSetIds as AdisGyroSetIds_0_2
from pus_tc.devs.gyros import L3gGyroSetIds as L3gGyroSetIds_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.controllers import pack_cmd_ctrl_to_off, pack_cmd_ctrl_to_nml
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
@ -34,13 +43,13 @@ from pus_tc.devs.reaction_wheels import pack_rw_ass_cmds, pack_set_speed_command
class OpCodes:
HEATER = ["0", "heater"]
TV_SETUP_TCS_FT_ON = ["setup", "tcs-ft-on"]
RESET_SCHED = ["reset-sched", "rs"]
HEATER = ["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"]
@ -48,16 +57,17 @@ class OpCodes:
STR_FT = ["str-ft"]
RW_FT_ONE_RW = ["rw-ft-one-rw"]
RW_FT_TWO_RWS = ["rw-ft-two-rws"]
TV_TEARDOWN_TCS_FT_OFF = ["teardown", "tcs-ft-off"]
class KeyAndInfo:
TV_SETUP_TCS_FT_ON = ["TCS Act.", "TCS functional test activation"]
RESET_SCHED = ["Reset Sched", "Reset/Clear TC schedule"]
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"]
@ -65,6 +75,7 @@ class KeyAndInfo:
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"]
TV_TEARDOWN_TCS_FT_OFF = ["TCS Deact.", "TCS functional test deactivation"]
KAI = KeyAndInfo
@ -74,21 +85,41 @@ PROC_INFO_DICT = {
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.TV_SETUP_TCS_FT_ON[0]: [OpCodes.TV_SETUP_TCS_FT_ON, KAI.TV_SETUP_TCS_FT_ON[1], 120.0, 10.0],
KAI.TV_TEARDOWN_TCS_FT_OFF[0]: [
OpCodes.TV_TEARDOWN_TCS_FT_OFF,
KAI.TV_TEARDOWN_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
# collection_time for KAI.MGT_FT_DP maybe be reduced as a full 120
# seconds 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
# collection_time for KAI.RW_FT_ONE_RW maybe be reduced as a full 120
# seconds 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
# collection_time for KAI.RW_FT_ONE_RW maybe be reduced as a full 120
# seconds 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],
}
class GenericHkListeningCfg:
def __init__(self, mgt: bool = False, one_rw: bool = False, two_rws: bool = False):
self.use_tc_sched = False
self.mgt = mgt
self.one_rw = one_rw
self.two_rws = two_rws
@classmethod
def default(cls) -> GenericHkListeningCfg:
return GenericHkListeningCfg(False, False, False)
def generic_print(tc_queue: TcQueueT, info: dict):
tc_queue.appendleft(
(QueueCommands.PRINT, f"Executing {info[1]} Procedure (OpCodes: {info[0]})")
@ -96,7 +127,6 @@ def generic_print(tc_queue: TcQueueT, info: dict):
def add_proc_cmds(cmd_dict: ServiceOpCodeDictT):
op_code_dict = dict()
for proc_entry in PROC_INFO_DICT.values():
add_op_code_entry(
@ -115,9 +145,7 @@ def pack_generic_hk_listening_cmds(
proc_key: str,
sid_list: list[bytearray],
diag: bool,
mgt: bool,
one_rw: bool,
two_rws: bool,
cfg: GenericHkListeningCfg,
):
info = PROC_INFO_DICT[proc_key]
collection_time = info[2]
@ -131,37 +159,58 @@ def pack_generic_hk_listening_cmds(
sid=sid,
interval_seconds=info[3],
)
if not cfg.use_tc_sched:
tc_queue.appendleft((QueueCommands.WAIT, 2.0))
if mgt is True:
if cfg.mgt:
activate_mgts_alternately(
tc_queue=tc_queue,
)
elif one_rw is True:
elif cfg.one_rw:
activate_all_rws_in_sequence(
tc_queue=tc_queue, test_speed=20000, test_ramp_time=10000, init_ssc=0
)
elif two_rws is True:
elif cfg.two_rws:
activate_all_rws_two_consecutively(tc_queue=tc_queue, init_ssc=0)
else:
pass
if not cfg.use_tc_sched:
tc_queue.appendleft((QueueCommands.WAIT, collection_time))
disable_cmd_list = []
for sid in sid_list:
disable_listen_to_hk_for_x_seconds(
disable_cmd_list.append(
gen_disable_listen_to_hk_for_x_seconds(
diag=diag,
tc_queue=tc_queue,
device=proc_key,
sid=sid,
)
)
current_time = time.time()
current_time += collection_time
if not cfg.use_tc_sched:
for cmd in disable_cmd_list:
tc_queue.appendleft(cmd.pack_command_tuple())
else:
for cmd in disable_cmd_list:
tc_queue.appendleft(
generate_time_tagged_cmd(
release_time=struct.pack("!I", current_time),
tc_to_insert=cmd,
ssc=0,
)
)
sid_list.clear()
def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
sid_list = []
obj_id_dict = get_object_ids()
if op_code in OpCodes.RESET_SCHED:
tc_queue.appendleft((QueueCommands.PRINT, "Resetting/Clearing TC schedule"))
tc_queue.appendleft(generate_reset_tc_sched_cmd().pack_command_tuple())
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))
@ -170,9 +219,7 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
proc_key=key,
sid_list=sid_list,
diag=False,
mgt=False,
one_rw=False,
two_rws=False,
cfg=GenericHkListeningCfg.default(),
)
if op_code in OpCodes.CORE_FT:
@ -183,9 +230,7 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
proc_key=key,
sid_list=sid_list,
diag=False,
mgt=False,
one_rw=False,
two_rws=False,
cfg=GenericHkListeningCfg.default(),
)
if op_code in OpCodes.PCDU_FT:
@ -203,9 +248,7 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
proc_key=key,
sid_list=sid_list,
diag=False,
mgt=False,
one_rw=False,
two_rws=False,
cfg=GenericHkListeningCfg.default(),
)
if op_code in OpCodes.RAD_SEN_FT:
@ -216,56 +259,64 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
proc_key=key,
sid_list=sid_list,
diag=False,
mgt=False,
one_rw=False,
two_rws=False,
cfg=GenericHkListeningCfg.default(),
)
if op_code in OpCodes.TCS_FT_ON:
if op_code in OpCodes.TV_SETUP_TCS_FT_ON:
# Enable scheduling
tc_queue.appendleft(generate_enable_tc_sched_cmd().pack_command_tuple())
# 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")
pack_tcs_sys_commands(
tc_queue=tc_queue, op_code=TcsOpCodes.TCS_BOARD_ASS_NORMAL[0]
)
pack_cmd_ctrl_to_nml(
tc_queue=tc_queue, object_id=obj_id_dict.get(oids.THERMAL_CONTROLLER_ID)
)
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.TV_TEARDOWN_TCS_FT_OFF:
# TCS board shold always be on anyway, do not command it off here
pack_cmd_ctrl_to_off(
tc_queue=tc_queue, object_id=obj_id_dict.get(oids.THERMAL_CONTROLLER_ID)
)
if op_code in OpCodes.ACS_FT:
key = KAI.ACS_FT[0]
a_side_pairs = [
(oids.MGM_0_LIS3_HANDLER_ID, MgmLis3SetIds_0_2.CORE_HK),
(oids.MGM_1_RM3100_HANDLER_ID, MgmRm3100SetIds_1_3.CORE_HK),
(oids.GYRO_0_ADIS_HANDLER_ID, AdisGyroSetIds_0_2.CORE_HK),
(oids.GYRO_1_L3G_HANDLER_ID, L3gGyroSetIds_1_3.CORE_HK),
(oids.GPS_CONTROLLER, GpsSetIds.HK),
]
b_side_pairs = [
(oids.MGM_2_LIS3_HANDLER_ID, MgmLis3SetIds_0_2.CORE_HK),
(oids.MGM_3_RM3100_HANDLER_ID, MgmRm3100SetIds_1_3.CORE_HK),
(oids.GYRO_2_ADIS_HANDLER_ID, AdisGyroSetIds_0_2.CORE_HK),
(oids.GYRO_3_L3G_HANDLER_ID, L3gGyroSetIds_1_3.CORE_HK),
(oids.GPS_CONTROLLER, GpsSetIds.HK),
]
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))
for a_side_dev in a_side_pairs:
oid = a_side_dev[0]
set_id = a_side_dev[1]
sid_list.append(make_sid(oid, set_id))
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,
cfg=GenericHkListeningCfg.default(),
)
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))
for b_side_dev in b_side_pairs:
oid = b_side_dev[0]
set_id = b_side_dev[1]
sid_list.append(make_sid(oid, set_id))
# GNSS1
sid_list.append(make_sid(oids.GPS_HANDLER_1_ID, GpsSetIds.HK))
pack_generic_hk_listening_cmds(
@ -273,9 +324,7 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
proc_key=key,
sid_list=sid_list,
diag=False,
mgt=False,
one_rw=False,
two_rws=False,
cfg=GenericHkListeningCfg.default(),
)
pack_acs_command(tc_queue=tc_queue, op_code="acs-off")
@ -283,28 +332,38 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
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))
sid_list.append(make_sid(oids.MGM_0_LIS3_HANDLER_ID, MgmLis3SetIds_0_2.CORE_HK))
sid_list.append(
make_sid(oids.MGM_1_RM3100_HANDLER_ID, MgmRm3100SetIds_1_3.CORE_HK)
)
sid_list.append(make_sid(oids.MGM_2_LIS3_HANDLER_ID, MgmLis3SetIds_0_2.CORE_HK))
sid_list.append(
make_sid(oids.MGM_3_RM3100_HANDLER_ID, MgmRm3100SetIds_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))
sid_list.append(
make_sid(oids.GYRO_0_ADIS_HANDLER_ID, AdisGyroSetIds_0_2.CORE_HK)
)
sid_list.append(
make_sid(oids.GYRO_0_ADIS_HANDLER_ID, AdisGyroSetIds_0_2.CFG_HK)
)
sid_list.append(make_sid(oids.GYRO_1_L3G_HANDLER_ID, L3gGyroSetIds_1_3.CORE_HK))
sid_list.append(
make_sid(oids.GYRO_2_ADIS_HANDLER_ID, AdisGyroSetIds_0_2.CORE_HK)
)
sid_list.append(
make_sid(oids.GYRO_2_ADIS_HANDLER_ID, AdisGyroSetIds_0_2.CFG_HK)
)
sid_list.append(make_sid(oids.GYRO_3_L3G_HANDLER_ID, L3gGyroSetIds_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_CONTROLLER, 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,
cfg=GenericHkListeningCfg.default(),
)
pack_acs_command(tc_queue=tc_queue, op_code="acs-off")
@ -318,14 +377,10 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
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))
cfg = GenericHkListeningCfg.default()
cfg.mgt = True
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,
tc_queue=tc_queue, proc_key=key, sid_list=sid_list, diag=False, cfg=cfg
)
pack_imtq_test_into(oids.IMTQ_HANDLER_ID, tc_queue=tc_queue, op_code="0")
@ -347,9 +402,7 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
proc_key=key,
sid_list=sid_list,
diag=False,
mgt=True,
one_rw=False,
two_rws=False,
cfg=GenericHkListeningCfg.default(),
)
pack_imtq_test_into(oids.IMTQ_HANDLER_ID, tc_queue=tc_queue, op_code="0")
@ -384,9 +437,7 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
proc_key=key,
sid_list=sid_list,
diag=False,
mgt=False,
one_rw=False,
two_rws=False,
cfg=GenericHkListeningCfg.default(),
)
pack_acs_command(tc_queue=tc_queue, op_code="sus-off")
@ -401,9 +452,7 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
proc_key=key,
sid_list=sid_list,
diag=False,
mgt=False,
one_rw=False,
two_rws=False,
cfg=GenericHkListeningCfg.default(),
)
pack_acs_command(tc_queue=tc_queue, op_code="sus-off")
@ -420,9 +469,7 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
proc_key=key,
sid_list=sid_list,
diag=False,
mgt=False,
one_rw=False,
two_rws=False,
cfg=GenericHkListeningCfg.default(),
)
pack_acs_command(tc_queue=tc_queue, op_code="sus-off")
@ -441,9 +488,7 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
proc_key=key,
sid_list=sid_list,
diag=False,
mgt=False,
one_rw=False,
two_rws=False,
cfg=GenericHkListeningCfg.default(),
)
pack_star_tracker_commands(
@ -478,9 +523,7 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
proc_key=key,
sid_list=sid_list,
diag=False,
mgt=False,
one_rw=True,
two_rws=False,
cfg=GenericHkListeningCfg(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")
@ -513,40 +556,12 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
proc_key=key,
sid_list=sid_list,
diag=False,
mgt=False,
one_rw=False,
two_rws=True,
cfg=GenericHkListeningCfg(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,
@ -561,19 +576,15 @@ def enable_listen_to_hk_for_x_seconds(
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(
def gen_disable_listen_to_hk_for_x_seconds(
tc_queue: TcQueueT,
diag: bool,
device: str,
sid: bytes,
):
) -> PusTelecommand:
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()
)
return disable_periodic_hk_command(diag=diag, sid=sid, ssc=0)
def activate_mgts_alternately(

8
pus_tc/tc_packer_hook.py
View File

@ -42,7 +42,7 @@ from pus_tc.devs.plpcdu import pack_pl_pcdu_commands
from pus_tc.devs.str_img_helper import pack_str_img_helper_command
from pus_tc.system.tcs import pack_tcs_sys_commands
from pus_tc.system.proc import pack_proc_commands
from pus_tc.system.controllers import pack_controller_commands
from pus_tc.system.controllers import pack_cmd_ctrl_to_prompted_mode
from config.definitions import CustomServiceList
from config.object_ids import (
get_object_ids,
@ -63,6 +63,7 @@ from config.object_ids import (
PLOC_SUPV_ID,
STAR_TRACKER_ID,
PLOC_MEMORY_DUMPER_ID,
GPS_CONTROLLER,
CCSDS_HANDLER_ID,
PDEC_HANDLER_ID,
STR_IMG_HELPER_ID,
@ -241,8 +242,13 @@ def pack_service_queue_user(
tc_queue=service_queue, object_id=RW_ASSEMBLY, op_code=op_code
)
if service == CustomServiceList.CONTROLLERS.value:
<<<<<<< HEAD
return pack_controller_commands(tc_queue=service_queue, op_code=op_code)
LOGGER.warning(f"Invalid Service {service}")
=======
return pack_cmd_ctrl_to_prompted_mode(tc_queue=service_queue, op_code=op_code)
LOGGER.warning("Invalid Service !")
>>>>>>> origin/develop
def create_total_tc_queue_user() -> TcQueueT:

6
pus_tm/hk_handling.py
View File

@ -98,9 +98,9 @@ def handle_regular_hk_print(
LOGGER.info("Service 3 TM: IMTQ handler reply with unknown set id")
elif objb == obj_ids.GPS_CONTROLLER:
return handle_gps_data(printer=printer, hk_data=hk_data)
if objb == obj_ids.BPX_HANDLER_ID:
elif objb == obj_ids.BPX_HANDLER_ID:
handle_bpx_hk_data(hk_data=hk_data, set_id=set_id, printer=printer)
if objb == obj_ids.CORE_CONTROLLER_ID:
elif objb == obj_ids.CORE_CONTROLLER_ID:
return handle_core_hk_data(printer=printer, hk_data=hk_data, set_id=set_id)
elif objb == obj_ids.PDU_1_HANDLER_ID:
return handle_pdu_data(
@ -135,7 +135,7 @@ def handle_regular_hk_print(
handle_sus_hk(
object_id=object_id, hk_data=hk_data, printer=printer, set_id=set_id
)
if objb == obj_ids.P60_DOCK_HANDLER:
elif objb == obj_ids.P60_DOCK_HANDLER:
handle_p60_hk_data(printer=printer, set_id=set_id, hk_data=hk_data)
elif objb in [
obj_ids.GYRO_0_ADIS_HANDLER_ID,

2
tmtccmd

@ -1 +1 @@
Subproject commit 2354f5d2778c7681ddf6602766dd4e68b943e1c0
Subproject commit eaed332a69885c19912827eb71440e8562935d36