Merge pull request 'Bump tmtccmd and fix CFDP' (#297) from bump-tmtccmd-fix-cfdp into main
All checks were successful
EIVE/-/pipeline/head This commit looks good

Reviewed-on: #297
Reviewed-by: Marius Eggert <eggertm@irs.uni-stuttgart.de>
This commit is contained in:
Robin Müller 2024-05-07 14:17:40 +02:00
commit 5cf76c07e9
13 changed files with 136 additions and 108 deletions

View File

@ -62,13 +62,7 @@ class CfdpHandler:
) )
def put_request(self, request: PutRequest): def put_request(self, request: PutRequest):
if not self.remote_cfg_table.get_cfg(request.destination_id): self.source_handler.put_request(request)
raise ValueError(
f"No remote CFDP config found for entity ID {request.destination_id}"
)
self.source_handler.put_request(
request, self.remote_cfg_table.get_cfg(request.destination_id) # type: ignore
)
def pull_next_source_packet(self) -> Optional[PduHolder]: def pull_next_source_packet(self) -> Optional[PduHolder]:
res = self.source_handler.state_machine() res = self.source_handler.state_machine()

View File

@ -4,7 +4,7 @@
import logging import logging
from typing import List, cast from typing import List, cast
from tmtccmd import DefaultProcedureInfo from tmtccmd import TreeCommandingProcedure
from tmtccmd.tmtc import DefaultPusQueueHelper from tmtccmd.tmtc import DefaultPusQueueHelper
from tmtccmd.util import ObjectIdU32 from tmtccmd.util import ObjectIdU32
@ -78,7 +78,7 @@ from eive_tmtc.utility.input_helper import InputHelper
def handle_pus_procedure( def handle_pus_procedure(
info: DefaultProcedureInfo, info: TreeCommandingProcedure,
queue_helper: DefaultPusQueueHelper, queue_helper: DefaultPusQueueHelper,
): ):
cmd_path = info.cmd_path cmd_path = info.cmd_path

View File

@ -68,8 +68,8 @@ class TcHandler(TcHandlerBase):
def feed_cb(self, info: ProcedureWrapper, wrapper: FeedWrapper): def feed_cb(self, info: ProcedureWrapper, wrapper: FeedWrapper):
self.queue_helper.queue_wrapper = wrapper.queue_wrapper self.queue_helper.queue_wrapper = wrapper.queue_wrapper
if info.proc_type == TcProcedureType.DEFAULT: if info.proc_type == TcProcedureType.TREE_COMMANDING:
handle_pus_procedure(info.to_def_procedure(), self.queue_helper) handle_pus_procedure(info.to_tree_commanding_procedure(), self.queue_helper)
elif info.proc_type == TcProcedureType.CFDP: elif info.proc_type == TcProcedureType.CFDP:
self.handle_cfdp_procedure(info) self.handle_cfdp_procedure(info)
@ -155,7 +155,7 @@ class TcHandler(TcHandlerBase):
def queue_finished_cb(self, info: ProcedureWrapper): def queue_finished_cb(self, info: ProcedureWrapper):
if info is not None: if info is not None:
if info.proc_type == TcQueueEntryType.PUS_TC: if info.proc_type == TcQueueEntryType.PUS_TC:
def_proc = info.to_def_procedure() def_proc = info.to_tree_commanding_procedure()
_LOGGER.info(f"Finished queue for command {def_proc.cmd_path}") _LOGGER.info(f"Finished queue for command {def_proc.cmd_path}")
elif info.proc_type == TcProcedureType.CFDP: elif info.proc_type == TcProcedureType.CFDP:
_LOGGER.info("Finished CFDP queue") _LOGGER.info("Finished CFDP queue")

View File

@ -1,5 +1,8 @@
import logging import logging
import struct import struct
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,
@ -20,56 +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 = Service8FsfwTm.unpack( tm_packet = PusTm.unpack(raw_tm, timestamp_len=CdsShortTimestamp.TIMESTAMP_SIZE)
raw_telemetry=raw_tm, time_reader=CdsShortTimestamp.empty() if len(tm_packet.source_data) < 8:
) _LOGGER.warning(
object_id = obj_id_dict.get(tm_packet.source_object_id_as_bytes) "received action service reply with source data smaller than 8 bytes"
pw = PrintWrapper(printer.file_logger) )
custom_data = tm_packet.custom_data return
action_id = tm_packet.action_id object_id_raw = struct.unpack("!I", tm_packet.source_data[0:4])[0]
generic_print_str = printer.generic_action_packet_tm_print( action_id = struct.unpack("!I", tm_packet.source_data[4:8])[0]
packet=tm_packet, obj_id=object_id object_id = obj_id_dict.get(object_id_raw)
) custom_data = tm_packet.source_data[8:]
pw.dlog(generic_print_str) if object_id is None:
if object_id.as_bytes == IMTQ_HANDLER_ID: object_id = ObjectIdU32(object_id_raw, "Unknown ID")
return handle_imtq_replies(action_id, pw, custom_data) if tm_packet.subservice == CustomSubservice.TM_DATA_REPLY:
elif object_id.as_bytes == PLOC_MPSOC_ID: if object_id.as_bytes == IMTQ_HANDLER_ID:
return handle_mpsoc_data_reply(action_id, pw, custom_data) return handle_imtq_replies(action_id, pw, custom_data)
elif object_id.as_bytes == PLOC_SUPV_ID: elif object_id.as_bytes == PLOC_MPSOC_ID:
return handle_supervisor_replies(action_id, pw, custom_data) return handle_mpsoc_data_reply(action_id, pw, custom_data)
elif object_id.as_bytes == CORE_CONTROLLER_ID: elif object_id.as_bytes == PLOC_SUPV_ID:
return handle_core_ctrl_action_replies(action_id, pw, custom_data) return handle_supervisor_replies(action_id, pw, custom_data)
elif object_id.as_bytes == STAR_TRACKER_ID: elif object_id.as_bytes == CORE_CONTROLLER_ID:
return handle_star_tracker_action_replies(action_id, pw, custom_data) return handle_core_ctrl_action_replies(action_id, pw, custom_data)
elif object_id.as_bytes == ACS_CONTROLLER: elif object_id.as_bytes == STAR_TRACKER_ID:
return handle_acs_ctrl_action_replies(action_id, pw, custom_data) return handle_star_tracker_action_replies(action_id, pw, custom_data)
elif object_id.as_bytes in [ elif object_id.as_bytes == ACS_CONTROLLER:
ACU_HANDLER_ID, return handle_acs_ctrl_action_replies(action_id, pw, custom_data)
PDU_1_HANDLER_ID, elif object_id.as_bytes in [
PDU_2_HANDLER_ID, ACU_HANDLER_ID,
P60_DOCK_HANDLER, PDU_1_HANDLER_ID,
]: PDU_2_HANDLER_ID,
return handle_get_param_data_reply(object_id, action_id, pw, custom_data) P60_DOCK_HANDLER,
]:
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",
@ -82,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

@ -25,7 +25,7 @@ def handle_event_packet( # noqa C901: Complexity okay here
): # noqa C901: Complexity okay here ): # noqa C901: Complexity okay here
if PRINT_RAW_EVENTS_B64_STR: if PRINT_RAW_EVENTS_B64_STR:
print(f"PUS Event TM Base64: {base64.b64encode(raw_tm)}") print(f"PUS Event TM Base64: {base64.b64encode(raw_tm)}")
tm = Service5Tm.unpack(data=raw_tm, time_reader=CdsShortTimestamp.empty()) tm = Service5Tm.unpack(data=raw_tm, timestamp_len=CdsShortTimestamp.TIMESTAMP_SIZE)
event_dict = get_event_dict() event_dict = get_event_dict()
event_def = tm.event_definition event_def = tm.event_definition
info = event_dict.get(event_def.event_id) info = event_dict.get(event_def.event_id)
@ -40,10 +40,10 @@ def handle_event_packet( # noqa C901: Complexity okay here
obj_name = event_def.reporter_id.hex(sep=",") obj_name = event_def.reporter_id.hex(sep=",")
else: else:
obj_name = obj_id_obj.name obj_name = obj_id_obj.name
assert tm.time_provider is not None timestamp = CdsShortTimestamp.unpack(tm.timestamp)
generic_event_string = ( generic_event_string = (
f"Object {obj_name} generated Event {info.name} (ID: {event_def.event_id:#04x})" f"Object {obj_name} generated Event {info.name} (ID: {event_def.event_id:#04x})"
f" at {tm.time_provider.as_date_time()}" f" at {timestamp.as_date_time()}"
) )
_LOGGER.info(generic_event_string) _LOGGER.info(generic_event_string)
pw.file_logger.info( pw.file_logger.info(

View File

@ -2,20 +2,17 @@ import uuid
import dataclasses import dataclasses
import datetime import datetime
import sqlite3 import sqlite3
from tmtccmd.pus.tm.s3_fsfw_hk import Service3FsfwTm from spacepackets.ecss.tm import CdsShortTimestamp, PusTm
@dataclasses.dataclass @dataclasses.dataclass
class HkTmInfo: class HkTmInfo:
packet_uuid: uuid.UUID packet_uuid: uuid.UUID
hk_packet: Service3FsfwTm hk_packet: PusTm
set_id: int
db_con: sqlite3.Connection db_con: sqlite3.Connection
hk_data: bytes hk_data: bytes
@property @property
def packet_datetime(self) -> datetime.datetime: def packet_datetime(self) -> datetime.datetime:
return self.hk_packet.pus_tm.time_provider.as_datetime() return CdsShortTimestamp.unpack(self.hk_packet.timestamp).as_datetime()
@property
def set_id(self) -> int:
return self.hk_packet.set_id

View File

@ -2,10 +2,14 @@
import dataclasses import dataclasses
import logging import logging
import base64 # noqa import base64
import sqlite3 import sqlite3
import struct
from typing import List, cast from typing import List, cast
from uuid import UUID from uuid import UUID
from spacepackets.ccsds.time import CdsShortTimestamp
from spacepackets.ecss import PusTm
from eive_tmtc.config.definitions import PRINT_RAW_HK_B64_STR from eive_tmtc.config.definitions import PRINT_RAW_HK_B64_STR
from eive_tmtc.pus_tm.hk import HkTmInfo from eive_tmtc.pus_tm.hk import HkTmInfo
@ -23,9 +27,6 @@ from eive_tmtc.tmtc.power.pwr_ctrl import handle_pwr_ctrl_hk_data
from eive_tmtc.tmtc.com.syrlinks_handler import handle_syrlinks_hk_data from eive_tmtc.tmtc.com.syrlinks_handler import handle_syrlinks_hk_data
from eive_tmtc.tmtc.tcs import handle_thermal_controller_hk_data from eive_tmtc.tmtc.tcs import handle_thermal_controller_hk_data
from eive_tmtc.tmtc.tcs.tmp1075 import handle_tmp_1075_hk_data from eive_tmtc.tmtc.tcs.tmp1075 import handle_tmp_1075_hk_data
from tmtccmd.pus.tm.s3_fsfw_hk import (
Service3FsfwTm,
)
from tmtccmd.pus.tm.s3_hk_base import HkContentType from tmtccmd.pus.tm.s3_hk_base import HkContentType
from tmtccmd.util.obj_id import ObjectIdU32, ObjectIdDictT from tmtccmd.util.obj_id import ObjectIdU32, ObjectIdDictT
@ -68,17 +69,20 @@ def handle_hk_packet(
hk_level: int, hk_level: int,
db_con: sqlite3.Connection, db_con: sqlite3.Connection,
): ):
tm_packet = Service3FsfwTm.unpack(raw_telemetry=raw_tm, custom_hk_handling=False) tm_packet = PusTm.unpack(raw_tm, CdsShortTimestamp.TIMESTAMP_SIZE)
named_obj_id = cast(ObjectIdU32, obj_id_dict.get(tm_packet.object_id.as_bytes)) obj_id_raw = struct.unpack("!I", tm_packet.tm_data[0:4])[0]
named_obj_id = cast(ObjectIdU32, obj_id_dict.get(obj_id_raw))
if named_obj_id is None: if named_obj_id is None:
named_obj_id = tm_packet.object_id named_obj_id = ObjectIdU32(obj_id_raw, "Unknown ID")
if tm_packet.subservice == 25 or tm_packet.subservice == 26: if tm_packet.subservice == 25 or tm_packet.subservice == 26:
set_id = struct.unpack("!I", tm_packet.tm_data[4:8])[0]
hk_data = tm_packet.tm_data[8:] hk_data = tm_packet.tm_data[8:]
if named_obj_id.as_bytes in hk_filter.object_ids: if named_obj_id.as_bytes in hk_filter.object_ids:
if PRINT_RAW_HK_B64_STR: if PRINT_RAW_HK_B64_STR:
print(f"PUS TM Base64: {base64.b64encode(raw_tm)}") print(f"PUS TM Base64: {base64.b64encode(raw_tm)}")
handle_regular_hk_print( handle_regular_hk_print(
printer=printer, printer=printer,
set_id=set_id,
packet_uuid=packet_uuid, packet_uuid=packet_uuid,
object_id=named_obj_id, object_id=named_obj_id,
hk_packet=tm_packet, hk_packet=tm_packet,
@ -89,11 +93,12 @@ def handle_hk_packet(
try: try:
if hk_level >= 1: if hk_level >= 1:
printer.generic_hk_tm_print( printer.generic_hk_tm_print(
HkContentType.HK, named_obj_id, tm_packet.set_id, hk_data HkContentType.HK, named_obj_id, set_id, hk_data
) )
if hk_level >= 1: if hk_level >= 1:
handle_regular_hk_print( handle_regular_hk_print(
printer=printer, printer=printer,
set_id=set_id,
packet_uuid=packet_uuid, packet_uuid=packet_uuid,
object_id=named_obj_id, object_id=named_obj_id,
hk_packet=tm_packet, hk_packet=tm_packet,
@ -109,7 +114,8 @@ def handle_hk_packet(
def handle_regular_hk_print( # noqa C901: Complexity okay here def handle_regular_hk_print( # noqa C901: Complexity okay here
hk_packet: Service3FsfwTm, hk_packet: PusTm,
set_id: int,
packet_uuid: UUID, packet_uuid: UUID,
hk_data: bytes, hk_data: bytes,
db: sqlite3.Connection, db: sqlite3.Connection,
@ -117,12 +123,15 @@ def handle_regular_hk_print( # noqa C901: Complexity okay here
printer: FsfwTmTcPrinter, printer: FsfwTmTcPrinter,
): ):
objb = object_id.as_bytes objb = object_id.as_bytes
set_id = hk_packet.set_id
hk_info = HkTmInfo( hk_info = HkTmInfo(
packet_uuid=packet_uuid, hk_packet=hk_packet, db_con=db, hk_data=hk_data packet_uuid=packet_uuid,
hk_packet=hk_packet,
db_con=db,
hk_data=hk_data,
set_id=set_id,
) )
assert hk_packet.pus_tm.time_provider is not None timestamp = CdsShortTimestamp.unpack(hk_packet.timestamp)
packet_dt = hk_packet.pus_tm.time_provider.as_date_time() packet_dt = timestamp.as_datetime()
pw = PrintWrapper(printer.file_logger) pw = PrintWrapper(printer.file_logger)
"""This function is called when a Service 3 Housekeeping packet is received.""" """This function is called when a Service 3 Housekeeping packet is received."""
if objb in [obj_ids.RW1_ID, obj_ids.RW2_ID, obj_ids.RW3_ID, obj_ids.RW4_ID]: if objb in [obj_ids.RW1_ID, obj_ids.RW2_ID, obj_ids.RW3_ID, obj_ids.RW4_ID]:
@ -139,7 +148,7 @@ def handle_regular_hk_print( # noqa C901: Complexity okay here
pw=pw, pw=pw,
set_id=set_id, set_id=set_id,
hk_data=hk_data, hk_data=hk_data,
packet_time=packet_dt, packet_time=timestamp.as_datetime(),
) )
elif objb == obj_ids.PCDU_HANDLER_ID: elif objb == obj_ids.PCDU_HANDLER_ID:
return handle_pcdu_hk(pw=pw, set_id=set_id, hk_data=hk_data) return handle_pcdu_hk(pw=pw, set_id=set_id, hk_data=hk_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
@ -63,15 +63,22 @@ class PusHandler(SpecificApidHandlerBase):
_LOGGER.warning("Detected packet shorter than 8 bytes!") _LOGGER.warning("Detected packet shorter than 8 bytes!")
return return
try: try:
tm_packet = PusTelemetry.unpack(packet, CdsShortTimestamp.empty()) tm_packet = PusTelemetry.unpack(packet, CdsShortTimestamp.TIMESTAMP_SIZE)
# _LOGGER.info(f"Sequence count: {tm_packet.seq_count}") # _LOGGER.info(f"Sequence count: {tm_packet.seq_count}")
except ValueError as value_error: except ValueError as value_error:
_LOGGER.warning(f"{value_error}") _LOGGER.warning(f"{value_error}")
_LOGGER.warning("Could not generate PUS TM object from raw data") _LOGGER.warning("Could not generate PUS TM object from raw data")
_LOGGER.warning(f"Raw Packet: [{packet.hex(sep=',')}], REPR: {packet!r}") _LOGGER.warning(f"Raw Packet: [{packet.hex(sep=',')}], REPR: {packet!r}")
return return
timestamp = CdsShortTimestamp.unpack(tm_packet.timestamp)
db_con = sqlite3.connect(TM_DB_PATH) db_con = sqlite3.connect(TM_DB_PATH)
self._store_packet_in_db(db_con, packet, tm_packet, packet_uuid) self._store_packet_in_db(
db_con=db_con,
packet=packet,
tm_packet=tm_packet,
timestamp=timestamp,
packet_uuid=packet_uuid,
)
service = tm_packet.service service = tm_packet.service
dedicated_handler = True dedicated_handler = True
if service == 1: if service == 1:
@ -89,12 +96,12 @@ 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(
data=packet, time_reader=CdsShortTimestamp.empty() data=packet, timestamp_len=CdsShortTimestamp.TIMESTAMP_SIZE
) )
if pus17_tm.subservice == 2: if pus17_tm.subservice == 2:
self.verif_wrapper.dlog("Received Ping Reply TM[17,2]") self.verif_wrapper.dlog("Received Ping Reply TM[17,2]")
@ -119,11 +126,11 @@ class PusHandler(SpecificApidHandlerBase):
self, self,
db_con: sqlite3.Connection, db_con: sqlite3.Connection,
packet: bytes, packet: bytes,
timestamp: CdsShortTimestamp,
tm_packet: PusTelemetry, tm_packet: PusTelemetry,
packet_uuid: uuid.UUID, packet_uuid: uuid.UUID,
): ):
cursor = db_con.cursor() cursor = db_con.cursor()
assert tm_packet.time_provider is not None
cursor.execute( cursor.execute(
""" """
CREATE TABLE IF NOT EXISTS pus_tm( CREATE TABLE IF NOT EXISTS pus_tm(
@ -139,7 +146,7 @@ class PusHandler(SpecificApidHandlerBase):
"INSERT INTO pus_tm VALUES(?, ?, ?, ?, ?, ?)", "INSERT INTO pus_tm VALUES(?, ?, ?, ?, ?, ?)",
( (
str(packet_uuid), str(packet_uuid),
tm_packet.time_provider.as_datetime(), timestamp.as_datetime(),
tm_packet.service, tm_packet.service,
tm_packet.subservice, tm_packet.subservice,
len(packet), len(packet),
@ -150,7 +157,7 @@ class PusHandler(SpecificApidHandlerBase):
def _handle_param_packet(self, raw_data: bytes, tm_packet: PusTelemetry): def _handle_param_packet(self, raw_data: bytes, tm_packet: PusTelemetry):
param_packet = Service20FsfwTm.unpack( param_packet = Service20FsfwTm.unpack(
raw_telemetry=raw_data, time_reader=CdsShortTimestamp.empty() raw_telemetry=raw_data, timestamp_len=CdsShortTimestamp.TIMESTAMP_SIZE
) )
if tm_packet.subservice == ParamSubservice.TM_DUMP_REPLY: if tm_packet.subservice == ParamSubservice.TM_DUMP_REPLY:
param_wrapper = Service20ParamDumpWrapper(param_tm=param_packet) param_wrapper = Service20ParamDumpWrapper(param_tm=param_packet)

View File

@ -17,7 +17,7 @@ def handle_service_1_fsfw_packet(wrapper: VerificationWrapper, raw_tm: bytes):
) )
# Error code with length 2 is FSFW specific # Error code with length 2 is FSFW specific
tm_packet = Service1Tm.unpack( tm_packet = Service1Tm.unpack(
data=raw_tm, params=UnpackParams(CdsShortTimestamp.empty(), 1, 2) data=raw_tm, params=UnpackParams(CdsShortTimestamp.TIMESTAMP_SIZE, 1, 2)
) )
fsfw_wrapper = Service1FsfwWrapper(tm_packet) fsfw_wrapper = Service1FsfwWrapper(tm_packet)
res = wrapper.verificator.add_tm(tm_packet) res = wrapper.verificator.add_tm(tm_packet)

View File

@ -717,7 +717,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

@ -6,32 +6,32 @@ build-backend = "setuptools.build_meta"
name = "eive-tmtc" name = "eive-tmtc"
description = "TMTC Commander EIVE" description = "TMTC Commander EIVE"
readme = "README.md" readme = "README.md"
version = "7.0.0" version = "7.1.0"
requires-python = ">=3.10" requires-python = ">=3.10"
license = {text = "Apache-2.0"} license = {text = "Apache-2.0"}
authors = [ authors = [
{name = "Robin Mueller", email = "muellerr@irs.uni-stuttgart.de"}, {name = "Robin Mueller", email = "muellerr@irs.uni-stuttgart.de"},
{name = "Jakob Meier", email = "meierj@irs.uni-stuttgart.de"}, {name = "Jakob Meier", email = "meierj@irs.uni-stuttgart.de"},
] ]
keywords = ["eive", "space", "communication", "commanding"] keywords = ["eive", "space", "communication", "commanding"]
classifiers = [ classifiers = [
"Development Status :: 5 - Production/Stable", "Development Status :: 5 - Production/Stable",
"License :: OSI Approved :: Apache Software License", "License :: OSI Approved :: Apache Software License",
"Natural Language :: English", "Natural Language :: English",
"Operating System :: POSIX", "Operating System :: POSIX",
"Operating System :: Microsoft :: Windows", "Operating System :: Microsoft :: Windows",
"Programming Language :: Python :: 3", "Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.9",
"Topic :: Communications", "Topic :: Communications",
"Topic :: Software Development :: Libraries", "Topic :: Software Development :: Libraries",
"Topic :: Software Development :: Libraries :: Python Modules", "Topic :: Software Development :: Libraries :: Python Modules",
"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@main", # "tmtccmd @ git+https://github.com/robamu-org/tmtccmd@prep_v8.0.0",
"python-dateutil ~= 2.8", "python-dateutil ~= 2.8",
] ]
@ -45,7 +45,7 @@ include-package-data = true
[tool.setuptools.packages] [tool.setuptools.packages]
find = {} find = {}
[tool.ruff] [tool.ruff.lint]
ignore = ["E501"] ignore = ["E501"]
[tool.ruff.extend-per-file-ignores] [tool.ruff.lint.extend-per-file-ignores]
"__init__.py" = ["F401"] "__init__.py" = ["F401"]

View File

@ -97,8 +97,8 @@ def setup_params() -> Tuple[SetupWrapper, int]:
hk_level = int(post_arg_parsing_wrapper.args_raw.hk) hk_level = int(post_arg_parsing_wrapper.args_raw.hk)
else: else:
hk_level = 0 hk_level = 0
if params.app_params.print_tree: if params.cmd_params.print_tree:
perform_tree_printout(params.app_params, hook_obj.get_command_definitions()) perform_tree_printout(params.cmd_params, hook_obj.get_command_definitions())
sys.exit(0) sys.exit(0)
params.apid = PUS_APID params.apid = PUS_APID
if params.com_if is None: if params.com_if is None: