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/main' into mueller/add-acs-ss-cmds

Browse Source
This commit is contained in:
Robin Müller 2022-10-20 17:03:41 +02:00
commit 90c9cf297b
44 changed files with 1406 additions and 1070 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.run/IMTQ_Test_UDP.run.xml → .run/IMTQ Dipole.run.xml
View File

@ -1,5 +1,5 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="IMTQ Test UDP" type="PythonConfigurationType" factoryName="Python" folderName="Devices">
<configuration default="false" name="IMTQ Dipole" type="PythonConfigurationType" factoryName="Python" folderName="IMTQ">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
@ -13,7 +13,7 @@
<option name="ADD_SOURCE_ROOTS" value="true" />
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtcc.py" />
<option name="PARAMETERS" value="-m seqcmd -c udp -s imtq -l -d 4" />
<option name="PARAMETERS" value="-s imtq -o set_dipole -l" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="true" />
<option name="MODULE_MODE" value="false" />

2
.run/IMTQ_Get_Commanded_Dipole.run.xml
View File

@ -1,5 +1,5 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="IMTQ Get Commanded Dipole" type="PythonConfigurationType" factoryName="Python" folderName="Devices">
<configuration default="false" name="IMTQ Get Commanded Dipole" type="PythonConfigurationType" factoryName="Python" folderName="IMTQ">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />

2
.run/IMTQ_Neg_X_Test.run.xml
View File

@ -1,5 +1,5 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="IMTQ Neg X Test" type="PythonConfigurationType" factoryName="Python" folderName="Devices">
<configuration default="false" name="IMTQ Neg X Test" type="PythonConfigurationType" factoryName="Python" folderName="IMTQ">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />

2
.run/IMTQ_Neg_Y_Test.run.xml
View File

@ -1,5 +1,5 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="IMTQ Neg Y Test" type="PythonConfigurationType" factoryName="Python" folderName="Devices">
<configuration default="false" name="IMTQ Neg Y Test" type="PythonConfigurationType" factoryName="Python" folderName="IMTQ">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />

2
.run/IMTQ_Neg_Z_Test.run.xml
View File

@ -1,5 +1,5 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="IMTQ Neg Z Test" type="PythonConfigurationType" factoryName="Python" folderName="Devices">
<configuration default="false" name="IMTQ Neg Z Test" type="PythonConfigurationType" factoryName="Python" folderName="IMTQ">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />

2
.run/IMTQ_Pos_X_Test.run.xml
View File

@ -1,5 +1,5 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="IMTQ Pos X Test" type="PythonConfigurationType" factoryName="Python" folderName="Devices">
<configuration default="false" name="IMTQ Pos X Test" type="PythonConfigurationType" factoryName="Python" folderName="IMTQ">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />

2
.run/IMTQ_Pos_Y_Test.run.xml
View File

@ -1,5 +1,5 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="IMTQ Pos Y Test" type="PythonConfigurationType" factoryName="Python" folderName="Devices">
<configuration default="false" name="IMTQ Pos Y Test" type="PythonConfigurationType" factoryName="Python" folderName="IMTQ">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />

2
.run/IMTQ_Pos_Z_Test.run.xml
View File

@ -1,5 +1,5 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="IMTQ Pos Z Test" type="PythonConfigurationType" factoryName="Python" folderName="Devices">
<configuration default="false" name="IMTQ Pos Z Test" type="PythonConfigurationType" factoryName="Python" folderName="IMTQ">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />

2
.run/P60_Dock_Commanding.run.xml
View File

@ -1,5 +1,5 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="P60 Dock Commanding" type="PythonConfigurationType" factoryName="Python" folderName="Power">
<configuration default="false" name="P60 Dock" type="PythonConfigurationType" factoryName="Python" folderName="Power">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />

2
.run/PDU1 Commanding.run.xml → .run/PDU1.run.xml
View File

@ -1,5 +1,5 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="PDU1 Commanding" type="PythonConfigurationType" factoryName="Python" folderName="Power">
<configuration default="false" name="PDU1" type="PythonConfigurationType" factoryName="Python" folderName="Power">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />

2
.run/PDU2 Commanding.run.xml → .run/PDU2.run.xml
View File

@ -1,5 +1,5 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="PDU2 Commanding" type="PythonConfigurationType" factoryName="Python" folderName="Power">
<configuration default="false" name="PDU2" type="PythonConfigurationType" factoryName="Python" folderName="Power">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />

4
.run/_IMTQ_Command_Dipole.run.xml → .run/RW All Off.run.xml
View File

@ -1,5 +1,5 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name=" IMTQ Command Dipole" type="PythonConfigurationType" factoryName="Python" folderName="Devices">
<configuration default="false" name="RW All Off" type="PythonConfigurationType" factoryName="Python" folderName="ACS">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
@ -13,7 +13,7 @@
<option name="ADD_SOURCE_ROOTS" value="true" />
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtcc.py" />
<option name="PARAMETERS" value="-c udp -s imtq -o 7 -l -d 4 --hk" />
<option name="PARAMETERS" value="-s rw_ass -o speed_off -d 0.1 -l" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="true" />
<option name="MODULE_MODE" value="false" />

24
.run/RW All On.run.xml Normal file
View File

@ -0,0 +1,24 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="RW All On" type="PythonConfigurationType" factoryName="Python" folderName="ACS">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />
</envs>
<option name="SDK_HOME" value="" />
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
<option name="IS_MODULE_SDK" value="true" />
<option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" />
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtcc.py" />
<option name="PARAMETERS" value="-s rw_ass -o speed_up -d 0.1 -l" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="true" />
<option name="MODULE_MODE" value="false" />
<option name="REDIRECT_INPUT" value="false" />
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
</component>

24
.run/RW1.run.xml Normal file
View File

@ -0,0 +1,24 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="RW1" type="PythonConfigurationType" factoryName="Python" folderName="ACS">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />
</envs>
<option name="SDK_HOME" value="" />
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
<option name="IS_MODULE_SDK" value="true" />
<option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" />
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtcc.py" />
<option name="PARAMETERS" value="-s rw_1 -l" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="true" />
<option name="MODULE_MODE" value="false" />
<option name="REDIRECT_INPUT" value="false" />
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
</component>

24
.run/RW2.run.xml Normal file
View File

@ -0,0 +1,24 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="RW2" type="PythonConfigurationType" factoryName="Python" folderName="ACS">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />
</envs>
<option name="SDK_HOME" value="" />
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
<option name="IS_MODULE_SDK" value="true" />
<option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" />
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtcc.py" />
<option name="PARAMETERS" value="-s rw_2 -l" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="true" />
<option name="MODULE_MODE" value="false" />
<option name="REDIRECT_INPUT" value="false" />
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
</component>

24
.run/RW3.run.xml Normal file
View File

@ -0,0 +1,24 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="RW3" type="PythonConfigurationType" factoryName="Python" folderName="ACS">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />
</envs>
<option name="SDK_HOME" value="" />
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
<option name="IS_MODULE_SDK" value="true" />
<option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" />
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtcc.py" />
<option name="PARAMETERS" value="-s rw_3 -l" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="true" />
<option name="MODULE_MODE" value="false" />
<option name="REDIRECT_INPUT" value="false" />
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
</component>

24
.run/RW4.run.xml Normal file
View File

@ -0,0 +1,24 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="RW4" type="PythonConfigurationType" factoryName="Python" folderName="ACS">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />
</envs>
<option name="SDK_HOME" value="" />
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
<option name="IS_MODULE_SDK" value="true" />
<option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" />
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtcc.py" />
<option name="PARAMETERS" value="-s rw_4 -l" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="true" />
<option name="MODULE_MODE" value="false" />
<option name="REDIRECT_INPUT" value="false" />
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
</component>

24
.run/SA Deployment.run.xml Normal file
View File

@ -0,0 +1,24 @@
<component name="ProjectRunConfigurationManager">
<configuration default="false" name="SA Deployment" type="PythonConfigurationType" factoryName="Python" folderName="Core">
<module name="tmtc" />
<option name="INTERPRETER_OPTIONS" value="" />
<option name="PARENT_ENVS" value="true" />
<envs>
<env name="PYTHONUNBUFFERED" value="1" />
</envs>
<option name="SDK_HOME" value="" />
<option name="WORKING_DIRECTORY" value="$PROJECT_DIR$" />
<option name="IS_MODULE_SDK" value="true" />
<option name="ADD_CONTENT_ROOTS" value="true" />
<option name="ADD_SOURCE_ROOTS" value="true" />
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtcc.py" />
<option name="PARAMETERS" value="-s sa_depl -l" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="true" />
<option name="MODULE_MODE" value="false" />
<option name="REDIRECT_INPUT" value="false" />
<option name="INPUT_FILE" value="" />
<method v="2" />
</configuration>
</component>

6
CHANGELOG.md
View File

@ -10,7 +10,11 @@ list yields a list of all related PRs for each release.
# [unreleased]
# [v1.13.0]
- Bugfixes for IMTQ TM handling
- Updates S/A deployment command
PR: https://egit.irs.uni-stuttgart.de/eive/eive-tmtc/pulls/118
# [v1.13.0] 13.10.2022
- CFDP integration
PR: https://egit.irs.uni-stuttgart.de/eive/eive-tmtc/pulls/113

4
config/__init__.py
View File

@ -1,6 +1,6 @@
SW_NAME = "eive"
VERSION_MAJOR = 1
VERSION_MINOR = 11
VERSION_MINOR = 13
VERSION_SUBMINOR = 0
__version__ = "1.11.0"
__version__ = "1.13.0"

2
deps/tmtccmd vendored

@ -1 +1 @@
Subproject commit a3dcf7125c04440c4cb7612051ebbbd2b7c6b800
Subproject commit 229a2b5032b0c1becd93fe37f34ddc93a6b425c4

40
gomspace/gomspace_common.py
View File

@ -31,46 +31,6 @@ class GomspaceDeviceActionIds(enum.IntEnum):
PRINT_LATCHUPS = 33
class GomspaceOpCodes:
# Request HK
REQUEST_CORE_HK_ONCE = ["hk_core"]
REQUEST_AUX_HK_ONCE = ["hk_aux"]
PRINT_SWITCH_V_I = ["print_switch_vi"]
PRINT_LATCHUPS = ["print_latchups"]
GET_PARAM = ["get_param"]
SET_INTEGER_PARAM = ["set_int_param"]
SAVE_TABLE = ["save_table"]
RESET_GND_WATCHDOG = ["reset_gnd_wdt"]
SAVE_TABLE_DEFAULT = ["save_table_default"]
LOAD_TABLE = ["load_table"]
REQUEST_CONFIG_TABLE = ["cfg_table"]
class GsInfo:
REQUEST_CORE_HK_ONCE = "Requesting Core HK once"
REQUEST_AUX_HK_ONCE = "Requesting Aux HK once"
PRINT_SWITCH_V_I = "Print Switch V I Info"
PRINT_LATCHUPS = "Print latchups"
GET_PARAMETER = "Get parameter"
SET_PARAMETER = "Set integer parameter"
REQUEST_CONFIG_TABLE = "Request Config Table"
RESET_GND_WATCHDOG = "Reset GND watchdog"
SAVE_TABLE = "Save table non-volatile (file)"
SAVE_TABLE_DEFAULT = "Save table non-volatile (default)"
LOAD_TABLE = "Load Table"
class SetIds:
PDU_1_CORE = 1
PDU_1_AUX = 2
PDU_2_CORE = 3
PDU_2_AUX = 4
P60_CORE = 5
P60_AUX = 6
ACU_CORE = 7
ACU_AUX = 8
class ParamTypes(enum.Enum):
U8 = 0
U16 = 1

20
pus_tc/cmd_definitions.py
View File

@ -185,26 +185,6 @@ def add_time_cmds(defs: TmtcDefinitionWrapper):
)
@tmtc_definitions_provider
def add_imtq_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
oce.add("0", "Mode Off")
oce.add("1", "Mode On")
oce.add("2", "Mode Normal")
oce.add("3", "IMTQ perform pos X self test")
oce.add("4", "IMTQ perform neg X self test")
oce.add("5", "IMTQ perform pos Y self test")
oce.add("6", "IMTQ perform neg Y self test")
oce.add("7", "IMTQ perform pos Z self test")
oce.add("8", "IMTQ perform neg Z self test")
oce.add("9", "IMTQ command dipole")
oce.add("10", "IMTQ get commanded dipole")
oce.add("11", "IMTQ get engineering hk set")
oce.add("12", "IMTQ get calibrated MTM measurement one shot")
oce.add("13", "IMTQ get raw MTM measurement one shot")
defs.add_service(CustomServiceList.IMTQ.value, "IMQT Device", oce)
@tmtc_definitions_provider
def add_system_cmds(defs: TmtcDefinitionWrapper):
import pus_tc.system.controllers as controllers

202
pus_tc/devs/imtq.py
View File

@ -1,202 +0,0 @@
# -*- coding: utf-8 -*-
"""
@file imtq.py
@brief Tests for the ISIS IMTQ (Magnettorquer) device handler
@author J. Meier
@date 25.03.2021
"""
import struct
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import (
make_sid,
generate_one_diag_command,
generate_one_hk_command,
)
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes
from tmtccmd.util import ObjectIdU32
class ImtqSetIds:
ENG_HK_SET = 1
CAL_MTM_SET = 2
RAW_MTM_SET = 3
POSITIVE_X_TEST = 4
NEGATIVE_X_TEST = 5
POSITIVE_Y_TEST = 6
NEGATIVE_Y_TEST = 7
POSITIVE_Z_TEST = 8
NEGATIVE_Z_TEST = 9
class ImtqActionIds:
start_actuation_dipole = bytearray([0x0, 0x0, 0x0, 0x02])
get_commanded_dipole = bytearray([0x0, 0x0, 0x0, 0x03])
perform_positive_x_test = bytearray([0x0, 0x0, 0x0, 0x07])
perform_negative_x_test = bytearray([0x0, 0x0, 0x0, 0x08])
perform_positive_y_test = bytearray([0x0, 0x0, 0x0, 0x09])
perform_negative_y_test = bytearray([0x0, 0x0, 0x0, 0x0A])
perform_positive_z_test = bytearray([0x0, 0x0, 0x0, 0x0B])
perform_negative_z_test = bytearray([0x0, 0x0, 0x0, 0x0C])
# Initiates the reading of the last performed self test. After sending this command the results
# can be downlinked via the housekeeping service by using the appropriate set ids listed above.
read_self_test_results = bytearray([0x0, 0x0, 0x0, 0x0D])
def pack_imtq_test_into(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code: str):
q.add_log_cmd(
f"Testing ISIS IMTQ handler with object id: {object_id.as_hex_string}"
)
if op_code == "0":
q.add_log_cmd("IMTQ: Set mode off")
command = pack_mode_data(object_id.as_bytes, Modes.OFF, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code == "1":
q.add_log_cmd("IMTQ: Set mode on")
command = pack_mode_data(object_id.as_bytes, Modes.ON, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code == "2":
q.add_log_cmd("IMTQ: Mode Normal")
command = pack_mode_data(object_id.as_bytes, Modes.NORMAL, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code == "3":
q.add_log_cmd("IMTQ: Perform positive x self test")
command = object_id.as_bytes + ImtqActionIds.perform_positive_x_test
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Initiate reading of positive x self test results")
command = object_id.as_bytes + ImtqActionIds.read_self_test_results
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Request dataset with positive x self test results")
sid = make_sid(object_id.as_bytes, ImtqSetIds.POSITIVE_X_TEST)
q.add_pus_tc(generate_one_hk_command(sid))
if op_code == "4":
q.add_log_cmd("IMTQ: Perform negative x self test")
command = object_id.as_bytes + ImtqActionIds.perform_negative_x_test
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Initiate reading of negative x self test results")
command = object_id.as_bytes + ImtqActionIds.read_self_test_results
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Request dataset with negative x self test results")
sid = make_sid(object_id.as_bytes, ImtqSetIds.NEGATIVE_X_TEST)
q.add_pus_tc(generate_one_hk_command(sid))
if op_code == "5":
q.add_log_cmd("IMTQ: Perform positive y self test")
command = object_id.as_bytes + ImtqActionIds.perform_positive_y_test
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Initiate reading of positive y self test results")
command = object_id.as_bytes + ImtqActionIds.read_self_test_results
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Request dataset with positive y self test results")
sid = make_sid(object_id.as_bytes, ImtqSetIds.POSITIVE_Y_TEST)
q.add_pus_tc(generate_one_hk_command(sid))
if op_code == "6":
q.add_log_cmd("IMTQ: Perform negative y self test")
command = object_id.as_bytes + ImtqActionIds.perform_negative_y_test
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Initiate reading of negative y self test results")
command = object_id.as_bytes + ImtqActionIds.read_self_test_results
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Request dataset with negative y self test results")
sid = make_sid(object_id.as_bytes, ImtqSetIds.NEGATIVE_Y_TEST)
q.add_pus_tc(generate_one_hk_command(sid))
if op_code == "7":
q.add_log_cmd("IMTQ: Perform positive z self test")
command = object_id.as_bytes + ImtqActionIds.perform_positive_z_test
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Initiate reading of positive z self test results")
command = object_id.as_bytes + ImtqActionIds.read_self_test_results
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Request dataset with positive z self test results")
sid = make_sid(object_id.as_bytes, ImtqSetIds.POSITIVE_Y_TEST)
q.add_pus_tc(generate_one_hk_command(sid))
if op_code == "8":
q.add_log_cmd("IMTQ: Perform negative z self test")
command = object_id.as_bytes + ImtqActionIds.perform_negative_z_test
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Initiate reading of negative z self test results")
command = object_id.as_bytes + ImtqActionIds.read_self_test_results
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Request dataset with negative z self test results")
sid = make_sid(object_id.as_bytes, ImtqSetIds.NEGATIVE_Z_TEST)
q.add_pus_tc(generate_one_hk_command(sid))
if op_code == "9":
q.add_log_cmd("IMTQ: Commanding dipole")
x_dipole = 0
y_dipole = 0
z_dipole = 0
duration = 0 # ms
q.add_pus_tc(
pack_dipole_command(
object_id.as_bytes, x_dipole, y_dipole, z_dipole, duration
)
)
if op_code == "10":
q.add_log_cmd("IMTQ: Get commanded dipole")
command = object_id.as_bytes + ImtqActionIds.get_commanded_dipole
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "11":
q.add_log_cmd("IMTQ: Get engineering hk set")
q.add_pus_tc(
generate_one_diag_command(
sid=make_sid(object_id=object_id.as_bytes, set_id=ImtqSetIds.ENG_HK_SET)
)
)
if op_code == "12":
q.add_log_cmd("IMTQ: Get calibrated MTM hk set")
q.add_pus_tc(
generate_one_diag_command(
sid=make_sid(
object_id=object_id.as_bytes, set_id=ImtqSetIds.CAL_MTM_SET
)
)
)
if op_code == "13":
q.add_log_cmd("IMTQ: Get raw MTM hk set")
q.add_pus_tc(
generate_one_diag_command(
sid=make_sid(
object_id=object_id.as_bytes, set_id=ImtqSetIds.RAW_MTM_SET
)
)
)
def pack_dipole_command(
object_id: bytes, x_dipole: int, y_dipole: int, z_dipole: int, duration: int
) -> PusTelecommand:
"""This function packs the command causing the ISIS IMTQ to generate a dipole.
@param object_id The object id of the IMTQ handler.
@param x_dipole The dipole of the x coil in 10^-4*Am^2 (max. 2000)
@param y_dipole The dipole of the y coil in 10^-4*Am^2 (max. 2000)
@param z_dipole The dipole of the z coil in 10^-4*Am^2 (max. 2000)
@param duration The duration in milliseconds the dipole will be generated by the coils.
When set to 0, the dipole will be generated until a new dipole actuation
command is sent.
"""
action_id = ImtqActionIds.start_actuation_dipole
command = object_id + action_id
command += struct.pack("!h", x_dipole)
command += struct.pack("!h", y_dipole)
command += struct.pack("!h", z_dipole)
command += struct.pack("!h", duration)
command = PusTelecommand(service=8, subservice=128, app_data=command)
return command

59
pus_tc/devs/plpcdu.py
View File

@ -11,6 +11,8 @@ from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import (
make_sid,
generate_one_diag_command,
enable_periodic_hk_command_with_interval,
disable_periodic_hk_command,
)
from tmtccmd.tc.pus_11_tc_sched import (
generate_enable_tc_sched_cmd,
@ -30,31 +32,32 @@ LOGGER = get_console_logger()
class OpCodes:
SWITCH_HPA_ON_PROC = ["0", "proc-hpa"]
SWITCH_HPA_ON_PROC = ["0", "proc_hpa"]
SWITCH_ON = ["2", "on"]
SWITCH_OFF = ["3", "off"]
NORMAL_SSR = ["4", "nml-ssr"]
NORMAL_DRO = ["5", "nml-dro"]
NORMAL_X8 = ["6", "nml-x8"]
NORMAL_TX = ["7", "nml-tx"]
NORMAL_MPA = ["8", "nml-mpa"]
NORMAL_HPA = ["9", "nml-hpa"]
NORMAL_SSR = ["4", "nml_ssr"]
NORMAL_DRO = ["5", "nml_dro"]
NORMAL_X8 = ["6", "nml_x8"]
NORMAL_TX = ["7", "nml_tx"]
NORMAL_MPA = ["8", "nml_mpa"]
NORMAL_HPA = ["9", "nml_hpa"]
REQ_OS_HK = ["8", "hk-os"]
ENABLE_HK = ["enable_hk"]
DISABLE_HK = ["disable_hk"]
REQ_OS_HK = ["hk_os"]
INJECT_SSR_TO_DRO_FAILURE = ["10", "inject-ssr-dro-fault"]
INJECT_DRO_TO_X8_FAILURE = ["11", "inject-dro-x8-fault"]
INJECT_X8_TO_TX_FAILURE = ["12", "inject-x8-tx-fault"]
INJECT_TX_TO_MPA_FAILURE = ["13", "inject-tx-mpa-fault"]
INJECT_MPA_TO_HPA_FAILURE = ["14", "inject-mpa-hpa-fault"]
INJECT_ALL_ON_FAILURE = ["15", "inject-all-on-fault"]
INJECT_SSR_TO_DRO_FAILURE = ["10", "inject_ssr_dro_fault"]
INJECT_DRO_TO_X8_FAILURE = ["11", "inject_dro_x8_fault"]
INJECT_X8_TO_TX_FAILURE = ["12", "inject_x8_tx_fault"]
INJECT_TX_TO_MPA_FAILURE = ["13", "inject_tx_mpa_fault"]
INJECT_MPA_TO_HPA_FAILURE = ["14", "inject_mpa_hpa_fault"]
INJECT_ALL_ON_FAILURE = ["15", "inject_all_on_fault"]
class Info:
NORMAL = "PL PCDU ADC modules normal"
SWITCH_HPA_ON_PROC = "Switch HPA on procedure"
SWITCH_ON = "Switching PL PCDU on"
SWITCH_OFF = "Switching PL PCDU off"
NORMAL = "ADC modules normal"
SWITCH_ON = "Switching on"
SWITCH_OFF = "Switching off"
NORMAL_SSR = f"{NORMAL}, SSR on"
NORMAL_DRO = f"{NORMAL},DRO on"
NORMAL_X8 = f"{NORMAL}, X8 on"
@ -62,6 +65,9 @@ class Info:
NORMAL_MPA = f"{NORMAL}, MPA on"
NORMAL_HPA = f"{NORMAL}, HPA on"
REQ_OS_HK = "Request One Shot HK"
SWITCH_HPA_ON_PROC = "Switch HPA on procedure"
ENABLE_HK = "Enable HK"
DISABLE_HK = "Disable HK"
class SetIds(enum.IntEnum):
@ -128,6 +134,7 @@ def add_pl_pcdu_cmds(defs: TmtcDefinitionWrapper):
oce.add(keys=OpCodes.NORMAL_MPA, info=Info.NORMAL_MPA)
oce.add(keys=OpCodes.NORMAL_HPA, info=Info.NORMAL_HPA)
oce.add(keys=OpCodes.REQ_OS_HK, info=Info.REQ_OS_HK)
oce.add(keys=OpCodes.ENABLE_HK, info=Info.ENABLE_HK)
oce.add(
keys=OpCodes.INJECT_SSR_TO_DRO_FAILURE,
info="Inject failure SSR to DRO transition",
@ -157,6 +164,22 @@ def pack_pl_pcdu_commands(q: DefaultPusQueueHelper, op_code: str):
pack_pl_pcdu_mode_cmd(q=q, info=Info.SWITCH_ON, mode=Modes.ON, submode=0)
if op_code in OpCodes.SWITCH_OFF:
pack_pl_pcdu_mode_cmd(q=q, info=Info.SWITCH_OFF, mode=Modes.OFF, submode=0)
if op_code in OpCodes.ENABLE_HK:
interval = float(
input("Please enter HK collection interval in floating point seconds: ")
)
cmds = enable_periodic_hk_command_with_interval(
diag=True, sid=make_sid(PL_PCDU_ID, SetIds.ADC), interval_seconds=interval
)
q.add_log_cmd(f"Enable PL PCDU HK with interval of {interval} seconds")
for cmd in cmds:
q.add_pus_tc(cmd)
if op_code in OpCodes.DISABLE_HK:
cmd = disable_periodic_hk_command(
diag=True, sid=make_sid(PL_PCDU_ID, SetIds.ADC)
)
q.add_log_cmd("Disabling PL PCDU HK")
q.add_pus_tc(cmd)
if op_code in OpCodes.NORMAL_SSR:
pack_pl_pcdu_mode_cmd(
q=q,

219
pus_tc/devs/reaction_wheels.py
View File

@ -1,219 +0,0 @@
# -*- coding: utf-8 -*-
"""reaction_wheels.py
@brief Tests for the reaction wheel handler
@author J. Meier
@date 20.06.2021
"""
import struct
from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import (
generate_one_hk_command,
generate_one_diag_command,
make_sid,
)
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes, Subservices
from config.definitions import CustomServiceList
class OpCodesDevs:
SPEED = ["0", "speed"]
ON = ["1", "on"]
NML = ["2", "nml"]
OFF = ["3", "off"]
GET_STATUS = ["4", "status"]
GET_TM = ["5", "tm"]
class InfoDevs:
SPEED = "Set speed"
ON = "Set On"
NML = "Set Normal"
OFF = "Set Off"
GET_STATUS = "Get Status HK"
GET_TM = "Get TM HK"
class OpCodesAss:
ON = ["0", "on"]
NML = ["1", "nml"]
OFF = ["2", "off"]
class InfoAss:
ON = "Mode On: 3/4 RWs min. on"
NML = "Mode Normal: 3/4 RWs min. normal"
OFF = "Mode Off: All RWs off"
class RwSetIds:
STATUS_SET_ID = 4
TEMPERATURE_SET_ID = 8
LAST_RESET = 2
TM_SET = 9
class RwCommandIds:
RESET_MCU = bytearray([0x0, 0x0, 0x0, 0x01])
# Reads status information from reaction wheel into dataset with id 4
GET_RW_STATUS = bytearray([0x0, 0x0, 0x0, 0x04])
INIT_RW_CONTROLLER = bytearray([0x0, 0x0, 0x0, 0x05])
SET_SPEED = bytearray([0x0, 0x0, 0x0, 0x06])
# Reads temperature from reaction wheel into dataset with id 8
GET_TEMPERATURE = bytearray([0x0, 0x0, 0x0, 0x08])
GET_TM = bytearray([0x0, 0x0, 0x0, 0x09])
class SpeedDefinitions:
RPM_100 = 1000
RPM_5000 = 5000
class RampTime:
MS_1000 = 1000
@tmtc_definitions_provider
def add_rw_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
oce.add(info=InfoDevs.SPEED, keys=OpCodesDevs.SPEED)
oce.add(info=InfoDevs.ON, keys=OpCodesDevs.ON)
oce.add(info=InfoDevs.OFF, keys=OpCodesDevs.OFF)
oce.add(info=InfoDevs.NML, keys=OpCodesDevs.NML)
oce.add(info=InfoDevs.GET_STATUS, keys=OpCodesDevs.GET_STATUS)
oce.add(info=InfoDevs.GET_TM, keys=OpCodesDevs.GET_TM)
defs.add_service(
name=CustomServiceList.REACTION_WHEEL_1.value,
info="Reaction Wheel 1",
op_code_entry=oce,
)
defs.add_service(
name=CustomServiceList.REACTION_WHEEL_2.value,
info="Reaction Wheel 2",
op_code_entry=oce,
)
defs.add_service(
name=CustomServiceList.REACTION_WHEEL_3.value,
info="Reaction Wheel 3",
op_code_entry=oce,
)
defs.add_service(
name=CustomServiceList.REACTION_WHEEL_4.value,
info="Reaction Wheel 4",
op_code_entry=oce,
)
oce = OpCodeEntry()
oce.add(info=InfoAss.ON, keys=OpCodesAss.ON)
oce.add(info=InfoAss.NML, keys=OpCodesAss.NML)
oce.add(info=InfoAss.OFF, keys=OpCodesAss.OFF)
defs.add_service(
name=CustomServiceList.RW_ASSEMBLY.value,
info="Reaction Wheel Assembly",
op_code_entry=oce,
)
def pack_single_rw_test_into(
object_id: bytes, rw_idx: int, q: DefaultPusQueueHelper, op_code: str
):
if op_code in OpCodesDevs.SPEED:
speed = int(input("Specify speed [0.1 RPM]: "))
ramp_time = int(input("Specify ramp time [ms]: "))
q.add_log_cmd(f"RW {rw_idx}: {InfoDevs.SPEED}")
q.add_pus_tc(pack_set_speed_command(object_id, speed, ramp_time))
if op_code in OpCodesDevs.ON:
q.add_log_cmd(f"RW {rw_idx}: {InfoDevs.ON}")
mode_data = pack_mode_data(object_id, Modes.ON, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
if op_code in OpCodesDevs.NML:
q.add_log_cmd(f"RW {rw_idx}: {InfoDevs.NML}")
mode_data = pack_mode_data(object_id, Modes.NORMAL, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
if op_code in OpCodesDevs.OFF:
q.add_log_cmd(f"RW {rw_idx}: {InfoDevs.OFF}")
mode_data = pack_mode_data(object_id, Modes.OFF, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
if op_code in OpCodesDevs.GET_TM:
q.add_log_cmd(f"RW {rw_idx}: {InfoDevs.GET_TM}")
q.add_pus_tc(
generate_one_hk_command(
sid=make_sid(object_id=object_id, set_id=RwSetIds.TM_SET)
)
)
if op_code in OpCodesDevs.GET_STATUS:
q.add_log_cmd(f"RW {rw_idx}: {InfoDevs.GET_STATUS}")
q.add_pus_tc(
generate_one_diag_command(
sid=make_sid(object_id=object_id, set_id=RwSetIds.STATUS_SET_ID)
)
)
def pack_rw_ass_cmds(q: DefaultPusQueueHelper, object_id: bytes, op_code: str):
if op_code in OpCodesAss.OFF:
data = pack_mode_data(object_id=object_id, mode=Modes.OFF, submode=0)
q.add_pus_tc(
PusTelecommand(
service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=data
)
)
if op_code in OpCodesAss.ON:
data = pack_mode_data(object_id=object_id, mode=Modes.ON, submode=0)
q.add_pus_tc(
PusTelecommand(
service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=data
)
)
if op_code in OpCodesAss.NML:
data = pack_mode_data(object_id=object_id, mode=Modes.NORMAL, submode=0)
q.add_pus_tc(
PusTelecommand(
service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=data
)
)
def pack_set_speed_command(
object_id: bytes, speed: int, ramp_time_ms: int
) -> PusTelecommand:
"""With this function a command is packed to set the speed of a reaction wheel
:param object_id: The object id of the reaction wheel handler.
:param speed: Valid speeds are [-65000, -1000] and [1000, 65000]. Values are
specified in 0.1 * RPM
:param ramp_time_ms: The time after which the reaction wheel will reach the commanded speed.
Valid times are 10 - 10000 ms
"""
if speed > 0:
if speed < 1000 or speed > 65000:
raise ValueError(
"Invalid RW speed specified. "
"Allowed range is [1000, 65000] 0.1 * RPM"
)
elif speed < 0:
if speed < -65000 or speed > -1000:
raise ValueError(
"Invalid RW speed specified. "
"Allowed range is [-65000, -1000] 0.1 * RPM"
)
else:
# Speed is 0
pass
if ramp_time_ms < 0 or (
ramp_time_ms > 0 and (ramp_time_ms > 10000 or ramp_time_ms < 10)
):
raise ValueError("Invalid Ramp Speed time. Allowed range is [10-10000] ms")
command_id = RwCommandIds.SET_SPEED
command = bytearray()
command += object_id + command_id
command = command + struct.pack("!i", speed)
command = command + ramp_time_ms.to_bytes(length=2, byteorder="big")
command = PusTelecommand(service=8, subservice=128, app_data=command)
return command

22
pus_tc/devs/solar_array_deployment.py
View File

@ -1,22 +0,0 @@
# -*- coding: utf-8 -*-
"""
@file solar_array_deployment.py
@brief The test function in this file simply returns a command which triggers the solar array deployment.
@author J. Meier
@date 15.02.2021
"""
from spacepackets.ecss import PusTelecommand
from tmtccmd.tc import DefaultPusQueueHelper
class ActionIds:
DEPLOY_SOLAR_ARRAYS = bytearray([0x0, 0x0, 0x0, 0x5])
def pack_solar_array_deployment_test_into(
object_id: bytearray, q: DefaultPusQueueHelper
):
q.add_log_cmd("Testing S/A Deployment")
command = object_id + ActionIds.DEPLOY_SOLAR_ARRAYS
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))

10
pus_tc/procedure_packer.py
View File

@ -27,11 +27,10 @@ from tmtc.power.p60dock import pack_p60dock_cmds
from tmtc.power.pdu2 import pack_pdu2_commands
from tmtc.power.pdu1 import pack_pdu1_commands
from tmtc.power.acu import pack_acu_commands
from pus_tc.devs.solar_array_deployment import pack_solar_array_deployment_test_into
from pus_tc.devs.imtq import pack_imtq_test_into
from tmtc.acs.imtq import pack_imtq_test_into
from pus_tc.devs.tmp1075 import pack_tmp1075_test_into
from pus_tc.devs.heater import pack_heater_cmds
from pus_tc.devs.reaction_wheels import pack_single_rw_test_into, pack_rw_ass_cmds
from tmtc.acs.reaction_wheels import pack_single_rw_test_into, pack_rw_ass_cmds
from pus_tc.devs.rad_sensor import pack_rad_sensor_test_into
from tmtc.ploc_memory_dumper import pack_ploc_memory_dumper_cmd
from pus_tc.devs.ccsds_handler import pack_ccsds_handler_test
@ -66,7 +65,6 @@ from config.object_ids import (
PDEC_HANDLER_ID,
STR_IMG_HELPER_ID,
SYRLINKS_HANDLER_ID,
SOLAR_ARRAY_DEPLOYMENT_ID,
RW_ASSEMBLY,
get_object_ids,
)
@ -186,10 +184,6 @@ def handle_default_procedure(
return pack_syrlinks_command(
object_id=object_id, q=queue_helper, op_code=op_code
)
if service == CustomServiceList.SA_DEPLYOMENT.value:
return pack_solar_array_deployment_test_into(
object_id=SOLAR_ARRAY_DEPLOYMENT_ID, q=queue_helper
)
if service == CustomServiceList.PROCEDURE.value:
return pack_proc_commands(q=queue_helper, op_code=op_code)
if service == CustomServiceList.PL_PCDU.value:

51
pus_tc/system/proc.py
View File

@ -2,7 +2,6 @@ from __future__ import annotations
import time
from datetime import timedelta
from typing import List
from config.definitions import CustomServiceList
from config.object_ids import get_object_ids
@ -22,7 +21,7 @@ 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 tmtc.power.common_power import SetIds as GsSetIds
from pus_tc.devs.rad_sensor import SetIds as RadSetIds
from pus_tc.devs.mgms import MgmLis3SetIds as MgmLis3SetIds_0_2
from pus_tc.devs.mgms import MgmRm3100SetIds as MgmRm3100SetIds_1_3
@ -30,16 +29,16 @@ 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.syrlinks_hk_handler import SetIds as SyrlinksSetIds
from pus_tc.devs.gps import SetIds as GpsSetIds
from pus_tc.devs.imtq import ImtqSetIds
from tmtc.acs.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 tmtc.acs.reaction_wheels import RwSetIds, rw_speed_up_cmd_consec
from pus_tc.system.controllers import pack_cmd_ctrl_to_off, pack_cmd_ctrl_to_nml
from tmtc.acs_board import pack_acs_command
from tmtc.sus_board import pack_sus_cmds
from pus_tc.devs.imtq import pack_imtq_test_into, pack_dipole_command
from tmtc.acs.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
from tmtc.acs.reaction_wheels import pack_rw_ass_cmds, pack_set_speed_command
class OpCodes:
@ -246,14 +245,14 @@ def pack_proc_commands(q: DefaultPusQueueHelper, op_code: str):
if op_code in OpCodes.PCDU_FT:
key = KAI.PCDU_FT[0]
pcdu_pairs = [
(oids.P60_DOCK_HANDLER, GsSetIds.P60_CORE),
(oids.PDU_1_HANDLER_ID, GsSetIds.PDU_1_CORE),
(oids.PDU_2_HANDLER_ID, GsSetIds.PDU_2_CORE),
(oids.ACU_HANDLER_ID, GsSetIds.ACU_CORE),
(oids.P60_DOCK_HANDLER, GsSetIds.P60_AUX),
(oids.PDU_1_HANDLER_ID, GsSetIds.PDU_1_AUX),
(oids.PDU_2_HANDLER_ID, GsSetIds.PDU_2_AUX),
(oids.ACU_HANDLER_ID, GsSetIds.ACU_AUX),
(oids.P60_DOCK_HANDLER, GsSetIds.CORE),
(oids.PDU_1_HANDLER_ID, GsSetIds.CORE),
(oids.PDU_2_HANDLER_ID, GsSetIds.CORE),
(oids.ACU_HANDLER_ID, GsSetIds.CORE),
(oids.P60_DOCK_HANDLER, GsSetIds.AUX),
(oids.PDU_1_HANDLER_ID, GsSetIds.AUX),
(oids.PDU_2_HANDLER_ID, GsSetIds.AUX),
(oids.ACU_HANDLER_ID, GsSetIds.AUX),
]
diag_list = [
@ -586,10 +585,10 @@ def pack_proc_commands(q: DefaultPusQueueHelper, op_code: str):
]
# HK listening
pack_generic_hk_listening_cmds(
tc_queue=tc_queue,
q=q,
proc_key=key,
sid_list=sid_list,
diag=False,
diag_list=[False],
cfg=GenericHkListeningCfg.default(),
)
if op_code in OpCodes.STR_FT:
@ -725,7 +724,7 @@ def enable_listen_to_hk_for_x_seconds(
):
q.add_log_cmd(f"Enabling periodic HK for {device}")
cmd_tuple = enable_periodic_hk_command_with_interval(
diag=diag, sid=sid, interval_seconds=interval_seconds, ssc=0
diag=diag, sid=sid, interval_seconds=interval_seconds
)
for cmd in cmd_tuple:
q.add_pus_tc(cmd)
@ -826,24 +825,6 @@ def rw_speed_cmd_single(
q.add_pus_tc(pack_set_speed_command(object_id=oid, speed=0, ramp_time_ms=ramp_time))
def rw_speed_up_cmd_consec(
q: DefaultPusQueueHelper, obids: List[bytes], speed: int, ramp_time: int
):
for oid in obids:
q.add_pus_tc(
pack_set_speed_command(object_id=oid, speed=speed, ramp_time_ms=ramp_time)
)
def rw_speed_down_cmd_consec(
q: DefaultPusQueueHelper, obids: List[bytes], ramp_time: int
):
for oid in obids:
q.add_pus_tc(
pack_set_speed_command(object_id=oid, speed=0, ramp_time_ms=ramp_time)
)
def activate_all_rws_in_sequence(
q: DefaultPusQueueHelper, init_ssc: int, test_speed: int, test_ramp_time: int
):

2
pus_tm/action_reply_handler.py
View File

@ -1,6 +1,6 @@
import struct
from config.object_ids import *
from pus_tc.devs.imtq import ImtqActionIds
from tmtc.acs.imtq import ImtqActionIds
from pus_tm.defs import PrintWrapper
from tmtc.ploc_mpsoc import PlocReplyIds
from tmtc.ploc_supervisor import SupvActionIds

228
pus_tm/devs/imtq_mgt.py
View File

@ -1,228 +0,0 @@
import struct
from pus_tm.defs import PrintWrapper
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
def handle_eng_set(printer: FsfwTmTcPrinter, hk_data: bytes):
pw = PrintWrapper(printer)
header_list = [
"Digital Voltage [mV]",
"Analog Voltage [mV]",
"Digital Current [mA]",
"Analog Current [mA]",
"Coil Current X [mA]",
"Coil Current Y [mA]",
"Coil Current Z [mA]",
"Coil X Temperature [°C]",
"Coil Y Temperature [°C]",
"Coil Z Temperature [°C]",
"Coil Z Temperature [°C]",
"MCU Temperature [°C]",
]
digital_voltage = struct.unpack("!H", hk_data[0:2])[0]
analog_voltage = struct.unpack("!H", hk_data[2:4])[0]
digital_current = struct.unpack("!f", hk_data[4:8])[0]
analog_current = struct.unpack("!f", hk_data[8:12])[0]
coil_x_current = struct.unpack("!f", hk_data[12:16])[0]
coil_y_current = struct.unpack("!f", hk_data[16:20])[0]
coil_z_current = struct.unpack("!f", hk_data[20:24])[0]
coil_x_temperature = struct.unpack("!H", hk_data[24:26])[0]
coil_y_temperature = struct.unpack("!H", hk_data[26:28])[0]
coil_z_temperature = struct.unpack("!H", hk_data[30:32])[0]
mcu_temperature = struct.unpack("!H", hk_data[32:34])[0]
validity_buffer = hk_data[42:]
content_list = [
digital_voltage,
analog_voltage,
digital_current,
analog_current,
coil_x_current,
coil_y_current,
coil_z_current,
coil_x_temperature,
coil_y_temperature,
coil_z_temperature,
mcu_temperature,
]
num_of_vars = len(header_list)
pw.dlog(str(header_list))
pw.dlog(str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=num_of_vars)
def handle_calibrated_mtm_measurement(printer: FsfwTmTcPrinter, hk_data: bytes):
pw = PrintWrapper(printer)
header_list = [
"Calibrated MTM X [nT]",
"Calibrated MTM Y [nT]",
"Calibrated MTM Z [nT]",
"Coild actuation status",
]
mtm_x = struct.unpack("!I", hk_data[0:4])[0]
mtm_y = struct.unpack("!I", hk_data[4:8])[0]
mtm_z = struct.unpack("!I", hk_data[8:12])[0]
coil_actuation_status = hk_data[12]
validity_buffer = hk_data[12:]
content_list = [mtm_x, mtm_y, mtm_z, coil_actuation_status]
num_of_vars = len(header_list)
pw.dlog(str(header_list))
pw.dlog(str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=num_of_vars)
def handle_raw_mtm_measurement(printer: FsfwTmTcPrinter, hk_data: bytes):
pw = PrintWrapper(printer)
header_list = [
"Raw MTM X [nT]",
"Raw MTM Y [nT]",
"Raw MTM Z [nT]",
"Coild actuation status",
]
mtm_x = struct.unpack("!f", hk_data[0:4])[0]
mtm_y = struct.unpack("!f", hk_data[4:8])[0]
mtm_z = struct.unpack("!f", hk_data[8:12])[0]
coil_actuation_status = hk_data[12]
validity_buffer = hk_data[12:]
content_list = [mtm_x, mtm_y, mtm_z, coil_actuation_status]
num_of_vars = len(header_list)
pw.dlog(str(header_list))
pw.dlog(str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=num_of_vars)
def handle_self_test_data(printer: FsfwTmTcPrinter, hk_data: bytes):
pw = PrintWrapper(printer)
header_list = [
"Init Err",
"Init Raw Mag X [nT]",
"Init Raw Mag Y [nT]",
"Init Raw Mag Z [nT]",
"Init Cal Mag X [nT]",
"Init Cal Mag Y [nT]",
"Init Cal Mag Z [nT]",
"Init Coil X Current [mA]",
"Init Coil Y Current [mA]",
"Init Coil Z Current [mA]",
"Init Coil X Temperature [°C]",
"Init Coil Y Temperature [°C]",
"Init Coil Z Temperature [°C]",
"Err",
"Raw Mag X [nT]",
"Raw Mag Y [nT]",
"Raw Mag Z [nT]",
"Cal Mag X [nT]",
"Cal Mag Y [nT]",
"Cal Mag Z [nT]",
"Coil X Current [mA]",
"Coil Y Current [mA]",
"Coil Z Current [mA]",
"Coil X Temperature [°C]",
"Coil Y Temperature [°C]",
"Coil Z Temperature [°C]",
"Fina Err",
"Fina Raw Mag X [nT]",
"Fina Raw Mag Y [nT]",
"Fina Raw Mag Z [nT]",
"Fina Cal Mag X [nT]",
"Fina Cal Mag Y [nT]",
"Fina Cal Mag Z [nT]",
"Fina Coil X Current [mA]",
"Fina Coil Y Current [mA]",
"Fina Coil Z Current [mA]",
"Fina Coil X Temperature [°C]",
"Fina Coil Y Temperature [°C]",
"Fina Coil Z Temperature [°C]",
]
# INIT step (no coil actuation)
init_err = hk_data[0]
init_raw_mag_x = struct.unpack("!f", hk_data[1:5])[0]
init_raw_mag_y = struct.unpack("!f", hk_data[5:9])[0]
init_raw_mag_z = struct.unpack("!f", hk_data[9:13])[0]
init_cal_mag_x = struct.unpack("!f", hk_data[13:17])[0]
init_cal_mag_y = struct.unpack("!f", hk_data[17:21])[0]
init_cal_mag_z = struct.unpack("!f", hk_data[21:25])[0]
init_coil_x_current = struct.unpack("!f", hk_data[25:29])[0]
init_coil_y_current = struct.unpack("!f", hk_data[29:33])[0]
init_coil_z_current = struct.unpack("!f", hk_data[33:37])[0]
init_coil_x_temperature = struct.unpack("!H", hk_data[37:39])[0]
init_coil_y_temperature = struct.unpack("!H", hk_data[39:41])[0]
init_coil_z_temperature = struct.unpack("!H", hk_data[41:43])[0]
# Actuation step
err = hk_data[43]
raw_mag_x = struct.unpack("!f", hk_data[44:48])[0]
raw_mag_y = struct.unpack("!f", hk_data[48:52])[0]
raw_mag_z = struct.unpack("!f", hk_data[52:56])[0]
cal_mag_x = struct.unpack("!f", hk_data[56:60])[0]
cal_mag_y = struct.unpack("!f", hk_data[60:64])[0]
cal_mag_z = struct.unpack("!f", hk_data[64:68])[0]
coil_x_current = struct.unpack("!f", hk_data[68:72])[0]
coil_y_current = struct.unpack("!f", hk_data[72:76])[0]
coil_z_current = struct.unpack("!f", hk_data[76:80])[0]
coil_x_temperature = struct.unpack("!H", hk_data[80:82])[0]
coil_y_temperature = struct.unpack("!H", hk_data[82:84])[0]
coil_z_temperature = struct.unpack("!H", hk_data[84:86])[0]
# FINA step (no coil actuation)
fina_err = hk_data[86]
fina_raw_mag_x = struct.unpack("!f", hk_data[87:91])[0]
fina_raw_mag_y = struct.unpack("!f", hk_data[91:95])[0]
fina_raw_mag_z = struct.unpack("!f", hk_data[95:99])[0]
fina_cal_mag_x = struct.unpack("!f", hk_data[99:103])[0]
fina_cal_mag_y = struct.unpack("!f", hk_data[103:107])[0]
fina_cal_mag_z = struct.unpack("!f", hk_data[107:111])[0]
fina_coil_x_current = struct.unpack("!f", hk_data[111:115])[0]
fina_coil_y_current = struct.unpack("!f", hk_data[115:119])[0]
fina_coil_z_current = struct.unpack("!f", hk_data[119:123])[0]
fina_coil_x_temperature = struct.unpack("!H", hk_data[123:125])[0]
fina_coil_y_temperature = struct.unpack("!H", hk_data[125:127])[0]
fina_coil_z_temperature = struct.unpack("!H", hk_data[127:129])[0]
validity_buffer = hk_data[129:]
content_list = [
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,
err,
raw_mag_x,
init_raw_mag_y,
raw_mag_z,
cal_mag_x,
cal_mag_y,
cal_mag_z,
coil_x_current,
coil_y_current,
coil_z_current,
coil_x_temperature,
coil_y_temperature,
coil_z_temperature,
fina_err,
fina_raw_mag_x,
fina_raw_mag_y,
fina_raw_mag_z,
fina_cal_mag_x,
fina_cal_mag_y,
fina_cal_mag_z,
fina_coil_x_current,
fina_coil_y_current,
fina_coil_z_current,
fina_coil_x_temperature,
fina_coil_y_temperature,
fina_coil_z_temperature,
]
num_of_vars = len(header_list)
pw.dlog(str(header_list))
pw.dlog(str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=num_of_vars)

124
pus_tm/devs/reaction_wheels.py
View File

@ -1,124 +0,0 @@
import struct
from pus_tm.defs import PrintWrapper, FsfwTmTcPrinter
from tmtccmd.util.obj_id import ObjectIdU32
def handle_rw_hk_data(
printer: FsfwTmTcPrinter, object_id: ObjectIdU32, set_id: int, hk_data: bytes
):
from pus_tc.devs.reaction_wheels import RwSetIds
pw = PrintWrapper(printer)
current_idx = 0
if set_id == RwSetIds.STATUS_SET_ID:
pw.dlog(
f"Received Status HK (ID {set_id}) from Reaction Wheel {object_id.name}"
)
fmt_str = "!IiiBB"
inc_len = struct.calcsize(fmt_str)
(temp, speed, ref_speed, state, clc_mode) = struct.unpack(
fmt_str, hk_data[current_idx : current_idx + inc_len]
)
current_idx += inc_len
speed_rpm = speed / 10.0
ref_speed_rpm = ref_speed / 10.0
pw.dlog(
f"Temperature {temp} C | Speed {speed_rpm} rpm | Reference Speed {ref_speed_rpm} rpm"
)
pw.dlog(
f"State {state}. 0: Error, 1: Idle, 2: Coasting, 3: Running, speed stable, "
f"4: Running, speed changing"
)
pw.dlog(
f"Current Limit Control mode {clc_mode}. 0: Low Current Mode (0.3 A), "
f"1: High Current Mode (0.6 A)"
)
printer.print_validity_buffer(hk_data[current_idx:], 5)
if set_id == RwSetIds.LAST_RESET:
pw.dlog(
f"Received Last Reset HK (ID {set_id}) from Reaction Wheel {object_id.name}"
)
fmt_str = "!BB"
inc_len = struct.calcsize(fmt_str)
(last_not_cleared_reset_status, current_reset_status) = struct.unpack(
fmt_str, hk_data[current_idx : current_idx + inc_len]
)
current_idx += inc_len
pw.dlog(
f"Last Non-Cleared (Cached) Reset Status {last_not_cleared_reset_status} | "
f"Current Reset Status {current_reset_status}"
)
if set_id == RwSetIds.TM_SET:
pw.dlog(f"Received TM HK (ID {set_id}) from Reaction Wheel {object_id.name}")
fmt_str = "!BiffBBiiIIIIIIIIIIIIIIII"
inc_len = struct.calcsize(fmt_str)
(
last_reset_status,
mcu_temp,
pressure_sens_temp,
pressure,
state,
clc_mode,
current_speed,
ref_speed,
num_invalid_crc_packets,
num_invalid_len_packets,
num_invalid_cmd_packets,
num_of_cmd_executed_requests,
num_of_cmd_replies,
uart_num_of_bytes_written,
uart_num_of_bytes_read,
uart_num_parity_errors,
uart_num_noise_errors,
uart_num_frame_errors,
uart_num_reg_overrun_errors,
uart_total_num_errors,
spi_num_bytes_written,
spi_num_bytes_read,
spi_num_reg_overrun_errors,
spi_total_num_errors,
) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
pw.dlog(
f"MCU Temperature {mcu_temp} | Pressure Sensore Temperature {pressure_sens_temp} C"
)
pw.dlog(f"Last Reset Status {last_reset_status}")
pw.dlog(
f"Current Limit Control mode {clc_mode}. 0: Low Current Mode (0.3 A), "
f"1: High Current Mode (0.6 A)"
)
pw.dlog(f"Speed {current_speed} rpm | Reference Speed {ref_speed} rpm")
pw.dlog(
f"State {state}. 0: Error, 1: Idle, 2: Coasting, 3: Running, speed stable, "
f"4: Running, speed changing"
)
pw.dlog(f"Number Of Invalid Packets:")
pw.dlog("CRC | Length | CMD")
pw.dlog(
f"{num_invalid_crc_packets} | {num_invalid_len_packets} | {num_invalid_cmd_packets}"
)
pw.dlog(
f"Num Of CMD Executed Requests {num_of_cmd_executed_requests} | "
f"Num of CMD Replies {num_of_cmd_replies}"
)
pw.dlog("UART COM information:")
pw.dlog(
f"NumBytesWritten | NumBytesRead | ParityErrs | NoiseErrs | FrameErrs | "
f"RegOverrunErrs | TotalErrs"
)
pw.dlog(
f"{uart_num_of_bytes_written} | {uart_num_of_bytes_read} | {uart_num_parity_errors} | "
f"{uart_num_noise_errors} | {uart_num_frame_errors} | {uart_num_reg_overrun_errors} | "
f"{uart_total_num_errors}"
)
pw.dlog("SPI COM Info:")
pw.dlog(f"NumBytesWritten | NumBytesRead | RegOverrunErrs | TotalErrs")
pw.dlog(
f"{spi_num_bytes_written} | {spi_num_bytes_read} | {spi_num_reg_overrun_errors} | "
f"{spi_total_num_errors}"
)
if current_idx > 0:
printer.print_validity_buffer(
validity_buffer=hk_data[current_idx:], num_vars=27
)

11
pus_tm/hk_handling.py
View File

@ -6,6 +6,7 @@ from pus_tm.devs.rad_sensor import handle_rad_sensor_data
from pus_tm.devs.sus import handle_sus_hk
from pus_tm.system.tcs import handle_thermal_controller_hk_data
from tmtc.ploc_supervisor import handle_supv_hk_data
from tmtc.acs.reaction_wheels import handle_rw_hk_data
from tmtccmd.tm.pus_3_fsfw_hk import (
Service3Base,
HkContentType,
@ -17,16 +18,10 @@ from tmtccmd.logging import get_console_logger
from pus_tm.devs.bpx_bat import handle_bpx_hk_data
from pus_tm.devs.gps import handle_gps_data
from pus_tm.devs.gyros import handle_gyros_hk_data
from pus_tm.devs.imtq_mgt import (
handle_self_test_data,
handle_eng_set,
handle_calibrated_mtm_measurement,
handle_raw_mtm_measurement,
)
from tmtc.power.tm import handle_pdu_data, handle_p60_hk_data, handle_acu_hk_data
from pus_tm.devs.syrlinks import handle_syrlinks_hk_data
from pus_tc.devs.imtq import ImtqSetIds
from pus_tm.devs.reaction_wheels import handle_rw_hk_data
from tmtc.acs.imtq import ImtqSetIds, handle_self_test_data, handle_eng_set, \
handle_calibrated_mtm_measurement, handle_raw_mtm_measurement
from pus_tm.defs import FsfwTmTcPrinter
from pus_tm.system.core import handle_core_hk_data
from pus_tm.devs.mgms import handle_mgm_hk_data

0
tmtc/acs/__init__.py Normal file
View File

488
tmtc/acs/imtq.py Normal file
View File

@ -0,0 +1,488 @@
# -*- coding: utf-8 -*-
"""
@file imtq.py
@brief Tests for the ISIS IMTQ (Magnettorquer) device handler
@author J. Meier
@date 25.03.2021
"""
import struct
from config.definitions import CustomServiceList
from pus_tm.defs import PrintWrapper
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.config.tmtc import (
tmtc_definitions_provider,
OpCodeEntry,
TmtcDefinitionWrapper,
)
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import (
make_sid,
generate_one_diag_command,
generate_one_hk_command,
)
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes
from tmtccmd.util import ObjectIdU32
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
class OpCodes:
ON = ["on"]
NORMAL = ["normal"]
OFF = ["off"]
SET_DIPOLE = ["set_dipole"]
class ImtqSetIds:
ENG_HK_SET = 1
CAL_MTM_SET = 2
RAW_MTM_SET = 3
POSITIVE_X_TEST = 4
NEGATIVE_X_TEST = 5
POSITIVE_Y_TEST = 6
NEGATIVE_Y_TEST = 7
POSITIVE_Z_TEST = 8
NEGATIVE_Z_TEST = 9
class ImtqActionIds:
start_actuation_dipole = bytearray([0x0, 0x0, 0x0, 0x02])
get_commanded_dipole = bytearray([0x0, 0x0, 0x0, 0x03])
perform_positive_x_test = bytearray([0x0, 0x0, 0x0, 0x07])
perform_negative_x_test = bytearray([0x0, 0x0, 0x0, 0x08])
perform_positive_y_test = bytearray([0x0, 0x0, 0x0, 0x09])
perform_negative_y_test = bytearray([0x0, 0x0, 0x0, 0x0A])
perform_positive_z_test = bytearray([0x0, 0x0, 0x0, 0x0B])
perform_negative_z_test = bytearray([0x0, 0x0, 0x0, 0x0C])
# Initiates the reading of the last performed self test. After sending this command the results
# can be downlinked via the housekeeping service by using the appropriate set ids listed above.
read_self_test_results = bytearray([0x0, 0x0, 0x0, 0x0D])
@tmtc_definitions_provider
def add_imtq_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
oce.add(OpCodes.OFF, "Mode Off")
oce.add(OpCodes.ON, "Mode On")
oce.add(OpCodes.NORMAL, "Mode Normal")
oce.add("3", "IMTQ perform pos X self test")
oce.add("4", "IMTQ perform neg X self test")
oce.add("5", "IMTQ perform pos Y self test")
oce.add("6", "IMTQ perform neg Y self test")
oce.add("7", "IMTQ perform pos Z self test")
oce.add("8", "IMTQ perform neg Z self test")
oce.add(OpCodes.SET_DIPOLE, "IMTQ command dipole")
oce.add("10", "IMTQ get commanded dipole")
oce.add("11", "IMTQ get engineering hk set")
oce.add("12", "IMTQ get calibrated MTM measurement one shot")
oce.add("13", "IMTQ get raw MTM measurement one shot")
defs.add_service(CustomServiceList.IMTQ.value, "IMQT Device", oce)
def pack_imtq_test_into(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code: str):
q.add_log_cmd(
f"Testing ISIS IMTQ handler with object id: {object_id.as_hex_string}"
)
if op_code in OpCodes.OFF:
q.add_log_cmd("IMTQ: Set mode off")
command = pack_mode_data(object_id.as_bytes, Modes.OFF, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code in OpCodes.ON:
q.add_log_cmd("IMTQ: Set mode on")
command = pack_mode_data(object_id.as_bytes, Modes.ON, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code in OpCodes.NORMAL:
q.add_log_cmd("IMTQ: Mode Normal")
command = pack_mode_data(object_id.as_bytes, Modes.NORMAL, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code == "3":
q.add_log_cmd("IMTQ: Perform positive x self test")
command = object_id.as_bytes + ImtqActionIds.perform_positive_x_test
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Initiate reading of positive x self test results")
command = object_id.as_bytes + ImtqActionIds.read_self_test_results
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Request dataset with positive x self test results")
sid = make_sid(object_id.as_bytes, ImtqSetIds.POSITIVE_X_TEST)
q.add_pus_tc(generate_one_hk_command(sid))
if op_code == "4":
q.add_log_cmd("IMTQ: Perform negative x self test")
command = object_id.as_bytes + ImtqActionIds.perform_negative_x_test
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Initiate reading of negative x self test results")
command = object_id.as_bytes + ImtqActionIds.read_self_test_results
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Request dataset with negative x self test results")
sid = make_sid(object_id.as_bytes, ImtqSetIds.NEGATIVE_X_TEST)
q.add_pus_tc(generate_one_hk_command(sid))
if op_code == "5":
q.add_log_cmd("IMTQ: Perform positive y self test")
command = object_id.as_bytes + ImtqActionIds.perform_positive_y_test
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Initiate reading of positive y self test results")
command = object_id.as_bytes + ImtqActionIds.read_self_test_results
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Request dataset with positive y self test results")
sid = make_sid(object_id.as_bytes, ImtqSetIds.POSITIVE_Y_TEST)
q.add_pus_tc(generate_one_hk_command(sid))
if op_code == "6":
q.add_log_cmd("IMTQ: Perform negative y self test")
command = object_id.as_bytes + ImtqActionIds.perform_negative_y_test
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Initiate reading of negative y self test results")
command = object_id.as_bytes + ImtqActionIds.read_self_test_results
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Request dataset with negative y self test results")
sid = make_sid(object_id.as_bytes, ImtqSetIds.NEGATIVE_Y_TEST)
q.add_pus_tc(generate_one_hk_command(sid))
if op_code == "7":
q.add_log_cmd("IMTQ: Perform positive z self test")
command = object_id.as_bytes + ImtqActionIds.perform_positive_z_test
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Initiate reading of positive z self test results")
command = object_id.as_bytes + ImtqActionIds.read_self_test_results
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Request dataset with positive z self test results")
sid = make_sid(object_id.as_bytes, ImtqSetIds.POSITIVE_Y_TEST)
q.add_pus_tc(generate_one_hk_command(sid))
if op_code == "8":
q.add_log_cmd("IMTQ: Perform negative z self test")
command = object_id.as_bytes + ImtqActionIds.perform_negative_z_test
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Initiate reading of negative z self test results")
command = object_id.as_bytes + ImtqActionIds.read_self_test_results
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
q.add_log_cmd("IMTQ: Request dataset with negative z self test results")
sid = make_sid(object_id.as_bytes, ImtqSetIds.NEGATIVE_Z_TEST)
q.add_pus_tc(generate_one_hk_command(sid))
if op_code in OpCodes.SET_DIPOLE:
x_dipole = int(input("Specify X dipole [range [0, 2000] * 10^-4 * Am^2]: "))
y_dipole = int(input("Specify Y dipole [range [0, 2000] * 10^-4 * Am^2]: "))
z_dipole = int(input("Specify Z dipole [range [0, 2000] * 10^-4 * Am^2]: "))
duration = int(
input(
f"Specify torque duration [range [0, {pow(2, 16) - 1}, "
f"0 for continuous generation until update]: "
)
)
dur_str = "infinite" if duration == 0 else str(duration)
q.add_log_cmd(
f"IMTQ: Commanding dipole X={x_dipole}, Y={y_dipole}, Z={y_dipole}, duration={dur_str}"
)
q.add_pus_tc(
pack_dipole_command(
object_id.as_bytes, x_dipole, y_dipole, z_dipole, duration
)
)
if op_code == "10":
q.add_log_cmd("IMTQ: Get commanded dipole")
command = object_id.as_bytes + ImtqActionIds.get_commanded_dipole
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
if op_code == "11":
q.add_log_cmd("IMTQ: Get engineering hk set")
q.add_pus_tc(
generate_one_diag_command(
sid=make_sid(object_id=object_id.as_bytes, set_id=ImtqSetIds.ENG_HK_SET)
)
)
if op_code == "12":
q.add_log_cmd("IMTQ: Get calibrated MTM hk set")
q.add_pus_tc(
generate_one_diag_command(
sid=make_sid(
object_id=object_id.as_bytes, set_id=ImtqSetIds.CAL_MTM_SET
)
)
)
if op_code == "13":
q.add_log_cmd("IMTQ: Get raw MTM hk set")
q.add_pus_tc(
generate_one_diag_command(
sid=make_sid(
object_id=object_id.as_bytes, set_id=ImtqSetIds.RAW_MTM_SET
)
)
)
def pack_dipole_command(
object_id: bytes, x_dipole: int, y_dipole: int, z_dipole: int, duration: int
) -> PusTelecommand:
"""This function packs the command causing the ISIS IMTQ to generate a dipole.
:param object_id: The object id of the IMTQ handler.
:param x_dipole: The dipole of the x coil in 10^-4*Am^2 (max. 2000)
:param y_dipole: The dipole of the y coil in 10^-4*Am^2 (max. 2000)
:param z_dipole: The dipole of the z coil in 10^-4*Am^2 (max. 2000)
:param duration: The duration in milliseconds the dipole will be generated by the coils.
When set to 0, the dipole will be generated until a new dipole actuation
command is sent.
"""
action_id = ImtqActionIds.start_actuation_dipole
command = object_id + action_id
x_dipole = int(round(x_dipole))
y_dipole = int(round(y_dipole))
z_dipole = int(round(z_dipole))
if x_dipole < -2000 or x_dipole > 2000:
raise_dipole_error("X dipole", x_dipole)
if y_dipole < -2000 or y_dipole > 2000:
raise_dipole_error("Y dipole", y_dipole)
if z_dipole < -2000 or z_dipole > 2000:
raise_dipole_error("Z dipole", z_dipole)
duration = int(round(duration))
if duration < 0 or duration > pow(2, 16) - 1:
raise ValueError(
f"Duration in ms of {duration} smaller than 0 or larger than allowed {pow(2, 16) - 1}"
)
command += struct.pack("!h", x_dipole)
command += struct.pack("!h", y_dipole)
command += struct.pack("!h", z_dipole)
command += struct.pack("!H", duration)
command = PusTelecommand(service=8, subservice=128, app_data=command)
return command
def raise_dipole_error(dipole_str: str, value: int):
raise ValueError(
f"{dipole_str} {value} negative or larger than maximum allowed 2000 * 10^-4*Am^2"
)
def handle_eng_set(printer: FsfwTmTcPrinter, hk_data: bytes):
pw = PrintWrapper(printer)
header_list = [
"Digital Voltage [mV]",
"Analog Voltage [mV]",
"Digital Current [mA]",
"Analog Current [mA]",
"Coil Current X [mA]",
"Coil Current Y [mA]",
"Coil Current Z [mA]",
"Coil X Temperature [°C]",
"Coil Y Temperature [°C]",
"Coil Z Temperature [°C]",
"MCU Temperature [°C]",
]
digital_voltage = struct.unpack("!H", hk_data[0:2])[0]
analog_voltage = struct.unpack("!H", hk_data[2:4])[0]
digital_current = struct.unpack("!f", hk_data[4:8])[0]
analog_current = struct.unpack("!f", hk_data[8:12])[0]
coil_x_current = struct.unpack("!f", hk_data[12:16])[0]
coil_y_current = struct.unpack("!f", hk_data[16:20])[0]
coil_z_current = struct.unpack("!f", hk_data[20:24])[0]
coil_x_temperature = struct.unpack("!h", hk_data[24:26])[0]
coil_y_temperature = struct.unpack("!h", hk_data[26:28])[0]
coil_z_temperature = struct.unpack("!h", hk_data[28:30])[0]
mcu_temperature = struct.unpack("!h", hk_data[30:32])[0]
validity_buffer = hk_data[32:]
content_list = [
digital_voltage,
analog_voltage,
digital_current,
analog_current,
coil_x_current,
coil_y_current,
coil_z_current,
coil_x_temperature,
coil_y_temperature,
coil_z_temperature,
mcu_temperature,
]
num_of_vars = len(header_list)
pw.dlog(str(header_list))
pw.dlog(str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=num_of_vars)
def handle_calibrated_mtm_measurement(printer: FsfwTmTcPrinter, hk_data: bytes):
pw = PrintWrapper(printer)
header_list = [
"Calibrated MTM X [nT]",
"Calibrated MTM Y [nT]",
"Calibrated MTM Z [nT]",
"Coil actuation status",
]
mtm_x = struct.unpack("!I", hk_data[0:4])[0]
mtm_y = struct.unpack("!I", hk_data[4:8])[0]
mtm_z = struct.unpack("!I", hk_data[8:12])[0]
coil_actuation_status = hk_data[12]
validity_buffer = hk_data[12:]
content_list = [mtm_x, mtm_y, mtm_z, coil_actuation_status]
num_of_vars = len(header_list)
pw.dlog(str(header_list))
pw.dlog(str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=num_of_vars)
def handle_raw_mtm_measurement(printer: FsfwTmTcPrinter, hk_data: bytes):
pw = PrintWrapper(printer)
header_list = [
"Raw MTM X [nT]",
"Raw MTM Y [nT]",
"Raw MTM Z [nT]",
"Coil actuation status",
]
mtm_x = struct.unpack("!f", hk_data[0:4])[0]
mtm_y = struct.unpack("!f", hk_data[4:8])[0]
mtm_z = struct.unpack("!f", hk_data[8:12])[0]
coil_actuation_status = hk_data[12]
validity_buffer = hk_data[13:]
content_list = [mtm_x, mtm_y, mtm_z, coil_actuation_status]
num_of_vars = 2
pw.dlog(str(header_list))
pw.dlog(str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=num_of_vars)
def handle_self_test_data(printer: FsfwTmTcPrinter, hk_data: bytes):
pw = PrintWrapper(printer)
header_list = [
"Init Err",
"Init Raw Mag X [nT]",
"Init Raw Mag Y [nT]",
"Init Raw Mag Z [nT]",
"Init Cal Mag X [nT]",
"Init Cal Mag Y [nT]",
"Init Cal Mag Z [nT]",
"Init Coil X Current [mA]",
"Init Coil Y Current [mA]",
"Init Coil Z Current [mA]",
"Init Coil X Temperature [°C]",
"Init Coil Y Temperature [°C]",
"Init Coil Z Temperature [°C]",
"Err",
"Raw Mag X [nT]",
"Raw Mag Y [nT]",
"Raw Mag Z [nT]",
"Cal Mag X [nT]",
"Cal Mag Y [nT]",
"Cal Mag Z [nT]",
"Coil X Current [mA]",
"Coil Y Current [mA]",
"Coil Z Current [mA]",
"Coil X Temperature [°C]",
"Coil Y Temperature [°C]",
"Coil Z Temperature [°C]",
"Fina Err",
"Fina Raw Mag X [nT]",
"Fina Raw Mag Y [nT]",
"Fina Raw Mag Z [nT]",
"Fina Cal Mag X [nT]",
"Fina Cal Mag Y [nT]",
"Fina Cal Mag Z [nT]",
"Fina Coil X Current [mA]",
"Fina Coil Y Current [mA]",
"Fina Coil Z Current [mA]",
"Fina Coil X Temperature [°C]",
"Fina Coil Y Temperature [°C]",
"Fina Coil Z Temperature [°C]",
]
# INIT step (no coil actuation)
init_err = hk_data[0]
init_raw_mag_x = struct.unpack("!f", hk_data[1:5])[0]
init_raw_mag_y = struct.unpack("!f", hk_data[5:9])[0]
init_raw_mag_z = struct.unpack("!f", hk_data[9:13])[0]
init_cal_mag_x = struct.unpack("!f", hk_data[13:17])[0]
init_cal_mag_y = struct.unpack("!f", hk_data[17:21])[0]
init_cal_mag_z = struct.unpack("!f", hk_data[21:25])[0]
init_coil_x_current = struct.unpack("!f", hk_data[25:29])[0]
init_coil_y_current = struct.unpack("!f", hk_data[29:33])[0]
init_coil_z_current = struct.unpack("!f", hk_data[33:37])[0]
init_coil_x_temperature = struct.unpack("!H", hk_data[37:39])[0]
init_coil_y_temperature = struct.unpack("!H", hk_data[39:41])[0]
init_coil_z_temperature = struct.unpack("!H", hk_data[41:43])[0]
# Actuation step
err = hk_data[43]
raw_mag_x = struct.unpack("!f", hk_data[44:48])[0]
raw_mag_y = struct.unpack("!f", hk_data[48:52])[0]
raw_mag_z = struct.unpack("!f", hk_data[52:56])[0]
cal_mag_x = struct.unpack("!f", hk_data[56:60])[0]
cal_mag_y = struct.unpack("!f", hk_data[60:64])[0]
cal_mag_z = struct.unpack("!f", hk_data[64:68])[0]
coil_x_current = struct.unpack("!f", hk_data[68:72])[0]
coil_y_current = struct.unpack("!f", hk_data[72:76])[0]
coil_z_current = struct.unpack("!f", hk_data[76:80])[0]
coil_x_temperature = struct.unpack("!H", hk_data[80:82])[0]
coil_y_temperature = struct.unpack("!H", hk_data[82:84])[0]
coil_z_temperature = struct.unpack("!H", hk_data[84:86])[0]
# FINA step (no coil actuation)
fina_err = hk_data[86]
fina_raw_mag_x = struct.unpack("!f", hk_data[87:91])[0]
fina_raw_mag_y = struct.unpack("!f", hk_data[91:95])[0]
fina_raw_mag_z = struct.unpack("!f", hk_data[95:99])[0]
fina_cal_mag_x = struct.unpack("!f", hk_data[99:103])[0]
fina_cal_mag_y = struct.unpack("!f", hk_data[103:107])[0]
fina_cal_mag_z = struct.unpack("!f", hk_data[107:111])[0]
fina_coil_x_current = struct.unpack("!f", hk_data[111:115])[0]
fina_coil_y_current = struct.unpack("!f", hk_data[115:119])[0]
fina_coil_z_current = struct.unpack("!f", hk_data[119:123])[0]
fina_coil_x_temperature = struct.unpack("!H", hk_data[123:125])[0]
fina_coil_y_temperature = struct.unpack("!H", hk_data[125:127])[0]
fina_coil_z_temperature = struct.unpack("!H", hk_data[127:129])[0]
validity_buffer = hk_data[129:]
content_list = [
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,
err,
raw_mag_x,
init_raw_mag_y,
raw_mag_z,
cal_mag_x,
cal_mag_y,
cal_mag_z,
coil_x_current,
coil_y_current,
coil_z_current,
coil_x_temperature,
coil_y_temperature,
coil_z_temperature,
fina_err,
fina_raw_mag_x,
fina_raw_mag_y,
fina_raw_mag_z,
fina_cal_mag_x,
fina_cal_mag_y,
fina_cal_mag_z,
fina_coil_x_current,
fina_coil_y_current,
fina_coil_z_current,
fina_coil_x_temperature,
fina_coil_y_temperature,
fina_coil_z_temperature,
]
num_of_vars = len(header_list)
pw.dlog(str(header_list))
pw.dlog(str(content_list))
printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=num_of_vars)

410
tmtc/acs/reaction_wheels.py Normal file
View File

@ -0,0 +1,410 @@
# -*- coding: utf-8 -*-
"""reaction_wheels.py
@brief Tests for the reaction wheel handler
@author J. Meier
@date 20.06.2021
"""
import struct
from typing import List
from pus_tm.defs import PrintWrapper
from config.object_ids import RW1_ID, RW2_ID, RW3_ID, RW4_ID
from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import (
generate_one_hk_command,
generate_one_diag_command,
make_sid,
enable_periodic_hk_command_with_interval,
disable_periodic_hk_command,
)
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes, Subservices
from config.definitions import CustomServiceList
from tmtccmd.util import ObjectIdU32
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
class OpCodesDevs:
SPEED = ["0", "speed"]
ON = ["1", "on"]
NML = ["2", "nml"]
OFF = ["3", "off"]
GET_STATUS = ["4", "status"]
GET_TM = ["5", "tm"]
ENABLE_STATUS_HK = ["6", "enable_status_hk"]
DISABLE_STATUS_HK = ["7", "disable_status_hk"]
class InfoDevs:
SPEED = "Set speed"
ON = "Set On"
NML = "Set Normal"
OFF = "Set Off"
GET_STATUS = "Get Status HK"
GET_TM = "Get TM HK"
ENABLE_STATUS_HK = "Enable Status HK"
DISABLE_STATUS_HK = "Disable Status HK"
class OpCodesAss:
ON = ["0", "on"]
NML = ["1", "nml"]
OFF = ["2", "off"]
ALL_SPEED_UP = ["3", "speed_up"]
ALL_SPEED_OFF = ["4", "speed_off"]
class InfoAss:
ON = "Mode On: 3/4 RWs min. on"
NML = "Mode Normal: 3/4 RWs min. normal"
OFF = "Mode Off: All RWs off"
ALL_SPEED_UP = "Speed up consecutively"
ALL_SPEED_OFF = "Speed down to 0"
class RwSetIds:
STATUS_SET_ID = 4
TEMPERATURE_SET_ID = 8
LAST_RESET = 2
TM_SET = 9
class RwCommandIds:
RESET_MCU = bytearray([0x0, 0x0, 0x0, 0x01])
# Reads status information from reaction wheel into dataset with id 4
GET_RW_STATUS = bytearray([0x0, 0x0, 0x0, 0x04])
INIT_RW_CONTROLLER = bytearray([0x0, 0x0, 0x0, 0x05])
SET_SPEED = bytearray([0x0, 0x0, 0x0, 0x06])
# Reads temperature from reaction wheel into dataset with id 8
GET_TEMPERATURE = bytearray([0x0, 0x0, 0x0, 0x08])
GET_TM = bytearray([0x0, 0x0, 0x0, 0x09])
class SpeedDefinitions:
RPM_100 = 1000
RPM_5000 = 5000
class RampTime:
MS_1000 = 1000
@tmtc_definitions_provider
def add_rw_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
oce.add(info=InfoDevs.SPEED, keys=OpCodesDevs.SPEED)
oce.add(info=InfoDevs.ON, keys=OpCodesDevs.ON)
oce.add(info=InfoDevs.OFF, keys=OpCodesDevs.OFF)
oce.add(info=InfoDevs.NML, keys=OpCodesDevs.NML)
oce.add(info=InfoDevs.GET_STATUS, keys=OpCodesDevs.GET_STATUS)
oce.add(info=InfoDevs.GET_TM, keys=OpCodesDevs.GET_TM)
oce.add(info=InfoDevs.ENABLE_STATUS_HK, keys=OpCodesDevs.ENABLE_STATUS_HK)
oce.add(info=InfoDevs.DISABLE_STATUS_HK, keys=OpCodesDevs.DISABLE_STATUS_HK)
defs.add_service(
name=CustomServiceList.REACTION_WHEEL_1.value,
info="Reaction Wheel 1",
op_code_entry=oce,
)
defs.add_service(
name=CustomServiceList.REACTION_WHEEL_2.value,
info="Reaction Wheel 2",
op_code_entry=oce,
)
defs.add_service(
name=CustomServiceList.REACTION_WHEEL_3.value,
info="Reaction Wheel 3",
op_code_entry=oce,
)
defs.add_service(
name=CustomServiceList.REACTION_WHEEL_4.value,
info="Reaction Wheel 4",
op_code_entry=oce,
)
oce = OpCodeEntry()
oce.add(info=InfoAss.ON, keys=OpCodesAss.ON)
oce.add(info=InfoAss.NML, keys=OpCodesAss.NML)
oce.add(info=InfoAss.OFF, keys=OpCodesAss.OFF)
oce.add(info=InfoAss.ALL_SPEED_UP, keys=OpCodesAss.ALL_SPEED_UP)
oce.add(info=InfoAss.ALL_SPEED_OFF, keys=OpCodesAss.ALL_SPEED_OFF)
defs.add_service(
name=CustomServiceList.RW_ASSEMBLY.value,
info="Reaction Wheel Assembly",
op_code_entry=oce,
)
def pack_single_rw_test_into(
object_id: bytes, rw_idx: int, q: DefaultPusQueueHelper, op_code: str
):
if op_code in OpCodesDevs.SPEED:
speed, ramp_time = prompt_speed_ramp_time()
q.add_log_cmd(
f"RW {rw_idx}: {InfoDevs.SPEED} with target "
f"speed {speed / 10.0} RPM and {ramp_time} ms ramp time"
)
q.add_pus_tc(pack_set_speed_command(object_id, speed, ramp_time))
if op_code in OpCodesDevs.ON:
q.add_log_cmd(f"RW {rw_idx}: {InfoDevs.ON}")
mode_data = pack_mode_data(object_id, Modes.ON, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
if op_code in OpCodesDevs.NML:
q.add_log_cmd(f"RW {rw_idx}: {InfoDevs.NML}")
mode_data = pack_mode_data(object_id, Modes.NORMAL, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
if op_code in OpCodesDevs.OFF:
q.add_log_cmd(f"RW {rw_idx}: {InfoDevs.OFF}")
mode_data = pack_mode_data(object_id, Modes.OFF, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
if op_code in OpCodesDevs.GET_TM:
q.add_log_cmd(f"RW {rw_idx}: {InfoDevs.GET_TM}")
q.add_pus_tc(
generate_one_hk_command(
sid=make_sid(object_id=object_id, set_id=RwSetIds.TM_SET)
)
)
if op_code in OpCodesDevs.GET_STATUS:
q.add_log_cmd(f"RW {rw_idx}: {InfoDevs.GET_STATUS}")
q.add_pus_tc(
generate_one_diag_command(
sid=make_sid(object_id=object_id, set_id=RwSetIds.STATUS_SET_ID)
)
)
if op_code in OpCodesDevs.ENABLE_STATUS_HK:
q.add_log_cmd(f"RW {rw_idx}: {InfoDevs.ENABLE_STATUS_HK}")
interval = float(input("Please enter HK interval in floating point seconds: "))
cmds = enable_periodic_hk_command_with_interval(
True, make_sid(object_id, RwSetIds.STATUS_SET_ID), interval
)
for cmd in cmds:
q.add_pus_tc(cmd)
if op_code in OpCodesDevs.DISABLE_STATUS_HK:
q.add_log_cmd(f"RW {rw_idx}: {InfoDevs.DISABLE_STATUS_HK}")
q.add_pus_tc(
disable_periodic_hk_command(
True, make_sid(object_id, RwSetIds.STATUS_SET_ID)
)
)
def pack_rw_ass_cmds(q: DefaultPusQueueHelper, object_id: bytes, op_code: str):
if op_code in OpCodesAss.OFF:
data = pack_mode_data(object_id=object_id, mode=Modes.OFF, submode=0)
q.add_pus_tc(
PusTelecommand(
service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=data
)
)
if op_code in OpCodesAss.ON:
data = pack_mode_data(object_id=object_id, mode=Modes.ON, submode=0)
q.add_pus_tc(
PusTelecommand(
service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=data
)
)
if op_code in OpCodesAss.NML:
data = pack_mode_data(object_id=object_id, mode=Modes.NORMAL, submode=0)
q.add_pus_tc(
PusTelecommand(
service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=data
)
)
if op_code in OpCodesAss.ALL_SPEED_UP:
speed, ramp_time = prompt_speed_ramp_time()
rw_speed_up_cmd_consec(q, [RW1_ID, RW2_ID, RW3_ID, RW4_ID], speed, ramp_time)
if op_code in OpCodesAss.ALL_SPEED_OFF:
rw_speed_down_cmd_consec(
q, [RW1_ID, RW2_ID, RW3_ID, RW4_ID], prompt_ramp_time()
)
def prompt_speed_ramp_time() -> (int, int):
speed = int(
input("Specify speed [0.1 RPM, 0 or range [-65000, -1000] and [1000, 65000]: ")
)
return speed, prompt_ramp_time()
def prompt_ramp_time() -> int:
return int(input("Specify ramp time [ms, range [10, 20000]]: "))
def pack_set_speed_command(
object_id: bytes, speed: int, ramp_time_ms: int
) -> PusTelecommand:
"""With this function a command is packed to set the speed of a reaction wheel
:param object_id: The object id of the reaction wheel handler.
:param speed: Valid speeds are 0, [-65000, -1000] and [1000, 65000]. Values are
specified in 0.1 * RPM
:param ramp_time_ms: The time after which the reaction wheel will reach the commanded speed.
Valid times are 10 - 20000 ms
"""
if speed > 0:
if speed < 1000 or speed > 65000:
raise ValueError(
"Invalid RW speed specified. "
"Allowed range is [1000, 65000] 0.1 * RPM"
)
elif speed < 0:
if speed < -65000 or speed > -1000:
raise ValueError(
"Invalid RW speed specified. "
"Allowed range is [-65000, -1000] 0.1 * RPM"
)
else:
# Speed is 0
pass
if ramp_time_ms < 0 or (
ramp_time_ms > 0 and (ramp_time_ms > 20000 or ramp_time_ms < 10)
):
raise ValueError("Invalid Ramp Speed time. Allowed range is [10-20000] ms")
command_id = RwCommandIds.SET_SPEED
command = bytearray()
command += object_id + command_id
command = command + struct.pack("!i", speed)
command = command + ramp_time_ms.to_bytes(length=2, byteorder="big")
command = PusTelecommand(service=8, subservice=128, app_data=command)
return command
def handle_rw_hk_data(
printer: FsfwTmTcPrinter, object_id: ObjectIdU32, set_id: int, hk_data: bytes
):
pw = PrintWrapper(printer)
current_idx = 0
if set_id == RwSetIds.STATUS_SET_ID:
pw.dlog(
f"Received Status HK (ID {set_id}) from Reaction Wheel {object_id.name}"
)
fmt_str = "!IiiBB"
inc_len = struct.calcsize(fmt_str)
(temp, speed, ref_speed, state, clc_mode) = struct.unpack(
fmt_str, hk_data[current_idx : current_idx + inc_len]
)
current_idx += inc_len
speed_rpm = speed / 10.0
ref_speed_rpm = ref_speed / 10.0
pw.dlog(
f"Temperature {temp} C | Speed {speed_rpm} rpm | Reference Speed {ref_speed_rpm} rpm"
)
pw.dlog(
f"State {state}. 0: Error, 1: Idle, 2: Coasting, 3: Running, speed stable, "
f"4: Running, speed changing"
)
pw.dlog(
f"Current Limit Control mode {clc_mode}. 0: Low Current Mode (0.3 A), "
f"1: High Current Mode (0.6 A)"
)
printer.print_validity_buffer(hk_data[current_idx:], 5)
if set_id == RwSetIds.LAST_RESET:
pw.dlog(
f"Received Last Reset HK (ID {set_id}) from Reaction Wheel {object_id.name}"
)
fmt_str = "!BB"
inc_len = struct.calcsize(fmt_str)
(last_not_cleared_reset_status, current_reset_status) = struct.unpack(
fmt_str, hk_data[current_idx : current_idx + inc_len]
)
current_idx += inc_len
pw.dlog(
f"Last Non-Cleared (Cached) Reset Status {last_not_cleared_reset_status} | "
f"Current Reset Status {current_reset_status}"
)
if set_id == RwSetIds.TM_SET:
pw.dlog(f"Received TM HK (ID {set_id}) from Reaction Wheel {object_id.name}")
fmt_str = "!BiffBBiiIIIIIIIIIIIIIIII"
inc_len = struct.calcsize(fmt_str)
(
last_reset_status,
mcu_temp,
pressure_sens_temp,
pressure,
state,
clc_mode,
current_speed,
ref_speed,
num_invalid_crc_packets,
num_invalid_len_packets,
num_invalid_cmd_packets,
num_of_cmd_executed_requests,
num_of_cmd_replies,
uart_num_of_bytes_written,
uart_num_of_bytes_read,
uart_num_parity_errors,
uart_num_noise_errors,
uart_num_frame_errors,
uart_num_reg_overrun_errors,
uart_total_num_errors,
spi_num_bytes_written,
spi_num_bytes_read,
spi_num_reg_overrun_errors,
spi_total_num_errors,
) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
pw.dlog(
f"MCU Temperature {mcu_temp} | Pressure Sensore Temperature {pressure_sens_temp} C"
)
pw.dlog(f"Last Reset Status {last_reset_status}")
pw.dlog(
f"Current Limit Control mode {clc_mode}. 0: Low Current Mode (0.3 A), "
f"1: High Current Mode (0.6 A)"
)
pw.dlog(f"Speed {current_speed} rpm | Reference Speed {ref_speed} rpm")
pw.dlog(
f"State {state}. 0: Error, 1: Idle, 2: Coasting, 3: Running, speed stable, "
f"4: Running, speed changing"
)
pw.dlog(f"Number Of Invalid Packets:")
pw.dlog("CRC | Length | CMD")
pw.dlog(
f"{num_invalid_crc_packets} | {num_invalid_len_packets} | {num_invalid_cmd_packets}"
)
pw.dlog(
f"Num Of CMD Executed Requests {num_of_cmd_executed_requests} | "
f"Num of CMD Replies {num_of_cmd_replies}"
)
pw.dlog("UART COM information:")
pw.dlog(
f"NumBytesWritten | NumBytesRead | ParityErrs | NoiseErrs | FrameErrs | "
f"RegOverrunErrs | TotalErrs"
)
pw.dlog(
f"{uart_num_of_bytes_written} | {uart_num_of_bytes_read} | {uart_num_parity_errors} | "
f"{uart_num_noise_errors} | {uart_num_frame_errors} | {uart_num_reg_overrun_errors} | "
f"{uart_total_num_errors}"
)
pw.dlog("SPI COM Info:")
pw.dlog(f"NumBytesWritten | NumBytesRead | RegOverrunErrs | TotalErrs")
pw.dlog(
f"{spi_num_bytes_written} | {spi_num_bytes_read} | {spi_num_reg_overrun_errors} | "
f"{spi_total_num_errors}"
)
if current_idx > 0:
printer.print_validity_buffer(
validity_buffer=hk_data[current_idx:], num_vars=27
)
def rw_speed_up_cmd_consec(
q: DefaultPusQueueHelper, obids: List[bytes], speed: int, ramp_time: int
):
for oid in obids:
q.add_pus_tc(
pack_set_speed_command(object_id=oid, speed=speed, ramp_time_ms=ramp_time)
)
def rw_speed_down_cmd_consec(
q: DefaultPusQueueHelper, obids: List[bytes], ramp_time: int
):
for oid in obids:
q.add_pus_tc(
pack_set_speed_command(object_id=oid, speed=0, ramp_time_ms=ramp_time)
)

30
tmtc/power/acu.py
View File

@ -6,25 +6,18 @@
from config.definitions import CustomServiceList
from tmtc.power.common_power import (
pack_gomspace_cmds,
pack_common_gomspace_cmds,
add_gomspace_cmd_defs,
req_hk_cmds,
pack_common_power_cmds,
SetIds,
)
from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import (
make_sid,
generate_one_diag_command,
generate_one_hk_command,
)
import gomspace.gomspace_common as gs
from gomspace.gomspace_common import GomspaceOpCodes
from gomspace.gomspace_common import GsInfo as GsInfo
from config.object_ids import ACU_HANDLER_ID
from tmtc.power.p60dock import P60DockConfigTable
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
from tmtccmd.util import ObjectIdU32
@ -68,15 +61,14 @@ def add_acu_cmds(defs: TmtcDefinitionWrapper):
def pack_acu_commands(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code: str):
q.add_log_cmd("Handling ACU command")
pack_gomspace_cmds("ACU", object_id, q, op_code)
pack_common_power_cmds("ACU", object_id, q, op_code)
pack_common_gomspace_cmds("ACU", object_id, q, op_code)
acu_req_hk_cmds(q, op_code)
pack_test_cmds(object_id=object_id, q=q)
def acu_req_hk_cmds(q: DefaultPusQueueHelper, op_code: str):
req_hk_cmds(
"ACU", q, op_code, ACU_HANDLER_ID, [gs.SetIds.ACU_CORE, gs.SetIds.ACU_AUX]
)
req_hk_cmds("ACU", q, op_code, ACU_HANDLER_ID, [SetIds.CORE, SetIds.AUX])
class ACUTestProcedure:
@ -173,13 +165,3 @@ def pack_test_cmds(object_id: ObjectIdU32, q: DefaultPusQueueHelper):
ACUConfigTable.ov_mode.parameter_size,
)
)
if ACUTestProcedure.all or ACUTestProcedure.off:
q.add_log_cmd("P60 Dock: Turning off ACU")
q.add_pus_tc(
gs.pack_set_param_command(
ACU_HANDLER_ID,
P60DockConfigTable.out_en_0.parameter_address,
P60DockConfigTable.out_en_0.parameter_size,
gs.Channel.off,
)
)

95
tmtc/power/common_power.py
View File

@ -1,8 +1,6 @@
from gomspace.gomspace_common import (
pack_set_u8_param_command,
Channel,
GomspaceOpCodes,
GsInfo,
GomspaceDeviceActionIds,
prompt_and_pack_set_integer_param_command,
prompt_and_pack_get_param_command,
@ -18,16 +16,44 @@ from tmtccmd.tc.pus_3_fsfw_hk import (
make_sid,
generate_one_diag_command,
generate_one_hk_command,
enable_periodic_hk_command_with_interval,
disable_periodic_hk_command,
)
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
from tmtccmd.util import ObjectIdU32, ObjectIdBase
class GomspaceOpCodes:
GET_PARAM = ["get_param"]
SET_INTEGER_PARAM = ["set_int_param"]
SAVE_TABLE = ["save_table"]
RESET_GND_WATCHDOG = ["reset_gnd_wdt"]
SAVE_TABLE_DEFAULT = ["save_table_default"]
LOAD_TABLE = ["load_table"]
REQUEST_CONFIG_TABLE = ["cfg_table"]
class GsInfo:
GET_PARAMETER = "Get parameter"
SET_PARAMETER = "Set integer parameter"
REQUEST_CONFIG_TABLE = "Request Config Table"
RESET_GND_WATCHDOG = "Reset GND watchdog"
SAVE_TABLE = "Save table non-volatile (file)"
SAVE_TABLE_DEFAULT = "Save table non-volatile (default)"
LOAD_TABLE = "Load Table"
class PowerInfo:
INFO_CORE = "Core Information"
INFO_AUX = "Auxiliary Information"
INFO_ALL = "All Information"
ENABLE_INFO_HK = "Enable Core Info HK"
DISABLE_INFO_HK = "Disable Core Info HK"
RESET_ALL_GND_WDTS = "Reset all Ground Watchdogs"
REQUEST_CORE_HK_ONCE = "Requesting Core HK once"
REQUEST_AUX_HK_ONCE = "Requesting Aux HK once"
PRINT_SWITCH_V_I = "Print Switch V I Info"
PRINT_LATCHUPS = "Print latchups"
class PowerOpCodes:
@ -68,24 +94,54 @@ class PowerOpCodes:
PL_CAM_OFF = ["cam_off"]
INFO_CORE = ["info"]
ENABLE_INFO_HK = ["info_hk_on"]
DISABLE_INFO_HK = ["info_hk_off"]
INFO_AUX = ["info_aux"]
INFO_ALL = ["info_all"]
RESET_ALL_GND_WDTS = ["reset_gnd_wdts"]
# Request HK
REQUEST_CORE_HK_ONCE = ["hk_core"]
REQUEST_AUX_HK_ONCE = ["hk_aux"]
PRINT_SWITCH_V_I = ["print_switch_vi"]
PRINT_LATCHUPS = ["print_latchups"]
def pack_gomspace_cmds(
class SetIds:
CORE = 1
AUX = 2
CONFIG = 3
def pack_common_power_cmds(
prefix: str, object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code: str
):
objb = object_id.as_bytes
if op_code in GomspaceOpCodes.PRINT_SWITCH_V_I:
q.add_log_cmd(f"{prefix}: {GsInfo.PRINT_SWITCH_V_I}")
if op_code in PowerOpCodes.ENABLE_INFO_HK:
interval = float(input("Specify HK interval in floating point seconds: "))
q.add_log_cmd(f"{prefix}: {PowerInfo.ENABLE_INFO_HK} with interval {interval}")
cmds = enable_periodic_hk_command_with_interval(
True, make_sid(objb, SetIds.CORE), interval
)
for cmd in cmds:
q.add_pus_tc(cmd)
if op_code in PowerOpCodes.DISABLE_INFO_HK:
q.add_log_cmd(f"{prefix}: {PowerInfo.DISABLE_INFO_HK}")
q.add_pus_tc(disable_periodic_hk_command(True, make_sid(objb, SetIds.CORE)))
def pack_common_gomspace_cmds(
prefix: str, object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code: str
):
objb = object_id.as_bytes
if op_code in PowerOpCodes.PRINT_SWITCH_V_I:
q.add_log_cmd(f"{prefix}: {PowerInfo.PRINT_SWITCH_V_I}")
q.add_pus_tc(
make_fsfw_action_cmd(
object_id=objb, action_id=GomspaceDeviceActionIds.PRINT_SWITCH_V_I
)
)
if op_code in GomspaceOpCodes.PRINT_LATCHUPS:
q.add_log_cmd(f"{prefix}: {GsInfo.PRINT_LATCHUPS}")
if op_code in PowerOpCodes.PRINT_LATCHUPS:
q.add_log_cmd(f"{prefix}: {PowerInfo.PRINT_LATCHUPS}")
q.add_pus_tc(
make_fsfw_action_cmd(
object_id=objb, action_id=GomspaceDeviceActionIds.PRINT_LATCHUPS
@ -189,12 +245,12 @@ def req_hk_cmds(
obj_id: bytes,
set_id_pair: [int, int],
):
if op_code in GomspaceOpCodes.REQUEST_CORE_HK_ONCE:
q.add_log_cmd(f"{prefix}: {GsInfo.REQUEST_CORE_HK_ONCE}")
if op_code in PowerOpCodes.REQUEST_CORE_HK_ONCE:
q.add_log_cmd(f"{prefix}: {PowerInfo.REQUEST_CORE_HK_ONCE}")
hk_sid = make_sid(object_id=obj_id, set_id=set_id_pair[0])
q.add_pus_tc(generate_one_diag_command(sid=hk_sid))
if op_code in GomspaceOpCodes.REQUEST_AUX_HK_ONCE:
q.add_log_cmd(f"{prefix}: {GsInfo.REQUEST_AUX_HK_ONCE}")
if op_code in PowerOpCodes.REQUEST_AUX_HK_ONCE:
q.add_log_cmd(f"{prefix}: {PowerInfo.REQUEST_AUX_HK_ONCE}")
hk_sid = make_sid(object_id=obj_id, set_id=set_id_pair[1])
q.add_pus_tc(generate_one_hk_command(sid=hk_sid))
@ -229,17 +285,24 @@ def generic_off_cmd(
)
def add_common_power_defs(oce: OpCodeEntry):
oce.add(keys=PowerOpCodes.REQUEST_CORE_HK_ONCE, info=PowerInfo.REQUEST_CORE_HK_ONCE)
oce.add(keys=PowerOpCodes.REQUEST_AUX_HK_ONCE, info=PowerInfo.REQUEST_AUX_HK_ONCE)
oce.add(keys=PowerOpCodes.ENABLE_INFO_HK, info=PowerInfo.ENABLE_INFO_HK)
oce.add(keys=PowerOpCodes.DISABLE_INFO_HK, info=PowerInfo.DISABLE_INFO_HK)
def add_gomspace_cmd_defs(oce: OpCodeEntry):
oce.add(
keys=GomspaceOpCodes.REQUEST_CORE_HK_ONCE,
info=GsInfo.REQUEST_CORE_HK_ONCE,
keys=PowerOpCodes.REQUEST_CORE_HK_ONCE,
info=PowerInfo.REQUEST_CORE_HK_ONCE,
)
oce.add(
keys=GomspaceOpCodes.REQUEST_AUX_HK_ONCE,
info=GsInfo.REQUEST_AUX_HK_ONCE,
keys=PowerOpCodes.REQUEST_AUX_HK_ONCE,
info=PowerInfo.REQUEST_AUX_HK_ONCE,
)
oce.add(keys=PowerOpCodes.PRINT_LATCHUPS, info=PowerInfo.PRINT_LATCHUPS)
oce.add(keys=GomspaceOpCodes.GET_PARAM, info=GsInfo.GET_PARAMETER)
oce.add(keys=GomspaceOpCodes.PRINT_LATCHUPS, info=GsInfo.PRINT_LATCHUPS)
oce.add(keys=GomspaceOpCodes.SET_INTEGER_PARAM, info=GsInfo.SET_PARAMETER)
oce.add(keys=GomspaceOpCodes.REQUEST_CONFIG_TABLE, info=GsInfo.REQUEST_CONFIG_TABLE)
oce.add(keys=GomspaceOpCodes.SAVE_TABLE, info=GsInfo.SAVE_TABLE)

15
tmtc/power/p60dock.py
View File

@ -5,7 +5,12 @@
@author J. Meier
@date 13.12.2020
"""
from tmtc.power.common_power import pack_gomspace_cmds, req_hk_cmds
from tmtc.power.common_power import (
pack_common_gomspace_cmds,
req_hk_cmds,
pack_common_power_cmds,
SetIds,
)
from tmtccmd.tc import DefaultPusQueueHelper
from gomspace.gomspace_common import (
TableEntry,
@ -17,7 +22,6 @@ from gomspace.gomspace_common import (
pack_reboot_command,
pack_set_u8_param_command,
pack_set_u16_param_command,
SetIds,
)
from config.object_ids import P60_DOCK_HANDLER
from tmtccmd.util import ObjectIdU32
@ -96,7 +100,8 @@ class P60DockHkTable:
def pack_p60dock_cmds(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code: str):
objb = object_id.as_bytes
pack_gomspace_cmds("P60 Dock", object_id, q, op_code)
pack_common_power_cmds("P60 Dock", object_id, q, op_code)
pack_common_gomspace_cmds("P60 Dock", object_id, q, op_code)
p60_dock_req_hk_cmds(q, op_code)
if op_code in P60OpCodes.STACK_3V3_ON:
q.add_log_cmd(P60Info.STACK_3V3_ON)
@ -233,6 +238,4 @@ def pack_p60dock_cmds(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code:
def p60_dock_req_hk_cmds(q: DefaultPusQueueHelper, op_code: str):
req_hk_cmds(
"P60 Dock", q, op_code, P60_DOCK_HANDLER, [SetIds.P60_CORE, SetIds.P60_AUX]
)
req_hk_cmds("P60 Dock", q, op_code, P60_DOCK_HANDLER, [SetIds.CORE, SetIds.AUX])

43
tmtc/power/pdu1.py
View File

@ -3,18 +3,27 @@
@author J. Meier
@date 17.12.2020
"""
from config.definitions import CustomServiceList
from config.object_ids import PDU_1_HANDLER_ID
from tmtc.power.common_power import (
pack_gomspace_cmds,
pack_common_gomspace_cmds,
req_hk_cmds,
PowerOpCodes,
generic_on_cmd,
generic_off_cmd,
add_gomspace_cmd_defs,
pack_common_power_cmds,
GomspaceOpCodes,
GsInfo,
PowerInfo,
add_common_power_defs,
SetIds,
)
from gomspace.gomspace_common import *
from gomspace.gomspace_pdu_definitions import *
from tmtccmd.config import OpCodeEntry
from tmtccmd.config import OpCodeEntry, TmtcDefinitionWrapper
from tmtccmd.config.tmtc import tmtc_definitions_provider
class Pdu1InfoBase:
@ -59,9 +68,10 @@ class PDU1TestProcedure:
def pack_pdu1_commands(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code: str):
q.add_log_cmd("Commanding PDU1")
objb = object_id.as_bytes
pdu1_cmds(q, op_code)
pdu1_switch_cmds(q, op_code)
pdu1_req_hk_cmds(q, op_code)
pack_gomspace_cmds("PDU1", object_id, q, op_code)
pack_common_power_cmds("PDU1", object_id, q, op_code)
pack_common_gomspace_cmds("PDU1", object_id, q, op_code)
if PDU1TestProcedure.all or PDU1TestProcedure.ping:
q.add_log_cmd("PDU1: Ping Test")
ping_data = bytearray([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
@ -79,9 +89,7 @@ def pack_pdu1_commands(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code
def pdu1_req_hk_cmds(q: DefaultPusQueueHelper, op_code: str):
req_hk_cmds(
"PDU1", q, op_code, PDU_1_HANDLER_ID, [SetIds.PDU_1_CORE, SetIds.PDU_1_AUX]
)
req_hk_cmds("PDU1", q, op_code, PDU_1_HANDLER_ID, [SetIds.CORE, SetIds.AUX])
def info_on_pdu1(base: str) -> str:
@ -92,7 +100,7 @@ def info_off_pdu1(base: str) -> str:
return "PDU1: " + base + " off"
def pdu1_cmds(q: DefaultPusQueueHelper, op_code: str):
def pdu1_switch_cmds(q: DefaultPusQueueHelper, op_code: str):
if op_code in PowerOpCodes.TCS_ON:
tcs_on_cmd(q)
elif op_code in PowerOpCodes.TCS_OFF:
@ -146,6 +154,25 @@ def add_pdu1_common_defs(oce: OpCodeEntry):
oce.add(keys=PowerOpCodes.SCEX_OFF, info=info_off_pdu1(Pdu1InfoBase.SCEX))
@tmtc_definitions_provider
def add_pdu1_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
add_pdu1_common_defs(oce)
add_common_power_defs(oce)
add_gomspace_cmd_defs(oce)
oce.add(
keys=PowerOpCodes.PRINT_SWITCH_V_I,
info="PDU1: Print Switches, Voltages, Currents",
)
oce.add(keys=GomspaceOpCodes.GET_PARAM, info=GsInfo.GET_PARAMETER)
defs.add_service(
name=CustomServiceList.PDU1.value,
info="PDU1 Device",
op_code_entry=oce,
)
def tcs_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.TCS, Pdu1ChIndex.TCS)

37
tmtc/power/pdu2.py
View File

@ -8,15 +8,20 @@
"""
from config.object_ids import PDU_2_HANDLER_ID
from tmtc.power.common_power import (
pack_gomspace_cmds,
pack_common_gomspace_cmds,
req_hk_cmds,
PowerOpCodes,
generic_on_cmd,
generic_off_cmd,
add_gomspace_cmd_defs,
pack_common_power_cmds,
SetIds,
add_common_power_defs,
)
from gomspace.gomspace_common import *
from gomspace.gomspace_pdu_definitions import *
from tmtccmd.config import OpCodeEntry
from tmtccmd.config import OpCodeEntry, TmtcDefinitionWrapper
from tmtccmd.config.tmtc import tmtc_definitions_provider
class Pdu2InfoBase:
@ -74,7 +79,8 @@ def pack_pdu2_commands(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code
objb = object_id.as_bytes
pdu2_cmds(q, op_code)
pdu2_req_hk_cmds(q, op_code)
pack_gomspace_cmds("PDU2", object_id, q, op_code)
pack_common_power_cmds("PDU2", object_id, q, op_code)
pack_common_gomspace_cmds("PDU2", object_id, q, op_code)
if PDU2TestProcedure.all or PDU2TestProcedure.reboot:
q.add_log_cmd("PDU2: Reboot")
q.add_pus_tc(pack_reboot_command(object_id))
@ -130,6 +136,27 @@ def pack_pdu2_commands(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code
q.add_pus_tc(pack_request_full_hk_table_command(object_id))
@tmtc_definitions_provider
def add_pdu2_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
add_pdu2_common_defs(oce)
add_common_power_defs(oce)
add_gomspace_cmd_defs(oce)
oce.add(
keys=PowerOpCodes.PRINT_SWITCH_V_I,
info="PDU2: Print Switches, Voltages, Currents",
)
oce.add(
keys=PowerOpCodes.PRINT_LATCHUPS,
info="PDU2: Print Latchups",
)
defs.add_service(
name="pdu2",
info="PDU2 Device",
op_code_entry=oce,
)
def pdu2_cmds(q: DefaultPusQueueHelper, op_code: str):
if op_code in PowerOpCodes.PL_PCDU_VBAT_NOM_ON:
pl_pcdu_bat_nom_on_cmd(q)
@ -211,9 +238,7 @@ def add_pdu2_common_defs(oce: OpCodeEntry):
def pdu2_req_hk_cmds(q: DefaultPusQueueHelper, op_code: str):
req_hk_cmds(
"PDU2", q, op_code, PDU_2_HANDLER_ID, [SetIds.PDU_2_CORE, SetIds.PDU_2_AUX]
)
req_hk_cmds("PDU2", q, op_code, PDU_2_HANDLER_ID, [SetIds.CORE, SetIds.AUX])
def pl_pcdu_bat_nom_on_cmd(q: DefaultPusQueueHelper):

89
tmtc/power/power.py
View File

@ -1,4 +1,3 @@
from gomspace.gomspace_common import GsInfo, GomspaceOpCodes
from tmtc.power.common_power import (
PowerOpCodes,
PowerInfo,
@ -13,8 +12,18 @@ from config.object_ids import (
PDU_2_HANDLER_ID,
get_object_ids,
)
from tmtc.power.pdu1 import pdu1_req_hk_cmds, pdu1_cmds, add_pdu1_common_defs
from tmtc.power.pdu2 import pdu2_req_hk_cmds, add_pdu2_common_defs, pdu2_cmds
from tmtc.power.pdu1 import (
pdu1_req_hk_cmds,
pdu1_switch_cmds,
add_pdu1_common_defs,
add_pdu1_cmds,
)
from tmtc.power.pdu2 import (
pdu2_req_hk_cmds,
add_pdu2_common_defs,
pdu2_cmds,
add_pdu2_cmds,
)
from tmtccmd import get_console_logger
from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
@ -28,29 +37,29 @@ LOGGER = get_console_logger()
def pack_power_commands(q: DefaultPusQueueHelper, op_code: str):
pdu1_cmds(q, op_code)
pdu1_switch_cmds(q, op_code)
pdu2_cmds(q, op_code)
if op_code in PowerOpCodes.INFO_CORE:
pdu1_req_hk_cmds(q, GomspaceOpCodes.REQUEST_CORE_HK_ONCE[0])
pdu2_req_hk_cmds(q, GomspaceOpCodes.REQUEST_CORE_HK_ONCE[0])
p60_dock_req_hk_cmds(q, GomspaceOpCodes.REQUEST_CORE_HK_ONCE[0])
acu_req_hk_cmds(q, GomspaceOpCodes.REQUEST_CORE_HK_ONCE[0])
pdu1_req_hk_cmds(q, PowerOpCodes.REQUEST_CORE_HK_ONCE[0])
pdu2_req_hk_cmds(q, PowerOpCodes.REQUEST_CORE_HK_ONCE[0])
p60_dock_req_hk_cmds(q, PowerOpCodes.REQUEST_CORE_HK_ONCE[0])
acu_req_hk_cmds(q, PowerOpCodes.REQUEST_CORE_HK_ONCE[0])
q.add_wait_seconds(8.0)
elif op_code in PowerOpCodes.INFO_AUX:
pdu1_req_hk_cmds(q, GomspaceOpCodes.REQUEST_AUX_HK_ONCE[0])
pdu2_req_hk_cmds(q, GomspaceOpCodes.REQUEST_AUX_HK_ONCE[0])
p60_dock_req_hk_cmds(q, GomspaceOpCodes.REQUEST_AUX_HK_ONCE[0])
acu_req_hk_cmds(q, GomspaceOpCodes.REQUEST_AUX_HK_ONCE[0])
pdu1_req_hk_cmds(q, PowerOpCodes.REQUEST_AUX_HK_ONCE[0])
pdu2_req_hk_cmds(q, PowerOpCodes.REQUEST_AUX_HK_ONCE[0])
p60_dock_req_hk_cmds(q, PowerOpCodes.REQUEST_AUX_HK_ONCE[0])
acu_req_hk_cmds(q, PowerOpCodes.REQUEST_AUX_HK_ONCE[0])
q.add_wait_seconds(8.0)
elif op_code in PowerOpCodes.INFO_ALL:
pdu1_req_hk_cmds(q, GomspaceOpCodes.REQUEST_CORE_HK_ONCE[0])
pdu2_req_hk_cmds(q, GomspaceOpCodes.REQUEST_CORE_HK_ONCE[0])
pdu1_req_hk_cmds(q, GomspaceOpCodes.REQUEST_AUX_HK_ONCE[0])
pdu2_req_hk_cmds(q, GomspaceOpCodes.REQUEST_AUX_HK_ONCE[0])
p60_dock_req_hk_cmds(q, GomspaceOpCodes.REQUEST_CORE_HK_ONCE[0])
p60_dock_req_hk_cmds(q, GomspaceOpCodes.REQUEST_AUX_HK_ONCE[0])
acu_req_hk_cmds(q, GomspaceOpCodes.REQUEST_CORE_HK_ONCE[0])
acu_req_hk_cmds(q, GomspaceOpCodes.REQUEST_AUX_HK_ONCE[0])
pdu1_req_hk_cmds(q, PowerOpCodes.REQUEST_CORE_HK_ONCE[0])
pdu2_req_hk_cmds(q, PowerOpCodes.REQUEST_CORE_HK_ONCE[0])
pdu1_req_hk_cmds(q, PowerOpCodes.REQUEST_AUX_HK_ONCE[0])
pdu2_req_hk_cmds(q, PowerOpCodes.REQUEST_AUX_HK_ONCE[0])
p60_dock_req_hk_cmds(q, PowerOpCodes.REQUEST_CORE_HK_ONCE[0])
p60_dock_req_hk_cmds(q, PowerOpCodes.REQUEST_AUX_HK_ONCE[0])
acu_req_hk_cmds(q, PowerOpCodes.REQUEST_CORE_HK_ONCE[0])
acu_req_hk_cmds(q, PowerOpCodes.REQUEST_AUX_HK_ONCE[0])
q.add_wait_seconds(8.0)
elif op_code in PowerOpCodes.RESET_ALL_GND_WDTS:
oids = get_object_ids()
@ -93,46 +102,6 @@ def add_power_cmd_defs(defs: TmtcDefinitionWrapper):
)
@tmtc_definitions_provider
def add_pdu1_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
add_pdu1_common_defs(oce)
add_gomspace_cmd_defs(oce)
oce.add(keys=GomspaceOpCodes.REQUEST_CORE_HK_ONCE, info=GsInfo.REQUEST_CORE_HK_ONCE)
oce.add(keys=GomspaceOpCodes.REQUEST_AUX_HK_ONCE, info=GsInfo.REQUEST_AUX_HK_ONCE)
oce.add(
keys=GomspaceOpCodes.PRINT_SWITCH_V_I,
info="PDU1: Print Switches, Voltages, Currents",
)
oce.add(keys=GomspaceOpCodes.GET_PARAM, info=GsInfo.GET_PARAMETER)
defs.add_service(
name=CustomServiceList.PDU1.value,
info="PDU1 Device",
op_code_entry=oce,
)
@tmtc_definitions_provider
def add_pdu2_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
add_pdu2_common_defs(oce)
add_gomspace_cmd_defs(oce)
oce.add(
keys=GomspaceOpCodes.PRINT_SWITCH_V_I,
info="PDU2: Print Switches, Voltages, Currents",
)
oce.add(
keys=GomspaceOpCodes.PRINT_LATCHUPS,
info="PDU2: Print Latchups",
)
defs.add_service(
name="pdu2",
info="PDU2 Device",
op_code_entry=oce,
)
def add_pcdu_cmds(defs: TmtcDefinitionWrapper):
add_p60_cmds(defs)
add_pdu1_cmds(defs)

19
tmtc/power/tm.py
View File

@ -1,10 +1,11 @@
import struct
from typing import List, Tuple
from tmtc.power.common_power import SetIds
from tmtccmd.util import ObjectIdBase
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
from pus_tm.defs import PrintWrapper
from gomspace.gomspace_common import SetIds, GomspaceDeviceActionIds
from gomspace.gomspace_common import GomspaceDeviceActionIds
from config.object_ids import (
PDU_1_HANDLER_ID,
PDU_2_HANDLER_ID,
@ -145,7 +146,7 @@ def handle_pdu_data(
pw = PrintWrapper(printer=printer)
current_idx = 0
priv_idx = pdu_idx - 1
if set_id == SetIds.PDU_1_AUX or set_id == SetIds.PDU_2_AUX:
if set_id == SetIds.AUX or set_id == SetIds.AUX:
fmt_str = "!hhBBBIIH"
inc_len = struct.calcsize(fmt_str)
(
@ -182,7 +183,7 @@ def handle_pdu_data(
wdt.print()
pw.dlog(f"PDU Device Types: 0:FRAM|1:ADC|2:ADC|3:TempSens|4,5,6,7:Reserved")
dev_parser.print(pw=pw)
if set_id == SetIds.PDU_1_CORE or set_id == SetIds.PDU_2_CORE:
if set_id == SetIds.CORE or set_id == SetIds.CORE:
pw.dlog(f"Received PDU HK from PDU {pdu_idx}")
current_list = []
for idx in range(len(PDU1_CHANNELS_NAMES)):
@ -223,7 +224,7 @@ def handle_pdu_data(
def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
pw = PrintWrapper(printer=printer)
if set_id == SetIds.P60_CORE:
if set_id == SetIds.CORE:
pw.dlog("Received P60 Core HK. Voltages in mV, currents in mA")
current_idx = 0
current_list = []
@ -270,7 +271,7 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
pw.dlog(temps)
pw.dlog(batt_info)
printer.print_validity_buffer(validity_buffer=hk_data[current_idx:], num_vars=9)
if set_id == SetIds.P60_AUX:
if set_id == SetIds.AUX:
pw.dlog("Received P60 AUX HK. Voltages in mV, currents in mA")
current_idx = 0
latchup_list = []
@ -349,7 +350,7 @@ def gen_six_entry_u16_list(hk_data: bytes, current_idx: int) -> Tuple[int, List[
def handle_acu_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
pw = PrintWrapper(printer=printer)
if set_id == SetIds.ACU_CORE:
if set_id == SetIds.CORE:
mppt_mode = hk_data[0]
current_idx = 1
current_idx, currents = gen_six_entry_u16_list(
@ -393,7 +394,7 @@ def handle_acu_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
printer.print_validity_buffer(
validity_buffer=hk_data[current_idx:], num_vars=12
)
if set_id == SetIds.ACU_AUX:
if set_id == SetIds.AUX:
current_idx = 0
fmt_str = "!BBB"
inc_len = struct.calcsize(fmt_str)
@ -540,6 +541,8 @@ def p60_dock_config_table_handler(pw: PrintWrapper, custom_data: bytes):
init_out_safe = unpack_array_in_data(custom_data, 0xB7, 1, 13, "B")
init_on_dly = unpack_array_in_data(custom_data, 0xC4, 2, 13, "H")
init_off_dly = unpack_array_in_data(custom_data, 0xDE, 2, 13, "H")
acu_channel_addrs = unpack_array_in_data(custom_data, 0x180, 1, 2, "B")
pdu_channel_addrs = unpack_array_in_data(custom_data, 0x186, 1, 4, "B")
pw.dlog(f"Ch Names: {ch_names}")
pw.dlog(f"{'out_on_cnt'.ljust(15)}: {out_on_cnt}")
pw.dlog(f"{'out_off_cnt'.ljust(15)}: {out_off_cnt}")
@ -547,6 +550,8 @@ def p60_dock_config_table_handler(pw: PrintWrapper, custom_data: bytes):
pw.dlog(f"{'init_out_safe'.ljust(15)}: {init_out_safe}")
pw.dlog(f"{'init_on_dly'.ljust(15)}: {init_on_dly}")
pw.dlog(f"{'init_off_dly'.ljust(15)}: {init_off_dly}")
pw.dlog(f"{'p60acu_addr'.ljust(15)}: {acu_channel_addrs}")
pw.dlog(f"{'p60pdu_addr'.ljust(15)}: {pdu_channel_addrs}")
def parse_name_list(data: bytes, name_len: int):

78
tmtc/solar_array_deployment.py Normal file
View File

@ -0,0 +1,78 @@
# -*- coding: utf-8 -*-
"""
@file solar_array_deployment.py
@brief The test function in this file simply returns a command which triggers the solar array deployment.
@author J. Meier
@date 15.02.2021
"""
import struct
from config.definitions import CustomServiceList
from config.object_ids import SOLAR_ARRAY_DEPLOYMENT_ID
from tmtccmd.config.tmtc import (
tmtc_definitions_provider,
TmtcDefinitionWrapper,
OpCodeEntry,
)
from tmtccmd.tc import service_provider
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
from tmtccmd.tc.decorator import ServiceProviderParams
from tmtccmd import get_console_logger
LOGGER = get_console_logger()
class OpCodes:
MANUAL_DEPLOYMENT = "man_depl"
class Info:
MANUAL_DEPLOYMENT = "Manual Solar Array Deployment"
class ActionIds:
MANUAL_DEPLOYMENT = 5
@tmtc_definitions_provider
def pack_sa_depl_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
oce.add(keys=OpCodes.MANUAL_DEPLOYMENT, info=Info.MANUAL_DEPLOYMENT)
defs.add_service(
name=CustomServiceList.SA_DEPLYOMENT,
info="Solar Array Deployment",
op_code_entry=oce,
)
@service_provider(CustomServiceList.SA_DEPLYOMENT)
def pack_solar_array_deployment_test_into(p: ServiceProviderParams):
q = p.queue_helper
user_data = bytearray()
while True:
burn_time = int(input("Please specify burn time in seconds [0-120 secs]: "))
if burn_time < 0 or burn_time > 120:
LOGGER.warning(f"Invalid burn time {burn_time}")
continue
user_data.extend(struct.pack("!I", burn_time))
break
while True:
dry_run = input("Dry run? [y/n]: ")
if dry_run in ["yes", "y", "1"]:
dry_run = 1
elif dry_run in ["no", "n", "0"]:
dry_run = 0
else:
LOGGER.warning("Invalid input for dry run parameter")
continue
user_data.append(dry_run)
break
if dry_run == 1:
dry_run_str = " as dry run"
else:
dry_run_str = ""
q.add_log_cmd(f"Testing S/A Deployment with burn time {burn_time}{dry_run_str}")
command = make_fsfw_action_cmd(
SOLAR_ARRAY_DEPLOYMENT_ID, ActionIds.MANUAL_DEPLOYMENT, user_data
)
q.add_pus_tc(command)