eive/eive-tmtc
eive
/
eive-tmtc
13
0
Fork 0
Code Issues Pull Requests 1 Projects Releases 96 Wiki Activity

Merge remote-tracking branch 'origin/main' into irini

This commit is contained in:
Robin Müller 2022-08-31 15:46:27 +02:00
parent c84f2c2b35 05dd173860
commit 395dbe9df3
27 changed files with 1027 additions and 900 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.run/PLOC_SUPV.run.xml
View File

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

4
.run/Reboot_0_0.run.xml → .run/Reboot 0 0.run.xml
View File

@ -12,8 +12,8 @@
<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$/tmtccli.py" />
<option name="PARAMETERS" value="-s core -o reboot_0_0 -d 4" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtcc.py" />
<option name="PARAMETERS" value="tmtc -s core -o reboot_00 -d 4" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="false" />
<option name="MODULE_MODE" value="false" />

2
.run/Reboot_0_1.run.xml → .run/Reboot 0 1.run.xml
View File

@ -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="-s core -o reboot_0_1 -d 4" />
<option name="PARAMETERS" value="tmtc -s core -o reboot_01 -d 4" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="false" />
<option name="MODULE_MODE" value="false" />

2
.run/Reboot_1_0.run.xml → .run/Reboot 1 0.run.xml
View File

@ -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="-s core -o reboot_1_0 -d 4" />
<option name="PARAMETERS" value="tmtc -s core -o reboot_10 -d 4" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="false" />
<option name="MODULE_MODE" value="false" />

2
.run/Reboot_1_1.run.xml → .run/Reboot 1 1.run.xml
View File

@ -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="-s core -o reboot_1_1 -d 4" />
<option name="PARAMETERS" value="tmtc -s core -o reboot_11 -d 4" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="false" />
<option name="MODULE_MODE" value="false" />

2
.run/tmtccli.run.xml
View File

@ -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="-d 6" />
<option name="PARAMETERS" value="tmtc -d 6" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="true" />
<option name="MODULE_MODE" value="false" />

3
CHANGELOG.md
View File

@ -15,6 +15,9 @@ list yields a list of all related PRs for each release.
- Major Update for `tmtccmd` and `spacepackets` dependencies which improves user API significantly.
PR: https://egit.irs.uni-stuttgart.de/eive/eive-tmtc/pulls/102
- Add commands to request MGM HK or enable/disable periodic HK for it
- Update power commands: Extend param get and set to all
gomspace devices.
PR: https://egit.irs.uni-stuttgart.de/eive/eive-tmtc/pulls/108
- Update PLOC supervisor commands
PR: https://egit.irs.uni-stuttgart.de/eive/eive-tmtc/pulls/107

55
gomspace/gomspace_common.py
View File

@ -11,6 +11,7 @@ import struct
from typing import Union
from spacepackets.ecss import PusTelecommand
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
from tmtccmd.util import ObjectIdU32
@ -22,26 +23,30 @@ class GomspaceDeviceActionIds(enum.IntEnum):
PARAM_SET = 255
WDT_RESET = 9
REQUEST_HK_TABLE = 16
REQUEST_CONFIG_TABLE = 17
PRINT_SWITCH_V_I = 32
PRINT_LATCHUPS = 33
class GomspaceOpCodes:
# Request HK
REQUEST_CORE_HK_ONCE = ["hk-core", "128"]
REQUEST_AUX_HK_ONCE = ["hk-aux", "129"]
PRINT_SWITCH_V_I = ["print-switch-vi", "130"]
PRINT_LATCHUPS = ["print-latchups", "131"]
GET_PARAM = ["get-param", "132"]
SET_PARAM = ["set-param", "133"]
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_PARAM = ["set_param"]
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 parameter"
REQUEST_CONFIG_TABLE = "Request Config Table"
class SetIds:
@ -75,6 +80,12 @@ class Channel:
off = 0
def pack_request_config_command(object_id: bytes) -> PusTelecommand:
return make_fsfw_action_cmd(
object_id=object_id, action_id=GomspaceDeviceActionIds.REQUEST_CONFIG_TABLE
)
def pack_get_param_command(
object_id: bytes,
table_id: int,
@ -103,11 +114,7 @@ def pack_get_param_command(
def pack_set_param_command(
object_id: bytes,
memory_address: bytes,
parameter_size: int,
parameter: int,
ssc: int = 0,
object_id: bytes, memory_address: bytes, parameter_size: int, parameter: int
) -> PusTelecommand:
"""Function to generate a command to set a parameter
:param object_id: The object id of the gomspace device handler.
@ -115,7 +122,6 @@ def pack_set_param_command(
:param parameter: The parameter value to set.
:param parameter_size: Size of the value to set. There are uint8_t, uint16_t and uint32_t
in the device tables.
:param ssc:
:return: The command as bytearray.
"""
action_id = GomspaceDeviceActionIds.PARAM_SET
@ -136,7 +142,30 @@ def pack_set_param_command(
app_data.append(byte_three)
app_data.append(byte_four)
return make_fsfw_action_cmd(
object_id=object_id, action_id=action_id, user_data=app_data, ssc=ssc
object_id=object_id, action_id=action_id, user_data=app_data
)
def prompt_and_pack_get_param_command(q: DefaultPusQueueHelper, object_id: ObjectIdU32):
table_id = int(input("Specify table ID: "))
memory_address = int(input("Specify memory address: 0x"), 16)
parameter_size = int(input("Specify parameter size: "))
q.add_pus_tc(
pack_get_param_command(
object_id.as_bytes, table_id, memory_address, parameter_size
)
)
def prompt_and_pack_set_param_command(q: DefaultPusQueueHelper, object_id: ObjectIdU32):
memory_address = int(input("Specify memory address: 0x"), 16)
memory_address = struct.pack("!H", memory_address)
parameter_size = int(input("Specify parameter size: "))
parameter = int(input("Specify parameter: "))
q.add_pus_tc(
pack_set_param_command(
object_id.as_bytes, memory_address, parameter_size, parameter
)
)

2
pus_tc/devs/__init__.py
View File

@ -1 +1 @@
from . import power

442
pus_tc/devs/common_power.py
View File

@ -1,442 +0,0 @@
import enum
from config.object_ids import PDU_1_HANDLER_ID, PDU_2_HANDLER_ID
from gomspace.gomspace_common import (
pack_set_param_command,
Channel,
GomspaceOpCodes,
GsInfo,
SetIds,
)
from gomspace.gomspace_pdu_definitions import PDU_CONFIG_LIST
from tmtccmd.config import OpCodeEntry
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import (
make_sid,
generate_one_diag_command,
generate_one_hk_command,
)
class Pdu1ChIndex(enum.IntEnum):
TCS = 0
SYRLINKS = 1
STR = 2
MGT = 3
SUS_N = 4
SCEX = 5
PLOC = 6
ACS_A = 7
class Pdu1InfoBase:
TCS = "Switch TCS Board"
SYRLINKS = "Switch Syrlinks (COM)"
STR = "Switch Startracker"
MGT = "Switch Magnetorquer"
SUS_N = "Switch Sun Sensor Board Nominal"
SCEX = "Switch Solar Cell Experiment"
PLOC = "Switch Payload On-Board Computer"
ACS_A = "Switch ACS Board A-Side"
class Pdu2InfoBase:
PL_PCDU_BAT_NOM = "Switch PL PCDU Nominal Battery Channel"
RW = "Switch Reaction Wheel"
HEATER = "Switch Heater"
SUS_R = "Switch Sun Sensor Board Redundant"
SOLAR_ARRAY_DEPL = "Switch Solar Array Deployment"
PL_PCDU_BAT_RED = "Switch PL PCDU Redundant Battery Channel"
ACS_B = "Switch ACS Board B-Side"
PL_CAM = "Switch Payload Camera"
class PowerInfo:
INFO_CORE = "Core Information"
INFO_AUX = "Auxiliary Information"
INFO_ALL = "All Information"
class Pdu2ChIndex(enum.IntEnum):
PL_PCDU_BAT_NOM = 1
RW = 2
HEATER = 3
SUS_R = 4
SOLAR_ARRAY_DEPL = 5
PL_PCDU_BAT_RED = 6
ACS_B = 7
PL_CAM = 8
class PowerOpCodes:
# PDU 1
TCS_ON = ["tcs-on"]
TCS_OFF = ["tcs-off"]
SYRLINKS_ON = ["syrlinks-on"]
SYRLINKS_OFF = ["syrlinks-off"]
STAR_TRACKER_ON = ["str-on"]
STAR_TRACKER_OFF = ["str-off"]
MGT_ON = ["mgt-on"]
MGT_OFF = ["mgt-off"]
SUS_N_ON = ["sus-nom-on"]
SUS_N_OFF = ["sus-nom-off"]
SCEX_ON = ["scex-on"]
SCEX_OFF = ["scex-off"]
PLOC_ON = ["ploc-on"]
PLOC_OFF = ["ploc-off"]
ACS_A_ON = ["acs-a-on"]
ACS_A_OFF = ["acs-a-off"]
# PDU 2
PL_PCDU_VBAT_NOM_ON = ["plpcdu-vbat-nom-on"]
PL_PCDU_VBAT_NOM_OFF = ["plpcdu-vbat-nom-off"]
RW_ON = ["rw-on"]
RW_OFF = ["rw-off"]
HEATER_ON = ["heater-on"]
HEATER_OFF = ["heater-off"]
SUS_R_ON = ["sus-red-on"]
SUS_R_OFF = ["sus-red-off"]
SOLAR_ARRAY_DEPL_ON = ["sa-depl-on"]
SOLAR_ARRAY_DEPL_OFF = ["sa-depl-off"]
PL_PCDU_VBAT_RED_ON = ["plpcdu-vbat-red-on"]
PL_PCDU_VBAT_RED_OFF = ["plpcdu-vbat-red-off"]
ACS_B_ON = ["acs-b-on"]
ACS_B_OFF = ["acs-b-off"]
PL_CAM_ON = ["cam-on"]
PL_CAM_OFF = ["cam-off"]
INFO_CORE = ["info"]
INFO_AUX = ["info-aux"]
INFO_ALL = ["info-all"]
def info_on_pdu1(base: str) -> str:
return "PDU1: " + base + " on"
def info_off_pdu1(base: str) -> str:
return "PDU1: " + base + " off"
def info_on_pdu2(base: str) -> str:
return "PDU2: " + base + " on"
def info_off_pdu2(base: str) -> str:
return "PDU2: " + base + " off"
def add_pdu1_common_defs(oce: OpCodeEntry):
oce.add(keys=PowerOpCodes.TCS_ON, info=info_on_pdu1(Pdu1InfoBase.TCS))
oce.add(keys=PowerOpCodes.TCS_OFF, info=info_off_pdu1(Pdu1InfoBase.TCS))
oce.add(keys=PowerOpCodes.STAR_TRACKER_ON, info=info_on_pdu1(Pdu1InfoBase.STR))
oce.add(keys=PowerOpCodes.STAR_TRACKER_OFF, info=info_off_pdu1(Pdu1InfoBase.STR))
oce.add(keys=PowerOpCodes.SUS_N_ON, info=info_on_pdu1(Pdu1InfoBase.SUS_N))
oce.add(keys=PowerOpCodes.SUS_N_OFF, info=info_off_pdu1(Pdu1InfoBase.SUS_N))
oce.add(keys=PowerOpCodes.ACS_A_ON, info=info_on_pdu1(Pdu1InfoBase.ACS_A))
oce.add(keys=PowerOpCodes.ACS_A_OFF, info=info_off_pdu1(Pdu1InfoBase.ACS_A))
oce.add(keys=PowerOpCodes.SYRLINKS_ON, info=info_on_pdu1(Pdu1InfoBase.SYRLINKS))
oce.add(keys=PowerOpCodes.SYRLINKS_OFF, info=info_off_pdu1(Pdu1InfoBase.SYRLINKS))
oce.add(keys=PowerOpCodes.MGT_ON, info=info_on_pdu1(Pdu1InfoBase.MGT))
oce.add(keys=PowerOpCodes.MGT_OFF, info=info_off_pdu1(Pdu1InfoBase.MGT))
oce.add(keys=PowerOpCodes.PLOC_ON, info=info_on_pdu1(Pdu1InfoBase.PLOC))
oce.add(keys=PowerOpCodes.PLOC_OFF, info=info_off_pdu1(Pdu1InfoBase.PLOC))
oce.add(keys=PowerOpCodes.SCEX_ON, info=info_on_pdu1(Pdu1InfoBase.SCEX))
oce.add(keys=PowerOpCodes.SCEX_OFF, info=info_off_pdu1(Pdu1InfoBase.SCEX))
def add_pdu2_common_defs(oce: OpCodeEntry):
oce.add(keys=PowerOpCodes.ACS_B_ON, info=info_on_pdu2(Pdu2InfoBase.ACS_B))
oce.add(keys=PowerOpCodes.ACS_B_OFF, info=info_off_pdu2(Pdu2InfoBase.ACS_B))
oce.add(keys=PowerOpCodes.SUS_R_ON, info=info_on_pdu2(Pdu2InfoBase.SUS_R))
oce.add(keys=PowerOpCodes.SUS_R_OFF, info=info_off_pdu2(Pdu2InfoBase.SUS_R))
oce.add(keys=PowerOpCodes.RW_ON, info=info_on_pdu2(Pdu2InfoBase.RW))
oce.add(keys=PowerOpCodes.RW_OFF, info=info_off_pdu2(Pdu2InfoBase.RW))
oce.add(
keys=PowerOpCodes.PL_PCDU_VBAT_NOM_ON,
info=info_on_pdu2(Pdu2InfoBase.PL_PCDU_BAT_NOM),
)
oce.add(
keys=PowerOpCodes.PL_PCDU_VBAT_NOM_OFF,
info=info_off_pdu2(Pdu2InfoBase.PL_PCDU_BAT_NOM),
)
oce.add(
keys=PowerOpCodes.PL_PCDU_VBAT_RED_ON,
info=info_on_pdu2(Pdu2InfoBase.PL_PCDU_BAT_RED),
)
oce.add(
keys=PowerOpCodes.PL_PCDU_VBAT_RED_OFF,
info=info_off_pdu2(Pdu2InfoBase.PL_PCDU_BAT_RED),
)
oce.add(keys=PowerOpCodes.HEATER_ON, info=info_on_pdu2(Pdu2InfoBase.HEATER))
oce.add(keys=PowerOpCodes.HEATER_OFF, info=info_off_pdu2(Pdu2InfoBase.HEATER))
oce.add(
keys=PowerOpCodes.SOLAR_ARRAY_DEPL_ON,
info=info_on_pdu2(Pdu2InfoBase.SOLAR_ARRAY_DEPL),
)
oce.add(
keys=PowerOpCodes.SOLAR_ARRAY_DEPL_OFF,
info=info_off_pdu2(Pdu2InfoBase.SOLAR_ARRAY_DEPL),
)
oce.add(keys=PowerOpCodes.PL_CAM_ON, info=info_on_pdu2(Pdu2InfoBase.PL_CAM))
oce.add(keys=PowerOpCodes.PL_CAM_OFF, info=info_off_pdu2(Pdu2InfoBase.PL_CAM))
def pdu1_cmds(q: DefaultPusQueueHelper, op_code: str):
if op_code in PowerOpCodes.TCS_ON:
tcs_on_cmd(q)
elif op_code in PowerOpCodes.TCS_OFF:
tcs_off_cmd(q)
elif op_code in PowerOpCodes.SYRLINKS_ON:
syrlinks_on_cmd(q)
elif op_code in PowerOpCodes.SYRLINKS_OFF:
syrlinks_off_cmd(q)
elif op_code in PowerOpCodes.STAR_TRACKER_ON:
startracker_on_cmd(q)
elif op_code in PowerOpCodes.STAR_TRACKER_OFF:
startracker_off_cmd(q)
elif op_code in PowerOpCodes.MGT_ON:
mgt_on_cmd(q)
elif op_code in PowerOpCodes.MGT_OFF:
mgt_off_cmd(q)
elif op_code in PowerOpCodes.SUS_N_ON:
sun_sensor_nominal_on_cmd(q)
elif op_code in PowerOpCodes.SUS_N_OFF:
sun_sensor_nominal_off_cmd(q)
elif op_code in PowerOpCodes.SCEX_ON:
solar_cell_experiment_on_cmd(q)
elif op_code in PowerOpCodes.SCEX_OFF:
solar_cell_experiment_off_cmd(q)
elif op_code in PowerOpCodes.PLOC_ON:
ploc_on_cmd(q)
elif op_code in PowerOpCodes.PLOC_OFF:
ploc_off_cmd(q)
elif op_code in PowerOpCodes.ACS_A_ON:
acs_board_a_on_cmd(q)
elif op_code in PowerOpCodes.ACS_A_OFF:
acs_board_a_off_cmd(q)
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)
elif op_code in PowerOpCodes.PL_PCDU_VBAT_NOM_OFF:
pl_pcdu_bat_nom_off_cmd(q)
elif op_code in PowerOpCodes.RW_ON:
reaction_wheel_on_cmd(q)
elif op_code in PowerOpCodes.RW_OFF:
reaction_wheel_off_cmd(q)
elif op_code in PowerOpCodes.HEATER_ON:
heater_on_cmd(q)
elif op_code in PowerOpCodes.HEATER_OFF:
heater_off_cmd(q)
elif op_code in PowerOpCodes.SUS_R_ON:
sus_red_on_cmd(q)
elif op_code in PowerOpCodes.SUS_R_OFF:
sus_red_off_cmd(q)
elif op_code in PowerOpCodes.SOLAR_ARRAY_DEPL_ON:
solar_array_deployment_on_cmd(q)
elif op_code in PowerOpCodes.SOLAR_ARRAY_DEPL_OFF:
solar_array_deployment_off_cmd(q)
elif op_code in PowerOpCodes.PL_PCDU_VBAT_RED_ON:
pl_pcdu_bat_red_on_cmd(q)
elif op_code in PowerOpCodes.PL_PCDU_VBAT_RED_OFF:
pl_pcdu_bat_nom_off_cmd(q)
elif op_code in PowerOpCodes.ACS_B_ON:
acs_board_b_side_on_cmd(q)
elif op_code in PowerOpCodes.ACS_B_OFF:
acs_board_b_side_off_cmd(q)
elif op_code in PowerOpCodes.PL_CAM_ON:
payload_camera_on_cmd(q)
elif op_code in PowerOpCodes.PL_CAM_OFF:
payload_camera_off_cmd(q)
def pdu1_req_hk_cmds(q: DefaultPusQueueHelper, op_code: str):
if op_code in GomspaceOpCodes.REQUEST_CORE_HK_ONCE:
q.add_log_cmd(f"PDU1: {GsInfo.REQUEST_CORE_HK_ONCE}")
hk_sid = make_sid(object_id=PDU_1_HANDLER_ID, set_id=SetIds.PDU_1_CORE)
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"PDU1: {GsInfo.REQUEST_AUX_HK_ONCE}")
hk_sid = make_sid(object_id=PDU_1_HANDLER_ID, set_id=SetIds.PDU_1_AUX)
q.add_pus_tc(generate_one_hk_command(sid=hk_sid))
def pdu2_req_hk_cmds(q: DefaultPusQueueHelper, op_code: str):
if op_code in GomspaceOpCodes.REQUEST_CORE_HK_ONCE:
q.add_log_cmd(f"PDU2: {GsInfo.REQUEST_CORE_HK_ONCE}")
hk_sid = make_sid(object_id=PDU_2_HANDLER_ID, set_id=SetIds.PDU_2_CORE)
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"PDU2: {GsInfo.REQUEST_AUX_HK_ONCE}")
hk_sid = make_sid(object_id=PDU_2_HANDLER_ID, set_id=SetIds.PDU_2_AUX)
q.add_pus_tc(generate_one_hk_command(sid=hk_sid))
def tcs_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.TCS, Pdu1ChIndex.TCS)
def tcs_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.TCS, Pdu1ChIndex.TCS)
def syrlinks_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SYRLINKS, Pdu1ChIndex.SYRLINKS)
def syrlinks_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SYRLINKS, Pdu1ChIndex.SYRLINKS)
def startracker_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.STR, Pdu1ChIndex.STR)
def startracker_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.STR, Pdu1ChIndex.STR)
def mgt_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.MGT, Pdu1ChIndex.MGT)
def mgt_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.MGT, Pdu1ChIndex.MGT)
def sun_sensor_nominal_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SUS_N, Pdu1ChIndex.SUS_N)
def sun_sensor_nominal_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SUS_N, Pdu1ChIndex.SUS_N)
def solar_cell_experiment_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SCEX, Pdu1ChIndex.SCEX)
def solar_cell_experiment_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SCEX, Pdu1ChIndex.SCEX)
def ploc_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.PLOC, Pdu1ChIndex.PLOC)
def ploc_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.PLOC, Pdu1ChIndex.PLOC)
def acs_board_a_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.ACS_A, Pdu1ChIndex.ACS_A)
def acs_board_a_off_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.ACS_A, Pdu1ChIndex.ACS_A)
def pl_pcdu_bat_nom_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(
PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_PCDU_BAT_NOM, Pdu2ChIndex.PL_PCDU_BAT_NOM
)
def pl_pcdu_bat_nom_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(
PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_PCDU_BAT_NOM, Pdu2ChIndex.PL_PCDU_BAT_NOM
)
def reaction_wheel_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.RW, Pdu2ChIndex.RW)
def reaction_wheel_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.RW, Pdu2ChIndex.RW)
def heater_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.HEATER, Pdu2ChIndex.HEATER)
def heater_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.HEATER, Pdu2ChIndex.HEATER)
def sus_red_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.SUS_R, Pdu2ChIndex.SUS_R)
def sus_red_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.SUS_R, Pdu2ChIndex.SUS_R)
def solar_array_deployment_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(
PDU_2_HANDLER_ID, q, Pdu2InfoBase.SOLAR_ARRAY_DEPL, Pdu2ChIndex.SOLAR_ARRAY_DEPL
)
def solar_array_deployment_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(
PDU_2_HANDLER_ID, q, Pdu2InfoBase.SOLAR_ARRAY_DEPL, Pdu2ChIndex.SOLAR_ARRAY_DEPL
)
def pl_pcdu_bat_red_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(
PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_PCDU_BAT_RED, Pdu2ChIndex.PL_PCDU_BAT_RED
)
def pl_pcdu_bat_red_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(
PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_PCDU_BAT_RED, Pdu2ChIndex.PL_PCDU_BAT_RED
)
def acs_board_b_side_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.ACS_B, Pdu2ChIndex.ACS_B)
def acs_board_b_side_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.ACS_B, Pdu2ChIndex.ACS_B)
def payload_camera_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_CAM, Pdu2ChIndex.PL_CAM)
def payload_camera_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_CAM, Pdu2ChIndex.PL_CAM)
def generic_on_cmd(
object_id: bytes, q: DefaultPusQueueHelper, info_str: str, out_idx: int
):
q.add_log_cmd(info_str + " on")
q.add_pus_tc(
pack_set_param_command(
object_id,
PDU_CONFIG_LIST[out_idx].parameter_address,
PDU_CONFIG_LIST[out_idx].parameter_size,
Channel.on,
)
)
def generic_off_cmd(
object_id: bytes, q: DefaultPusQueueHelper, info_str: str, out_idx: int
):
q.add_log_cmd(info_str + " off")
q.add_pus_tc(
pack_set_param_command(
object_id,
PDU_CONFIG_LIST[out_idx].parameter_address,
PDU_CONFIG_LIST[out_idx].parameter_size,
Channel.off,
)
)

120
pus_tc/devs/pdu1.py
View File

@ -1,120 +0,0 @@
# -*- coding: utf-8 -*-
"""PDU1 is mounted on the X2 slot of the P60 dock
@author J. Meier
@date 17.12.2020
"""
import gomspace.gomspace_common as gs
from pus_tc.devs.common_power import pdu1_cmds, pdu1_req_hk_cmds
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import (
generate_one_hk_command,
make_sid,
generate_one_diag_command,
)
from gomspace.gomspace_common import *
from gomspace.gomspace_pdu_definitions import *
from config.object_ids import PDU_1_HANDLER_ID
class PDU1TestProcedure:
"""
@brief Use this class to define the tests to perform for the PDU2.
@details Setting all to True will run all tests.
Setting all to False will only run the tests set to True.
"""
all = False
reboot = False
ping = False
read_temperature = False
turn_channel_2_on = False # Star Tracker connected to this channel (5V)
turn_channel_2_off = False
turn_channel_3_on = False # MTQ connected to this channel (5V)
turn_channel_3_off = False
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_req_hk_cmds(q, op_code)
if op_code in GomspaceOpCodes.PRINT_SWITCH_V_I:
q.add_log_cmd("PDU1: Print Switches, Voltages, Currents")
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("PDU1: Print Latchups")
q.add_pus_tc(
make_fsfw_action_cmd(
object_id=objb, action_id=GomspaceDeviceActionIds.PRINT_LATCHUPS
)
)
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])
q.add_pus_tc(pack_ping_command(object_id, ping_data))
if PDU1TestProcedure.all or PDU1TestProcedure.read_temperature:
q.add_log_cmd("PDU1: Testing temperature reading")
q.add_pus_tc(
pack_get_param_command(
objb,
TableIds.hk,
PduHkTable.temperature.parameter_address,
PduHkTable.temperature.parameter_size,
)
)
if PDU1TestProcedure.all or PDU1TestProcedure.turn_channel_2_on:
q.add_log_cmd("PDU1: Turn channel 2 on (Star Tracker)")
q.add_pus_tc(
pack_set_param_command(
objb,
PduConfigTable.out_en_2.parameter_address,
PduConfigTable.out_en_2.parameter_size,
Channel.on,
)
)
if PDU1TestProcedure.all or PDU1TestProcedure.turn_channel_2_off:
q.add_log_cmd("PDU1: Turn channel 2 off (Star Tracker)")
q.add_pus_tc(
pack_set_param_command(
objb,
PduConfigTable.out_en_2.parameter_address,
PduConfigTable.out_en_2.parameter_size,
Channel.off,
)
)
if PDU1TestProcedure.all or PDU1TestProcedure.turn_channel_3_on:
q.add_log_cmd("PDU1: Turn channel 3 on (MTQ)")
q.add_pus_tc(
pack_set_param_command(
objb,
PduConfigTable.out_en_3.parameter_address,
PduConfigTable.out_en_3.parameter_size,
Channel.on,
)
)
if PDU1TestProcedure.all or PDU1TestProcedure.turn_channel_3_off:
q.add_log_cmd("PDU1: Turn channel 3 off (MTQ)")
q.add_pus_tc(
pack_set_param_command(
objb,
PduConfigTable.out_en_3.parameter_address,
PduConfigTable.out_en_3.parameter_size,
Channel.off,
)
)
if op_code in GomspaceOpCodes.SET_PARAM:
q.add_log_cmd(f"PDU1: {GsInfo.SET_PARAMETER}")
memory_address = int(input("Specify memory address: 0x"), 16)
memory_address = struct.pack("!H", memory_address)
parameter_size = int(input("Specify parameter size: "))
parameter = int(input("Specify parameter: "))
q.add_pus_tc(
gs.pack_set_param_command(
object_id.as_bytes, memory_address, parameter_size, parameter
)
)

139
pus_tc/devs/pdu2.py
View File

@ -1,139 +0,0 @@
# -*- coding: utf-8 -*-
"""
@file tmtcc_tc_pdu2.py
@brief PDU2 tests
@details PDU2 is mounted on the X4 slot of the P60 dock
@author J. Meier
@date 17.12.2020
"""
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import (
generate_one_hk_command,
generate_one_diag_command,
make_sid,
)
from pus_tc.devs.common_power import pdu2_cmds, pdu2_req_hk_cmds
from gomspace.gomspace_common import *
from gomspace.gomspace_pdu_definitions import *
from config.object_ids import PDU_2_HANDLER_ID
class PDU2TestProcedure:
"""
@brief Use this class to define the tests to perform for the PDU2.
@details Setting all to True will run all tests.
Setting all to False will only run the tests set to True.
"""
all = False
reboot = False
read_gnd_wdt = False
gnd_wdt_reset = False
ping = False
channel_2_off = False # Reaction wheels 5V
read_temperature = False
read_channel_2_state = False # Reaction wheels 5V
read_cur_lu_lim_0 = False # OBC
channel_2_on = False # Reaction wheels 5V
invalid_table_id_test = (
False # Test to check if software properly handles invalid table ids
)
invalid_address_test = (
False # Test to check if software properly handles invalid addresses
)
invalid_parameter_size_test = False
request_hk_table = False
def pack_pdu2_commands(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code: str):
q.add_log_cmd("Testing PDU2")
objb = object_id.as_bytes
pdu2_cmds(q, op_code)
pdu2_req_hk_cmds(q, op_code)
if op_code in GomspaceOpCodes.PRINT_SWITCH_V_I:
q.add_log_cmd(f"PDU2: {GsInfo.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("PDU2: Print Latchups")
q.add_pus_tc(
make_fsfw_action_cmd(
object_id=objb, action_id=GomspaceDeviceActionIds.PRINT_LATCHUPS
)
)
if PDU2TestProcedure.all or PDU2TestProcedure.reboot:
q.add_log_cmd("PDU2: Reboot")
q.add_pus_tc(pack_reboot_command(object_id))
if PDU2TestProcedure.all or PDU2TestProcedure.read_gnd_wdt:
q.add_log_cmd("PDU2: Reading ground watchdog timer value")
q.add_pus_tc(
pack_get_param_command(
objb,
TableIds.hk,
PduHkTable.wdt_gnd_left.parameter_address,
PduHkTable.wdt_gnd_left.parameter_size,
)
)
if PDU2TestProcedure.all or PDU2TestProcedure.gnd_wdt_reset:
q.add_log_cmd("PDU2: Testing ground watchdog reset")
q.add_pus_tc(pack_gnd_wdt_reset_command(object_id))
if PDU2TestProcedure.all or PDU2TestProcedure.ping:
q.add_log_cmd("PDU2: Ping Test")
ping_data = bytearray([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
q.add_pus_tc(pack_ping_command(object_id, ping_data))
if PDU2TestProcedure.all or PDU2TestProcedure.channel_2_on:
q.add_log_cmd("PDU2: Testing setting output channel 2 on (TCS Heater)")
q.add_pus_tc(
pack_set_param_command(
objb,
PduConfigTable.out_en_2.parameter_address,
PduConfigTable.out_en_2.parameter_size,
Channel.on,
)
)
if PDU2TestProcedure.all or PDU2TestProcedure.read_temperature:
q.add_log_cmd("PDU2: Testing temperature reading")
q.add_pus_tc(
pack_get_param_command(
objb,
TableIds.hk,
PduHkTable.temperature.parameter_address,
PduHkTable.temperature.parameter_size,
)
)
if PDU2TestProcedure.all or PDU2TestProcedure.read_channel_2_state:
q.add_log_cmd("PDU2: Reading output channel 2 state (TCS Heater)")
q.add_pus_tc(
pack_get_param_command(
objb,
TableIds.config,
PduConfigTable.out_en_2.parameter_address,
PduConfigTable.out_en_2.parameter_size,
)
)
if PDU2TestProcedure.all or PDU2TestProcedure.read_cur_lu_lim_0:
q.add_log_cmd("PDU2: Reading current limit value of output channel 0 (OBC)")
q.add_pus_tc(
pack_get_param_command(
objb,
TableIds.config,
PduConfigTable.cur_lu_lim_0.parameter_address,
PduConfigTable.cur_lu_lim_0.parameter_size,
)
)
if PDU2TestProcedure.all or PDU2TestProcedure.channel_2_off:
q.add_log_cmd("PDU2: Testing setting output channel 2 off")
q.add_pus_tc(
pack_set_param_command(
objb,
PduConfigTable.out_en_2.parameter_address,
PduConfigTable.out_en_2.parameter_size,
Channel.off,
)
)
if PDU2TestProcedure.all or PDU2TestProcedure.request_hk_table:
q.add_log_cmd("PDU2: Requesting housekeeping table")
q.add_pus_tc(pack_request_full_hk_table_command(object_id))

4
pus_tc/devs/star_tracker.py
View File

@ -85,6 +85,10 @@ class StarTrackerActionIds:
FIRMWARE_UPDATE = 84
class OpCodes:
NORMAL = ["2", "nml"]
class SetIds:
TEMPERATURE = 25

4
pus_tc/devs/syrlinks_hk_handler.py
View File

@ -19,6 +19,10 @@ class SetIds:
TX_REGISTERS_DATASET = 2
class OpCodes:
NORMAL = ["2", "nml"]
class CommandIds:
READ_RX_STATUS_REGISTERS = 2
SET_TX_MODE_STANDBY = 3

12
pus_tc/procedure_packer.py
View File

@ -2,7 +2,7 @@
"""
from typing import cast
from pus_tc.devs.power import pack_power_commands
from tmtc.power.power import pack_power_commands
from pus_tc.devs.rtd import pack_rtd_commands
from pus_tc.devs.scex import pack_scex_cmds
from pus_tc.system.controllers import (
@ -23,14 +23,13 @@ from tmtccmd.tc.pus_5_event import (
from tmtccmd.pus.pus_17_test import pack_service_17_ping_command
from pus_tc.service_200_mode import pack_service_200_test_into
from pus_tc.devs.p60dock import pack_p60dock_cmds
from pus_tc.devs.pdu2 import pack_pdu2_commands
from pus_tc.devs.pdu1 import pack_pdu1_commands
from pus_tc.devs.acu import pack_acu_commands
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 pus_tc.devs.tmp1075 import pack_tmp1075_test_into
from tmtc.ploc_mpsoc import pack_ploc_mpsoc_commands
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 pus_tc.devs.rad_sensor import pack_rad_sensor_test_into
@ -55,7 +54,6 @@ from config.object_ids import (
TMP_1075_2_HANDLER_ID,
HEATER_ID,
IMTQ_HANDLER_ID,
PLOC_MPSOC_ID,
RW1_ID,
RW2_ID,
RW3_ID,

33
pus_tc/system/core.py
View File

@ -32,22 +32,22 @@ class SetIds(enum.IntEnum):
class OpCodes:
REBOOT_XSC = ["0", "reboot-xsc"]
XSC_REBOOT_SELF = ["1", "reboot-self"]
XSC_REBOOT_0_0 = ["2", "reboot-00"]
XSC_REBOOT_0_1 = ["3", "reboot-01"]
XSC_REBOOT_1_0 = ["4", "reboot-10"]
XSC_REBOOT_1_1 = ["5", "reboot-11"]
REBOOT_FULL = ["6", "reboot-regular"]
GET_HK = ["7", "get-hk"]
ENABLE_REBOOT_FILE_HANDLING = ["32", "rbh-off"]
DISABLE_REBOOT_FILE_HANDLING = ["33", "rbh-on"]
RESET_ALL_REBOOT_COUNTERS = ["34", "rbh-reset-a"]
RESET_REBOOT_COUNTER_00 = ["35", "rbh-reset-00"]
RESET_REBOOT_COUNTER_01 = ["36", "rbh-reset-01"]
RESET_REBOOT_COUNTER_10 = ["37", "rbh-reset-10"]
RESET_REBOOT_COUNTER_11 = ["38", "rbh-reset-11"]
SET_MAX_REBOOT_CNT = ["39", "rbh-max-cnt"]
REBOOT_XSC = ["0", "reboot_xsc"]
XSC_REBOOT_SELF = ["1", "reboot_self"]
XSC_REBOOT_0_0 = ["2", "reboot_00"]
XSC_REBOOT_0_1 = ["3", "reboot_01"]
XSC_REBOOT_1_0 = ["4", "reboot_10"]
XSC_REBOOT_1_1 = ["5", "reboot_11"]
REBOOT_FULL = ["6", "reboot_regular"]
GET_HK = ["7", "get_hk"]
ENABLE_REBOOT_FILE_HANDLING = ["32", "rbh_off"]
DISABLE_REBOOT_FILE_HANDLING = ["33", "rbh_on"]
RESET_ALL_REBOOT_COUNTERS = ["34", "rbh_reset_a"]
RESET_REBOOT_COUNTER_00 = ["35", "rbh_reset_00"]
RESET_REBOOT_COUNTER_01 = ["36", "rbh_reset_01"]
RESET_REBOOT_COUNTER_10 = ["37", "rbh_reset_10"]
RESET_REBOOT_COUNTER_11 = ["38", "rbh_reset_11"]
SET_MAX_REBOOT_CNT = ["39", "rbh_max_cnt"]
class Info:
@ -251,6 +251,5 @@ def perform_reboot_cmd(
object_id=CORE_CONTROLLER_ID,
action_id=ActionIds.XSC_REBOOT,
user_data=tc_data,
ssc=0,
)
)

21
pus_tc/system/proc.py
View File

@ -28,6 +28,9 @@ from pus_tc.devs.mgms import MgmLis3SetIds as MgmLis3SetIds_0_2
from pus_tc.devs.mgms import MgmRm3100SetIds as MgmRm3100SetIds_1_3
from pus_tc.devs.gyros import AdisGyroSetIds as AdisGyroSetIds_0_2
from pus_tc.devs.gyros import L3gGyroSetIds as L3gGyroSetIds_1_3
from pus_tc.devs.syrlinks_hk_handler import OpCodes as SyrlinksOpCodes
from pus_tc.devs.syrlinks_hk_handler import SetIds as SyrlinksSetIds
from pus_tc.devs.star_tracker import OpCodes as StrOpCodes
from pus_tc.devs.gps import SetIds as GpsSetIds
from pus_tc.devs.imtq import ImtqSetIds
from pus_tc.devs.sus import SetIds
@ -57,7 +60,8 @@ class OpCodes:
STR_FT = ["str-ft"]
RW_FT_ONE_RW = ["rw-ft-one-rw"]
RW_FT_TWO_RWS = ["rw-ft-two-rws"]
TV_TEARDOWN_TCS_FT_OFF = ["t", "tcs-ft-off"]
SYRLINKS_FT = ["syrlinks-ft"]
TV_TEARDOWN_TCS_FT_OFF = ["teardown", "tcs-ft-off"]
class KeyAndInfo:
@ -67,6 +71,7 @@ class KeyAndInfo:
BAT_FT = ["BPX Battery", "battery functional test"]
CORE_FT = ["OBC", "OBC functional test"]
PCDU_FT = ["PCDU", "PCDU functional test"]
SYRLINKS_FT = ["Syrlinks", "Syrlinks functional test"]
RAD_SEN_FT = ["Radiation Sensor", "Radiation Sensor functional test"]
ACS_FT = ["ACS", "ACS functional test"]
MGT_FT = ["MGT", "MGT functional test"]
@ -574,6 +579,20 @@ def pack_proc_commands(q: DefaultPusQueueHelper, op_code: str):
pack_acs_command(q=q, op_code="sus-off")
if op_code in OpCodes.SYRLINKS_FT:
key = KAI.SYRLINKS_FT[0]
sid_list = [
make_sid(oids.SYRLINKS_HANDLER_ID, SyrlinksSetIds.RX_REGISTERS_DATASET),
make_sid(oids.SYRLINKS_HANDLER_ID, SyrlinksSetIds.TX_REGISTERS_DATASET),
]
# HK listening
pack_generic_hk_listening_cmds(
tc_queue=tc_queue,
proc_key=key,
sid_list=sid_list,
diag=False,
cfg=GenericHkListeningCfg.default(),
)
if op_code in OpCodes.STR_FT:
key = KAI.STR_FT[0]

43
pus_tm/action_reply_handler.py
View File

@ -1,10 +1,11 @@
import struct
from config.object_ids import *
from pus_tc.devs.imtq import ImtqActionIds
from pus_tm.defs import PrintWrapper
from tmtc.ploc_mpsoc import PlocReplyIds
from tmtc.ploc_supervisor import SupvActionIds
from pus_tc.devs.star_tracker import StarTrackerActionIds
from gomspace.gomspace_common import GomspaceDeviceActionIds
from tmtc.power.tm import handle_get_param_data_reply
from tmtccmd.logging import get_console_logger
from tmtccmd.tm import Service8FsfwTm
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
@ -21,13 +22,13 @@ def handle_action_reply(
tm_packet = Service8FsfwTm.unpack(raw_telemetry=raw_tm)
printer.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
object_id = obj_id_dict.get(tm_packet.source_object_id_as_bytes)
pw = PrintWrapper(printer)
custom_data = tm_packet.custom_data
action_id = tm_packet.action_id
generic_print_str = printer.generic_action_packet_tm_print(
packet=tm_packet, obj_id=object_id
)
print(generic_print_str)
printer.file_logger.info(generic_print_str)
pw.dlog(generic_print_str)
if object_id.as_bytes == IMTQ_HANDLER_ID:
return handle_imtq_replies(action_id, printer, custom_data)
elif object_id.as_bytes == PLOC_MPSOC_ID:
@ -36,8 +37,16 @@ def handle_action_reply(
return handle_supervisor_replies(action_id, printer, custom_data)
elif object_id.as_bytes == STAR_TRACKER_ID:
return handle_startracker_replies(action_id, printer, custom_data)
elif object_id.as_bytes == ACU_HANDLER_ID:
return handle_acu_replies(action_id, printer, custom_data)
elif object_id.as_bytes in [
ACU_HANDLER_ID,
PDU_1_HANDLER_ID,
PDU_2_HANDLER_ID,
P60_DOCK_HANDLER,
]:
return handle_get_param_data_reply(object_id, action_id, pw, custom_data)
else:
pw.dlog(f"No dedicated action reply handler found for reply from {object_id}")
pw.dlog(f"Raw Data: {tm_packet.custom_data.hex(sep=',')}")
def handle_imtq_replies(
@ -123,27 +132,3 @@ def handle_startracker_replies(
print(content_list)
printer.file_logger.info(header_list)
printer.file_logger.info(content_list)
def handle_acu_replies(
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
):
if action_id == GomspaceDeviceActionIds.PARAM_GET:
header_list = [
"Gomspace action ID" "Table ID",
"Memory Address",
"Payload length" "Payload",
]
fmt_str = "!BBHH"
(action, table_id, address, payload_length) = struct.unpack(
fmt_str, custom_data[:6]
)
content_list = [
action,
table_id,
"0x" + hex(address),
payload_length,
"0x" + custom_data[6:].hex(),
]
print(header_list)
print(content_list)

2
pus_tm/hk_handling.py
View File

@ -23,7 +23,7 @@ from pus_tm.devs.imtq_mgt import (
handle_calibrated_mtm_measurement,
handle_raw_mtm_measurement,
)
from pus_tm.system.power import handle_pdu_data, handle_p60_hk_data, handle_acu_hk_data
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

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

75
pus_tc/devs/acu.py → tmtc/power/acu.py
View File

@ -3,9 +3,13 @@
@author J. Meier, R. Mueller
@date 21.12.2020
"""
import struct
from config.definitions import CustomServiceList
from tmtc.power.common_power import (
add_gomspace_cmds,
add_gomspace_cmd_defs,
req_hk_cmds,
)
from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
from tmtccmd.config.tmtc import tmtc_definitions_provider
@ -19,7 +23,7 @@ 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 pus_tc.devs.p60dock import P60DockConfigTable
from tmtc.power.p60dock import P60DockConfigTable
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
from tmtccmd.util import ObjectIdU32
@ -52,26 +56,8 @@ class Info:
@tmtc_definitions_provider
def add_acu_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
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.REQUEST_AUX_HK_ONCE,
info=GsInfo.REQUEST_AUX_HK_ONCE,
)
oce.add(
keys=GomspaceOpCodes.GET_PARAM,
info=GsInfo.GET_PARAMETER,
)
oce.add(
keys=GomspaceOpCodes.SET_PARAM,
info=GsInfo.SET_PARAMETER,
)
add_gomspace_cmd_defs(oce)
oce.add(keys=OpCodes.TEST, info=Info.TEST)
defs.add_service(
name=CustomServiceList.ACU.value,
@ -82,46 +68,17 @@ 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")
if op_code in GomspaceOpCodes.PRINT_SWITCH_V_I:
q.add_log_cmd("ACU: Print channel stats")
q.add_pus_tc(
make_fsfw_action_cmd(
object_id=object_id.as_bytes,
action_id=gs.GomspaceDeviceActionIds.PRINT_SWITCH_V_I,
)
)
if op_code in GomspaceOpCodes.REQUEST_CORE_HK_ONCE:
q.add_log_cmd(f"PDU1: {GsInfo.REQUEST_CORE_HK_ONCE}")
hk_sid = make_sid(object_id=object_id.as_bytes, set_id=gs.SetIds.ACU_CORE)
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"PDU1: {GsInfo.REQUEST_AUX_HK_ONCE}")
hk_sid = make_sid(object_id=object_id.as_bytes, set_id=gs.SetIds.ACU_AUX)
q.add_pus_tc(generate_one_hk_command(sid=hk_sid))
if op_code in GomspaceOpCodes.GET_PARAM:
q.add_log_cmd(f"PDU1: {GsInfo.GET_PARAMETER}")
table_id = int(input("Specify table ID: "))
memory_address = int(input("Specify memory address: 0x"), 16)
parameter_size = int(input("Specify parameter size: "))
q.add_pus_tc(
gs.pack_get_param_command(
object_id.as_bytes, table_id, memory_address, parameter_size
)
)
if op_code in GomspaceOpCodes.SET_PARAM:
q.add_log_cmd(f"PDU1: {GsInfo.SET_PARAMETER}")
memory_address = int(input("Specify memory address: 0x"), 16)
memory_address = struct.pack("!H", memory_address)
parameter_size = int(input("Specify parameter size: "))
parameter = int(input("Specify parameter: "))
q.add_pus_tc(
gs.pack_set_param_command(
object_id.as_bytes, memory_address, parameter_size, parameter
)
)
add_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]
)
class ACUTestProcedure:
"""
@brief Use this class to define the tests to perform for the ACU.

157
tmtc/power/common_power.py Normal file
View File

@ -0,0 +1,157 @@
from gomspace.gomspace_common import (
pack_set_param_command,
Channel,
GomspaceOpCodes,
GsInfo,
GomspaceDeviceActionIds,
prompt_and_pack_set_param_command,
prompt_and_pack_get_param_command,
pack_request_config_command,
)
from gomspace.gomspace_pdu_definitions import PDU_CONFIG_LIST
from tmtccmd.config import OpCodeEntry
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_8_funccmd import make_fsfw_action_cmd
from tmtccmd.util import ObjectIdU32
class PowerInfo:
INFO_CORE = "Core Information"
INFO_AUX = "Auxiliary Information"
INFO_ALL = "All Information"
class PowerOpCodes:
# PDU 1
TCS_ON = ["tcs-on"]
TCS_OFF = ["tcs-off"]
SYRLINKS_ON = ["syrlinks-on"]
SYRLINKS_OFF = ["syrlinks-off"]
STAR_TRACKER_ON = ["str-on"]
STAR_TRACKER_OFF = ["str-off"]
MGT_ON = ["mgt-on"]
MGT_OFF = ["mgt-off"]
SUS_N_ON = ["sus-nom-on"]
SUS_N_OFF = ["sus-nom-off"]
SCEX_ON = ["scex-on"]
SCEX_OFF = ["scex-off"]
PLOC_ON = ["ploc-on"]
PLOC_OFF = ["ploc-off"]
ACS_A_ON = ["acs-a-on"]
ACS_A_OFF = ["acs-a-off"]
# PDU 2
PL_PCDU_VBAT_NOM_ON = ["plpcdu-vbat-nom-on"]
PL_PCDU_VBAT_NOM_OFF = ["plpcdu-vbat-nom-off"]
RW_ON = ["rw-on"]
RW_OFF = ["rw-off"]
HEATER_ON = ["heater-on"]
HEATER_OFF = ["heater-off"]
SUS_R_ON = ["sus-red-on"]
SUS_R_OFF = ["sus-red-off"]
SOLAR_ARRAY_DEPL_ON = ["sa-depl-on"]
SOLAR_ARRAY_DEPL_OFF = ["sa-depl-off"]
PL_PCDU_VBAT_RED_ON = ["plpcdu-vbat-red-on"]
PL_PCDU_VBAT_RED_OFF = ["plpcdu-vbat-red-off"]
ACS_B_ON = ["acs-b-on"]
ACS_B_OFF = ["acs-b-off"]
PL_CAM_ON = ["cam-on"]
PL_CAM_OFF = ["cam-off"]
INFO_CORE = ["info"]
INFO_AUX = ["info_aux"]
INFO_ALL = ["info_all"]
def add_gomspace_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}")
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}")
q.add_pus_tc(
make_fsfw_action_cmd(
object_id=objb, action_id=GomspaceDeviceActionIds.PRINT_LATCHUPS
)
)
if op_code in GomspaceOpCodes.SET_PARAM:
q.add_log_cmd(f"{prefix}: {GsInfo.SET_PARAMETER}")
prompt_and_pack_set_param_command(q, object_id)
if op_code in GomspaceOpCodes.GET_PARAM:
q.add_log_cmd(f"{prefix}: {GsInfo.GET_PARAMETER}")
prompt_and_pack_get_param_command(q, object_id)
if op_code in GomspaceOpCodes.REQUEST_CONFIG_TABLE:
q.add_log_cmd(f"{prefix}: {GsInfo.REQUEST_CONFIG_TABLE}")
q.add_pus_tc(pack_request_config_command(object_id.as_bytes))
def req_hk_cmds(
prefix: str,
q: DefaultPusQueueHelper,
op_code: str,
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}")
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}")
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))
def generic_on_cmd(
object_id: bytes, q: DefaultPusQueueHelper, info_str: str, out_idx: int
):
q.add_log_cmd(info_str + " on")
q.add_pus_tc(
pack_set_param_command(
object_id,
PDU_CONFIG_LIST[out_idx].parameter_address,
PDU_CONFIG_LIST[out_idx].parameter_size,
Channel.on,
)
)
def generic_off_cmd(
object_id: bytes, q: DefaultPusQueueHelper, info_str: str, out_idx: int
):
q.add_log_cmd(info_str + " off")
q.add_pus_tc(
pack_set_param_command(
object_id,
PDU_CONFIG_LIST[out_idx].parameter_address,
PDU_CONFIG_LIST[out_idx].parameter_size,
Channel.off,
)
)
def add_gomspace_cmd_defs(oce: OpCodeEntry):
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.GET_PARAM, info=GsInfo.GET_PARAMETER)
oce.add(keys=GomspaceOpCodes.PRINT_LATCHUPS, info=GsInfo.PRINT_LATCHUPS)
oce.add(keys=GomspaceOpCodes.SET_PARAM, info=GsInfo.SET_PARAMETER)
oce.add(keys=GomspaceOpCodes.REQUEST_CONFIG_TABLE, info=GsInfo.REQUEST_CONFIG_TABLE)

36
pus_tc/devs/p60dock.py → tmtc/power/p60dock.py
View File

@ -5,11 +5,9 @@
@author J. Meier
@date 13.12.2020
"""
from tmtc.power.common_power import add_gomspace_cmds, req_hk_cmds
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import generate_one_hk_command, make_sid
from gomspace.gomspace_common import (
GsInfo,
GomspaceOpCodes,
TableEntry,
Channel,
pack_set_param_command,
@ -17,12 +15,10 @@ from gomspace.gomspace_common import (
pack_get_param_command,
pack_gnd_wdt_reset_command,
pack_ping_command,
GomspaceDeviceActionIds,
pack_reboot_command,
SetIds,
)
from config.object_ids import P60_DOCK_HANDLER
from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd
from tmtccmd.util import ObjectIdU32
@ -99,6 +95,8 @@ class P60DockHkTable:
def pack_p60dock_cmds(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code: str):
objb = object_id.as_bytes
add_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)
q.add_pus_tc(
@ -139,28 +137,6 @@ def pack_p60dock_cmds(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code:
Channel.off,
)
)
if op_code in GomspaceOpCodes.REQUEST_CORE_HK_ONCE:
q.add_log_cmd("P60 Dock: Requesting HK Core HK Once")
hk_sid = make_sid(object_id=P60_DOCK_HANDLER, set_id=SetIds.P60_CORE)
q.add_pus_tc(generate_one_hk_command(sid=hk_sid))
if op_code in GomspaceOpCodes.REQUEST_AUX_HK_ONCE:
q.add_log_cmd("P60 Dock: Requesting HK Aux HK Once")
hk_sid = make_sid(object_id=P60_DOCK_HANDLER, set_id=SetIds.P60_AUX)
q.add_pus_tc(generate_one_hk_command(sid=hk_sid))
if op_code in GomspaceOpCodes.PRINT_SWITCH_V_I:
q.add_log_cmd("P60 Dock: Print Switches, Voltages, Currents")
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("P60 Dock: Print Latchups")
q.add_pus_tc(
make_fsfw_action_cmd(
object_id=objb, action_id=GomspaceDeviceActionIds.PRINT_LATCHUPS
)
)
if P60DockTestProcedure.all or P60DockTestProcedure.reboot:
q.add_log_cmd("P60 Dock: Reboot")
q.add_pus_tc(pack_reboot_command(object_id))
@ -287,3 +263,9 @@ def pack_p60dock_cmds(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code:
parameter,
)
)
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]
)

250
tmtc/power/pdu1.py Normal file
View File

@ -0,0 +1,250 @@
# -*- coding: utf-8 -*-
"""PDU1 is mounted on the X2 slot of the P60 dock
@author J. Meier
@date 17.12.2020
"""
from config.object_ids import PDU_1_HANDLER_ID
from tmtc.power.common_power import (
add_gomspace_cmds,
req_hk_cmds,
PowerOpCodes,
generic_on_cmd,
generic_off_cmd,
)
from gomspace.gomspace_common import *
from gomspace.gomspace_pdu_definitions import *
from tmtccmd.config import OpCodeEntry
class Pdu1InfoBase:
TCS = "Switch TCS Board"
SYRLINKS = "Switch Syrlinks (COM)"
STR = "Switch Startracker"
MGT = "Switch Magnetorquer"
SUS_N = "Switch Sun Sensor Board Nominal"
SCEX = "Switch Solar Cell Experiment"
PLOC = "Switch Payload On-Board Computer"
ACS_A = "Switch ACS Board A-Side"
class Pdu1ChIndex(enum.IntEnum):
TCS = 0
SYRLINKS = 1
STR = 2
MGT = 3
SUS_N = 4
SCEX = 5
PLOC = 6
ACS_A = 7
class PDU1TestProcedure:
"""
@brief Use this class to define the tests to perform for the PDU2.
@details Setting all to True will run all tests.
Setting all to False will only run the tests set to True.
"""
all = False
reboot = False
ping = False
read_temperature = False
turn_channel_2_on = False # Star Tracker connected to this channel (5V)
turn_channel_2_off = False
turn_channel_3_on = False # MTQ connected to this channel (5V)
turn_channel_3_off = False
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_req_hk_cmds(q, op_code)
add_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])
q.add_pus_tc(pack_ping_command(object_id, ping_data))
if PDU1TestProcedure.all or PDU1TestProcedure.read_temperature:
q.add_log_cmd("PDU1: Testing temperature reading")
q.add_pus_tc(
pack_get_param_command(
objb,
TableIds.hk,
PduHkTable.temperature.parameter_address,
PduHkTable.temperature.parameter_size,
)
)
if PDU1TestProcedure.all or PDU1TestProcedure.turn_channel_2_on:
q.add_log_cmd("PDU1: Turn channel 2 on (Star Tracker)")
q.add_pus_tc(
pack_set_param_command(
objb,
PduConfigTable.out_en_2.parameter_address,
PduConfigTable.out_en_2.parameter_size,
Channel.on,
)
)
if PDU1TestProcedure.all or PDU1TestProcedure.turn_channel_2_off:
q.add_log_cmd("PDU1: Turn channel 2 off (Star Tracker)")
q.add_pus_tc(
pack_set_param_command(
objb,
PduConfigTable.out_en_2.parameter_address,
PduConfigTable.out_en_2.parameter_size,
Channel.off,
)
)
if PDU1TestProcedure.all or PDU1TestProcedure.turn_channel_3_on:
q.add_log_cmd("PDU1: Turn channel 3 on (MTQ)")
q.add_pus_tc(
pack_set_param_command(
objb,
PduConfigTable.out_en_3.parameter_address,
PduConfigTable.out_en_3.parameter_size,
Channel.on,
)
)
if PDU1TestProcedure.all or PDU1TestProcedure.turn_channel_3_off:
q.add_log_cmd("PDU1: Turn channel 3 off (MTQ)")
q.add_pus_tc(
pack_set_param_command(
objb,
PduConfigTable.out_en_3.parameter_address,
PduConfigTable.out_en_3.parameter_size,
Channel.off,
)
)
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]
)
def info_on_pdu1(base: str) -> str:
return "PDU1: " + base + " on"
def info_off_pdu1(base: str) -> str:
return "PDU1: " + base + " off"
def pdu1_cmds(q: DefaultPusQueueHelper, op_code: str):
if op_code in PowerOpCodes.TCS_ON:
tcs_on_cmd(q)
elif op_code in PowerOpCodes.TCS_OFF:
tcs_off_cmd(q)
elif op_code in PowerOpCodes.SYRLINKS_ON:
syrlinks_on_cmd(q)
elif op_code in PowerOpCodes.SYRLINKS_OFF:
syrlinks_off_cmd(q)
elif op_code in PowerOpCodes.STAR_TRACKER_ON:
startracker_on_cmd(q)
elif op_code in PowerOpCodes.STAR_TRACKER_OFF:
startracker_off_cmd(q)
elif op_code in PowerOpCodes.MGT_ON:
mgt_on_cmd(q)
elif op_code in PowerOpCodes.MGT_OFF:
mgt_off_cmd(q)
elif op_code in PowerOpCodes.SUS_N_ON:
sun_sensor_nominal_on_cmd(q)
elif op_code in PowerOpCodes.SUS_N_OFF:
sun_sensor_nominal_off_cmd(q)
elif op_code in PowerOpCodes.SCEX_ON:
solar_cell_experiment_on_cmd(q)
elif op_code in PowerOpCodes.SCEX_OFF:
solar_cell_experiment_off_cmd(q)
elif op_code in PowerOpCodes.PLOC_ON:
ploc_on_cmd(q)
elif op_code in PowerOpCodes.PLOC_OFF:
ploc_off_cmd(q)
elif op_code in PowerOpCodes.ACS_A_ON:
acs_board_a_on_cmd(q)
elif op_code in PowerOpCodes.ACS_A_OFF:
acs_board_a_off_cmd(q)
def add_pdu1_common_defs(oce: OpCodeEntry):
oce.add(keys=PowerOpCodes.TCS_ON, info=info_on_pdu1(Pdu1InfoBase.TCS))
oce.add(keys=PowerOpCodes.TCS_OFF, info=info_off_pdu1(Pdu1InfoBase.TCS))
oce.add(keys=PowerOpCodes.STAR_TRACKER_ON, info=info_on_pdu1(Pdu1InfoBase.STR))
oce.add(keys=PowerOpCodes.STAR_TRACKER_OFF, info=info_off_pdu1(Pdu1InfoBase.STR))
oce.add(keys=PowerOpCodes.SUS_N_ON, info=info_on_pdu1(Pdu1InfoBase.SUS_N))
oce.add(keys=PowerOpCodes.SUS_N_OFF, info=info_off_pdu1(Pdu1InfoBase.SUS_N))
oce.add(keys=PowerOpCodes.ACS_A_ON, info=info_on_pdu1(Pdu1InfoBase.ACS_A))
oce.add(keys=PowerOpCodes.ACS_A_OFF, info=info_off_pdu1(Pdu1InfoBase.ACS_A))
oce.add(keys=PowerOpCodes.SYRLINKS_ON, info=info_on_pdu1(Pdu1InfoBase.SYRLINKS))
oce.add(keys=PowerOpCodes.SYRLINKS_OFF, info=info_off_pdu1(Pdu1InfoBase.SYRLINKS))
oce.add(keys=PowerOpCodes.MGT_ON, info=info_on_pdu1(Pdu1InfoBase.MGT))
oce.add(keys=PowerOpCodes.MGT_OFF, info=info_off_pdu1(Pdu1InfoBase.MGT))
oce.add(keys=PowerOpCodes.PLOC_ON, info=info_on_pdu1(Pdu1InfoBase.PLOC))
oce.add(keys=PowerOpCodes.PLOC_OFF, info=info_off_pdu1(Pdu1InfoBase.PLOC))
oce.add(keys=PowerOpCodes.SCEX_ON, info=info_on_pdu1(Pdu1InfoBase.SCEX))
oce.add(keys=PowerOpCodes.SCEX_OFF, info=info_off_pdu1(Pdu1InfoBase.SCEX))
def tcs_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.TCS, Pdu1ChIndex.TCS)
def tcs_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.TCS, Pdu1ChIndex.TCS)
def syrlinks_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SYRLINKS, Pdu1ChIndex.SYRLINKS)
def syrlinks_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SYRLINKS, Pdu1ChIndex.SYRLINKS)
def startracker_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.STR, Pdu1ChIndex.STR)
def startracker_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.STR, Pdu1ChIndex.STR)
def mgt_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.MGT, Pdu1ChIndex.MGT)
def mgt_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.MGT, Pdu1ChIndex.MGT)
def sun_sensor_nominal_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SUS_N, Pdu1ChIndex.SUS_N)
def sun_sensor_nominal_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SUS_N, Pdu1ChIndex.SUS_N)
def solar_cell_experiment_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SCEX, Pdu1ChIndex.SCEX)
def solar_cell_experiment_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SCEX, Pdu1ChIndex.SCEX)
def ploc_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.PLOC, Pdu1ChIndex.PLOC)
def ploc_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.PLOC, Pdu1ChIndex.PLOC)
def acs_board_a_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.ACS_A, Pdu1ChIndex.ACS_A)
def acs_board_a_off_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.ACS_A, Pdu1ChIndex.ACS_A)

311
tmtc/power/pdu2.py Normal file
View File

@ -0,0 +1,311 @@
# -*- coding: utf-8 -*-
"""
@file tmtcc_tc_pdu2.py
@brief PDU2 tests
@details PDU2 is mounted on the X4 slot of the P60 dock
@author J. Meier
@date 17.12.2020
"""
from config.object_ids import PDU_2_HANDLER_ID
from tmtc.power.common_power import (
add_gomspace_cmds,
req_hk_cmds,
PowerOpCodes,
generic_on_cmd,
generic_off_cmd,
)
from gomspace.gomspace_common import *
from gomspace.gomspace_pdu_definitions import *
from tmtccmd.config import OpCodeEntry
class Pdu2InfoBase:
PL_PCDU_BAT_NOM = "Switch PL PCDU Nominal Battery Channel"
RW = "Switch Reaction Wheel"
HEATER = "Switch Heater"
SUS_R = "Switch Sun Sensor Board Redundant"
SOLAR_ARRAY_DEPL = "Switch Solar Array Deployment"
PL_PCDU_BAT_RED = "Switch PL PCDU Redundant Battery Channel"
ACS_B = "Switch ACS Board B-Side"
PL_CAM = "Switch Payload Camera"
class Pdu2ChIndex(enum.IntEnum):
PL_PCDU_BAT_NOM = 1
RW = 2
HEATER = 3
SUS_R = 4
SOLAR_ARRAY_DEPL = 5
PL_PCDU_BAT_RED = 6
ACS_B = 7
PL_CAM = 8
class PDU2TestProcedure:
"""
@brief Use this class to define the tests to perform for the PDU2.
@details Setting all to True will run all tests.
Setting all to False will only run the tests set to True.
"""
all = False
reboot = False
read_gnd_wdt = False
gnd_wdt_reset = False
ping = False
channel_2_off = False # Reaction wheels 5V
read_temperature = False
read_channel_2_state = False # Reaction wheels 5V
read_cur_lu_lim_0 = False # OBC
channel_2_on = False # Reaction wheels 5V
invalid_table_id_test = (
False # Test to check if software properly handles invalid table ids
)
invalid_address_test = (
False # Test to check if software properly handles invalid addresses
)
invalid_parameter_size_test = False
request_hk_table = False
def pack_pdu2_commands(object_id: ObjectIdU32, q: DefaultPusQueueHelper, op_code: str):
q.add_log_cmd("Testing PDU2")
objb = object_id.as_bytes
pdu2_cmds(q, op_code)
pdu2_req_hk_cmds(q, op_code)
add_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))
if PDU2TestProcedure.all or PDU2TestProcedure.read_gnd_wdt:
q.add_log_cmd("PDU2: Reading ground watchdog timer value")
q.add_pus_tc(
pack_get_param_command(
objb,
TableIds.hk,
PduHkTable.wdt_gnd_left.parameter_address,
PduHkTable.wdt_gnd_left.parameter_size,
)
)
if PDU2TestProcedure.all or PDU2TestProcedure.gnd_wdt_reset:
q.add_log_cmd("PDU2: Testing ground watchdog reset")
q.add_pus_tc(pack_gnd_wdt_reset_command(object_id))
if PDU2TestProcedure.all or PDU2TestProcedure.ping:
q.add_log_cmd("PDU2: Ping Test")
ping_data = bytearray([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
q.add_pus_tc(pack_ping_command(object_id, ping_data))
if PDU2TestProcedure.all or PDU2TestProcedure.channel_2_on:
q.add_log_cmd("PDU2: Testing setting output channel 2 on (TCS Heater)")
q.add_pus_tc(
pack_set_param_command(
objb,
PduConfigTable.out_en_2.parameter_address,
PduConfigTable.out_en_2.parameter_size,
Channel.on,
)
)
if PDU2TestProcedure.all or PDU2TestProcedure.read_temperature:
q.add_log_cmd("PDU2: Testing temperature reading")
q.add_pus_tc(
pack_get_param_command(
objb,
TableIds.hk,
PduHkTable.temperature.parameter_address,
PduHkTable.temperature.parameter_size,
)
)
if PDU2TestProcedure.all or PDU2TestProcedure.read_channel_2_state:
q.add_log_cmd("PDU2: Reading output channel 2 state (TCS Heater)")
q.add_pus_tc(
pack_get_param_command(
objb,
TableIds.config,
PduConfigTable.out_en_2.parameter_address,
PduConfigTable.out_en_2.parameter_size,
)
)
if PDU2TestProcedure.all or PDU2TestProcedure.read_cur_lu_lim_0:
q.add_log_cmd("PDU2: Reading current limit value of output channel 0 (OBC)")
q.add_pus_tc(
pack_get_param_command(
objb,
TableIds.config,
PduConfigTable.cur_lu_lim_0.parameter_address,
PduConfigTable.cur_lu_lim_0.parameter_size,
)
)
if PDU2TestProcedure.all or PDU2TestProcedure.channel_2_off:
q.add_log_cmd("PDU2: Testing setting output channel 2 off")
q.add_pus_tc(
pack_set_param_command(
objb,
PduConfigTable.out_en_2.parameter_address,
PduConfigTable.out_en_2.parameter_size,
Channel.off,
)
)
if PDU2TestProcedure.all or PDU2TestProcedure.request_hk_table:
q.add_log_cmd("PDU2: Requesting housekeeping table")
q.add_pus_tc(pack_request_full_hk_table_command(object_id))
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)
elif op_code in PowerOpCodes.PL_PCDU_VBAT_NOM_OFF:
pl_pcdu_bat_nom_off_cmd(q)
elif op_code in PowerOpCodes.RW_ON:
reaction_wheel_on_cmd(q)
elif op_code in PowerOpCodes.RW_OFF:
reaction_wheel_off_cmd(q)
elif op_code in PowerOpCodes.HEATER_ON:
heater_on_cmd(q)
elif op_code in PowerOpCodes.HEATER_OFF:
heater_off_cmd(q)
elif op_code in PowerOpCodes.SUS_R_ON:
sus_red_on_cmd(q)
elif op_code in PowerOpCodes.SUS_R_OFF:
sus_red_off_cmd(q)
elif op_code in PowerOpCodes.SOLAR_ARRAY_DEPL_ON:
solar_array_deployment_on_cmd(q)
elif op_code in PowerOpCodes.SOLAR_ARRAY_DEPL_OFF:
solar_array_deployment_off_cmd(q)
elif op_code in PowerOpCodes.PL_PCDU_VBAT_RED_ON:
pl_pcdu_bat_red_on_cmd(q)
elif op_code in PowerOpCodes.PL_PCDU_VBAT_RED_OFF:
pl_pcdu_bat_nom_off_cmd(q)
elif op_code in PowerOpCodes.ACS_B_ON:
acs_board_b_side_on_cmd(q)
elif op_code in PowerOpCodes.ACS_B_OFF:
acs_board_b_side_off_cmd(q)
elif op_code in PowerOpCodes.PL_CAM_ON:
payload_camera_on_cmd(q)
elif op_code in PowerOpCodes.PL_CAM_OFF:
payload_camera_off_cmd(q)
def info_on_pdu2(base: str) -> str:
return "PDU2: " + base + " on"
def info_off_pdu2(base: str) -> str:
return "PDU2: " + base + " off"
def add_pdu2_common_defs(oce: OpCodeEntry):
oce.add(keys=PowerOpCodes.ACS_B_ON, info=info_on_pdu2(Pdu2InfoBase.ACS_B))
oce.add(keys=PowerOpCodes.ACS_B_OFF, info=info_off_pdu2(Pdu2InfoBase.ACS_B))
oce.add(keys=PowerOpCodes.SUS_R_ON, info=info_on_pdu2(Pdu2InfoBase.SUS_R))
oce.add(keys=PowerOpCodes.SUS_R_OFF, info=info_off_pdu2(Pdu2InfoBase.SUS_R))
oce.add(keys=PowerOpCodes.RW_ON, info=info_on_pdu2(Pdu2InfoBase.RW))
oce.add(keys=PowerOpCodes.RW_OFF, info=info_off_pdu2(Pdu2InfoBase.RW))
oce.add(
keys=PowerOpCodes.PL_PCDU_VBAT_NOM_ON,
info=info_on_pdu2(Pdu2InfoBase.PL_PCDU_BAT_NOM),
)
oce.add(
keys=PowerOpCodes.PL_PCDU_VBAT_NOM_OFF,
info=info_off_pdu2(Pdu2InfoBase.PL_PCDU_BAT_NOM),
)
oce.add(
keys=PowerOpCodes.PL_PCDU_VBAT_RED_ON,
info=info_on_pdu2(Pdu2InfoBase.PL_PCDU_BAT_RED),
)
oce.add(
keys=PowerOpCodes.PL_PCDU_VBAT_RED_OFF,
info=info_off_pdu2(Pdu2InfoBase.PL_PCDU_BAT_RED),
)
oce.add(keys=PowerOpCodes.HEATER_ON, info=info_on_pdu2(Pdu2InfoBase.HEATER))
oce.add(keys=PowerOpCodes.HEATER_OFF, info=info_off_pdu2(Pdu2InfoBase.HEATER))
oce.add(
keys=PowerOpCodes.SOLAR_ARRAY_DEPL_ON,
info=info_on_pdu2(Pdu2InfoBase.SOLAR_ARRAY_DEPL),
)
oce.add(
keys=PowerOpCodes.SOLAR_ARRAY_DEPL_OFF,
info=info_off_pdu2(Pdu2InfoBase.SOLAR_ARRAY_DEPL),
)
oce.add(keys=PowerOpCodes.PL_CAM_ON, info=info_on_pdu2(Pdu2InfoBase.PL_CAM))
oce.add(keys=PowerOpCodes.PL_CAM_OFF, info=info_off_pdu2(Pdu2InfoBase.PL_CAM))
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]
)
def pl_pcdu_bat_nom_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(
PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_PCDU_BAT_NOM, Pdu2ChIndex.PL_PCDU_BAT_NOM
)
def pl_pcdu_bat_nom_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(
PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_PCDU_BAT_NOM, Pdu2ChIndex.PL_PCDU_BAT_NOM
)
def reaction_wheel_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.RW, Pdu2ChIndex.RW)
def reaction_wheel_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.RW, Pdu2ChIndex.RW)
def heater_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.HEATER, Pdu2ChIndex.HEATER)
def heater_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.HEATER, Pdu2ChIndex.HEATER)
def sus_red_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.SUS_R, Pdu2ChIndex.SUS_R)
def sus_red_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.SUS_R, Pdu2ChIndex.SUS_R)
def solar_array_deployment_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(
PDU_2_HANDLER_ID, q, Pdu2InfoBase.SOLAR_ARRAY_DEPL, Pdu2ChIndex.SOLAR_ARRAY_DEPL
)
def solar_array_deployment_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(
PDU_2_HANDLER_ID, q, Pdu2InfoBase.SOLAR_ARRAY_DEPL, Pdu2ChIndex.SOLAR_ARRAY_DEPL
)
def pl_pcdu_bat_red_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(
PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_PCDU_BAT_RED, Pdu2ChIndex.PL_PCDU_BAT_RED
)
def pl_pcdu_bat_red_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(
PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_PCDU_BAT_RED, Pdu2ChIndex.PL_PCDU_BAT_RED
)
def acs_board_b_side_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.ACS_B, Pdu2ChIndex.ACS_B)
def acs_board_b_side_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.ACS_B, Pdu2ChIndex.ACS_B)
def payload_camera_on_cmd(q: DefaultPusQueueHelper):
generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_CAM, Pdu2ChIndex.PL_CAM)
def payload_camera_off_cmd(q: DefaultPusQueueHelper):
generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_CAM, Pdu2ChIndex.PL_CAM)

53
pus_tc/devs/power.py → tmtc/power/power.py
View File

@ -1,20 +1,17 @@
from gomspace.gomspace_common import GsInfo
from pus_tc.devs.common_power import (
from gomspace.gomspace_common import GsInfo, GomspaceOpCodes
from tmtc.power.common_power import (
PowerOpCodes,
add_pdu1_common_defs,
add_pdu2_common_defs,
PowerInfo,
pdu1_cmds,
pdu2_cmds,
pdu1_req_hk_cmds,
pdu2_req_hk_cmds,
add_gomspace_cmd_defs,
)
from config.definitions import CustomServiceList
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 tmtccmd import get_console_logger
from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
from pus_tc.devs.p60dock import P60OpCodes, GomspaceOpCodes, P60Info
from pus_tc.devs.acu import add_acu_cmds
from tmtc.power.p60dock import P60OpCodes, P60Info, p60_dock_req_hk_cmds
from tmtc.power.acu import add_acu_cmds, acu_req_hk_cmds
from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc import DefaultPusQueueHelper
@ -28,17 +25,25 @@ def pack_power_commands(q: DefaultPusQueueHelper, op_code: str):
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])
q.add_wait_seconds(3.0)
p60_dock_req_hk_cmds(q, GomspaceOpCodes.REQUEST_CORE_HK_ONCE[0])
acu_req_hk_cmds(q, GomspaceOpCodes.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])
q.add_wait_seconds(3.0)
p60_dock_req_hk_cmds(q, GomspaceOpCodes.REQUEST_AUX_HK_ONCE[0])
acu_req_hk_cmds(q, GomspaceOpCodes.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])
q.add_wait_seconds(3.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])
q.add_wait_seconds(8.0)
if q.empty():
LOGGER.info(f"Queue is empty, no stack for op code {op_code}")
@ -50,13 +55,7 @@ def add_p60_cmds(defs: TmtcDefinitionWrapper):
oce.add(keys=P60OpCodes.STACK_3V3_OFF, info=P60Info.STACK_3V3_OFF)
oce.add(keys=P60OpCodes.STACK_5V_ON, info=P60Info.STACK_5V_ON)
oce.add(keys=P60OpCodes.STACK_5V_OFF, info=P60Info.STACK_5V_OFF)
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="P60 Dock: Print Switches, Voltages, Currents",
)
oce.add(keys=GomspaceOpCodes.PRINT_LATCHUPS, info="P60 Dock: Print Latchups")
add_gomspace_cmd_defs(oce)
oce.add(keys=P60OpCodes.TEST, info="P60 Tests")
defs.add_service(
name=CustomServiceList.P60DOCK.value, info="P60 Device", op_code_entry=oce
@ -82,14 +81,15 @@ def add_power_cmd_defs(defs: TmtcDefinitionWrapper):
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.PRINT_LATCHUPS, info="PDU1: Print Latchups")
oce.add(keys=GomspaceOpCodes.SET_PARAM, info="Set parameter")
oce.add(keys=GomspaceOpCodes.GET_PARAM, info=GsInfo.GET_PARAMETER)
defs.add_service(
name=CustomServiceList.PDU1.value,
info="PDU1 Device",
@ -101,14 +101,7 @@ def add_pdu1_cmds(defs: TmtcDefinitionWrapper):
def add_pdu2_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()
add_pdu2_common_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,
)
add_gomspace_cmd_defs(oce)
oce.add(
keys=GomspaceOpCodes.PRINT_SWITCH_V_I,
info="PDU2: Print Switches, Voltages, Currents",

151
pus_tm/system/power.py → tmtc/power/tm.py
View File

@ -1,9 +1,16 @@
import struct
from typing import List, Tuple
from tmtccmd.util import ObjectIdBase
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
from pus_tm.defs import PrintWrapper
from gomspace.gomspace_common import SetIds
from gomspace.gomspace_common import SetIds, GomspaceDeviceActionIds
from config.object_ids import (
PDU_1_HANDLER_ID,
PDU_2_HANDLER_ID,
P60_DOCK_HANDLER,
ACU_HANDLER_ID,
)
P60_INDEX_LIST = [
"ACU VCC",
@ -244,7 +251,7 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
f"{voltage_list[idx]:05} | {current_list[idx]:04}"
)
pw.dlog(content_line)
fmt_str = "!IBhHhh"
fmt_str = "!IBhHff"
inc_len = struct.calcsize(fmt_str)
(
boot_count,
@ -259,7 +266,7 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
f"Batt: Mode {batt_mode} | Boot Count {boot_count} | "
f"Charge current {batt_current} | Voltage {batt_voltage}"
)
temps = f"In C: Temp 0 {temp_0 / 10.0} | Temp 1 {temp_1 / 10.0} | "
temps = f"In C: Temp 0 {temp_0} | Temp 1 {temp_1} | "
pw.dlog(temps)
pw.dlog(batt_info)
printer.print_validity_buffer(validity_buffer=hk_data[current_idx:], num_vars=9)
@ -361,7 +368,7 @@ def handle_acu_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
current_idx, powers = gen_six_entry_u16_list(
hk_data=hk_data, current_idx=current_idx
)
fmt_str = "!HHHIIHH"
fmt_str = "!fffIIHH"
inc_len = struct.calcsize(fmt_str)
(tmp0, tmp1, tmp2, bootcnt, uptime, mppt_time, mppt_period) = struct.unpack(
fmt_str, hk_data[current_idx : current_idx + inc_len]
@ -378,9 +385,7 @@ def handle_acu_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
f"{i} | {str(voltages[i]).ljust(4)} | {str(currents[i]).ljust(4)} | "
f"{str(vboosts[i]).ljust(4)} | {str(powers[i]).ljust(2)}"
)
pw.dlog(
f"Temperatures in C: Ch0 {tmp0/10.0} | Ch1 {tmp1/10.0} | Ch2 {tmp2/10.0}"
)
pw.dlog(f"Temperatures in C: Ch0 {tmp0} | Ch1 {tmp1} | Ch2 {tmp2}")
pw.dlog(
f"Boot Count {bootcnt} | Uptime {uptime} sec | "
f"MPPT Time {mppt_time} msec | MPPT Period {mppt_period} msec"
@ -423,3 +428,135 @@ def handle_acu_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
)
dev_parser.print(pw=pw)
printer.print_validity_buffer(validity_buffer=hk_data[current_idx:], num_vars=8)
OBC_ENDIANNESS = "<"
def handle_get_param_data_reply(
obj_id: ObjectIdBase, action_id: int, pw: PrintWrapper, custom_data: bytearray
):
if action_id == GomspaceDeviceActionIds.PARAM_GET:
pw.dlog(f"Parameter Get Request received for object {obj_id}")
header_list = [
"Gomspace Request Code",
"Table ID",
"Memory Address",
"Payload length",
"Payload",
]
fmt_str = "!BBHH"
(gs_request_code, table_id, address, payload_length) = struct.unpack(
fmt_str, custom_data[:6]
)
content_list = [
hex(gs_request_code),
table_id,
hex(address),
payload_length,
f"0x[{custom_data[6:].hex(sep=',')}]",
]
pw.dlog(f"{header_list}")
pw.dlog(f"{content_list}")
elif action_id == GomspaceDeviceActionIds.REQUEST_CONFIG_TABLE:
print(f"Received config table with size {len(custom_data)} for object {obj_id}")
if obj_id.as_bytes == PDU_1_HANDLER_ID or obj_id.as_bytes == PDU_2_HANDLER_ID:
pdu_config_table_handler(pw, custom_data)
elif obj_id.as_bytes == ACU_HANDLER_ID:
acu_config_table_handler(pw, custom_data)
elif obj_id.as_bytes == P60_DOCK_HANDLER:
p60_dock_config_table_handler(pw, custom_data)
def pdu_config_table_handler(pw: PrintWrapper, custom_data: bytes):
out_on_cnt = unpack_array_in_data(custom_data, 0x52, 2, 9, "H")
out_off_cnt = unpack_array_in_data(custom_data, 0x64, 2, 9, "H")
init_out_norm = unpack_array_in_data(custom_data, 0x76, 1, 9, "B")
init_out_safe = unpack_array_in_data(custom_data, 0x80, 1, 9, "B")
init_on_dly = unpack_array_in_data(custom_data, 0x8A, 2, 9, "H")
init_off_dly = unpack_array_in_data(custom_data, 0x9C, 2, 9, "H")
safe_off_dly = unpack_array_in_data(custom_data, 0xAE, 1, 9, "B")
batt_hwmax = struct.unpack(f"{OBC_ENDIANNESS}H", custom_data[0x11C : 0x11C + 2])[0]
batt_max = struct.unpack(f"{OBC_ENDIANNESS}H", custom_data[0x11E : 0x11E + 2])[0]
batt_norm = struct.unpack(f"{OBC_ENDIANNESS}H", custom_data[0x120 : 0x120 + 2])[0]
batt_safe = struct.unpack(f"{OBC_ENDIANNESS}H", custom_data[0x122 : 0x122 + 2])[0]
batt_crit = struct.unpack(f"{OBC_ENDIANNESS}H", custom_data[0x124 : 0x124 + 2])[0]
pw.dlog(f"{'out_on_cnt'.ljust(15)}: {out_on_cnt}")
pw.dlog(f"{'out_off_cnt'.ljust(15)}: {out_off_cnt}")
pw.dlog(f"{'init_out_norm'.ljust(15)}: {init_out_norm}")
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"{'safe_off_dly'.ljust(15)}: {safe_off_dly}")
pw.dlog(f"{'batt_hwmax'.ljust(15)}: {batt_hwmax}")
pw.dlog(f"{'batt_max'.ljust(15)}: {batt_max}")
pw.dlog(f"{'batt_norm'.ljust(15)}: {batt_norm}")
pw.dlog(f"{'batt_safe'.ljust(15)}: {batt_safe}")
pw.dlog(f"{'batt_crit'.ljust(15)}: {batt_crit}")
def acu_config_table_handler(pw: PrintWrapper, custom_data: bytes):
mppt_mode = custom_data[0]
mppt_delta_mode = custom_data[1]
vboost_list = unpack_array_in_data(custom_data, 0x02, 2, 6, "H")
vbat_max_hi = struct.unpack(f"{OBC_ENDIANNESS}H", custom_data[0x10 : 0x10 + 2])[0]
vbat_max_lo = struct.unpack(f"{OBC_ENDIANNESS}H", custom_data[0x12 : 0x12 + 2])[0]
mppt_period = struct.unpack(f"{OBC_ENDIANNESS}I", custom_data[0x14 : 0x14 + 4])[0]
max_dv = struct.unpack(f"{OBC_ENDIANNESS}H", custom_data[0x18 : 0x18 + 2])[0]
ov_mode = custom_data[0x1A]
pw.dlog(f"{'mppt_mode'.ljust(15)}: {mppt_mode}")
pw.dlog(f"{'mppt_delta_mode'.ljust(15)}: {mppt_delta_mode}")
pw.dlog(f"{'vboost_list'.ljust(15)}: {vboost_list}")
pw.dlog(f"{'vbat_max_hi'.ljust(15)}: {vbat_max_hi}")
pw.dlog(f"{'vbat_max_lo'.ljust(15)}: {vbat_max_lo}")
pw.dlog(f"{'mppt_period'.ljust(15)}: {mppt_period}")
pw.dlog(f"{'max_dv'.ljust(15)}: {max_dv}")
pw.dlog(f"{'ov_mode'.ljust(15)}: {ov_mode}")
def p60_dock_config_table_handler(pw: PrintWrapper, custom_data: bytes):
ch_names = parse_name_list(custom_data[0:0x68], 13)
out_on_cnt = unpack_array_in_data(custom_data, 0x76, 2, 13, "H")
out_off_cnt = unpack_array_in_data(custom_data, 0x90, 2, 13, "H")
init_out_norm = unpack_array_in_data(custom_data, 0xAA, 1, 13, "B")
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")
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}")
pw.dlog(f"{'init_out_norm'.ljust(15)}: {init_out_norm}")
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}")
def parse_name_list(data: bytes, name_len: int):
ch_list = []
idx = 0
while len(ch_list) < name_len:
next_byte = data[idx]
if next_byte != 0:
string_end_found = False
string_end_idx = idx
while not string_end_found:
string_end_idx += 1
if data[string_end_idx] == 0:
string_end_found = True
name = data[idx:string_end_idx].decode()
ch_list.append(name)
idx += len(name)
idx += 1
return ch_list
def unpack_array_in_data(
data: bytes, start_addr: int, width: int, entries: int, struct_spec: str
) -> List:
return [
struct.unpack(
f"{OBC_ENDIANNESS}{struct_spec}",
data[start_addr + (i * width) : start_addr + ((i + 1) * width)],
)[0]
for i in range(entries)
]