Compare commits

...

30 Commits

Author SHA1 Message Date
0c1bfc6fd3 bump changelog 2023-05-12 16:27:16 +02:00
04bbe057e7 flash c ontent report works now 2023-05-12 16:24:45 +02:00
e05a54b076 somethings wrong with the format 2023-05-12 16:08:57 +02:00
ef0adef04a start adding action reply handler for MPSoC 2023-05-12 15:51:17 +02:00
377e98b5c2 bugfix MPSoC HK parsing 2023-05-12 13:22:58 +02:00
87e5abe8eb add missing command list of PLOC MPSoC commands 2023-05-12 11:51:47 +02:00
13fd9a7d84 Merge pull request 'impl MPSoC HK parsing' (#188) from mpsoc_commands into main
Reviewed-on: #188
Reviewed-by: Marius Eggert <eggertm@irs.uni-stuttgart.de>
2023-05-04 15:30:49 +02:00
bbcc0f9de7 Merge branch 'mpsoc_commands' of https://egit.irs.uni-stuttgart.de/eive/eive-tmtc into mpsoc_commands 2023-05-04 15:17:12 +02:00
a0aa6525e4 fix 2023-05-04 15:17:00 +02:00
1ab8710040 Merge branch 'main' into mpsoc_commands 2023-05-04 14:51:13 +02:00
f480d86fbd impl MPSoC HK parsing 2023-05-04 14:50:38 +02:00
4d921e01af Merge pull request 'MPSoC module update' (#187) from mpsoc_commands into main
Reviewed-on: #187
Reviewed-by: Marius Eggert <eggertm@irs.uni-stuttgart.de>
2023-05-04 12:01:43 +02:00
b505524e0b Merge branch 'main' into mpsoc_commands 2023-05-04 12:01:33 +02:00
e0e9a310b9 add command to get flash dir content 2023-05-04 11:51:50 +02:00
0e9ebefc87 new mpsoc commands 2023-05-04 11:27:12 +02:00
e85d1a1966 Merge pull request 'most important bugfix' (#186) from most-important-bugfix into main
Reviewed-on: #186
2023-05-03 13:36:31 +02:00
4ff50b6559 bub 2023-05-03 13:34:14 +02:00
60fba8b6d9 Merge pull request 'more system modes' (#185) from more-system-modes into main
Reviewed-on: #185
2023-04-28 10:25:35 +02:00
1707f24612 more system modes 2023-04-19 15:10:44 +02:00
5fbd19bb6c Merge branch 'main' of https://egit.irs.uni-stuttgart.de/eive/eive-tmtc 2023-04-17 18:41:08 +02:00
4083a3090f added event handler for reboot events 2023-04-17 18:40:27 +02:00
0c6a9677e1 bump patch revision 2023-04-17 11:43:43 +02:00
7eaf6557eb update generated event file 2023-04-17 11:34:56 +02:00
3bb0a08e95 bump tmtc to v3.1.0 2023-04-16 11:53:06 +02:00
d623e83be8 done 2023-04-16 02:56:14 +02:00
5a49c4a6ce that should do the job 2023-04-16 02:00:50 +02:00
bb463aa05c changelog 2023-04-15 23:12:57 +02:00
77e90328a1 some fixes 2023-04-15 21:53:11 +02:00
37bb164cc4 add missing return statement 2023-04-15 20:51:19 +02:00
bebde054f4 add new action commands 2023-04-15 20:51:08 +02:00
10 changed files with 563 additions and 127 deletions

View File

@ -10,6 +10,24 @@ list yields a list of all related PRs for each release.
# [unreleased] # [unreleased]
## Added
- Event handling for reboot counter events.
- Start adding new MPSoC commands, improve MPSoC commanding module a bit.
- Handling for PLOC MPSoC flash content report.
# [v3.1.1] 2023-04-17
## Added
- Update generated event file.
# [v3.1.0] 2023-04-16
## Added
- New core controller file system helper commands.
# [v3.0.0] 2023-04-14 # [v3.0.0] 2023-04-14
## Fixed ## Fixed

View File

@ -1,12 +1,12 @@
__version__ = "3.0.0" __version__ = "3.1.1"
import logging import logging
from pathlib import Path from pathlib import Path
SW_NAME = "eive-tmtc" SW_NAME = "eive-tmtc"
VERSION_MAJOR = 3 VERSION_MAJOR = 3
VERSION_MINOR = 0 VERSION_MINOR = 1
VERSION_REVISION = 0 VERSION_REVISION = 1
EIVE_TMTC_ROOT = Path(__file__).parent EIVE_TMTC_ROOT = Path(__file__).parent
PACKAGE_ROOT = EIVE_TMTC_ROOT.parent PACKAGE_ROOT = EIVE_TMTC_ROOT.parent

View File

@ -274,6 +274,7 @@ Event ID (dec); Event ID (hex); Name; Severity; Description; File Path
14105;0x3719;CAMERA_OVERHEATING;HIGH;No description;mission/controller/tcsDefs.h 14105;0x3719;CAMERA_OVERHEATING;HIGH;No description;mission/controller/tcsDefs.h
14106;0x371a;PCDU_SYSTEM_OVERHEATING;HIGH;No description;mission/controller/tcsDefs.h 14106;0x371a;PCDU_SYSTEM_OVERHEATING;HIGH;No description;mission/controller/tcsDefs.h
14107;0x371b;HEATER_NOT_OFF_FOR_OFF_MODE;MEDIUM;No description;mission/controller/tcsDefs.h 14107;0x371b;HEATER_NOT_OFF_FOR_OFF_MODE;MEDIUM;No description;mission/controller/tcsDefs.h
14108;0x371c;MGT_OVERHEATING;MEDIUM;No description;mission/controller/tcsDefs.h
14201;0x3779;TX_TIMER_EXPIRED;INFO;The transmit timer to protect the Syrlinks expired P1: The current timer value;mission/system/com/ComSubsystem.h 14201;0x3779;TX_TIMER_EXPIRED;INFO;The transmit timer to protect the Syrlinks expired P1: The current timer value;mission/system/com/ComSubsystem.h
14202;0x377a;BIT_LOCK_TX_ON;INFO;Transmitter will be turned on due to detection of bitlock;mission/system/com/ComSubsystem.h 14202;0x377a;BIT_LOCK_TX_ON;INFO;Transmitter will be turned on due to detection of bitlock;mission/system/com/ComSubsystem.h
14300;0x37dc;POSSIBLE_FILE_CORRUPTION;LOW;P1: Result code of TM packet parser. P2: Timestamp of possibly corrupt file as a unix timestamp.;mission/persistentTmStoreDefs.h 14300;0x37dc;POSSIBLE_FILE_CORRUPTION;LOW;P1: Result code of TM packet parser. P2: Timestamp of possibly corrupt file as a unix timestamp.;mission/persistentTmStoreDefs.h

1 Event ID (dec) Event ID (hex) Name Severity Description File Path
274 14105 0x3719 CAMERA_OVERHEATING HIGH No description mission/controller/tcsDefs.h
275 14106 0x371a PCDU_SYSTEM_OVERHEATING HIGH No description mission/controller/tcsDefs.h
276 14107 0x371b HEATER_NOT_OFF_FOR_OFF_MODE MEDIUM No description mission/controller/tcsDefs.h
277 14108 0x371c MGT_OVERHEATING MEDIUM No description mission/controller/tcsDefs.h
278 14201 0x3779 TX_TIMER_EXPIRED INFO The transmit timer to protect the Syrlinks expired P1: The current timer value mission/system/com/ComSubsystem.h
279 14202 0x377a BIT_LOCK_TX_ON INFO Transmitter will be turned on due to detection of bitlock mission/system/com/ComSubsystem.h
280 14300 0x37dc POSSIBLE_FILE_CORRUPTION LOW P1: Result code of TM packet parser. P2: Timestamp of possibly corrupt file as a unix timestamp. mission/persistentTmStoreDefs.h

View File

@ -2,7 +2,8 @@ import struct
from eive_tmtc.config.object_ids import * from eive_tmtc.config.object_ids import *
from eive_tmtc.tmtc.acs.imtq import ImtqActionId from eive_tmtc.tmtc.acs.imtq import ImtqActionId
from eive_tmtc.pus_tm.defs import PrintWrapper from eive_tmtc.pus_tm.defs import PrintWrapper
from eive_tmtc.tmtc.payload.ploc_mpsoc import PlocReplyIds from eive_tmtc.tmtc.core import handle_core_ctrl_action_replies
from eive_tmtc.tmtc.payload.ploc_mpsoc import handle_mpsoc_data_reply
from eive_tmtc.tmtc.payload.ploc_supervisor import SupvActionId from eive_tmtc.tmtc.payload.ploc_supervisor import SupvActionId
from eive_tmtc.tmtc.acs.star_tracker import StarTrackerActionId from eive_tmtc.tmtc.acs.star_tracker import StarTrackerActionId
from eive_tmtc.tmtc.power.tm import handle_get_param_data_reply from eive_tmtc.tmtc.power.tm import handle_get_param_data_reply
@ -22,7 +23,6 @@ def handle_action_reply(
tm_packet = Service8FsfwTm.unpack( tm_packet = Service8FsfwTm.unpack(
raw_telemetry=raw_tm, time_reader=CdsShortTimestamp.empty() raw_telemetry=raw_tm, time_reader=CdsShortTimestamp.empty()
) )
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) object_id = obj_id_dict.get(tm_packet.source_object_id_as_bytes)
pw = PrintWrapper(printer) pw = PrintWrapper(printer)
custom_data = tm_packet.custom_data custom_data = tm_packet.custom_data
@ -34,9 +34,11 @@ def handle_action_reply(
if object_id.as_bytes == IMTQ_HANDLER_ID: if object_id.as_bytes == IMTQ_HANDLER_ID:
return handle_imtq_replies(action_id, printer, custom_data) return handle_imtq_replies(action_id, printer, custom_data)
elif object_id.as_bytes == PLOC_MPSOC_ID: elif object_id.as_bytes == PLOC_MPSOC_ID:
return handle_ploc_replies(action_id, printer, custom_data) return handle_mpsoc_data_reply(action_id, printer, custom_data)
elif object_id.as_bytes == PLOC_SUPV_ID: elif object_id.as_bytes == PLOC_SUPV_ID:
return handle_supervisor_replies(action_id, printer, custom_data) return handle_supervisor_replies(action_id, printer, custom_data)
elif object_id.as_bytes == CORE_CONTROLLER_ID:
return handle_core_ctrl_action_replies(action_id, printer, custom_data)
elif object_id.as_bytes == STAR_TRACKER_ID: elif object_id.as_bytes == STAR_TRACKER_ID:
return handle_startracker_replies(action_id, printer, custom_data) return handle_startracker_replies(action_id, printer, custom_data)
elif object_id.as_bytes in [ elif object_id.as_bytes in [
@ -68,36 +70,6 @@ def handle_imtq_replies(
printer.file_logger.info(content_list) printer.file_logger.info(content_list)
def handle_ploc_replies(
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
):
if action_id == PlocReplyIds.TM_MEM_READ_RPT:
header_list = [
"PLOC Memory Address",
"PLOC Mem Len",
"PLOC Read Memory Data",
]
content_list = [
"0x" + custom_data[:4].hex(),
struct.unpack("!H", custom_data[4:6])[0],
"0x" + custom_data[6:10].hex(),
]
print(header_list)
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( def handle_supervisor_replies(
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
): ):

View File

@ -80,6 +80,18 @@ def handle_event_packet(raw_tm: bytes, printer: FsfwTmTcPrinter):
pw.dlog( pw.dlog(
f"Mode Number {event_def.param1}, Mode Name {mode_name}, Submode: {event_def.param2}" f"Mode Number {event_def.param1}, Mode Name {mode_name}, Submode: {event_def.param2}"
) )
if info.name == "INDIVIDUAL_BOOT_COUNTS":
boot_count_00 = (event_def.param1 >> 16) & 0xFFFF
boot_count_01 = event_def.param1 & 0xFFFF
boot_count_10 = (event_def.param2 >> 16) & 0xFFFF
boot_count_11 = event_def.param2 & 0xFFFF
pw.dlog(f"Boot count 0 0: {boot_count_00}")
pw.dlog(f"Boot count 0 1: {boot_count_01}")
pw.dlog(f"Boot count 1 0: {boot_count_10}")
pw.dlog(f"Boot count 1 1: {boot_count_11}")
if info.name == "REBOOT_COUNTER":
boot_count = (event_def.param1 << 32) | event_def.param2
pw.dlog(f"Total boot count: {boot_count}")
if info.name == "VERSION_INFO": if info.name == "VERSION_INFO":
specific_handler = True specific_handler = True
ver_major = (event_def.param1 >> 24) & 0xFF ver_major = (event_def.param1 >> 24) & 0xFF

View File

@ -4,6 +4,7 @@ import logging
# from pus_tm.tcp_server_objects import TCP_SEVER_SENSOR_TEMPERATURES # from pus_tm.tcp_server_objects import TCP_SEVER_SENSOR_TEMPERATURES
from eive_tmtc.tmtc.acs.acs_ctrl import handle_acs_ctrl_hk_data from eive_tmtc.tmtc.acs.acs_ctrl import handle_acs_ctrl_hk_data
from eive_tmtc.tmtc.payload.ploc_mpsoc import handle_ploc_mpsoc_hk_data
from eive_tmtc.tmtc.tcs.rtd import RTD_NAMES, handle_rtd_hk from eive_tmtc.tmtc.tcs.rtd import RTD_NAMES, handle_rtd_hk
from eive_tmtc.tmtc.acs.star_tracker import handle_str_hk_data from eive_tmtc.tmtc.acs.star_tracker import handle_str_hk_data
from eive_tmtc.tmtc.power.plpcdu import handle_plpcdu_hk from eive_tmtc.tmtc.power.plpcdu import handle_plpcdu_hk
@ -117,6 +118,10 @@ def handle_regular_hk_print(
return handle_pdu_data( return handle_pdu_data(
printer=printer, pdu_idx=2, set_id=set_id, hk_data=hk_data printer=printer, pdu_idx=2, set_id=set_id, hk_data=hk_data
) )
elif objb == obj_ids.PLOC_MPSOC_ID:
return handle_ploc_mpsoc_hk_data(
printer=printer, hk_data=hk_data, set_id=set_id
)
elif objb == obj_ids.ACU_HANDLER_ID: elif objb == obj_ids.ACU_HANDLER_ID:
return handle_acu_hk_data(printer=printer, hk_data=hk_data, set_id=set_id) return handle_acu_hk_data(printer=printer, hk_data=hk_data, set_id=set_id)
elif objb == obj_ids.RAD_SENSOR_ID: elif objb == obj_ids.RAD_SENSOR_ID:

View File

@ -979,7 +979,7 @@ def handle_gps_data_processed(pw: PrintWrapper, hk_data: bytes):
pw.dlog(f"GPS Altitude: {alt} [m]") pw.dlog(f"GPS Altitude: {alt} [m]")
pw.dlog(f"GPS Position: {pos} [m]") pw.dlog(f"GPS Position: {pos} [m]")
pw.dlog(f"GPS Velocity: {velo} [m/s]") pw.dlog(f"GPS Velocity: {velo} [m/s]")
pw.printer.print_validity_buffer(hk_data[current_idx:], num_vars=4) pw.printer.print_validity_buffer(hk_data[current_idx:], num_vars=5)
def handle_mekf_data(pw: PrintWrapper, hk_data: bytes): def handle_mekf_data(pw: PrintWrapper, hk_data: bytes):

View File

@ -1,6 +1,8 @@
import enum import enum
import logging import logging
import os
import struct import struct
from pathlib import Path
from eive_tmtc.pus_tm.defs import PrintWrapper from eive_tmtc.pus_tm.defs import PrintWrapper
@ -23,7 +25,6 @@ _LOGGER = logging.getLogger(__name__)
class ActionId(enum.IntEnum): class ActionId(enum.IntEnum):
LIST_DIR_INTO_FILE = 0
ANNOUNCE_VERSION = 1 ANNOUNCE_VERSION = 1
ANNOUNCE_CURRENT_IMAGE = 2 ANNOUNCE_CURRENT_IMAGE = 2
ANNOUNCE_BOOT_COUNTS = 3 ANNOUNCE_BOOT_COUNTS = 3
@ -42,6 +43,12 @@ class ActionId(enum.IntEnum):
EXECUTE_SHELL_CMD_BLOCKING = 40 EXECUTE_SHELL_CMD_BLOCKING = 40
EXECUTE_SHELL_CMD_NON_BLOCKING = 41 EXECUTE_SHELL_CMD_NON_BLOCKING = 41
SYSTEMCTL_CMD_EXECUTOR = 42 SYSTEMCTL_CMD_EXECUTOR = 42
LIST_DIR_INTO_FILE = 50
LIST_DIR_DUMP_DIRECTLY = 51
CP_HELPER = 52
MV_HELPER = 53
RM_HELPER = 54
MKDIR_HELPER = 55
class ParamId(enum.IntEnum): class ParamId(enum.IntEnum):
@ -59,6 +66,12 @@ class OpCode:
EXECUTE_SHELL_CMD_BLOCKING = "exec_cmd_blocking" EXECUTE_SHELL_CMD_BLOCKING = "exec_cmd_blocking"
EXECUTE_SHELL_CMD_NON_BLOCKING = "exec_cmd_non_blocking" EXECUTE_SHELL_CMD_NON_BLOCKING = "exec_cmd_non_blocking"
SYSTEMCTL_CMD_EXECUTOR = "systemctl_cmd" SYSTEMCTL_CMD_EXECUTOR = "systemctl_cmd"
LIST_DIR_INTO_FILE = "list_dir_into_file"
LIST_DIR_DUMP_DIRECTLY = "list_dir_dump_directly"
CP_HELPER = "cp_helper"
MV_HELPER = "mv_helper"
RM_HELPER = "rm_helper"
MKDIR_HELPER = "mkdir_helper"
SET_PREF_SD = "set_pref_sd" SET_PREF_SD = "set_pref_sd"
REBOOT_XSC = ["reboot_xsc"] REBOOT_XSC = ["reboot_xsc"]
XSC_REBOOT_SELF = ["reboot_self"] XSC_REBOOT_SELF = ["reboot_self"]
@ -100,6 +113,12 @@ class Info:
SWITCH_TO_SD_0 = "Switch to SD card 0" SWITCH_TO_SD_0 = "Switch to SD card 0"
SWITCH_TO_SD_1 = "Switch to SD card 1" SWITCH_TO_SD_1 = "Switch to SD card 1"
SWITCH_TO_BOTH_SD_CARDS = "Switch to both SD cards with specified active card" SWITCH_TO_BOTH_SD_CARDS = "Switch to both SD cards with specified active card"
LIST_DIR_INTO_FILE = "List directory, dump output into file"
LIST_DIR_DUMP_DIRECTLY = "List directory, dump content directly"
CP_HELPER = "Filesystem Copy Helper"
MV_HELPER = "Filesystem Move Helper"
RM_HELPER = "Filesystem Removal Helper"
MKDIR_HELPER = "Filesystem Directory Creation Helper"
class Chip(enum.IntEnum): class Chip(enum.IntEnum):
@ -184,6 +203,12 @@ def add_core_controller_definitions(defs: TmtcDefinitionWrapper):
oce.add(keys=OpCode.SWITCH_TO_SD_0, info=Info.SWITCH_TO_SD_0) oce.add(keys=OpCode.SWITCH_TO_SD_0, info=Info.SWITCH_TO_SD_0)
oce.add(keys=OpCode.SWITCH_TO_SD_1, info=Info.SWITCH_TO_SD_1) oce.add(keys=OpCode.SWITCH_TO_SD_1, info=Info.SWITCH_TO_SD_1)
oce.add(keys=OpCode.SWITCH_TO_BOTH_SD_CARDS, info=Info.SWITCH_TO_BOTH_SD_CARDS) oce.add(keys=OpCode.SWITCH_TO_BOTH_SD_CARDS, info=Info.SWITCH_TO_BOTH_SD_CARDS)
oce.add(keys=OpCode.LIST_DIR_INTO_FILE, info=Info.LIST_DIR_INTO_FILE)
oce.add(keys=OpCode.LIST_DIR_DUMP_DIRECTLY, info=Info.LIST_DIR_DUMP_DIRECTLY)
oce.add(keys=OpCode.MV_HELPER, info=Info.MV_HELPER)
oce.add(keys=OpCode.CP_HELPER, info=Info.CP_HELPER)
oce.add(keys=OpCode.RM_HELPER, info=Info.RM_HELPER)
oce.add(keys=OpCode.MKDIR_HELPER, info=Info.MKDIR_HELPER)
defs.add_service(CustomServiceList.CORE.value, "Core Controller", oce) defs.add_service(CustomServiceList.CORE.value, "Core Controller", oce)
@ -369,12 +394,97 @@ def pack_core_commands(q: DefaultPusQueueHelper, op_code: str):
).pack() ).pack()
) )
) )
elif op_code == OpCode.CP_HELPER:
cp_recursive = int(input("Copy recursively (0/1) ?: "))
if cp_recursive not in [0, 1]:
raise ValueError("Invalid value, only 0 or 1 allowed")
user_data = bytearray([cp_recursive])
user_data.extend(packet_source_dest_path("Copy"))
q.add_log_cmd(Info.CP_HELPER)
q.add_pus_tc(
create_action_cmd(CORE_CONTROLLER_ID, ActionId.CP_HELPER, user_data)
)
elif op_code == OpCode.MV_HELPER:
user_data = packet_source_dest_path("Move")
q.add_log_cmd(Info.MV_HELPER)
q.add_pus_tc(
create_action_cmd(CORE_CONTROLLER_ID, ActionId.MV_HELPER, user_data)
)
elif op_code == OpCode.RM_HELPER:
rm_recursive = int(input("Remove with recursive (-r) option (0/1) ?: "))
if rm_recursive not in [0, 1]:
raise ValueError("Invalid value, only 0 or 1 allowed")
rm_force = int(input("Remove with force (-f) option (0/1) ?: "))
if rm_force not in [0, 1]:
raise ValueError("Invalid value, only 0 or 1 allowed")
user_data = bytearray([rm_recursive, rm_force])
removed_file_or_dir = input("Specify absolute path to be removed: ")
user_data.extend(removed_file_or_dir.encode())
user_data.append(0)
q.add_log_cmd(Info.RM_HELPER)
q.add_pus_tc(
create_action_cmd(CORE_CONTROLLER_ID, ActionId.RM_HELPER, user_data)
)
elif op_code == OpCode.LIST_DIR_INTO_FILE:
q.add_log_cmd(Info.LIST_DIR_INTO_FILE)
user_data = list_directory_base_user_data()
dest_file_path = input("Destination file path: ")
user_data.extend(dest_file_path.encode())
user_data.append(0)
q.add_pus_tc(
create_action_cmd(
CORE_CONTROLLER_ID, ActionId.LIST_DIR_INTO_FILE, user_data
)
)
elif op_code == OpCode.LIST_DIR_DUMP_DIRECTLY:
q.add_log_cmd(Info.LIST_DIR_DUMP_DIRECTLY)
user_data = list_directory_base_user_data()
q.add_pus_tc(
create_action_cmd(
CORE_CONTROLLER_ID, ActionId.LIST_DIR_DUMP_DIRECTLY, user_data
)
)
elif op_code == OpCode.MKDIR_HELPER:
q.add_log_cmd(Info.MKDIR_HELPER)
user_data = input("Specify absolute path of newly created directory: ")
user_data = bytearray(user_data.encode())
user_data.append(0)
q.add_pus_tc(
create_action_cmd(CORE_CONTROLLER_ID, ActionId.MKDIR_HELPER, user_data)
)
else: else:
_LOGGER.warning( _LOGGER.warning(
f"Unknown operation code {op_code} for core controller commands" f"Unknown operation code {op_code} for core controller commands"
) )
def list_directory_base_user_data() -> bytearray:
all_opt = int(input("Use all (-a) option (0/1) ?: "))
if all_opt not in [0, 1]:
raise ValueError("Invalid value, only 0 or 1 allowed")
recursive_opt = int(input("Use recursive (-R) option (0/1) ?: "))
if recursive_opt not in [0, 1]:
raise ValueError("Invalid value, only 0 or 1 allowed")
compression_opt = int(input("Compress target file (0/1) ?: "))
if compression_opt not in [0, 1]:
raise ValueError("Invalid value, only 0 or 1 allowed")
listing_path = input("Specify listing path (absolute path): ")
user_data = bytearray([all_opt, recursive_opt, compression_opt])
user_data.extend(listing_path.encode())
user_data.append(0)
return user_data
def packet_source_dest_path(context: str) -> bytes:
source = input(f"Specify {context} source file: ")
dest = input(f"Specify {context} destination file: ")
raw_src_dest = bytearray(source.encode())
raw_src_dest.append(0)
raw_src_dest.extend(dest.encode())
raw_src_dest.append(0)
return raw_src_dest
def reset_specific_boot_counter(q: DefaultPusQueueHelper, chip: int, copy: int): def reset_specific_boot_counter(q: DefaultPusQueueHelper, chip: int, copy: int):
q.add_log_cmd(f"Resetting boot counter {chip} {copy}") q.add_log_cmd(f"Resetting boot counter {chip} {copy}")
q.add_pus_tc( q.add_pus_tc(
@ -483,3 +593,49 @@ def handle_core_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
) )
pw.dlog(printout) pw.dlog(printout)
printer.print_validity_buffer(validity_buffer=hk_data[inc_len:], num_vars=3) printer.print_validity_buffer(validity_buffer=hk_data[inc_len:], num_vars=3)
def handle_core_ctrl_action_replies(
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytes
):
pw = PrintWrapper(printer)
if action_id == ActionId.LIST_DIR_DUMP_DIRECTLY:
if len(custom_data) < 4:
_LOGGER.warning("Data unexpectedly small")
return
seq_idx = struct.unpack("!I", custom_data[0:4])[0]
total_chunks = struct.unpack("!I", custom_data[4:8])[0]
compressed = custom_data[8]
ls_cmd = custom_data[9:].split(b"\x00")[0].decode()
# Include length of NULL termination
file_data_offset = 9 + len(ls_cmd) + 1
pw.dlog(
f"Received directory listing dump for ls command {ls_cmd}. "
f"Chunk {seq_idx + 1}/{total_chunks}"
)
def remove_if_exists_and_new(seq_idx_: int, path_: Path):
if seq_idx_ == 0 and path_.exists():
os.remove(path_)
if compressed:
path = Path("dir_listing.txt.gz")
remove_if_exists_and_new(seq_idx, path)
pw.dlog(
f"Compression option: {compressed}. Dumping file into dir_listing.txt.gz"
)
with open(path, "ab") as listing_file:
listing_file.write(custom_data[file_data_offset:])
else:
path = Path("dir_listing.txt")
remove_if_exists_and_new(seq_idx, path)
pw.dlog(
f"Compression option: {compressed}. Dumping file into dir_listing.txt"
)
with open(path, "a") as listing_file:
listing_file_str = custom_data[file_data_offset:].decode()
listing_file.write(listing_file_str)
if seq_idx + 1 == total_chunks:
pw.dlog("Full directory listing: ")
with open("dir_listing.txt", "r") as listing_file:
print(listing_file.read())

View File

@ -6,12 +6,14 @@
@author J. Meier @author J. Meier
@date 06.03.2021 @date 06.03.2021
""" """
import dataclasses
import logging import logging
import struct import struct
import enum import enum
from eive_tmtc.config.definitions import CustomServiceList from eive_tmtc.config.definitions import CustomServiceList
from eive_tmtc.config.object_ids import get_object_ids, PLOC_MPSOC_ID from eive_tmtc.config.object_ids import get_object_ids, PLOC_MPSOC_ID
from eive_tmtc.pus_tm.defs import PrintWrapper
from tmtccmd.config.tmtc import ( from tmtccmd.config.tmtc import (
tmtc_definitions_provider, tmtc_definitions_provider,
OpCodeEntry, OpCodeEntry,
@ -22,38 +24,44 @@ from tmtccmd.tc import service_provider
from tmtccmd.tc.decorator import ServiceProviderParams from tmtccmd.tc.decorator import ServiceProviderParams
from eive_tmtc.utility.input_helper import InputHelper from eive_tmtc.utility.input_helper import InputHelper
from tmtccmd.tc.pus_200_fsfw_mode import pack_mode_data, Mode from tmtccmd.tc.pus_200_fsfw_mode import pack_mode_data, Mode
from tmtccmd.tc.pus_8_fsfw_funccmd import create_action_cmd
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
MANUAL_INPUT = "1" MANUAL_INPUT = "1"
flash_write_file_dict = { FLASH_WRITE_FILE_DICT = {
MANUAL_INPUT: ["manual input", ""], MANUAL_INPUT: ("manual input", ""),
"2": ["/mnt/sd0/ploc/mpsoc/flash_write.bin", "/mnt/sd0/ploc/mpsoc/flash_write.bin"], "2": ("/mnt/sd0/ploc/mpsoc/flash_write.bin", "/mnt/sd0/ploc/mpsoc/flash_write.bin"),
} }
mpsoc_file_dict = { MPSOC_FILE_DICT = {
MANUAL_INPUT: ["manual input", ""], MANUAL_INPUT: ("manual input", ""),
"2": ["0:/flash", "0:/flash"], "2": ("0:/flash", "0:/flash"),
} }
SEQ_FILE_NAMES = ["0:/EM16/231", "0:/EQ04/E-75", "0:/EQ01/E130"] SEQ_FILE_NAMES = ["0:/EM16/231", "0:/EQ04/E-75", "0:/EQ01/E130"]
SEQ_FILE_DICT = { SEQ_FILE_DICT = {
MANUAL_INPUT: ["manual input", ""], MANUAL_INPUT: ("manual input", ""),
"2": [f"16QRM On Carrier 200 MBd ({SEQ_FILE_NAMES[0]})", f"{SEQ_FILE_NAMES[0]}"], "2": (f"16QRM On Carrier 200 MBd ({SEQ_FILE_NAMES[0]})", f"{SEQ_FILE_NAMES[0]}"),
"3": [f"QPSK On Carrier 780 MBd ({SEQ_FILE_NAMES[1]})", f"{SEQ_FILE_NAMES[1]}"], "3": (f"QPSK On Carrier 780 MBd ({SEQ_FILE_NAMES[1]})", f"{SEQ_FILE_NAMES[1]}"),
"4": [f"Maximum Bandwidth QPSK ({SEQ_FILE_NAMES[2]})", f"{SEQ_FILE_NAMES[2]}"], "4": (f"Maximum Bandwidth QPSK ({SEQ_FILE_NAMES[2]})", f"{SEQ_FILE_NAMES[2]}"),
} }
CARRIAGE_RETURN = 0xD CARRIAGE_RETURN = 0xD
class CommandId(enum.IntEnum): class SetId(enum.IntEnum):
HK_ID = 0
class ActionId(enum.IntEnum):
TC_MEM_WRITE = 1 TC_MEM_WRITE = 1
TC_MEM_READ = 2 TC_MEM_READ = 2
FLASH_WRITE = 9 TM_MEM_READ_RPT = 6
TC_FLASH_WRITE_FULL_FILE = 9
TC_FLASH_DELETE = 10 TC_FLASH_DELETE = 10
TC_REPLAY_START = 11 TC_REPLAY_START = 11
TC_REPLAY_STOP = 12 TC_REPLAY_STOP = 12
@ -64,28 +72,40 @@ class CommandId(enum.IntEnum):
TC_MODE_REPLAY = 16 TC_MODE_REPLAY = 16
TC_CAM_CMD_SEND = 17 TC_CAM_CMD_SEND = 17
TC_MODE_IDLE = 18 TC_MODE_IDLE = 18
TM_CAM_CMD_RPT = 19
SET_UART_TX_TRISTATE = 20 SET_UART_TX_TRISTATE = 20
RELEASE_UART_TX = 21 RELEASE_UART_TX = 21
TC_CAM_TAKE_PIC = 22 TC_CAM_TAKE_PIC = 22
TC_SIMPLEX_SEND_FILE = 23 TC_SIMPLEX_SEND_FILE = 23
TC_DOWNLINK_DATA_MODULATE = 24 TC_DOWNLINK_DATA_MODULATE = 24
TC_MODE_SNAPSHOT = 25 TC_MODE_SNAPSHOT = 25
TC_FLASH_DIR_GET_CONTENT = 28
TM_FLASH_DIRECTORY_CONTENT = 29
TC_FLASH_READ_FULL_FILE = 30
class OpCode: class OpCode:
ON = ["on"] ON = "on"
OFF = ["off"] OFF = "off"
NORMAL = ["normal"] NORMAL = "normal"
VERIFY_BOOT = ["verify_boot"] VERIFY_BOOT = "verify_boot"
MODE_REPLAY = ["mode_replay"] MODE_REPLAY = "mode_replay"
MODE_IDLE = ["mode_idle"] MODE_IDLE = "mode_idle"
REPLAY_WRITE_SEQ = ["replay_write"] REPLAY_WRITE_SEQ = "replay_write"
DOWNLINK_PWR_ON = ["downlink_pwr_on"] DOWNLINK_PWR_ON = "downlink_pwr_on"
REPLAY_START = ["replay_start"] MEM_WRITE = "memory_write"
CAM_TAKE_PIC = ["cam_take_pic"] MEM_READ = "memory_read"
SIMPLEX_SEND_FILE = ["simplex_send_file"] DOWNLINK_PWR_OFF = "downlink_pwr_off"
DOWNLINK_DATA_MODULATE = ["downlink_data_modulate"] FLASH_WRITE_FILE = "flash_write_file"
MODE_SNAPSHOT = ["mode_snapshot"] FLASH_READ_FILE = "flash_read_file"
FLASH_DELETE_FILE = "flash_delete_file"
FLASH_GET_DIR_CONTENT = "flash_get_dir_content"
REPLAY_STOP = "replay_stop"
REPLAY_START = "replay_start"
CAM_TAKE_PIC = "cam_take_pic"
SIMPLEX_SEND_FILE = "simplex_send_file"
DOWNLINK_DATA_MODULATE = "downlink_data_modulate"
MODE_SNAPSHOT = "mode_snapshot"
class Info: class Info:
@ -94,12 +114,18 @@ class Info:
NORMAL = "Normal" NORMAL = "Normal"
VERIFY_BOOT = "Verify boot by reading 0xdeadbeef from DEADBEEF address" VERIFY_BOOT = "Verify boot by reading 0xdeadbeef from DEADBEEF address"
MODE_REPLAY = "Switch to REPLAY mode" MODE_REPLAY = "Switch to REPLAY mode"
REPLAY_STOP = "Stop Replay"
MODE_IDLE = "Switch to IDLE mode" MODE_IDLE = "Switch to IDLE mode"
REPLAY_WRITE_SEQ = "Replay write sequence" REPLAY_WRITE_SEQ = "Replay write sequence"
DOWNLINK_PWR_ON = "Downlink Power On" DOWNLINK_PWR_ON = "Downlink Power On"
DOWNLINK_PWR_OFF = "Downlink Power Off"
REPLAY_START = "Replay Start" REPLAY_START = "Replay Start"
CAM_TAKE_PIC = "Cam Take Picture" CAM_TAKE_PIC = "Cam Take Picture"
SIMPLEX_SEND_FILE = "Simplex Send File" SIMPLEX_SEND_FILE = "Simplex Send File"
FLASH_READ_FILE = "Copy file from MPSoC to OBC"
FLASH_WRITE_FILE = "Copy file from OBC to MPSoC"
FLASH_DELETE_FILE = "Delete file on MPSoC"
FLASH_GET_DIR_CONTENT = "Get flash directory content on MPSoC"
DOWNLINK_DATA_MODULATE = "Downlink data modulate" DOWNLINK_DATA_MODULATE = "Downlink data modulate"
MODE_SNAPSHOT = "Mode Snapshot" MODE_SNAPSHOT = "Mode Snapshot"
@ -108,28 +134,25 @@ class MemAddresses(enum.IntEnum):
DEADBEEF = 0x40000004 DEADBEEF = 0x40000004
class PlocReplyIds(enum.IntEnum):
TM_MEM_READ_RPT = 6
TM_CAM_CMD_RPT = 19
@tmtc_definitions_provider @tmtc_definitions_provider
def add_ploc_mpsoc_cmds(defs: TmtcDefinitionWrapper): def add_ploc_mpsoc_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry() oce = OpCodeEntry()
oce.add(OpCode.OFF, Info.OFF) oce.add(OpCode.OFF, Info.OFF)
oce.add(OpCode.ON, Info.ON) oce.add(OpCode.ON, Info.ON)
oce.add(OpCode.NORMAL, Info.NORMAL) oce.add(OpCode.NORMAL, Info.NORMAL)
oce.add("3", "Ploc MPSoC: Memory write") oce.add(OpCode.MEM_WRITE, "Ploc MPSoC: Memory write")
oce.add("4", "Ploc MPSoC: Memory read") oce.add(OpCode.MEM_READ, "Ploc MPSoC: Memory read")
oce.add("5", "Ploc MPSoC: Flash write") oce.add(OpCode.FLASH_WRITE_FILE, Info.FLASH_WRITE_FILE)
oce.add("6", "Ploc MPSoC: Flash delete") oce.add(OpCode.FLASH_READ_FILE, Info.FLASH_READ_FILE)
oce.add(OpCode.FLASH_DELETE_FILE, Info.FLASH_DELETE_FILE)
oce.add(OpCode.FLASH_GET_DIR_CONTENT, Info.FLASH_GET_DIR_CONTENT)
oce.add(OpCode.REPLAY_START, Info.REPLAY_START) oce.add(OpCode.REPLAY_START, Info.REPLAY_START)
oce.add("8", "Ploc MPSoC: Replay stop") oce.add(OpCode.REPLAY_STOP, Info.REPLAY_STOP)
oce.add(OpCode.DOWNLINK_PWR_ON, Info.DOWNLINK_PWR_ON) oce.add(OpCode.DOWNLINK_PWR_ON, Info.DOWNLINK_PWR_ON)
oce.add("10", "Ploc MPSoC: Downlink pwr off") oce.add(OpCode.DOWNLINK_PWR_OFF, Info.DOWNLINK_PWR_OFF)
oce.add(OpCode.REPLAY_WRITE_SEQ, Info.REPLAY_WRITE_SEQ) oce.add(OpCode.REPLAY_WRITE_SEQ, Info.REPLAY_WRITE_SEQ)
oce.add("12", "Ploc MPSoC: OBSW reset sequence count") oce.add("12", "Ploc MPSoC: OBSW reset sequence count")
oce.add(OpCode.VERIFY_BOOT, "Ploc MPSoC: Read DEADBEEF address") oce.add(OpCode.VERIFY_BOOT, Info.VERIFY_BOOT)
oce.add(OpCode.MODE_REPLAY, Info.MODE_REPLAY) oce.add(OpCode.MODE_REPLAY, Info.MODE_REPLAY)
oce.add(OpCode.MODE_IDLE, Info.MODE_IDLE) oce.add(OpCode.MODE_IDLE, Info.MODE_IDLE)
oce.add("16", "Ploc MPSoC: Tc cam command send") oce.add("16", "Ploc MPSoC: Tc cam command send")
@ -152,19 +175,19 @@ def pack_ploc_mpsoc_commands(p: ServiceProviderParams):
f"Generate command for PLOC MPSoC with object id: {object_id.as_hex_string}" f"Generate command for PLOC MPSoC with object id: {object_id.as_hex_string}"
) )
obyt = object_id.as_bytes obyt = object_id.as_bytes
if op_code in OpCode.OFF: if op_code == OpCode.OFF:
q.add_log_cmd(f"{prefix}: {Info.OFF}") q.add_log_cmd(f"{prefix}: {Info.OFF}")
command = pack_mode_data(obyt, Mode.OFF, 0) command = pack_mode_data(obyt, Mode.OFF, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command)) q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
if op_code in OpCode.ON: if op_code == OpCode.ON:
q.add_log_cmd(f"{prefix}: {Info.ON}") q.add_log_cmd(f"{prefix}: {Info.ON}")
data = pack_mode_data(obyt, Mode.ON, 0) data = pack_mode_data(obyt, Mode.ON, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=data)) q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=data))
if op_code in OpCode.NORMAL: if op_code == OpCode.NORMAL:
q.add_log_cmd(f"{prefix}: {Info.NORMAL}") q.add_log_cmd(f"{prefix}: {Info.NORMAL}")
data = pack_mode_data(object_id.as_bytes, Mode.NORMAL, 0) data = pack_mode_data(object_id.as_bytes, Mode.NORMAL, 0)
q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=data)) q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=data))
if op_code == "3": if op_code == OpCode.MEM_WRITE:
q.add_log_cmd("PLOC MPSoC: TC mem write test") q.add_log_cmd("PLOC MPSoC: TC mem write test")
memory_address = int( memory_address = int(
input("PLOC MPSoC: Tc Mem Write: Type memory address: 0x"), 16 input("PLOC MPSoC: Tc Mem Write: Type memory address: 0x"), 16
@ -180,11 +203,15 @@ def pack_ploc_mpsoc_commands(p: ServiceProviderParams):
q.add_log_cmd("PLOC MPSoC: TC mem read test") q.add_log_cmd("PLOC MPSoC: TC mem read test")
data = prepare_mem_read_command(object_id=object_id.as_bytes) data = prepare_mem_read_command(object_id=object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code == "5": if op_code == OpCode.FLASH_WRITE_FILE:
q.add_log_cmd("PLOC MPSoC: Flash write") q.add_log_cmd(f"{prefix}: {Info.FLASH_WRITE_FILE}")
data = prepare_flash_write_cmd(object_id.as_bytes) data = prepare_flash_write_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code == "6": if op_code == OpCode.FLASH_READ_FILE:
q.add_log_cmd(f"{prefix}: {Info.FLASH_READ_FILE}")
data = prepare_flash_read_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code == OpCode.FLASH_DELETE_FILE:
q.add_log_cmd("PLOC MPSoC: Flash delete") q.add_log_cmd("PLOC MPSoC: Flash delete")
data = prepare_flash_delete_cmd(object_id.as_bytes) data = prepare_flash_delete_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
@ -192,76 +219,88 @@ def pack_ploc_mpsoc_commands(p: ServiceProviderParams):
q.add_log_cmd(f"{prefix}: {Info.REPLAY_START}") q.add_log_cmd(f"{prefix}: {Info.REPLAY_START}")
data = prepare_replay_start_cmd(object_id.as_bytes) data = prepare_replay_start_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code == "8": if op_code == OpCode.REPLAY_STOP:
q.add_log_cmd("PLOC MPSoC: Replay stop") q.add_log_cmd("PLOC MPSoC: Replay stop")
data = object_id.as_bytes + struct.pack("!I", CommandId.TC_REPLAY_STOP) data = object_id.as_bytes + struct.pack("!I", ActionId.TC_REPLAY_STOP)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code in OpCode.DOWNLINK_PWR_ON: if op_code == OpCode.DOWNLINK_PWR_ON:
q.add_log_cmd(f"{prefix}: {OpCode.DOWNLINK_PWR_ON}") q.add_log_cmd(f"{prefix}: {OpCode.DOWNLINK_PWR_ON}")
data = prepare_downlink_pwr_on_cmd(object_id.as_bytes) data = prepare_downlink_pwr_on_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code == "10": if op_code == OpCode.DOWNLINK_PWR_OFF:
q.add_log_cmd("PLOC MPSoC: Downlink pwr off") q.add_log_cmd("PLOC MPSoC: Downlink pwr off")
data = object_id.as_bytes + struct.pack("!I", CommandId.TC_DOWNLINK_PWR_OFF) data = object_id.as_bytes + struct.pack("!I", ActionId.TC_DOWNLINK_PWR_OFF)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code in OpCode.REPLAY_WRITE_SEQ: if op_code == OpCode.FLASH_GET_DIR_CONTENT:
q.add_log_cmd(f"{prefix}: {Info.FLASH_GET_DIR_CONTENT}")
dir_name = input("Please specify MPSoC directory name to get information for: ")
dir_name = bytearray(dir_name.encode("utf-8"))
dir_name.append(0)
q.add_pus_tc(
create_action_cmd(
object_id=object_id.as_bytes,
action_id=ActionId.TC_FLASH_DIR_GET_CONTENT,
user_data=dir_name,
)
)
if op_code == OpCode.REPLAY_WRITE_SEQ:
q.add_log_cmd(f"{prefix}: {Info.REPLAY_WRITE_SEQ}") q.add_log_cmd(f"{prefix}: {Info.REPLAY_WRITE_SEQ}")
data = prepare_replay_write_sequence_cmd(object_id.as_bytes) data = prepare_replay_write_sequence_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code == "12": if op_code == "12":
q.add_log_cmd("PLOC MPSoC: Reset OBSW sequence count") q.add_log_cmd("PLOC MPSoC: Reset OBSW sequence count")
data = object_id.as_bytes + struct.pack("!I", CommandId.OBSW_RESET_SEQ_COUNT) data = object_id.as_bytes + struct.pack("!I", ActionId.OBSW_RESET_SEQ_COUNT)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code in OpCode.VERIFY_BOOT: if op_code == OpCode.VERIFY_BOOT:
num_words = 1 num_words = 1
q.add_log_cmd(f"{prefix} {Info.VERIFY_BOOT}") q.add_log_cmd(f"{prefix} {Info.VERIFY_BOOT}")
data = ( data = (
object_id.as_bytes object_id.as_bytes
+ struct.pack("!I", CommandId.TC_MEM_READ) + struct.pack("!I", ActionId.TC_MEM_READ)
+ struct.pack("!I", MemAddresses.DEADBEEF) + struct.pack("!I", MemAddresses.DEADBEEF)
+ struct.pack("!H", num_words) + struct.pack("!H", num_words)
) )
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code in OpCode.MODE_REPLAY: if op_code == OpCode.MODE_REPLAY:
q.add_log_cmd("PLOC MPSoC: Tc mode replay") q.add_log_cmd("PLOC MPSoC: Tc mode replay")
data = object_id.as_bytes + struct.pack("!I", CommandId.TC_MODE_REPLAY) data = object_id.as_bytes + struct.pack("!I", ActionId.TC_MODE_REPLAY)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code in OpCode.MODE_IDLE: if op_code == OpCode.MODE_IDLE:
q.add_log_cmd("PLOC MPSoC: Tc mode idle") q.add_log_cmd("PLOC MPSoC: Tc mode idle")
data = object_id.as_bytes + struct.pack("!I", CommandId.TC_MODE_IDLE) data = object_id.as_bytes + struct.pack("!I", ActionId.TC_MODE_IDLE)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code == "16": if op_code == "16":
q.add_log_cmd("PLOC MPSoC: Tc cam command send") q.add_log_cmd("PLOC MPSoC: Tc cam command send")
cam_cmd = input("Specify cam command string: ") cam_cmd = input("Specify cam command string: ")
data = ( data = (
object_id.as_bytes object_id.as_bytes
+ struct.pack("!I", CommandId.TC_CAM_CMD_SEND) + struct.pack("!I", ActionId.TC_CAM_CMD_SEND)
+ bytearray(cam_cmd, "utf-8") + bytearray(cam_cmd, "utf-8")
) )
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code == "17": if op_code == "17":
q.add_log_cmd("PLOC MPSoC: Set UART TX tristate") q.add_log_cmd("PLOC MPSoC: Set UART TX tristate")
data = object_id.as_bytes + struct.pack("!I", CommandId.SET_UART_TX_TRISTATE) data = object_id.as_bytes + struct.pack("!I", ActionId.SET_UART_TX_TRISTATE)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code == "18": if op_code == "18":
q.add_log_cmd("PLOC MPSoC: Release UART TX") q.add_log_cmd("PLOC MPSoC: Release UART TX")
data = object_id.as_bytes + struct.pack("!I", CommandId.RELEASE_UART_TX) data = object_id.as_bytes + struct.pack("!I", ActionId.RELEASE_UART_TX)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code in OpCode.CAM_TAKE_PIC: if op_code == OpCode.CAM_TAKE_PIC:
q.add_log_cmd("PLOC MPSoC: Cam take picture") q.add_log_cmd("PLOC MPSoC: Cam take picture")
data = prepare_cam_take_pic_cmd(object_id.as_bytes) data = prepare_cam_take_pic_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code in OpCode.SIMPLEX_SEND_FILE: if op_code == OpCode.SIMPLEX_SEND_FILE:
q.add_log_cmd("PLOC MPSoC: Simplex send file") q.add_log_cmd("PLOC MPSoC: Simplex send file")
data = prepare_simplex_send_file_cmd(object_id.as_bytes) data = prepare_simplex_send_file_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code in OpCode.DOWNLINK_DATA_MODULATE: if op_code == OpCode.DOWNLINK_DATA_MODULATE:
q.add_log_cmd("PLOC MPSoC: Downlink data modulate") q.add_log_cmd("PLOC MPSoC: Downlink data modulate")
data = prepare_downlink_data_modulate_cmd(object_id.as_bytes) data = prepare_downlink_data_modulate_cmd(object_id.as_bytes)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code in OpCode.MODE_SNAPSHOT: if op_code == OpCode.MODE_SNAPSHOT:
q.add_log_cmd("PLOC MPSoC: Mode snapshot") q.add_log_cmd("PLOC MPSoC: Mode snapshot")
data = object_id.as_bytes + struct.pack("!I", CommandId.TC_MODE_SNAPSHOT) data = object_id.as_bytes + struct.pack("!I", ActionId.TC_MODE_SNAPSHOT)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data)) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
@ -278,7 +317,7 @@ def generate_write_mem_command(
""" """
command = ( command = (
object_id object_id
+ struct.pack("!I", CommandId.TC_MEM_WRITE) + struct.pack("!I", ActionId.TC_MEM_WRITE)
+ struct.pack("!I", memory_address) + struct.pack("!I", memory_address)
+ struct.pack("!H", mem_len) + struct.pack("!H", mem_len)
+ struct.pack("!I", memory_data) + struct.pack("!I", memory_data)
@ -291,31 +330,40 @@ def prepare_mem_read_command(object_id: bytes) -> bytearray:
num_words = int(input("PLOC MPSoC specify number of words (32-bit) to read: ")) num_words = int(input("PLOC MPSoC specify number of words (32-bit) to read: "))
command = ( command = (
object_id object_id
+ struct.pack("!I", CommandId.TC_MEM_READ) + struct.pack("!I", ActionId.TC_MEM_READ)
+ struct.pack("!I", memory_address) + struct.pack("!I", memory_address)
+ struct.pack("!H", num_words) + struct.pack("!H", num_words)
) )
return bytearray(command) return bytearray(command)
def prepare_flash_write_cmd(object_id: bytes) -> bytearray: def prepare_flash_base_cmd(action_id: int, object_id: bytes) -> bytearray:
obc_file = get_obc_file() obc_file = get_obc_file()
mpsoc_file = get_mpsoc_file() mpsoc_file = get_mpsoc_file()
command = ( command = bytearray(object_id)
object_id command.extend(struct.pack("!I", action_id))
+ struct.pack("!I", CommandId.FLASH_WRITE) command.extend(obc_file.encode("utf-8"))
+ bytearray(obc_file, "utf-8") command.append(0)
+ bytearray(mpsoc_file, "utf-8") command.extend(mpsoc_file.encode("utf-8"))
) command.append(0)
return bytearray(command) return command
def prepare_flash_write_cmd(object_id: bytes) -> bytearray:
return prepare_flash_base_cmd(ActionId.TC_FLASH_WRITE_FULL_FILE, object_id)
def prepare_flash_read_cmd(object_id: bytes) -> bytearray:
cmd = prepare_flash_base_cmd(ActionId.TC_FLASH_READ_FULL_FILE, object_id)
file_size = get_mpsoc_file_size()
cmd.extend(struct.pack("!I", file_size))
return cmd
def prepare_flash_delete_cmd(object_id: bytes) -> bytearray: def prepare_flash_delete_cmd(object_id: bytes) -> bytearray:
file = get_mpsoc_file() file = get_mpsoc_file()
command = ( command = (
object_id object_id + struct.pack("!I", ActionId.TC_FLASH_DELETE) + file.encode("utf-8")
+ struct.pack("!I", CommandId.TC_FLASH_DELETE)
+ bytearray(file, "utf-8")
) )
return bytearray(command) return bytearray(command)
@ -324,7 +372,7 @@ def prepare_replay_start_cmd(object_id: bytes) -> bytearray:
replay = int(input("Specify replay mode (0 - once, 1 - repeated): ")) replay = int(input("Specify replay mode (0 - once, 1 - repeated): "))
command = ( command = (
object_id object_id
+ struct.pack("!I", CommandId.TC_REPLAY_START) + struct.pack("!I", ActionId.TC_REPLAY_START)
+ struct.pack("!B", replay) + struct.pack("!B", replay)
) )
return bytearray(command) return bytearray(command)
@ -335,7 +383,7 @@ def prepare_downlink_pwr_on_cmd(object_id: bytes) -> bytearray:
lane_rate = int(input("Specify lane rate (0 - 9): ")) lane_rate = int(input("Specify lane rate (0 - 9): "))
command = ( command = (
object_id object_id
+ struct.pack("!I", CommandId.TC_DOWNLINK_PWR_ON) + struct.pack("!I", ActionId.TC_DOWNLINK_PWR_ON)
+ struct.pack("!B", mode) + struct.pack("!B", mode)
+ struct.pack("!B", lane_rate) + struct.pack("!B", lane_rate)
) )
@ -347,7 +395,7 @@ def prepare_replay_write_sequence_cmd(object_id: bytes) -> bytearray:
file = get_sequence_file() file = get_sequence_file()
command = ( command = (
object_id object_id
+ struct.pack("!I", CommandId.TC_REPLAY_WRITE_SEQUENCE) + struct.pack("!I", ActionId.TC_REPLAY_WRITE_SEQUENCE)
+ struct.pack("!B", use_decoding) + struct.pack("!B", use_decoding)
+ bytearray(file, "utf-8") + bytearray(file, "utf-8")
# + bytes([0]) # + bytes([0])
@ -377,7 +425,7 @@ def prepare_cam_take_pic_cmd(object_id: bytes) -> bytearray:
bypass_compressor = int(input("Specify bypassCompressor: ")) bypass_compressor = int(input("Specify bypassCompressor: "))
command = ( command = (
object_id object_id
+ struct.pack("!I", CommandId.TC_CAM_TAKE_PIC) + struct.pack("!I", ActionId.TC_CAM_TAKE_PIC)
+ bytearray(filename, "utf-8") + bytearray(filename, "utf-8")
+ bytes([0]) + bytes([0])
+ struct.pack("!B", encoder_setting_y) + struct.pack("!B", encoder_setting_y)
@ -395,7 +443,7 @@ def prepare_simplex_send_file_cmd(object_id: bytes) -> bytearray:
filename = input("Specify filename: ") filename = input("Specify filename: ")
command = ( command = (
object_id object_id
+ struct.pack("!I", CommandId.TC_SIMPLEX_SEND_FILE) + struct.pack("!I", ActionId.TC_SIMPLEX_SEND_FILE)
+ bytearray(filename, "utf-8") + bytearray(filename, "utf-8")
+ bytes([0]) + bytes([0])
) )
@ -409,7 +457,7 @@ def prepare_downlink_data_modulate_cmd(object_id: bytes) -> bytearray:
dest_mem_addr = int(input("Specify destMemAddr: ")) dest_mem_addr = int(input("Specify destMemAddr: "))
command = ( command = (
object_id object_id
+ struct.pack("!I", CommandId.TC_DOWNLINK_DATA_MODULATE) + struct.pack("!I", ActionId.TC_DOWNLINK_DATA_MODULATE)
+ struct.pack("!B", format) + struct.pack("!B", format)
+ struct.pack("!I", src_mem_addr) + struct.pack("!I", src_mem_addr)
+ struct.pack("!H", src_mem_len) + struct.pack("!H", src_mem_len)
@ -420,26 +468,33 @@ def prepare_downlink_data_modulate_cmd(object_id: bytes) -> bytearray:
def get_obc_file() -> str: def get_obc_file() -> str:
_LOGGER.info("Specify OBC file ") _LOGGER.info("Specify OBC file ")
input_helper = InputHelper(flash_write_file_dict) input_helper = InputHelper(FLASH_WRITE_FILE_DICT)
key = input_helper.get_key() key = input_helper.get_key()
if key == MANUAL_INPUT: if key == MANUAL_INPUT:
file = input("Ploc MPSoC: Specify absolute name of flash file: ") file = input("Ploc MPSoC: Specify absolute name of flash file: ")
else: else:
file = flash_write_file_dict[key][1] file = FLASH_WRITE_FILE_DICT[key][1]
return file return file
def get_mpsoc_file() -> str: def get_mpsoc_file() -> str:
_LOGGER.info("Specify MPSoC file") _LOGGER.info("Specify MPSoC file")
input_helper = InputHelper(mpsoc_file_dict) input_helper = InputHelper(MPSOC_FILE_DICT)
key = input_helper.get_key() key = input_helper.get_key()
if key == MANUAL_INPUT: if key == MANUAL_INPUT:
file = input("Ploc MPSoC: Specify absolute name file: ") file = input("Ploc MPSoC: Specify absolute name file: ")
else: else:
file = mpsoc_file_dict[key][1] file = MPSOC_FILE_DICT[key][1]
return file return file
def get_mpsoc_file_size() -> int:
file_size = int(input("Specify MPSoC file size"))
if file_size <= 0:
raise ValueError("Invalid file size")
return file_size
def get_sequence_file() -> str: def get_sequence_file() -> str:
_LOGGER.info("Specify sequence file") _LOGGER.info("Specify sequence file")
input_helper = InputHelper(SEQ_FILE_DICT) input_helper = InputHelper(SEQ_FILE_DICT)
@ -449,3 +504,192 @@ def get_sequence_file() -> str:
else: else:
file = SEQ_FILE_DICT[key][1] file = SEQ_FILE_DICT[key][1]
return file return file
def handle_ploc_mpsoc_hk_data(printer: FsfwTmTcPrinter, hk_data: bytes, set_id: int):
pw = PrintWrapper(printer)
if set_id == SetId.HK_ID:
fmt_str = "!IBBBBBBB"
current_idx = 0
inc_len = struct.calcsize(fmt_str)
(
status,
mode,
downlink_pwr_on,
downlink_reply_active,
downlink_jesd_sync_status,
downlink_dac_status,
cam_status,
cam_sdi_status,
) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
current_idx += inc_len
pw.ilog(_LOGGER, "Received MPSoC HK")
pw.dlog(f"Status: {status}")
pw.dlog(f"Mode: {mode}")
pw.dlog(f"Downlink Power On: {downlink_pwr_on}")
pw.dlog(f"Downlink Reply Active: {downlink_reply_active}")
pw.dlog(f"Downlink JESD Sync Status: {downlink_jesd_sync_status}")
pw.dlog(f"Downlink DAC Status: {downlink_dac_status}")
pw.dlog(f"CAM Status: {cam_status}")
pw.dlog(f"CAM SDI Status: {cam_sdi_status}")
fmt_str = "!fffffffff"
inc_len = struct.calcsize(fmt_str)
(
cam_fpga_temp,
cam_soc_temp,
sysmon_temp,
sysmon_vcc_int,
sysmon_vcc_aux,
sysmon_vcc_bram,
sysmon_vcc_paux,
sysmon_vcc_pint,
sysmon_vcc_pdro,
) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
current_idx += inc_len
pw.dlog(f"CAM FPGA Temperature: {cam_fpga_temp}")
pw.dlog(f"CAM SoC Temperature: {cam_soc_temp}")
pw.dlog(f"System Monitor Temperature: {sysmon_temp}")
pw.dlog(
f"SYSMON VCC INT {sysmon_vcc_int:.3f} | SYSMON VCC AUX {sysmon_vcc_aux:.3f} | "
f"SYSMON VCC BRAM {sysmon_vcc_bram:.3f}"
)
pw.dlog(
f"SYSMON VCC PAUX {sysmon_vcc_paux:.3f} | SYSMON VCC PINT {sysmon_vcc_pint:.3f} | "
f"SYSMON VCC PDRO {sysmon_vcc_pdro:.3f}"
)
fmt_str = "!fffffffffffff"
inc_len = struct.calcsize(fmt_str)
(
sysmon_mb_12v,
sysmon_mb_3v3,
sysmon_mb_1v8,
sysmon_vcc_12v,
sysmon_vcc_5v,
sysmon_vcc_3v3,
sysmon_vcc_3v3va,
sysmon_vcc_2v5ddr,
sysmon_vcc_1v2ddr,
sysmon_vcc_0v9,
sysmon_vcc_0v6vtt,
sysmon_safe_cotr_cur,
sysmon_nvm4_xo_cur,
) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
current_idx += inc_len
pw.dlog(
f"SYSMON MB 12V {sysmon_mb_12v:.3f} | SYSMON MB 3V3 {sysmon_mb_3v3:.3f} | "
f"SYSMON MBA 1V8 {sysmon_mb_1v8:.3f}"
)
pw.dlog(
f"SYSMON VCC 12V {sysmon_vcc_12v:.3f} | SYSMON VCC 5V {sysmon_vcc_5v:.3f} | "
f"SYSMON VCC 3V3 {sysmon_vcc_3v3:.3f} | SYSMON VCC 3V3VA {sysmon_vcc_3v3va}"
)
pw.dlog(
f"SYSMON VCC 2V5DDR {sysmon_vcc_2v5ddr:.3f} | "
f"SYSMON VCC 1V2DDR {sysmon_vcc_1v2ddr:.3f} | "
f"SYSMON VCC 0V9 {sysmon_vcc_0v9:.3f} | "
f"SYSMON VCC 0V6VTT {sysmon_vcc_0v6vtt}"
)
pw.dlog(
f"SYSMON SAFE COTS CURR: {sysmon_safe_cotr_cur} | "
f"SYSMON NVM4XO CURR {sysmon_nvm4_xo_cur}"
)
fmt_str = "!HHBB"
inc_len = struct.calcsize(fmt_str)
(
sem_uncorrectable_errs,
sem_correctable_errs,
sem_status,
reboot_mpsoc_required,
) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
pw.dlog(f"SEM IP Uncorrectable Errors: {sem_uncorrectable_errs}")
pw.dlog(f"SEM IP Correctable Errors: {sem_correctable_errs}")
pw.dlog(f"SEM IP Status: {sem_status}")
pw.dlog(f"Reboot MPSoC required: {reboot_mpsoc_required}")
else:
_LOGGER.warning(f"Unknown set ID {set_id} for MPSoC HK")
pass
@dataclasses.dataclass
class DirElement:
name: str
attr: int
size: int
def handle_mpsoc_data_reply(
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
):
pw = PrintWrapper(printer)
if action_id == ActionId.TM_MEM_READ_RPT:
header_list = [
"PLOC Memory Address",
"PLOC Mem Len",
"PLOC Read Memory Data",
]
content_list = [
"0x" + custom_data[:4].hex(),
struct.unpack("!H", custom_data[4:6])[0],
"0x" + custom_data[6:10].hex(),
]
print(header_list)
print(content_list)
printer.file_logger.info(header_list)
printer.file_logger.info(content_list)
elif action_id == ActionId.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)
elif action_id == ActionId.TM_FLASH_DIRECTORY_CONTENT:
if len(custom_data) < 16:
_LOGGER.warning(
"PLOC MPSoC flash directory data shorter than minimum 16 bytes"
)
current_idx = 0
dir_name_short = custom_data[current_idx : current_idx + 12].decode("utf-8")
current_idx += 12
num_elements = struct.unpack("!I", custom_data[current_idx : current_idx + 4])[
0
]
current_idx += 4
elem_names = []
elem_attrs = []
elem_sizes = []
expected_size = 16 + num_elements * 17
if len(custom_data) < expected_size:
_LOGGER.warning(
f"PLOC MPSoC flash directory data shorter than expected {expected_size}"
)
pw.dlog(
f"Received PLOC MPSoC flash directory content for path {dir_name_short} "
f"with {num_elements} elements"
)
# It is as weird as it looks..
for _ in range(num_elements):
end_of_str = custom_data[current_idx : current_idx + 12].index(b"\x00")
elem_name = custom_data[current_idx : current_idx + end_of_str].decode(
"utf-8"
)
current_idx += 12
elem_names.append(elem_name)
for _ in range(num_elements):
elem_attrs.append(custom_data[current_idx])
current_idx += 1
for _ in range(num_elements):
elem_sizes.append(
struct.unpack("!I", custom_data[current_idx : current_idx + 4])[0]
)
current_idx += 4
for i in range(num_elements):
pw.dlog(f"{DirElement(elem_names[i], elem_attrs[i], elem_sizes[i])}")

View File

@ -22,6 +22,10 @@ class SystemMode:
BOOT = 5 BOOT = 5
SAFE = AcsMode.SAFE SAFE = AcsMode.SAFE
IDLE = AcsMode.IDLE IDLE = AcsMode.IDLE
PTG_NADIR = AcsMode.PTG_NADIR
PTG_TARGET = AcsMode.PTG_TARGET
PTG_TARGET_GS = AcsMode.PTG_TARGET_GS
PTG_INERTIAL = AcsMode.PTG_INERTIAL
class ActionId(enum.IntEnum): class ActionId(enum.IntEnum):
@ -32,6 +36,10 @@ class OpCode:
BOOT_MODE = "boot" BOOT_MODE = "boot"
SAFE_MODE = "safe" SAFE_MODE = "safe"
IDLE_MODE = "idle" IDLE_MODE = "idle"
NADIR_MODE = "nadir"
TARGET_MODE = "ptg_target"
TARGET_GS_MODE = "ptg_target_gs"
INERTIAL_MODE = "ptg_inertial"
ANNOUNCE_MODES = "announce_modes" ANNOUNCE_MODES = "announce_modes"
REBOOT_I2C = "reboot_i2c" REBOOT_I2C = "reboot_i2c"
@ -40,6 +48,10 @@ class Info:
BOOT_MODE = "Command System into Boot Mode" BOOT_MODE = "Command System into Boot Mode"
SAFE_MODE = "Command System into Safe Mode" SAFE_MODE = "Command System into Safe Mode"
IDLE_MODE = "Command System into Idle Pointing Mode" IDLE_MODE = "Command System into Idle Pointing Mode"
NADIR_MODE = "Command System into Nadir Pointing Mode"
TARGET_MODE = "Command System into Target Pointing Mode"
TARGET_GS_MODE = "Command System into Target Groundstation Pointing Mode"
INERTIAL_MODE = "Command System into Inertial Pointing Mode"
ANNOUNCE_MODES = "Announce mode recursively" ANNOUNCE_MODES = "Announce mode recursively"
REBOOT_I2C = "Reboot I2C bus" REBOOT_I2C = "Reboot I2C bus"
@ -55,6 +67,18 @@ def build_system_cmds(p: ServiceProviderParams):
elif o == OpCode.IDLE_MODE: elif o == OpCode.IDLE_MODE:
q.add_log_cmd(f"{prefix}: {Info.IDLE_MODE}") q.add_log_cmd(f"{prefix}: {Info.IDLE_MODE}")
q.add_pus_tc(create_mode_command(EIVE_SYSTEM_ID, SystemMode.IDLE, 0)) q.add_pus_tc(create_mode_command(EIVE_SYSTEM_ID, SystemMode.IDLE, 0))
elif o == OpCode.NADIR_MODE:
q.add_log_cmd(f"{prefix}: {Info.NADIR_MODE}")
q.add_pus_tc(create_mode_command(EIVE_SYSTEM_ID, SystemMode.PTG_NADIR, 0))
elif o == OpCode.TARGET_MODE:
q.add_log_cmd(f"{prefix}: {Info.TARGET_MODE}")
q.add_pus_tc(create_mode_command(EIVE_SYSTEM_ID, SystemMode.PTG_TARGET, 0))
elif o == OpCode.TARGET_GS_MODE:
q.add_log_cmd(f"{prefix}: {Info.TARGET_GS_MODE}")
q.add_pus_tc(create_mode_command(EIVE_SYSTEM_ID, SystemMode.PTG_TARGET_GS, 0))
elif o == OpCode.INERTIAL_MODE:
q.add_log_cmd(f"{prefix}: {Info.INERTIAL_MODE}")
q.add_pus_tc(create_mode_command(EIVE_SYSTEM_ID, SystemMode.PTG_INERTIAL, 0))
elif o == OpCode.ANNOUNCE_MODES: elif o == OpCode.ANNOUNCE_MODES:
q.add_log_cmd(f"{prefix}: {Info.ANNOUNCE_MODES}") q.add_log_cmd(f"{prefix}: {Info.ANNOUNCE_MODES}")
q.add_pus_tc(create_announce_mode_recursive_command(EIVE_SYSTEM_ID)) q.add_pus_tc(create_announce_mode_recursive_command(EIVE_SYSTEM_ID))
@ -72,6 +96,10 @@ def add_system_cmd_defs(defs: TmtcDefinitionWrapper):
oce.add(keys=OpCode.BOOT_MODE, info=Info.BOOT_MODE) oce.add(keys=OpCode.BOOT_MODE, info=Info.BOOT_MODE)
oce.add(keys=OpCode.SAFE_MODE, info=Info.SAFE_MODE) oce.add(keys=OpCode.SAFE_MODE, info=Info.SAFE_MODE)
oce.add(keys=OpCode.IDLE_MODE, info=Info.IDLE_MODE) oce.add(keys=OpCode.IDLE_MODE, info=Info.IDLE_MODE)
oce.add(keys=OpCode.NADIR_MODE, info=Info.NADIR_MODE)
oce.add(keys=OpCode.TARGET_MODE, info=Info.TARGET_MODE)
oce.add(keys=OpCode.TARGET_GS_MODE, info=Info.TARGET_GS_MODE)
oce.add(keys=OpCode.INERTIAL_MODE, info=Info.INERTIAL_MODE)
oce.add(keys=OpCode.ANNOUNCE_MODES, info=Info.ANNOUNCE_MODES) oce.add(keys=OpCode.ANNOUNCE_MODES, info=Info.ANNOUNCE_MODES)
oce.add(keys=OpCode.REBOOT_I2C, info=Info.REBOOT_I2C) oce.add(keys=OpCode.REBOOT_I2C, info=Info.REBOOT_I2C)
defs.add_service( defs.add_service(