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

works
All checks were successful
EIVE/-/pipeline/head This commit looks good
Details

Browse Source
This commit is contained in:
Robin Müller 2023-06-22 14:52:28 +02:00
parent b28d938468
commit 8239e610cc
No known key found for this signature in database
GPG Key ID: 11D4952C8CCEF814
2 changed files with 19 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
eive_tmtc/pus_tm/hk_handling.py
View File

@ -44,6 +44,8 @@ _LOGGER = logging.getLogger(__name__)
FORWARD_SENSOR_TEMPS = False FORWARD_SENSOR_TEMPS = False
# TODO: Transform this into a CLI argument
HK_OUTPUT_LEVEL = 1
def handle_hk_packet( def handle_hk_packet(
@ -70,6 +72,7 @@ def handle_hk_packet(
hk_data=hk_data, hk_data=hk_data,
) )
try: try:
if HK_OUTPUT_LEVEL > 0:
handle_regular_hk_print( handle_regular_hk_print(
printer=printer, printer=printer,
object_id=named_obj_id, object_id=named_obj_id,

18
eive_tmtc/tmtc/core.py
View File

@ -157,7 +157,9 @@ def add_core_controller_definitions(defs: TmtcDefinitionWrapper):
oce.add(keys=OpCode.XSC_REBOOT_1_0, info="Reboot 1 0") oce.add(keys=OpCode.XSC_REBOOT_1_0, info="Reboot 1 0")
oce.add(keys=OpCode.XSC_REBOOT_1_1, info="Reboot 1 1") oce.add(keys=OpCode.XSC_REBOOT_1_1, info="Reboot 1 1")
oce.add(keys=OpCode.SET_PREF_SD, info=Info.SET_PREF_SD) oce.add(keys=OpCode.SET_PREF_SD, info=Info.SET_PREF_SD)
oce.add(keys=OpCode.READ_REBOOT_MECHANISM_INFO, info=Info.READ_REBOOT_MECHANISM_INFO) oce.add(
keys=OpCode.READ_REBOOT_MECHANISM_INFO, info=Info.READ_REBOOT_MECHANISM_INFO
)
oce.add(keys=OpCode.OBSW_UPDATE_FROM_TMP, info=Info.OBSW_UPDATE_FROM_TMP) oce.add(keys=OpCode.OBSW_UPDATE_FROM_TMP, info=Info.OBSW_UPDATE_FROM_TMP)
oce.add(keys=OpCode.OBSW_UPDATE_FROM_SD_0, info=Info.OBSW_UPDATE_FROM_SD_0) oce.add(keys=OpCode.OBSW_UPDATE_FROM_SD_0, info=Info.OBSW_UPDATE_FROM_SD_0)
oce.add(keys=OpCode.OBSW_UPDATE_FROM_SD_1, info=Info.OBSW_UPDATE_FROM_SD_1) oce.add(keys=OpCode.OBSW_UPDATE_FROM_SD_1, info=Info.OBSW_UPDATE_FROM_SD_1)
@ -303,7 +305,7 @@ def pack_core_commands( # noqa C901
q.add_pus_tc( q.add_pus_tc(
create_action_cmd( create_action_cmd(
object_id=CORE_CONTROLLER_ID, object_id=CORE_CONTROLLER_ID,
action_id=ActionId.READ_REBOOT_MECHANISM_INFO action_id=ActionId.READ_REBOOT_MECHANISM_INFO,
) )
) )
elif op_code == OpCode.DISABLE_REBOOT_FILE_HANDLING: elif op_code == OpCode.DISABLE_REBOOT_FILE_HANDLING:
@ -622,7 +624,7 @@ def handle_core_ctrl_action_replies(
handle_list_dir_dump_reply(pw, custom_data) handle_list_dir_dump_reply(pw, custom_data)
def handle_reboot_mechanism_info_reply(pw: PrintWrapper, custom_data : bytes): def handle_reboot_mechanism_info_reply(pw: PrintWrapper, custom_data: bytes):
pw.dlog("Received reboot mechansm information") pw.dlog("Received reboot mechansm information")
fmt_str = "!BIIIIIBBBBBBBB" fmt_str = "!BIIIIIBBBBBBBB"
inc_len = struct.calcsize(fmt_str) inc_len = struct.calcsize(fmt_str)
@ -642,8 +644,8 @@ def handle_reboot_mechanism_info_reply(pw: PrintWrapper, custom_data : bytes):
last_chip, last_chip,
last_copy, last_copy,
next_chip, next_chip,
next_copy next_copy,
) = struct.unpack(fmt_str, custom_data[: inc_len]) ) = struct.unpack(fmt_str, custom_data[:inc_len])
pw.dlog(f"Enabled: {enabled}") pw.dlog(f"Enabled: {enabled}")
pw.dlog(f"Max Count: {max_count}") pw.dlog(f"Max Count: {max_count}")
pw.dlog(f"Count 00: {img00_count}") pw.dlog(f"Count 00: {img00_count}")
@ -660,7 +662,7 @@ def handle_reboot_mechanism_info_reply(pw: PrintWrapper, custom_data : bytes):
pw.dlog(f"Next Copy: {next_copy}") pw.dlog(f"Next Copy: {next_copy}")
def handle_list_dir_dump_reply(pw: PrintWrapper, custom_data : bytes): def handle_list_dir_dump_reply(pw: PrintWrapper, custom_data: bytes):
if len(custom_data) < 4: if len(custom_data) < 4:
_LOGGER.warning("Data unexpectedly small") _LOGGER.warning("Data unexpectedly small")
return return
@ -690,9 +692,7 @@ def handle_list_dir_dump_reply(pw: PrintWrapper, custom_data : bytes):
else: else:
path = Path("dir_listing.txt") path = Path("dir_listing.txt")
remove_if_exists_and_new(seq_idx, path) remove_if_exists_and_new(seq_idx, path)
pw.dlog( pw.dlog(f"Compression option: {compressed}. Dumping file into dir_listing.txt")
f"Compression option: {compressed}. Dumping file into dir_listing.txt"
)
with open(path, "a") as listing_file: with open(path, "a") as listing_file:
listing_file_str = custom_data[file_data_offset:].decode() listing_file_str = custom_data[file_data_offset:].decode()
listing_file.write(listing_file_str) listing_file.write(listing_file_str)