bump tmtccmd and fix CFDP code
Some checks are pending
EIVE/-/pipeline/head Build started...

This commit is contained in:
Robin Müller 2024-05-07 14:07:18 +02:00
parent 86a68e25f7
commit d6b879da67
5 changed files with 57 additions and 41 deletions

View File

@ -2,6 +2,7 @@ import logging
import struct import struct
from spacepackets.ecss import PusTm from spacepackets.ecss import PusTm
from tmtccmd.pus.s8_fsfw_action_defs import CustomSubservice
from eive_tmtc.config.object_ids import ( from eive_tmtc.config.object_ids import (
ACU_HANDLER_ID, ACU_HANDLER_ID,
PDU_1_HANDLER_ID, PDU_1_HANDLER_ID,
@ -22,59 +23,74 @@ from eive_tmtc.tmtc.payload.ploc_supervisor import SupvActionId
from eive_tmtc.tmtc.acs.star_tracker import handle_star_tracker_action_replies from eive_tmtc.tmtc.acs.star_tracker import handle_star_tracker_action_replies
from eive_tmtc.tmtc.acs.acs_ctrl import handle_acs_ctrl_action_replies from eive_tmtc.tmtc.acs.acs_ctrl import handle_acs_ctrl_action_replies
from eive_tmtc.tmtc.power.tm import handle_get_param_data_reply from eive_tmtc.tmtc.power.tm import handle_get_param_data_reply
from tmtccmd.pus.s8_fsfw_action import Service8FsfwTm
from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter
from spacepackets.ccsds.time import CdsShortTimestamp from spacepackets.ccsds.time import CdsShortTimestamp
from tmtccmd.util import ObjectIdDictT from tmtccmd.util import ObjectIdBase, ObjectIdDictT, ObjectIdU32
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
def handle_action_reply( _LOGGER = logging.getLogger(__name__)
raw_tm: bytes, printer: FsfwTmTcPrinter, obj_id_dict: ObjectIdDictT
def handle_action_service_tm(
raw_tm: bytes, pw: PrintWrapper, obj_id_dict: ObjectIdDictT
): ):
"""Core Action reply handler """Core Action reply handler
:return: :return:
""" """
tm_packet = PusTm.unpack(raw_tm, timestamp_len=CdsShortTimestamp.TIMESTAMP_SIZE) tm_packet = PusTm.unpack(raw_tm, timestamp_len=CdsShortTimestamp.TIMESTAMP_SIZE)
pw = PrintWrapper(printer.file_logger)
if len(tm_packet.source_data) < 8: if len(tm_packet.source_data) < 8:
_LOGGER.warn( _LOGGER.warning(
"invalid TM packet length for PUS 8 reply, must have at least 8 bytes" "received action service reply with source data smaller than 8 bytes"
) )
return return
object_id_raw = struct.unpack("!I", tm_packet.source_data[0:4])[0] object_id_raw = struct.unpack("!I", tm_packet.source_data[0:4])[0]
action_id = struct.unpack("!I", tm_packet.source_data[4:8])[0] action_id = struct.unpack("!I", tm_packet.source_data[4:8])[0]
object_id = obj_id_dict.get(object_id_raw)
custom_data = tm_packet.source_data[8:] custom_data = tm_packet.source_data[8:]
generic_print_str = printer.generic_action_packet_tm_print( if object_id is None:
packet=tm_packet, obj_id=object_id object_id = ObjectIdU32(object_id_raw, "Unknown ID")
) if tm_packet.subservice == CustomSubservice.TM_DATA_REPLY:
pw.dlog(generic_print_str) if object_id.as_bytes == IMTQ_HANDLER_ID:
if object_id.as_bytes == IMTQ_HANDLER_ID: return handle_imtq_replies(action_id, pw, custom_data)
return handle_imtq_replies(action_id, pw, custom_data) elif object_id.as_bytes == PLOC_MPSOC_ID:
elif object_id.as_bytes == PLOC_MPSOC_ID: return handle_mpsoc_data_reply(action_id, pw, custom_data)
return handle_mpsoc_data_reply(action_id, pw, custom_data) elif object_id.as_bytes == PLOC_SUPV_ID:
elif object_id.as_bytes == PLOC_SUPV_ID: return handle_supervisor_replies(action_id, pw, custom_data)
return handle_supervisor_replies(action_id, pw, custom_data) elif object_id.as_bytes == CORE_CONTROLLER_ID:
elif object_id.as_bytes == CORE_CONTROLLER_ID: return handle_core_ctrl_action_replies(action_id, pw, custom_data)
return handle_core_ctrl_action_replies(action_id, pw, custom_data) elif object_id.as_bytes == STAR_TRACKER_ID:
elif object_id.as_bytes == STAR_TRACKER_ID: return handle_star_tracker_action_replies(action_id, pw, custom_data)
return handle_star_tracker_action_replies(action_id, pw, custom_data) elif object_id.as_bytes == ACS_CONTROLLER:
elif object_id.as_bytes == ACS_CONTROLLER: return handle_acs_ctrl_action_replies(action_id, pw, custom_data)
return handle_acs_ctrl_action_replies(action_id, pw, custom_data) elif object_id.as_bytes in [
elif object_id.as_bytes in [ ACU_HANDLER_ID,
ACU_HANDLER_ID, PDU_1_HANDLER_ID,
PDU_1_HANDLER_ID, PDU_2_HANDLER_ID,
PDU_2_HANDLER_ID, P60_DOCK_HANDLER,
P60_DOCK_HANDLER, ]:
]: return handle_get_param_data_reply(object_id, action_id, pw, custom_data)
return handle_get_param_data_reply(object_id, action_id, pw, custom_data) else:
# TODO: Could add a handler here depending on action ID and object ID.
handle_action_data_reply(tm_packet, object_id, action_id, pw)
else: else:
pw.dlog(f"No dedicated action reply handler found for reply from {object_id}") pw.dlog(
pw.dlog(f"Raw Data: {tm_packet.custom_data.hex(sep=',')}") f"service 8 packet from {object_id} with action ID {action_id} "
f"and unknown subservice {tm_packet.subservice}"
)
def handle_imtq_replies(action_id: int, pw: PrintWrapper, custom_data: bytearray): def handle_action_data_reply(
tm_packet: PusTm, named_obj_id: ObjectIdBase, action_id: int, printer: PrintWrapper
):
print_string = (
f"service 8 data reply from {named_obj_id} with action ID {action_id} "
f"and data size {len(tm_packet.tm_data[8:])}"
)
printer.dlog(print_string)
def handle_imtq_replies(action_id: int, pw: PrintWrapper, custom_data: bytes):
if action_id == struct.unpack("!I", ImtqActionId.get_commanded_dipole)[0]: if action_id == struct.unpack("!I", ImtqActionId.get_commanded_dipole)[0]:
header_list = [ header_list = [
"Commanded X-Dipole", "Commanded X-Dipole",
@ -87,7 +103,7 @@ def handle_imtq_replies(action_id: int, pw: PrintWrapper, custom_data: bytearray
pw.dlog(f"{content_list}") pw.dlog(f"{content_list}")
def handle_supervisor_replies(action_id: int, pw: PrintWrapper, custom_data: bytearray): def handle_supervisor_replies(action_id: int, pw: PrintWrapper, custom_data: bytes):
if action_id == SupvActionId.DUMP_MRAM: if action_id == SupvActionId.DUMP_MRAM:
header_list = ["MRAM Dump"] header_list = ["MRAM Dump"]
content_list = [custom_data[: len(custom_data)]] content_list = [custom_data[: len(custom_data)]]

View File

@ -22,7 +22,7 @@ from eive_tmtc.config.definitions import TM_DB_PATH, PUS_APID
from eive_tmtc.config.object_ids import get_object_ids from eive_tmtc.config.object_ids import get_object_ids
from .action_reply_handler import handle_action_reply from .action_reply_handler import handle_action_service_tm
from .defs import PrintWrapper from .defs import PrintWrapper
from .event_handler import handle_event_packet from .event_handler import handle_event_packet
from .hk_handler import HkFilter, handle_hk_packet from .hk_handler import HkFilter, handle_hk_packet
@ -96,8 +96,8 @@ class PusHandler(SpecificApidHandlerBase):
elif service == 5: elif service == 5:
handle_event_packet(raw_tm=packet, pw=self.pw) handle_event_packet(raw_tm=packet, pw=self.pw)
elif service == 8: elif service == 8:
handle_action_reply( handle_action_service_tm(
raw_tm=packet, printer=self.printer, obj_id_dict=self.obj_id_dict raw_tm=packet, pw=self.pw, obj_id_dict=self.obj_id_dict
) )
elif service == 17: elif service == 17:
pus17_tm = Service17Tm.unpack( pus17_tm = Service17Tm.unpack(

View File

@ -686,7 +686,7 @@ class DirElement:
size: int size: int
def handle_mpsoc_data_reply(action_id: int, pw: PrintWrapper, custom_data: bytearray): def handle_mpsoc_data_reply(action_id: int, pw: PrintWrapper, custom_data: bytes):
if action_id == ActionId.TM_MEM_READ_RPT: if action_id == ActionId.TM_MEM_READ_RPT:
header_list = [ header_list = [
"PLOC Memory Address", "PLOC Memory Address",

View File

@ -537,7 +537,7 @@ def handle_acu_hk_data(pw: PrintWrapper, set_id: int, hk_data: bytes):
def handle_get_param_data_reply( def handle_get_param_data_reply(
obj_id: ObjectIdBase, action_id: int, pw: PrintWrapper, custom_data: bytearray obj_id: ObjectIdBase, action_id: int, pw: PrintWrapper, custom_data: bytes
): ):
if action_id == GomspaceDeviceActionId.PARAM_GET: if action_id == GomspaceDeviceActionId.PARAM_GET:
pw.dlog(f"Parameter Get Request received for object {obj_id}") pw.dlog(f"Parameter Get Request received for object {obj_id}")

View File

@ -29,7 +29,7 @@ classifiers = [
"Topic :: Scientific/Engineering" "Topic :: Scientific/Engineering"
] ]
dependencies = [ dependencies = [
"tmtccmd ~= 8.0.0rc1", "tmtccmd ~= 8.0",
"cfdp-py~=0.1.0", "cfdp-py~=0.1.0",
# "tmtccmd @ git+https://github.com/robamu-org/tmtccmd@prep_v8.0.0", # "tmtccmd @ git+https://github.com/robamu-org/tmtccmd@prep_v8.0.0",
"python-dateutil ~= 2.8", "python-dateutil ~= 2.8",