Compare commits

...

24 Commits

Author SHA1 Message Date
510ba82fef rw cmds cntd 2022-05-05 02:00:18 +02:00
bc4099c5bf added more rw cmds 2022-05-05 01:21:57 +02:00
dd34a049d3 Merge remote-tracking branch 'origin/develop' into mueller/rw-cmds 2022-05-05 00:40:11 +02:00
2e24f764a6 Merge pull request 'PLOC Commands' (#64) from meier/ploc into develop
Reviewed-on: #64
2022-05-05 00:37:14 +02:00
Jakob Meier
a4bf730b22 Merge branch 'develop' into meier/ploc 2022-05-04 19:09:28 +02:00
Jakob Meier
213e6709ac cam cmd reply handling 2022-05-04 19:08:54 +02:00
Jakob Meier
9c3a8e9ff2 removed unused adc monitor commands 2022-05-04 12:48:43 +02:00
Jakob Meier
b3f40b38fc Merge branch 'meier/develop' into meier/ploc 2022-05-03 19:12:50 +02:00
Jakob Meier
0eb4d0fb26 fixed conflicts 2022-05-03 19:12:37 +02:00
Jakob Meier
7413a5db5f supervisor read gpio reply handling 2022-05-03 19:09:23 +02:00
Jakob Meier
c7a19f7122 udpate gitignore 2022-05-03 15:27:43 +02:00
Jakob Meier
c5f99d80a2 fixed conflicts 2022-05-03 15:02:10 +02:00
Jakob Meier
fe833edc91 tmtccmd update 2022-05-02 13:47:44 +02:00
d7efe7d864 re-generate event list 2022-05-02 12:41:02 +02:00
f23ed26807 Merge remote-tracking branch 'origin/develop' into mueller/rw-cmds 2022-05-02 12:39:36 +02:00
ecd16da2f4 continued RW HK handling 2022-04-28 19:43:16 +02:00
3d1a4044fe run black afmt 2022-04-28 19:04:17 +02:00
981f4f0c3e add basic rw commanding framework 2022-04-28 18:59:15 +02:00
Jakob Meier
3c6b88cfcd service 17 run config 2022-03-09 10:50:21 +01:00
Jakob Meier
39dc222068 Merge branch 'develop' into meier/develop 2022-03-09 10:49:51 +01:00
Jakob.Meier
b0f6551c0a Merge branch 'develop' into meier/develop 2021-08-20 11:35:26 +02:00
Jakob.Meier
a47925859b Merge branch 'meier/plocUpdater' into develop 2021-08-20 11:34:43 +02:00
Jakob.Meier
77e46c12ba Merge branch 'develop' into meier/plocUpdater 2021-08-20 08:41:51 +02:00
Jakob.Meier
df8f86ebd6 Merge branch 'meier/pcduSwitchCommanding' into meier/develop 2021-08-18 08:27:37 +02:00
14 changed files with 346 additions and 325 deletions

2
.gitignore vendored
View File

@@ -5,4 +5,4 @@ __pycache__
log
/gps_log.txt
/config/*.json
tmtc_conf.json
tmtc_conf.json

View File

@@ -27,10 +27,11 @@ class CustomServiceList(enum.Enum):
PCDU = "pcdu"
PL_PCDU = "plpcdu"
SA_DEPLYOMENT = "sa_depl"
REACTION_WHEEL_1 = "reaction_wheel_1"
REACTION_WHEEL_2 = "reaction_wheel_2"
REACTION_WHEEL_3 = "reaction_wheel_3"
REACTION_WHEEL_4 = "reaction_wheel_4"
REACTION_WHEEL_1 = "rw-1"
REACTION_WHEEL_2 = "rw-2"
REACTION_WHEEL_3 = "rw-3"
REACTION_WHEEL_4 = "rw-4"
RW_ASSEMBLY = "rw-ass"
RAD_SENSOR = "rad_sensor"
PLOC_UPDATER = "ploc_updater"
GPS_0 = "gps0"

View File

@@ -80,7 +80,6 @@
11300;0x2c24;SWITCH_CMD_SENT;INFO;Indicates that a FSFW object requested setting a switch P1: 1 if on was requested, 0 for off | P2: Switch Index;mission/devices/devicedefinitions/powerDefinitions.h
11301;0x2c25;SWITCH_HAS_CHANGED;INFO;Indicated that a swithc state has changed P1: New switch state, 1 for on, 0 for off | P2: Switch Index;mission/devices/devicedefinitions/powerDefinitions.h
11302;0x2c26;SWITCHING_Q7S_DENIED;MEDIUM;;mission/devices/devicedefinitions/powerDefinitions.h
11303;0x2c27;FDIR_REACTION_IGNORED;MEDIUM;;mission/devices/devicedefinitions/powerDefinitions.h
11400;0x2c88;GPIO_PULL_HIGH_FAILED;LOW;;mission/devices/HeaterHandler.h
11401;0x2c89;GPIO_PULL_LOW_FAILED;LOW;;mission/devices/HeaterHandler.h
11402;0x2c8a;SWITCH_ALREADY_ON;LOW;;mission/devices/HeaterHandler.h
1 2200 0x0898 STORE_SEND_WRITE_FAILED LOW fsfw/src/fsfw/tmstorage/TmStoreBackendIF.h
80 11300 0x2c24 SWITCH_CMD_SENT INFO Indicates that a FSFW object requested setting a switch P1: 1 if on was requested, 0 for off | P2: Switch Index mission/devices/devicedefinitions/powerDefinitions.h
81 11301 0x2c25 SWITCH_HAS_CHANGED INFO Indicated that a swithc state has changed P1: New switch state, 1 for on, 0 for off | P2: Switch Index mission/devices/devicedefinitions/powerDefinitions.h
82 11302 0x2c26 SWITCHING_Q7S_DENIED MEDIUM mission/devices/devicedefinitions/powerDefinitions.h
11303 0x2c27 FDIR_REACTION_IGNORED MEDIUM mission/devices/devicedefinitions/powerDefinitions.h
83 11400 0x2c88 GPIO_PULL_HIGH_FAILED LOW mission/devices/HeaterHandler.h
84 11401 0x2c89 GPIO_PULL_LOW_FAILED LOW mission/devices/HeaterHandler.h
85 11402 0x2c8a SWITCH_ALREADY_ON LOW mission/devices/HeaterHandler.h

View File

@@ -65,7 +65,6 @@ def get_eive_service_op_code_dict(service_op_code_dict: ServiceOpCodeDictT):
add_pcdu_cmds,
add_pl_pcdu_cmds,
add_imtq_cmds,
add_rw_cmds,
add_rad_sens_cmds,
add_ploc_mpsoc_cmds,
add_ploc_supv_cmds,
@@ -73,6 +72,7 @@ def get_eive_service_op_code_dict(service_op_code_dict: ServiceOpCodeDictT):
add_time_cmds,
)
from pus_tc.devs.gps import GpsOpCodes
from pus_tc.devs.reaction_wheels import add_rw_cmds
add_bpx_cmd_definitions(cmd_dict=service_op_code_dict)
add_core_controller_definitions(cmd_dict=service_op_code_dict)

View File

@@ -71,6 +71,7 @@ PL_PCDU_ID = bytes([0x44, 0x30, 0x00, 0x00])
ACS_BOARD_ASS_ID = bytes([0x73, 0x00, 0x00, 0x01])
SUS_BOARD_ASS_ID = bytes([0x73, 0x00, 0x00, 0x02])
TCS_BOARD_ASS_ID = bytes([0x73, 0x00, 0x00, 0x03])
RW_ASSEMBLY = bytes([0x73, 0x00, 0x00, 0x04])
def get_object_ids() -> ObjectIdDictT:

View File

@@ -489,25 +489,6 @@ def add_imtq_cmds(cmd_dict: ServiceOpCodeDictT):
cmd_dict[CustomServiceList.IMTQ.value] = service_imtq_tuple
def add_rw_cmds(cmd_dict: ServiceOpCodeDictT):
op_code_dict_srv_rw = {
"0": ("Reaction Wheel: Run all commands", {OpCodeDictKeys.TIMEOUT: 2.0}),
"1": ("Reaction Wheel: Set speed", {OpCodeDictKeys.TIMEOUT: 2.0}),
"2": ("Reaction Wheel: Set mode on", {OpCodeDictKeys.TIMEOUT: 2.0}),
"3": ("Reaction Wheel: Set mode normal", {OpCodeDictKeys.TIMEOUT: 2.0}),
"4": ("Reaction Wheel: Set mode off", {OpCodeDictKeys.TIMEOUT: 2.0}),
"5": (
"Reaction Wheel: Send get-telemetry-command",
{OpCodeDictKeys.TIMEOUT: 2.0},
),
}
service_rw_tuple = ("Reaction Wheel", op_code_dict_srv_rw)
cmd_dict[CustomServiceList.REACTION_WHEEL_1.value] = service_rw_tuple
cmd_dict[CustomServiceList.REACTION_WHEEL_2.value] = service_rw_tuple
cmd_dict[CustomServiceList.REACTION_WHEEL_3.value] = service_rw_tuple
cmd_dict[CustomServiceList.REACTION_WHEEL_4.value] = service_rw_tuple
def add_rad_sens_cmds(cmd_dict: ServiceOpCodeDictT):
op_code_dict_srv_rad_sensor = {
"0": ("Radiation Sensor: Set mode on", {OpCodeDictKeys.TIMEOUT: 2.0}),
@@ -560,7 +541,6 @@ def add_ploc_supv_cmds(cmd_dict: ServiceOpCodeDictT):
"1": ("PLOC Supervisor: Set mode on", {OpCodeDictKeys.TIMEOUT: 2.0}),
"2": ("PLOC Supervisor: Set mode normal", {OpCodeDictKeys.TIMEOUT: 2.0}),
"3": ("PLOC Supervisor: Get HK Report", {OpCodeDictKeys.TIMEOUT: 2.0}),
"4": ("PLOC Supervisor: Restart MPSoC", {OpCodeDictKeys.TIMEOUT: 2.0}),
"5": ("PLOC Supervisor: Start MPSoC", {OpCodeDictKeys.TIMEOUT: 2.0}),
"6": ("PLOC Supervisor: Shutdown MPSoC", {OpCodeDictKeys.TIMEOUT: 2.0}),
"7": (
@@ -576,18 +556,9 @@ def add_ploc_supv_cmds(cmd_dict: ServiceOpCodeDictT):
"PLOC Supervisor: Request boot status report",
{OpCodeDictKeys.TIMEOUT: 2.0},
),
"14": ("PLOC Supervisor: Update available", {OpCodeDictKeys.TIMEOUT: 2.0}),
"15": ("PLOC Supervisor: Watchdogs enable", {OpCodeDictKeys.TIMEOUT: 2.0}),
"16": (
"PLOC Supervisor: Watchdog Configure Timeout",
{OpCodeDictKeys.TIMEOUT: 2.0},
),
"17": ("PLOC Supervisor: Enable latchup alert", {OpCodeDictKeys.TIMEOUT: 2.0}),
"18": ("PLOC Supervisor: Disable latchup alert", {OpCodeDictKeys.TIMEOUT: 2.0}),
"19": ("PLOC Supervisor: Auto calibrate alert", {OpCodeDictKeys.TIMEOUT: 2.0}),
"20": ("PLOC Supervisor: Set alert limit", {OpCodeDictKeys.TIMEOUT: 2.0}),
"21": ("PLOC Supervisor: Set alert irq filter", {OpCodeDictKeys.TIMEOUT: 2.0}),
"22": ("PLOC Supervisor: Set ADC sweep period", {OpCodeDictKeys.TIMEOUT: 2.0}),
"23": (
"PLOC Supervisor: Set ADC enabled channels",
{OpCodeDictKeys.TIMEOUT: 2.0},
@@ -604,8 +575,6 @@ def add_ploc_supv_cmds(cmd_dict: ServiceOpCodeDictT):
"27": ("PLOC Supervisor: Copy ADC data to MRAM", {OpCodeDictKeys.TIMEOUT: 2.0}),
"30": ("PLOC Supervisor: Run auto EM tests", {OpCodeDictKeys.TIMEOUT: 2.0}),
"31": ("PLOC Supervisor: MRAM Wipe", {OpCodeDictKeys.TIMEOUT: 2.0}),
"33": ("PLOC Supervisor: Print CPU stats", {OpCodeDictKeys.TIMEOUT: 2.0}),
"34": ("PLOC Supervisor: Set debug verbosity", {OpCodeDictKeys.TIMEOUT: 2.0}),
"35": ("PLOC Supervisor: Set GPIO", {OpCodeDictKeys.TIMEOUT: 2.0}),
"36": ("PLOC Supervisor: Read GPIO", {OpCodeDictKeys.TIMEOUT: 2.0}),
"37": ("PLOC Supervisor: Restart supervisor", {OpCodeDictKeys.TIMEOUT: 2.0}),
@@ -622,18 +591,28 @@ def add_ploc_supv_cmds(cmd_dict: ServiceOpCodeDictT):
{OpCodeDictKeys.TIMEOUT: 2.0},
),
"42": ("PLOC Supervisor: Perform update", {OpCodeDictKeys.TIMEOUT: 2.0}),
"43": ("PLOC Supervisor: Terminate supervisor process", {OpCodeDictKeys.TIMEOUT: 2.0}),
"43": (
"PLOC Supervisor: Terminate supervisor process",
{OpCodeDictKeys.TIMEOUT: 2.0},
),
"44": ("PLOC Supervisor: Start MPSoC quiet", {OpCodeDictKeys.TIMEOUT: 2.0}),
"45": ("PLOC Supervisor: Set shutdown timeout", {OpCodeDictKeys.TIMEOUT: 2.0}),
"46": ("PLOC Supervisor: Factory flash", {OpCodeDictKeys.TIMEOUT: 2.0}),
"47": ("PLOC Supervisor: Enable auto TM", {OpCodeDictKeys.TIMEOUT: 2.0}),
"48": ("PLOC Supervisor: Disable auto TM", {OpCodeDictKeys.TIMEOUT: 2.0}),
"49": ("PLOC Supervisor: Enable ADC monitor task", {OpCodeDictKeys.TIMEOUT: 2.0}),
"50": ("PLOC Supervisor: Disable ADC monitor task", {OpCodeDictKeys.TIMEOUT: 2.0}),
"51": ("PLOC Supervisor: Logging request event buffers", {OpCodeDictKeys.TIMEOUT: 2.0}),
"52": ("PLOC Supervisor: Logging clear counters", {OpCodeDictKeys.TIMEOUT: 2.0}),
"51": (
"PLOC Supervisor: Logging request event buffers",
{OpCodeDictKeys.TIMEOUT: 2.0},
),
"52": (
"PLOC Supervisor: Logging clear counters",
{OpCodeDictKeys.TIMEOUT: 2.0},
),
"53": ("PLOC Supervisor: Logging set topic", {OpCodeDictKeys.TIMEOUT: 2.0}),
"54": ("PLOC Supervisor: Logging request counters", {OpCodeDictKeys.TIMEOUT: 2.0}),
"54": (
"PLOC Supervisor: Logging request counters",
{OpCodeDictKeys.TIMEOUT: 2.0},
),
"55": ("PLOC Supervisor: Request ADC Report", {OpCodeDictKeys.TIMEOUT: 2.0}),
"56": ("PLOC Supervisor: Reset PL", {OpCodeDictKeys.TIMEOUT: 2.0}),
}

View File

@@ -59,8 +59,9 @@ class MemAddresses(enum.IntEnum):
DEADBEEF = 0x40000004
class PlocReplyIds:
tm_mem_read_report = 6
class PlocReplyIds(enum.IntEnum):
TM_MEM_READ_RPT = 6
TM_CAM_CMD_RPT = 19
def pack_ploc_mpsoc_commands(
@@ -172,7 +173,11 @@ def pack_ploc_mpsoc_commands(
elif op_code == "16":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: Tc cam command send"))
cam_cmd = input("Specify cam command string: ")
command = object_id + struct.pack("!I", CommandIds.TC_CAM_CMD_SEND) + bytearray(cam_cmd, 'utf-8')
command = (
object_id
+ struct.pack("!I", CommandIds.TC_CAM_CMD_SEND)
+ bytearray(cam_cmd, "utf-8")
)
command = PusTelecommand(service=8, subservice=128, ssc=32, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())

View File

@@ -32,8 +32,14 @@ MANUAL_INPUT = "1"
update_file_dict = {
MANUAL_INPUT: ["manual input", ""],
"2": ["/mnt/sd0/ploc/supervisor/update.bin", "/mnt/sd0/ploc/supervisor/update.bin"],
"3": ["/mnt/sd0/ploc/supervisor/update-large.bin", "/mnt/sd0/ploc/supervisor/update-large.bin"],
"4": ["/mnt/sd0/ploc/supervisor/update-small.bin", "/mnt/sd0/ploc/supervisor/update-small.bin"],
"3": [
"/mnt/sd0/ploc/supervisor/update-large.bin",
"/mnt/sd0/ploc/supervisor/update-large.bin",
],
"4": [
"/mnt/sd0/ploc/supervisor/update-small.bin",
"/mnt/sd0/ploc/supervisor/update-small.bin",
],
}
event_buffer_path_dict = {
@@ -44,7 +50,6 @@ event_buffer_path_dict = {
class SupvActionIds:
HK_REPORT = 1
RESTART_MPSOC = 2
START_MPSOC = 3
SHUTWOWN_MPSOC = 4
SEL_MPSOC_BOOT_IMAGE = 5
@@ -55,27 +60,19 @@ class SupvActionIds:
DISABLE_HK = 10
GET_BOOT_STATUS_REPORT = 11
UPDATE_AVAILABLE = 12
WATCHDOGS_ENABLE = 13
WATCHDOGS_CONFIG_TIMEOUT = 14
ENABLE_LATCHUP_ALERT = 15
DISABLE_LATCHUP_ALERT = 16
AUTO_CALIBRATE_ALERT = 17
SET_ALERT_LIMIT = 18
SET_ALERT_IRQ_FILTER = 19
SET_ADC_SWEEP_PERIOD = 20
SET_ADC_ENABLED_CHANNELS = 21
SET_ADC_WINDOW_AND_STRIDE = 22
SET_ADC_THRESHOLD = 23
GET_LATCHUP_STATUS_REPORT = 24
COPY_ADC_DATA_TO_MRAM = 25
ENABLE_NVMS = 26
SELECT_NVM = 27
RUN_AUTO_EM_TESTS = 28
WIPE_MRAM = 29
DUMP_MRAM = 30
SET_DBG_VERBOSITY = 31
CAN_LOOPBACK_TEST = 32
PRINT_CPU_STATS = 33
SET_GPIO = 34
READ_GPIO = 35
RESTART_SUPERVISOR = 36
@@ -91,8 +88,6 @@ class SupvActionIds:
TERMINATE_SUPV_HELPER = 49
ENABLE_AUTO_TM = 50
DISABLE_AUTO_TM = 51
ENABLE_ADC_MONITOR_TASK = 52
DISABLE_ADC_MONITOR_TASK = 53
LOGGING_REQUEST_EVENT_BUFFERS = 54
LOGGING_CLEAR_COUNTERS = 55
LOGGING_SET_TOPIC = 56
@@ -134,11 +129,6 @@ def pack_ploc_supv_commands(
command = object_id + struct.pack("!I", SupvActionIds.HK_REPORT)
command = PusTelecommand(service=8, subservice=128, ssc=20, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "4":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Restart MPSoC"))
command = object_id + struct.pack("!I", SupvActionIds.RESTART_MPSOC)
command = PusTelecommand(service=8, subservice=128, ssc=21, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "5":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Start MPSoC"))
command = object_id + struct.pack("!I", SupvActionIds.START_MPSOC)
@@ -162,9 +152,9 @@ def pack_ploc_supv_commands(
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "8":
tc_queue.appendleft(
(QueueCommands.PRINT, "PLOC Supervisor: Set max restart tries: ")
(QueueCommands.PRINT, "PLOC Supervisor: Set max restart tries")
)
restart_tries = int(input("Set maximum restart tries:"))
restart_tries = int(input("Specify maximum restart tries: "))
command = (
object_id
+ struct.pack("!I", SupvActionIds.SET_MAX_RESTART_TRIES)
@@ -186,7 +176,7 @@ def pack_ploc_supv_commands(
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "11":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set boot timeout"))
boot_timeout = int(input("Specify boot timeout [ms]:"))
boot_timeout = int(input("Specify boot timeout [ms]: "))
command = (
object_id
+ struct.pack("!I", SupvActionIds.SET_BOOT_TIMEOUT)
@@ -206,23 +196,6 @@ def pack_ploc_supv_commands(
command = object_id + struct.pack("!I", SupvActionIds.GET_BOOT_STATUS_REPORT)
command = PusTelecommand(service=8, subservice=128, ssc=30, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "14":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Update available"))
command = pack_update_available_cmd(object_id)
command = PusTelecommand(service=8, subservice=128, ssc=31, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "15":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Watchdogs Enable"))
command = pack_watchdogs_enable_cmd(object_id)
command = PusTelecommand(service=8, subservice=128, ssc=32, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "16":
tc_queue.appendleft(
(QueueCommands.PRINT, "PLOC Supervisor: Watchdog configure timeout")
)
command = pack_watchdog_config_timeout_cmd(object_id)
command = PusTelecommand(service=8, subservice=128, ssc=33, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "17":
tc_queue.appendleft(
(QueueCommands.PRINT, "PLOC Supervisor: Enable latchup alert")
@@ -237,32 +210,11 @@ def pack_ploc_supv_commands(
command = pack_lachtup_alert_cmd(object_id, False)
command = PusTelecommand(service=8, subservice=128, ssc=35, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "19":
tc_queue.appendleft(
(QueueCommands.PRINT, "PLOC Supervisor: Auto calibrate alert")
)
command = pack_auto_calibrate_alert_cmd(object_id)
command = PusTelecommand(service=8, subservice=128, ssc=36, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "20":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set alert limit"))
command = pack_set_alert_limit_cmd(object_id)
command = PusTelecommand(service=8, subservice=128, ssc=37, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "21":
tc_queue.appendleft(
(QueueCommands.PRINT, "PLOC Supervisor: Set alert irq filter")
)
command = pack_set_alert_irq_filter_cmd(object_id)
command = PusTelecommand(service=8, subservice=128, ssc=38, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "22":
tc_queue.appendleft(
(QueueCommands.PRINT, "PLOC Supervisor: Set ADC sweep period")
)
command = pack_set_adc_sweep_period_cmd(object_id)
command = PusTelecommand(service=8, subservice=128, ssc=39, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "23":
tc_queue.appendleft(
(QueueCommands.PRINT, "PLOC Supervisor: Set ADC enabled channels")
@@ -306,18 +258,6 @@ def pack_ploc_supv_commands(
command = pack_mram_wipe_cmd(object_id)
command = PusTelecommand(service=8, subservice=128, ssc=46, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "33":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Print CPU stats"))
command = pack_print_cpu_stats_cmd(object_id)
command = PusTelecommand(service=8, subservice=128, ssc=48, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "34":
tc_queue.appendleft(
(QueueCommands.PRINT, "PLOC Supervisor: Set debug verbosity")
)
command = pack_set_debug_verbosity_cmd(object_id)
command = PusTelecommand(service=8, subservice=128, ssc=49, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "35":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set GPIO command"))
command = pack_set_gpio_cmd(object_id)
@@ -373,69 +313,51 @@ def pack_ploc_supv_commands(
command = PusTelecommand(service=8, subservice=128, ssc=57, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "43":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Terminate supervisor process"))
command = object_id + struct.pack(
"!I", SupvActionIds.TERMINATE_SUPV_HELPER
tc_queue.appendleft(
(QueueCommands.PRINT, "PLOC Supervisor: Terminate supervisor process")
)
command = object_id + struct.pack("!I", SupvActionIds.TERMINATE_SUPV_HELPER)
command = PusTelecommand(service=8, subservice=128, ssc=58, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "44":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Start MPSoC quiet"))
command = object_id + struct.pack(
"!I", SupvActionIds.START_MPSOC_QUIET
)
command = object_id + struct.pack("!I", SupvActionIds.START_MPSOC_QUIET)
command = PusTelecommand(service=8, subservice=128, ssc=59, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "45":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set shutdown timeout"))
tc_queue.appendleft(
(QueueCommands.PRINT, "PLOC Supervisor: Set shutdown timeout")
)
command = pack_set_shutdown_timeout_command(object_id)
command = PusTelecommand(service=8, subservice=128, ssc=60, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "46":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Factory flash"))
command = object_id + struct.pack(
"!I", SupvActionIds.FACTORY_FLASH
)
command = object_id + struct.pack("!I", SupvActionIds.FACTORY_FLASH)
command = PusTelecommand(service=8, subservice=128, ssc=61, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "47":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Enable auto TM"))
command = object_id + struct.pack(
"!I", SupvActionIds.ENABLE_AUTO_TM
)
command = object_id + struct.pack("!I", SupvActionIds.ENABLE_AUTO_TM)
command = PusTelecommand(service=8, subservice=128, ssc=62, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "48":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Disable auto TM"))
command = object_id + struct.pack(
"!I", SupvActionIds.DISABLE_AUTO_TM
)
command = object_id + struct.pack("!I", SupvActionIds.DISABLE_AUTO_TM)
command = PusTelecommand(service=8, subservice=128, ssc=63, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "49":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Enable ADC monitor task"))
command = object_id + struct.pack(
"!I", SupvActionIds.ENABLE_ADC_MONITOR_TASK
)
command = PusTelecommand(service=8, subservice=128, ssc=64, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "50":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Disable ADC monitor task"))
command = object_id + struct.pack(
"!I", SupvActionIds.DISABLE_ADC_MONITOR_TASK
)
command = PusTelecommand(service=8, subservice=128, ssc=65, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "51":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Logging request event buffers"))
tc_queue.appendleft(
(QueueCommands.PRINT, "PLOC Supervisor: Logging request event buffers")
)
command = pack_logging_buffer_request(object_id)
command = PusTelecommand(service=8, subservice=128, ssc=66, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "52":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Logging clear counters"))
command = object_id + struct.pack(
"!I", SupvActionIds.LOGGING_CLEAR_COUNTERS
tc_queue.appendleft(
(QueueCommands.PRINT, "PLOC Supervisor: Logging clear counters")
)
command = object_id + struct.pack("!I", SupvActionIds.LOGGING_CLEAR_COUNTERS)
command = PusTelecommand(service=8, subservice=128, ssc=67, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "53":
@@ -444,18 +366,22 @@ def pack_ploc_supv_commands(
command = PusTelecommand(service=8, subservice=128, ssc=68, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "54":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Logging request counters"))
command = object_id + struct.pack('!I', SupvActionIds.LOGGING_REQUEST_COUNTERS)
tc_queue.appendleft(
(QueueCommands.PRINT, "PLOC Supervisor: Logging request counters")
)
command = object_id + struct.pack("!I", SupvActionIds.LOGGING_REQUEST_COUNTERS)
command = PusTelecommand(service=8, subservice=128, ssc=69, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "55":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Request ADC report"))
command = object_id + struct.pack('!I', SupvActionIds.REQUEST_ADC_REPORT)
tc_queue.appendleft(
(QueueCommands.PRINT, "PLOC Supervisor: Request ADC report")
)
command = object_id + struct.pack("!I", SupvActionIds.REQUEST_ADC_REPORT)
command = PusTelecommand(service=8, subservice=128, ssc=70, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
elif op_code == "56":
tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Reset PL"))
command = object_id + struct.pack('!I', SupvActionIds.RESET_PL)
command = object_id + struct.pack("!I", SupvActionIds.RESET_PL)
command = PusTelecommand(service=8, subservice=128, ssc=71, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
@@ -552,7 +478,7 @@ def pack_auto_calibrate_alert_cmd(object_id: bytearray) -> bytearray:
@param object_id The object id of the PLOC supervisor handler.
"""
latchup_id = get_latchup_id()
mg = int(input("Specify MG:"))
mg = int(input("Specify MG: "))
command = bytearray()
command = object_id + struct.pack("!I", SupvActionIds.AUTO_CALIBRATE_ALERT)
command = command + struct.pack("!B", latchup_id)
@@ -573,7 +499,7 @@ def get_latchup_id() -> int:
key_string = key.ljust(key_column_width)
description_string = latchup_id_dict[key].ljust(description_column_width)
print(f"{key_string} | {description_string}")
return int(input("Specify latchup ID:"))
return int(input("Specify latchup ID: "))
def pack_set_alert_limit_cmd(object_id: bytearray) -> bytearray:
@@ -582,7 +508,7 @@ def pack_set_alert_limit_cmd(object_id: bytearray) -> bytearray:
@param object_id The object id of the PLOC supervisor handler.
"""
latchup_id = get_latchup_id()
dutycycle = int(input("Specify dutycycle:"))
dutycycle = int(input("Specify dutycycle: "))
command = bytearray()
command = object_id + struct.pack("!I", SupvActionIds.SET_ALERT_LIMIT)
command = command + struct.pack("!B", latchup_id)
@@ -590,49 +516,20 @@ def pack_set_alert_limit_cmd(object_id: bytearray) -> bytearray:
return command
def pack_set_alert_irq_filter_cmd(object_id: bytearray) -> bytearray:
"""
@brief This function packs the command to configure the latchup alert irq filter.
@param object_id The object id of the PLOC supervisor handler.
"""
latchup_id = get_latchup_id()
tp = int(input("Specify filter type (TP):"))
div = int(input("Specify clock divider (DIV):"))
command = bytearray()
command = object_id + struct.pack("!I", SupvActionIds.SET_ALERT_IRQ_FILTER)
command = command + struct.pack("!B", latchup_id)
command = command + struct.pack("!B", tp)
command = command + struct.pack("!B", div)
return command
def pack_set_adc_sweep_period_cmd(object_id: bytearray) -> bytearray:
"""
@brief This function packs the command to set the limit of a latchup alert.
@param object_id The object id of the PLOC supervisor handler.
"""
sweep_period = int(input("Specify sweep period (min 21 us):"))
command = bytearray()
command = object_id + struct.pack("!I", SupvActionIds.SET_ADC_SWEEP_PERIOD)
command = command + struct.pack("!I", sweep_period)
return command
def pack_set_adc_enabled_channels_cmd(object_id: bytearray) -> bytearray:
"""
@brief This function packs the command to enable or disable channels of the ADC.
@param object_id The object id of the PLOC supervisor handler.
"""
ch = int(input("Specify ch: 0x"), 16)
command = bytearray()
command = object_id + struct.pack("!I", SupvActionIds.SET_ADC_ENABLED_CHANNELS)
command = command + struct.pack("!H", ch)
return command
cmd = object_id + struct.pack("!I", SupvActionIds.SET_ADC_ENABLED_CHANNELS)
cmd = cmd + struct.pack("!H", ch)
return cmd
def pack_set_adc_window_and_stride_cmd(object_id: bytearray) -> bytearray:
window_size = int(input("Specify window size:"))
striding_step_size = int(input("Specify striding step size:"))
window_size = int(input("Specify window size: "))
striding_step_size = int(input("Specify striding step size: "))
command = bytearray()
command = object_id + struct.pack("!I", SupvActionIds.SET_ADC_WINDOW_AND_STRIDE)
command = command + struct.pack("!H", window_size)
@@ -648,18 +545,8 @@ def pack_set_adc_threshold_cmd(object_id: bytearray) -> bytearray:
return command
def pack_enable_nvms_cmd(object_id: bytearray) -> bytearray:
n01 = int(input("NVM0/1 (0 - off, 1 - on):"))
n3 = int(input("NVM3 (0 - off, 1 - on):"))
command = bytearray()
command = object_id + struct.pack("!I", SupvActionIds.ENABLE_NVMS)
command = command + struct.pack("!B", n01)
command = command + struct.pack("!B", n3)
return command
def pack_select_nvm_cmd(object_id: bytearray) -> bytearray:
mem = int(input("Specify NVM (0 - NVM0, 1 - MVM1):"))
mem = int(input("Specify NVM (0 - NVM0, 1 - MVM1): "))
command = bytearray()
command = object_id + struct.pack("!I", SupvActionIds.SELECT_NVM)
command = command + struct.pack("!B", mem)
@@ -667,7 +554,7 @@ def pack_select_nvm_cmd(object_id: bytearray) -> bytearray:
def pack_auto_em_tests_cmd(object_id: bytearray) -> bytearray:
test = int(input("Specify test (1 - complete, 2 - short):"))
test = int(input("Specify test (1 - complete, 2 - short): "))
command = bytearray()
command = object_id + struct.pack("!I", SupvActionIds.RUN_AUTO_EM_TESTS)
command = command + struct.pack("!B", test)
@@ -684,80 +571,39 @@ def pack_mram_wipe_cmd(object_id: bytearray) -> bytearray:
return command
def pack_print_cpu_stats_cmd(object_id: bytearray) -> bytearray:
en = 1
command = bytearray()
command = object_id + struct.pack("!I", SupvActionIds.PRINT_CPU_STATS)
command = command + struct.pack("!B", en)
return command
def pack_set_debug_verbosity_cmd(object_id: bytearray) -> bytearray:
command = bytearray()
verbosity = get_debug_verbosity()
command = object_id + struct.pack("!I", SupvActionIds.SET_DBG_VERBOSITY)
command = command + struct.pack("!B", verbosity)
return command
def pack_update_command(object_id: bytearray) -> bytearray:
command = bytearray()
memory_id = int(input("Specify memory ID: "))
start_address = int(input("Specify start address: 0x"), 16)
update_file = get_update_file()
command += object_id
command += struct.pack('!I', SupvActionIds.PERFORM_UPDATE)
command += bytearray(update_file, 'utf-8')
command += struct.pack("!I", SupvActionIds.PERFORM_UPDATE)
command += bytearray(update_file, "utf-8")
# Adding null terminator
command += struct.pack('!B', 0)
command += struct.pack('!B', memory_id)
command += struct.pack('!I', start_address)
command += struct.pack("!B", 0)
command += struct.pack("!B", memory_id)
command += struct.pack("!I", start_address)
return command
def pack_set_shutdown_timeout_command(object_id: bytearray) -> bytearray:
command = bytearray()
command += object_id
command += struct.pack('!I', SupvActionIds.SET_SHUTDOWN_TIMEOUT)
command += struct.pack("!I", SupvActionIds.SET_SHUTDOWN_TIMEOUT)
timeout = int(input("Specify shutdown timeout (ms): "))
command += struct.pack('!I', timeout)
command += struct.pack("!I", timeout)
return command
def pack_logging_buffer_request(object_id: bytearray) -> bytearray:
command = bytearray()
command += object_id
command += struct.pack('!I', SupvActionIds.LOGGING_REQUEST_EVENT_BUFFERS)
command += struct.pack("!I", SupvActionIds.LOGGING_REQUEST_EVENT_BUFFERS)
path = get_event_buffer_path()
command += bytearray(path, 'utf-8')
command += bytearray(path, "utf-8")
return command
def get_debug_verbosity() -> int:
tries = 0
while tries < 3:
try:
print("Debug verbosity options")
verbosity_options_dict = {
0: "None",
1: "Error",
2: "Warn",
3: "Info",
}
print("{:<6} | {}".format("Key", "Description"))
for entry in verbosity_options_dict.items():
print("{:<6} | {}".format(entry[0], entry[1]))
verbosity = int(input("Specify verbosity key: "))
if verbosity > len(verbosity_options_dict) - 1:
raise ValueError
return verbosity
except ValueError:
LOGGER.warning("Invalid verbosity key specified")
tries = tries + 1
LOGGER.error("get_debug_verbosity: Exceeded max tries to input verbosity key")
quit()
def pack_set_gpio_cmd(object_id: bytearray) -> bytearray:
port = int(input("Specify port : 0x"), 16)
pin = int(input("Specify pin: 0x"), 16)
@@ -779,9 +625,9 @@ def pack_read_gpio_cmd(object_id: bytearray) -> bytearray:
def pack_logging_set_topic(objetc_id: bytearray) -> bytearray:
command = objetc_id + struct.pack('!I', SupvActionIds.LOGGING_SET_TOPIC)
command = objetc_id + struct.pack("!I", SupvActionIds.LOGGING_SET_TOPIC)
tpc = int(input("Specify logging topic: "))
command += struct.pack('!B', tpc)
command += struct.pack("!B", tpc)
return command

View File

@@ -1,21 +1,52 @@
# -*- coding: utf-8 -*-
"""
@file reaction_wheels.py
"""reaction_wheels.py
@brief Tests for the reaction wheel handler
@author J. Meier
@date 20.06.2021
"""
import struct
from tmtccmd.config.definitions import QueueCommands
from tmtccmd.config.definitions import QueueCommands, ServiceOpCodeDictT, OpCodeDictKeys
from tmtccmd.tc.service_3_housekeeping import generate_one_hk_command, make_sid
from tmtccmd.config.globals import add_op_code_entry, add_service_op_code_entry
from tmtccmd.tc.packer import TcQueueT
from spacepackets.ecss.tc import PusTelecommand
from pus_tc.service_200_mode import pack_mode_data
from tmtccmd.tc.service_200_mode import pack_mode_data, Modes
from config.definitions import CustomServiceList
class OpCodesDevs:
SPEED = ["0", "speed"]
ON = ["1", "on"]
NML = ["2", "nml"]
OFF = ["3", "off"]
GET_TM = ["4", "tm"]
class InfoDevs:
SPEED = "Set speed"
ON = "Set On"
NML = "Set Normal"
OFF = "Set Off"
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:
@@ -38,66 +69,116 @@ class RampTime:
MS_1000 = 1000
def pack_single_rw_test_into(
object_id: bytearray, tc_queue: TcQueueT, op_code: str
) -> TcQueueT:
tc_queue.appendleft(
(
QueueCommands.PRINT,
"Testing reaction wheel handler with object id: 0x" + object_id.hex(),
)
def add_rw_cmds(cmd_dict: ServiceOpCodeDictT):
op_code_dict = dict()
add_op_code_entry(
op_code_dict=op_code_dict, info=InfoDevs.SPEED, keys=OpCodesDevs.SPEED
)
add_op_code_entry(op_code_dict=op_code_dict, info=InfoDevs.ON, keys=OpCodesDevs.ON)
add_op_code_entry(
op_code_dict=op_code_dict, info=InfoDevs.OFF, keys=OpCodesDevs.OFF
)
add_op_code_entry(
op_code_dict=op_code_dict, info=InfoDevs.NML, keys=OpCodesDevs.NML
)
add_op_code_entry(
op_code_dict=op_code_dict, info=InfoDevs.GET_TM, keys=OpCodesDevs.GET_TM
)
if op_code == "0" or op_code == "1":
add_service_op_code_entry(
srv_op_code_dict=cmd_dict,
name=CustomServiceList.REACTION_WHEEL_1.value,
op_code_entry=op_code_dict,
info="Reaction Wheel 1",
)
add_service_op_code_entry(
srv_op_code_dict=cmd_dict,
name=CustomServiceList.REACTION_WHEEL_2.value,
op_code_entry=op_code_dict,
info="Reaction Wheel 2",
)
add_service_op_code_entry(
srv_op_code_dict=cmd_dict,
name=CustomServiceList.REACTION_WHEEL_3.value,
op_code_entry=op_code_dict,
info="Reaction Wheel 3",
)
add_service_op_code_entry(
srv_op_code_dict=cmd_dict,
name=CustomServiceList.REACTION_WHEEL_4.value,
op_code_entry=op_code_dict,
info="Reaction Wheel 4",
)
op_code_dict = dict()
add_op_code_entry(op_code_dict=op_code_dict, info=InfoAss.ON, keys=OpCodesAss.ON)
add_op_code_entry(op_code_dict=op_code_dict, info=InfoAss.NML, keys=OpCodesAss.NML)
add_op_code_entry(op_code_dict=op_code_dict, info=InfoAss.OFF, keys=OpCodesAss.OFF)
add_service_op_code_entry(
srv_op_code_dict=cmd_dict,
name=CustomServiceList.RW_ASSEMBLY.value,
op_code_entry=op_code_dict,
info="Reaction Wheel Assembly",
)
def pack_single_rw_test_into(
object_id: bytes, rw_idx: int, tc_queue: TcQueueT, op_code: str
) -> TcQueueT:
if op_code in OpCodesDevs.SPEED:
speed = int(input("Specify speed [0.1 RPM]: "))
ramp_time = int(input("Specify ramp time [ms]: "))
tc_queue.appendleft((QueueCommands.PRINT, "Reaction Wheel: Set speed"))
tc_queue.appendleft((QueueCommands.PRINT, f"RW {rw_idx}: {InfoDevs.SPEED}"))
command = pack_set_speed_command(object_id, speed, ramp_time)
command = PusTelecommand(service=8, subservice=128, ssc=40, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "2":
tc_queue.appendleft((QueueCommands.PRINT, "Reaction Wheel: Switch to mode on"))
mode_data = pack_mode_data(object_id, 1, 0)
if op_code in OpCodesDevs.ON:
tc_queue.appendleft((QueueCommands.PRINT, f"RW {rw_idx}: {InfoDevs.ON}"))
mode_data = pack_mode_data(object_id, Modes.ON, 0)
command = PusTelecommand(service=200, subservice=1, ssc=41, app_data=mode_data)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "3":
tc_queue.appendleft(
(QueueCommands.PRINT, "Reaction Wheel: Switch to mode normal")
)
mode_data = pack_mode_data(object_id, 2, 0)
if op_code in OpCodesDevs.NML:
tc_queue.appendleft((QueueCommands.PRINT, f"RW {rw_idx}: {InfoDevs.NML}"))
mode_data = pack_mode_data(object_id, Modes.NORMAL, 0)
command = PusTelecommand(service=200, subservice=1, ssc=42, app_data=mode_data)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "4":
tc_queue.appendleft((QueueCommands.PRINT, "Reaction Wheel: Switch to mode off"))
mode_data = pack_mode_data(object_id, 0, 0)
if op_code in OpCodesDevs.OFF:
tc_queue.appendleft((QueueCommands.PRINT, f"RW {rw_idx}: {InfoDevs.OFF}"))
mode_data = pack_mode_data(object_id, Modes.OFF, 0)
command = PusTelecommand(service=200, subservice=1, ssc=43, app_data=mode_data)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "5":
tc_queue.appendleft(
(QueueCommands.PRINT, "Reaction Wheel: Send get-telemetry-command")
if op_code in OpCodesDevs.GET_TM:
tc_queue.appendleft((QueueCommands.PRINT, f"RW {rw_idx}: {InfoDevs.GET_TM}"))
command = generate_one_hk_command(
sid=make_sid(object_id=object_id, set_id=RwSetIds.TM_SET), ssc=0
)
command = object_id + RwCommandIds.GET_TM
command = PusTelecommand(service=8, subservice=128, ssc=44, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
return tc_queue
def pack_set_speed_command(
object_id: bytearray, speed: int, ramp_time: int
) -> bytearray:
def pack_rw_ass_cmds(tc_queue: TcQueueT, object_id: bytes, op_code: str):
if op_code in OpCodesAss.OFF:
pass
if op_code in OpCodesAss.ON:
pass
if op_code in OpCodesAss.NML:
pass
def pack_set_speed_command(object_id: bytes, speed: int, ramp_time: int) -> bytearray:
"""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 The time after which the reaction wheel will reached the commanded speed. Valid times are
10 - 10000 ms
: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: The time after which the reaction wheel will reached the commanded speed.
Valid times are 10 - 10000 ms
"""
command_id = RwCommandIds.SET_SPEED
command = bytearray()
command = object_id + command_id
command += object_id + command_id
command = command + struct.pack("!i", speed)
command = command + ramp_time.to_bytes(length=2, byteorder="big")
return command

View File

@@ -64,17 +64,17 @@ def pack_syrlinks_command(
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "3":
tc_queue.appendleft((QueueCommands.PRINT, "syrlinks: Set TX mode standby"))
command = object_id + struct.pack('!I', CommandIds.SET_TX_MODE_STANDBY)
command = object_id + struct.pack("!I", CommandIds.SET_TX_MODE_STANDBY)
command = PusTelecommand(service=8, subservice=128, ssc=10, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "4":
tc_queue.appendleft((QueueCommands.PRINT, "syrlinks: Set TX mode modulation"))
command = object_id + struct.pack('!I', CommandIds.SET_TX_MODE_MODULATION)
command = object_id + struct.pack("!I", CommandIds.SET_TX_MODE_MODULATION)
command = PusTelecommand(service=8, subservice=128, ssc=11, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "5":
tc_queue.appendleft((QueueCommands.PRINT, "syrlinks: Set TX mode CW"))
command = object_id + struct.pack('!I', CommandIds.SET_TX_MODE_CW)
command = object_id + struct.pack("!I", CommandIds.SET_TX_MODE_CW)
command = PusTelecommand(service=8, subservice=128, ssc=12, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "6":
@@ -89,26 +89,26 @@ def pack_syrlinks_command(
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "8":
tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Read TX status"))
command = object_id + struct.pack('!I', CommandIds.READ_TX_STATUS)
command = object_id + struct.pack("!I", CommandIds.READ_TX_STATUS)
command = PusTelecommand(service=8, subservice=128, ssc=13, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "9":
tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Read TX waveform"))
command = object_id + struct.pack('!I', CommandIds.READ_TX_WAVEFORM)
command = object_id + struct.pack("!I", CommandIds.READ_TX_WAVEFORM)
command = PusTelecommand(service=8, subservice=128, ssc=14, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "10":
tc_queue.appendleft(
(QueueCommands.PRINT, "Syrlinks: Read TX AGC value high byte")
)
command = object_id + struct.pack('!I', CommandIds.READ_TX_AGC_VALUE_HIGH_BYTE)
command = object_id + struct.pack("!I", CommandIds.READ_TX_AGC_VALUE_HIGH_BYTE)
command = PusTelecommand(service=8, subservice=128, ssc=15, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "11":
tc_queue.appendleft(
(QueueCommands.PRINT, "Syrlinks: Read TX AGC value low byte")
)
command = object_id + struct.pack('!I', CommandIds.READ_TX_AGC_VALUE_LOW_BYTE)
command = object_id + struct.pack("!I", CommandIds.READ_TX_AGC_VALUE_LOW_BYTE)
command = PusTelecommand(service=8, subservice=128, ssc=16, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if op_code == "12":

View File

@@ -27,7 +27,7 @@ from pus_tc.devs.tmp1075 import pack_tmp1075_test_into
from pus_tc.devs.ploc_mpsoc import pack_ploc_mpsoc_commands
from pus_tc.devs.ploc_supervisor import pack_ploc_supv_commands
from pus_tc.devs.heater import pack_heater_test_into
from pus_tc.devs.reaction_wheels import pack_single_rw_test_into
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
from pus_tc.devs.ploc_upater import pack_ploc_updater_test_into
from pus_tc.devs.ploc_memory_dumper import pack_ploc_memory_dumper_cmd
@@ -68,6 +68,7 @@ from config.object_ids import (
STR_IMG_HELPER_ID,
SYRLINKS_HANDLER_ID,
SOLAR_ARRAY_DEPLOYMENT_ID,
RW_ASSEMBLY,
)
@@ -146,24 +147,20 @@ def pack_service_queue_user(
object_id=object_id, tc_queue=service_queue, op_code=op_code
)
if service == CustomServiceList.REACTION_WHEEL_1.value:
object_id = RW1_ID
return pack_single_rw_test_into(
object_id=object_id, tc_queue=service_queue, op_code=op_code
object_id=RW1_ID, rw_idx=1, tc_queue=service_queue, op_code=op_code
)
if service == CustomServiceList.REACTION_WHEEL_2.value:
object_id = RW2_ID
return pack_single_rw_test_into(
object_id=object_id, tc_queue=service_queue, op_code=op_code
object_id=RW2_ID, rw_idx=2, tc_queue=service_queue, op_code=op_code
)
if service == CustomServiceList.REACTION_WHEEL_3.value:
object_id = RW3_ID
return pack_single_rw_test_into(
object_id=object_id, tc_queue=service_queue, op_code=op_code
object_id=RW3_ID, rw_idx=3, tc_queue=service_queue, op_code=op_code
)
if service == CustomServiceList.REACTION_WHEEL_4.value:
object_id = RW4_ID
return pack_single_rw_test_into(
object_id=object_id, tc_queue=service_queue, op_code=op_code
object_id=RW4_ID, rw_idx=4, tc_queue=service_queue, op_code=op_code
)
if service == CustomServiceList.RAD_SENSOR.value:
object_id = RAD_SENSOR_ID
@@ -233,6 +230,10 @@ def pack_service_queue_user(
return pack_tcs_sys_commands(tc_queue=service_queue, op_code=op_code)
if service == CustomServiceList.TIME.value:
return pack_set_current_time_ascii_command(tc_queue=service_queue, ssc=0)
if service == CustomServiceList.RW_ASSEMBLY:
return pack_rw_ass_cmds(
tc_queue=service_queue, object_id=RW_ASSEMBLY, op_code=op_code
)
LOGGER.warning("Invalid Service !")

View File

@@ -58,7 +58,7 @@ def handle_imtq_replies(
def handle_ploc_replies(
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
):
if action_id == PlocReplyIds.tm_mem_read_report:
if action_id == PlocReplyIds.TM_MEM_READ_RPT:
header_list = [
"PLOC Memory Address",
"PLOC Mem Len",
@@ -73,6 +73,16 @@ def handle_ploc_replies(
print(content_list)
printer.file_logger.info(header_list)
printer.file_logger.info(content_list)
elif action_id == PlocReplyIds.TM_CAM_CMD_RPT:
header_list = ["Camera reply string", "ACK"]
content_list = [
custom_data[: len(custom_data) - 1].decode("utf-8"),
hex(custom_data[-1]),
]
print(header_list)
print(content_list)
printer.file_logger.info(header_list)
printer.file_logger.info(content_list)
def handle_supervisor_replies(
@@ -86,6 +96,13 @@ def handle_supervisor_replies(
print(content_list)
printer.file_logger.info(header_list)
printer.file_logger.info(content_list)
elif action_id == SupvActionIds.READ_GPIO:
header_list = ["GPIO state"]
content_list = [struct.unpack("!H", custom_data[:2])[0]]
print(header_list)
print(content_list)
printer.file_logger.info(header_list)
printer.file_logger.info(content_list)
def handle_startracker_replies(

View File

@@ -55,40 +55,46 @@ def handle_regular_hk_print(
hk_packet: Service3Base,
hk_data: bytes,
):
object_id = object_id.as_bytes
objb = object_id.as_bytes
set_id = hk_packet.set_id
"""This function is called when a Service 3 Housekeeping packet is received."""
if object_id == obj_ids.SYRLINKS_HANDLER_ID:
if object_id in [obj_ids.RW1_ID, obj_ids.RW2_ID, obj_ids.RW3_ID, obj_ids.RW4_ID]:
handle_rw_hk_data(printer, object_id, set_id, hk_data)
if objb == obj_ids.SYRLINKS_HANDLER_ID:
if set_id == SetIds.RX_REGISTERS_DATASET:
return handle_syrlinks_rx_registers_dataset(printer, hk_data)
elif set_id == SetIds.TX_REGISTERS_DATASET:
return handle_syrlinks_tx_registers_dataset(printer, hk_data)
else:
LOGGER.info("Service 3 TM: Syrlinks handler reply with unknown set id")
elif object_id == obj_ids.IMTQ_HANDLER_ID:
if objb == obj_ids.IMTQ_HANDLER_ID:
if (set_id >= ImtqSetIds.POSITIVE_X_TEST) and (
set_id <= ImtqSetIds.NEGATIVE_Z_TEST
):
return handle_self_test_data(printer, hk_data)
else:
LOGGER.info("Service 3 TM: Syrlinks handler reply with unknown set id")
elif object_id == obj_ids.GPS_HANDLER_0_ID or object_id == obj_ids.GPS_HANDLER_1_ID:
if objb == obj_ids.GPS_HANDLER_0_ID or object_id == obj_ids.GPS_HANDLER_1_ID:
handle_gps_data(printer=printer, hk_data=hk_data)
elif object_id == obj_ids.BPX_HANDLER_ID:
if objb == obj_ids.BPX_HANDLER_ID:
handle_bpx_hk_data(hk_data=hk_data, set_id=set_id, printer=printer)
elif object_id == obj_ids.CORE_CONTROLLER_ID:
if objb == obj_ids.CORE_CONTROLLER_ID:
return handle_core_hk_data(printer=printer, hk_data=hk_data)
elif object_id == obj_ids.PDU_1_HANDLER_ID:
if objb == obj_ids.PDU_1_HANDLER_ID:
return handle_pdu_data(
printer=printer, pdu_idx=1, set_id=set_id, hk_data=hk_data
)
elif object_id == obj_ids.PDU_2_HANDLER_ID:
if objb == obj_ids.PDU_2_HANDLER_ID:
return handle_pdu_data(
printer=printer, pdu_idx=2, set_id=set_id, hk_data=hk_data
)
elif object_id == obj_ids.P60_DOCK_HANDLER:
if objb in [obj_ids.RW1_ID, obj_ids.RW2_ID, obj_ids.RW3_ID, obj_ids.RW4_ID]:
return handle_rw_hk_data(
printer=printer, object_id=object_id, set_id=set_id, hk_data=hk_data
)
if objb == obj_ids.P60_DOCK_HANDLER:
handle_p60_hk_data(printer=printer, set_id=set_id, hk_data=hk_data)
elif object_id == obj_ids.PL_PCDU_ID:
if objb == obj_ids.PL_PCDU_ID:
log_to_both(printer, "Received PL PCDU HK data")
else:
LOGGER.info("Service 3 TM: Parsing for this SID has not been implemented.")
@@ -702,6 +708,91 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
)
def handle_rw_hk_data(
printer: FsfwTmTcPrinter, object_id: ObjectId, set_id: int, hk_data: bytes
):
from pus_tc.devs.reaction_wheels import RwSetIds
current_idx = 0
if set_id == RwSetIds.TEMPERATURE_SET_ID:
log_to_both(
printer,
f"Received Temperature HK (ID {set_id}) from Reaction Wheel {object_id.name}",
)
temp = struct.unpack("!I", hk_data[0:4])
log_to_both(printer, f"Temperature {temp}")
current_idx += 4
if set_id == RwSetIds.STATUS_SET_ID:
log_to_both(
printer,
f"Received Status HK (ID {set_id}) from Reaction Wheel {object_id.name}",
)
fmt_str = "!iiBB"
inc_len = struct.calcsize(fmt_str)
(speed, ref_speed, state, clc_mode) = struct.unpack(
fmt_str, hk_data[current_idx : current_idx + inc_len]
)
current_idx += inc_len
log_to_both(
printer,
f"Speed {speed} rpm | Reference Speed {ref_speed} rpm | State {state} | CLC Mode {clc_mode}",
)
if set_id == RwSetIds.LAST_RESET:
log_to_both(
printer,
f"Received Last Reset HK (ID {set_id}) from Reaction Wheel {object_id.name}",
)
fmt_str = "!BB"
inc_len = struct.calcsize(fmt_str)
(last_reset_status, curr_reset_status) = struct.unpack(
fmt_str, hk_data[current_idx : current_idx + inc_len]
)
current_idx += inc_len
log_to_both(
printer,
f"Last Reset Status {last_reset_status} | Current Reset Status {curr_reset_status}",
)
if set_id == RwSetIds.TM_SET:
log_to_both(
printer,
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])
pass
if current_idx > 0:
printer.print_validity_buffer(
validity_buffer=hk_data[current_idx:], num_vars=27
)
pass
def log_to_both(printer: FsfwTmTcPrinter, string: str):
print(string)
printer.file_logger.info(string)

Submodule tmtccmd updated: ea2992372f...9f566a739a