should be done
EIVE/-/pipeline/pr-main This commit looks good Details

This commit is contained in:
Marius Eggert 2023-10-19 16:52:08 +02:00
parent c12dfd06d8
commit ad530925d5
2 changed files with 68 additions and 21 deletions

View File

@ -16,7 +16,7 @@ from eive_tmtc.pus_tm.defs import PrintWrapper
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.acs.star_tracker import StarTrackerActionId
from eive_tmtc.tmtc.acs.star_tracker import handle_star_tracker_action_replies
from eive_tmtc.tmtc.power.tm import handle_get_param_data_reply
from tmtccmd.tm import Service8FsfwTm
from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter
@ -52,7 +52,7 @@ def handle_action_reply(
elif object_id.as_bytes == CORE_CONTROLLER_ID:
return handle_core_ctrl_action_replies(action_id, pw, custom_data)
elif object_id.as_bytes == STAR_TRACKER_ID:
return handle_startracker_replies(action_id, pw, custom_data)
return handle_star_tracker_action_replies(action_id, pw, custom_data)
elif object_id.as_bytes in [
ACU_HANDLER_ID,
PDU_1_HANDLER_ID,
@ -89,20 +89,3 @@ def handle_supervisor_replies(action_id: int, pw: PrintWrapper, custom_data: byt
content_list = [struct.unpack("!H", custom_data[:2])[0]]
pw.dlog(f"{header_list}")
pw.dlog(f"{content_list}")
def handle_startracker_replies(
action_id: int, pw: PrintWrapper, custom_data: bytearray
):
if action_id == StarTrackerActionId.CHECKSUM:
if len(custom_data) != 5:
_LOGGER.warning(
"Star tracker reply has invalid length {0}".format(len(custom_data))
)
return
header_list = ["Checksum", "Checksum valid"]
print(custom_data[4])
checksum_valid_flag = custom_data[4] >> 8
content_list = ["0x" + custom_data[:4].hex(), checksum_valid_flag]
pw.dlog(f"{header_list}")
pw.dlog(f"{content_list}")

View File

@ -23,6 +23,7 @@ from tmtccmd.tc.pus_3_fsfw_hk import (
disable_periodic_hk_command,
make_sid,
)
from tmtccmd.pus.s8_fsfw_funccmd import create_action_cmd
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_200_fsfw_mode import pack_mode_data, Mode
from tmtccmd.util import ObjectIdU32
@ -96,6 +97,9 @@ class StarTrackerActionId(enum.IntEnum):
DEBUG_CAMERA = 83
FIRMWARE_UPDATE = 84
SET_TIME_FROM_SYS_TIME = 87
ADD_SECONDARY_TM_TO_NORMAL_MODE = 95
RESET_SECONDARY_TM_SET = 96
READ_SECONDARY_TM_SET = 97
class OpCodes:
@ -107,7 +111,9 @@ class OpCodes:
ONE_SHOOT_HK = "one_shoot_hk"
ENABLE_HK = "enable_hk"
DISABLE_HK = "disable_hk"
REQUEST_SOLUTION_SET_ACTION = "action_req_sol"
ADD_SECONDARY_TM_TO_NORMAL_MODE = "add_secondary_tm"
RESET_SECONDARY_TM_SET = "reset_secondary_tm"
READ_SECONDARY_TM_SET = "read_secondary_tm"
TAKE_IMAGE = "take_image"
UPLOAD_IMAGE = "upload_image"
SET_IMG_PROCESSOR_MODE = "set_img_proc_mode"
@ -119,7 +125,9 @@ class Info:
ONE_SHOOT_HK = "One shoot HK Set"
ENABLE_HK = "Enable Periodic HK"
DISABLE_HK = "Disable Periodic HK"
REQUEST_SOLUTION_SET_ACTION = "Request Solution Set Action"
ADD_SECONDARY_TM_TO_NORMAL_MODE = "Add specific Dataset to secondary TM"
RESET_SECONDARY_TM_SET = "Reset secondary TM to Temperature Set only"
READ_SECONDARY_TM_SET = "Read list of secondary TM Sets"
UPLOAD_IMAGE = "Upload Image"
TAKE_IMAGE = "Take Image"
SET_IMG_PROCESSOR_MODE = "Set Image Processor Mode"
@ -641,6 +649,26 @@ def pack_star_tracker_commands( # noqa C901
+ firmware.encode()
)
q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
if op_code == OpCodes.ADD_SECONDARY_TM_TO_NORMAL_MODE:
q.add_log_cmd(f"{Info.ADD_SECONDARY_TM_TO_NORMAL_MODE}")
for val in SetId:
print("{:<2}: {:<20}".format(val, val.name))
setId = int(input("Specify the dataset \n" ""))
q.add_pus_tc(
create_action_cmd(
STAR_TRACKER_ID,
StarTrackerActionId.ADD_SECONDARY_TM_TO_NORMAL_MODE,
setId.to_bytes(4, byteorder="big"),
)
)
if op_code == OpCodes.RESET_SECONDARY_TM_SET:
q.add_log_cmd(f"{Info.RESET_SECONDARY_TM_SET}")
q.add_pus_tc(
create_action_cmd(
STAR_TRACKER_ID,
StarTrackerActionId.RESET_SECONDARY_TM_SET,
)
)
def request_dataset(q: DefaultPusQueueHelper, type: DataSetRequest):
@ -1018,6 +1046,42 @@ def handle_auto_blob_set(hk_data: bytes, pw: PrintWrapper):
FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=3)
def handle_star_tracker_action_replies(
action_id: int, pw: PrintWrapper, custom_data: bytes
):
if action_id == StarTrackerActionId.CHECKSUM:
handle_checksum(pw, custom_data)
elif action_id == StarTrackerActionId.READ_SECONDARY_TM_SET:
handle_read_secondary_tm_set(pw, custom_data)
def handle_checksum(pw: PrintWrapper, custom_data: bytes):
if len(custom_data) != 5:
_LOGGER.warning(
"Star tracker reply has invalid length {0}".format(len(custom_data))
)
return
header_list = ["Checksum", "Checksum valid"]
print(custom_data[4])
checksum_valid_flag = custom_data[4] >> 8
content_list = ["0x" + custom_data[:4].hex(), checksum_valid_flag]
pw.dlog(f"{header_list}")
pw.dlog(f"{content_list}")
def handle_read_secondary_tm_set(pw: PrintWrapper, custom_data: bytes):
pw.dlog("Received secondary TM Sets")
data_length = len(custom_data)
if data_length < 1: # lul
raise ValueError(f"Received data shorter than expected")
fmt_str = "!B" * data_length
inc_len = struct.calcsize(fmt_str)
setIds = struct.unpack(fmt_str, custom_data[:inc_len])
pw.dlog("The following Datasets are currently Part of the secondary TM list")
for setId in setIds:
pw.dlog(SetId[setId].name)
@tmtc_definitions_provider
def add_str_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry()