update common_tmtc

This commit is contained in:
Robin Müller 2022-05-18 23:40:13 +02:00
parent 59a80fed91
commit d308899920
No known key found for this signature in database
GPG Key ID: 11D4952C8CCEF814
21 changed files with 433 additions and 218 deletions

View File

@ -0,0 +1,6 @@
SW_NAME = "fsfw-tmtc"
SW_VERSION = 1
SW_SUBVERSION = 4
SW_SUBSUBVERSION = 0
__version__ = "1.4.0"

View File

@ -1 +1,2 @@
PUS_APID = 0xEF PUS_APID = 0xEF
TM_SP_IDS = ((0x08 << 8) | PUS_APID,)

View File

@ -1,76 +1,34 @@
import argparse from typing import Dict, Optional
from typing import Dict, Tuple, Optional
from spacepackets.ecss.conf import PusVersion
from tmtccmd.com_if.com_interface_base import CommunicationInterface from tmtccmd.com_if.com_interface_base import CommunicationInterface
from tmtccmd.config.definitions import ServiceOpCodeDictT from tmtccmd.config.definitions import ServiceOpCodeDictT
from tmtccmd.config.hook import TmTcHookBase from tmtccmd.config.hook import TmTcHookBase
from tmtccmd.core.backend import TmTcHandler from tmtccmd.core.backend import TmTcHandler
from tmtccmd.tc.definitions import TcQueueT from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.tm.service_3_base import Service3Base
from tmtccmd.utility.tmtc_printer import TmTcPrinter
from common_tmtc.config.definitions import PUS_APID from tmtccmd.utility.retval import RetvalDictT
from common_tmtc.config.definitions import TM_SP_IDS
from common_tmtc.pus_tc.cmd_definitions import get_fsfw_service_op_code_dict
class FsfwHookBase(TmTcHookBase): class FsfwHookBase(TmTcHookBase):
def get_json_config_file_path(self) -> str: def __init__(self, json_cfg_path: str):
return "config/tmtc_config.json" super().__init__(json_cfg_path=json_cfg_path)
def get_service_op_code_dictionary(self) -> ServiceOpCodeDictT: def get_service_op_code_dictionary(self) -> ServiceOpCodeDictT:
from tmtccmd.config.globals import ( return get_fsfw_service_op_code_dict()
get_default_service_op_code_dict,
add_service_op_code_entry,
add_op_code_entry,
OpCodeDictKeys,
)
def_dict = get_default_service_op_code_dict()
op_code = dict()
add_op_code_entry(op_code_dict=op_code, keys="test", info="Mode CMD Test")
add_op_code_entry(
op_code_dict=op_code,
keys=["0", "asm_to_normal"],
info="Command test assembly to normal mode",
options={OpCodeDictKeys.TIMEOUT: 6.0},
)
add_service_op_code_entry(
srv_op_code_dict=def_dict,
name="200",
info="Mode MGMT",
op_code_entry=op_code,
)
return def_dict
def add_globals_pre_args_parsing(self, gui: bool = False):
from tmtccmd.config.globals import set_default_globals_pre_args_parsing
set_default_globals_pre_args_parsing(
gui=gui,
pus_tm_version=PusVersion.PUS_C,
pus_tc_version=PusVersion.PUS_C,
tc_apid=PUS_APID,
tm_apid=PUS_APID,
)
def add_globals_post_args_parsing(self, args: argparse.Namespace):
from tmtccmd.config.globals import set_default_globals_post_args_parsing
set_default_globals_post_args_parsing(
args=args, json_cfg_path=self.get_json_config_file_path()
)
def assign_communication_interface( def assign_communication_interface(
self, com_if_key: str, tmtc_printer: TmTcPrinter self, com_if_key: str
) -> Optional[CommunicationInterface]: ) -> Optional[CommunicationInterface]:
from tmtccmd.config.com_if import create_communication_interface_default from tmtccmd.config.com_if import create_communication_interface_default
return create_communication_interface_default( return create_communication_interface_default(
com_if_key=com_if_key, com_if_key=com_if_key,
tmtc_printer=tmtc_printer, json_cfg_path=self.json_cfg_path,
json_cfg_path=self.get_json_config_file_path(), space_packet_ids=TM_SP_IDS,
space_packet_ids=(0x08EF,),
) )
def perform_mode_operation(self, tmtc_backend: TmTcHandler, mode: int): def perform_mode_operation(self, tmtc_backend: TmTcHandler, mode: int):
@ -88,25 +46,5 @@ class FsfwHookBase(TmTcHookBase):
return get_object_ids() return get_object_ids()
@staticmethod def get_retval_dict(self) -> RetvalDictT:
def handle_service_8_telemetry( return self.get_retval_dict()
object_id: int, action_id: int, custom_data: bytearray
) -> Tuple[list, list]:
from common_tmtc.pus_tm.service_8_handling import custom_service_8_handling
return custom_service_8_handling(
object_id=object_id, action_id=action_id, custom_data=custom_data
)
@staticmethod
def handle_service_3_housekeeping(
object_id: bytes, set_id: int, hk_data: bytearray, service3_packet: Service3Base
) -> Tuple[list, list, bytearray, int]:
from common_tmtc.pus_tm.service_3_hk_handling import service_3_hk_handling
return service_3_hk_handling(
object_id=object_id,
set_id=set_id,
hk_data=hk_data,
service3_packet=service3_packet,
)

View File

@ -3,7 +3,15 @@
@details Template configuration file. Copy this folder to the TMTC commander root and adapt @details Template configuration file. Copy this folder to the TMTC commander root and adapt
it to your needs. it to your needs.
""" """
from typing import Dict import os
from tmtccmd.fsfw import parse_fsfw_objects_csv
from tmtccmd.logging import get_console_logger
from tmtccmd.utility.obj_id import ObjectIdDictT
LOGGER = get_console_logger()
DEFAULT_OBJECTS_CSV_PATH = "config/objects.csv"
__OBJECT_ID_DICT = None
PUS_SERVICE_17_ID = bytes([0x53, 0x00, 0x00, 0x17]) PUS_SERVICE_17_ID = bytes([0x53, 0x00, 0x00, 0x17])
TEST_DEVICE_0_ID = bytes([0x44, 0x01, 0xAF, 0xFE]) TEST_DEVICE_0_ID = bytes([0x44, 0x01, 0xAF, 0xFE])
@ -11,11 +19,13 @@ TEST_DEVICE_1_ID = bytes([0x44, 0x02, 0xAF, 0xFE])
ASSEMBLY_ID = bytes([0x41, 0x00, 0xCA, 0xFE]) ASSEMBLY_ID = bytes([0x41, 0x00, 0xCA, 0xFE])
def get_object_ids() -> Dict[bytes, list]: def get_object_ids() -> ObjectIdDictT:
object_id_dict = { global __OBJECT_ID_DICT
PUS_SERVICE_17_ID: ["PUS Service 17"], if not os.path.exists(DEFAULT_OBJECTS_CSV_PATH):
TEST_DEVICE_0_ID: ["Test Device 0"], LOGGER.warning(f"No Objects CSV file found at {DEFAULT_OBJECTS_CSV_PATH}")
TEST_DEVICE_1_ID: ["Test Device 1"], if __OBJECT_ID_DICT is None:
ASSEMBLY_ID: ["Assembly ID"], if os.path.exists(DEFAULT_OBJECTS_CSV_PATH):
} __OBJECT_ID_DICT = parse_fsfw_objects_csv(csv_file=DEFAULT_OBJECTS_CSV_PATH)
return object_id_dict else:
__OBJECT_ID_DICT = dict()
return __OBJECT_ID_DICT

22
config/retvals.py Normal file
View File

@ -0,0 +1,22 @@
import os
from tmtccmd.fsfw import parse_fsfw_returnvalues_csv, RetvalDictT
from tmtccmd.logging import get_console_logger
DEFAULT_RETVAL_CSV_NAME = "config/returnvalues.csv"
__RETVAL_DICT = None
LOGGER = get_console_logger()
def get_retval_dict() -> RetvalDictT:
global __RETVAL_DICT
if __RETVAL_DICT is None:
if os.path.exists(DEFAULT_RETVAL_CSV_NAME):
__RETVAL_DICT = parse_fsfw_returnvalues_csv(
csv_file=DEFAULT_RETVAL_CSV_NAME
)
else:
LOGGER.warning(
f"No Return Value CSV file found at {DEFAULT_RETVAL_CSV_NAME}"
)
__RETVAL_DICT = dict()
return __RETVAL_DICT

View File

@ -1,4 +0,0 @@
SW_NAME = "fsfw-tmtc"
SW_VERSION = 1
SW_SUBVERSION = 4
SW_SUBSUBVERSION = 0

26
pus_tc/cmd_definitions.py Normal file
View File

@ -0,0 +1,26 @@
from tmtccmd.config import (
ServiceOpCodeDictT,
add_op_code_entry,
add_service_op_code_entry,
OpCodeDictKeys,
)
from tmtccmd.config.globals import get_default_service_op_code_dict
def get_fsfw_service_op_code_dict() -> ServiceOpCodeDictT:
service_op_code_dict = get_default_service_op_code_dict()
op_code = dict()
add_op_code_entry(op_code_dict=op_code, keys="test", info="Mode CMD Test")
add_op_code_entry(
op_code_dict=op_code,
keys=["0", "asm_to_normal"],
info="Command test assembly to normal mode",
options={OpCodeDictKeys.TIMEOUT: 6.0},
)
add_service_op_code_entry(
srv_op_code_dict=service_op_code_dict,
name="200",
info="Mode MGMT",
op_code_entry=op_code,
)
return service_op_code_dict

View File

@ -1,5 +1,5 @@
from tmtccmd.tc.definitions import TcQueueT from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.pus.service_17_test import ( from tmtccmd.pus.pus_17_test import (
pack_service_17_ping_command, pack_service_17_ping_command,
pack_generic_service17_test, pack_generic_service17_test,
) )

View File

@ -1,15 +1,9 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
"""
@file tmtcc_tc_service_200_mode.py
@brief PUS Service 200: PUS custom service 200: Mode commanding
@author R. Mueller
@date 02.05.2020
"""
from spacepackets.ecss.tc import PusTelecommand from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.config.definitions import QueueCommands from tmtccmd.config.definitions import QueueCommands
from tmtccmd.tc.packer import TcQueueT from tmtccmd.tc.packer import TcQueueT
from tmtccmd.tc.service_200_mode import pack_mode_data, Modes from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes
from common_tmtc.config.object_ids import TEST_DEVICE_0_ID, ASSEMBLY_ID from common_tmtc.config.object_ids import TEST_DEVICE_0_ID, ASSEMBLY_ID
@ -57,5 +51,4 @@ def pack_service_200_test_into(init_ssc: int, tc_queue: TcQueueT) -> int:
command = PusTelecommand(service=200, subservice=1, ssc=new_ssc, app_data=mode_data) command = PusTelecommand(service=200, subservice=1, ssc=new_ssc, app_data=mode_data)
tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft(command.pack_command_tuple())
new_ssc += 1 new_ssc += 1
tc_queue.appendleft((QueueCommands.EXPORT_LOG, "log/tmtc_log_service200.txt"))
return new_ssc return new_ssc

View File

@ -4,9 +4,9 @@ from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.config.definitions import QueueCommands from tmtccmd.config.definitions import QueueCommands
from tmtccmd.tc.definitions import TcQueueT from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.tc.service_20_parameter import pack_type_and_matrix_data, pack_parameter_id from tmtccmd.tc.pus_20_params import pack_type_and_matrix_data, pack_parameter_id
from tmtccmd.tc.service_200_mode import pack_mode_data from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes
from tmtccmd.utility.logger import get_console_logger from tmtccmd.logging import get_console_logger
from common_tmtc.config.object_ids import TEST_DEVICE_0_ID from common_tmtc.config.object_ids import TEST_DEVICE_0_ID
@ -25,7 +25,7 @@ def pack_service20_test_into(tc_queue: TcQueueT, called_externally: bool = False
# set mode normal # set mode normal
tc_queue.appendleft((QueueCommands.PRINT, "Testing Service 20: Set Normal Mode")) tc_queue.appendleft((QueueCommands.PRINT, "Testing Service 20: Set Normal Mode"))
mode_data = pack_mode_data(object_id, 2, 0) mode_data = pack_mode_data(object_id, Modes.NORMAL, 0)
command = PusTelecommand(service=200, subservice=1, ssc=2000, app_data=mode_data) command = PusTelecommand(service=200, subservice=1, ssc=2000, app_data=mode_data)
tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft(command.pack_command_tuple())
@ -33,9 +33,6 @@ def pack_service20_test_into(tc_queue: TcQueueT, called_externally: bool = False
load_param_1_simple_test_commands(tc_queue=tc_queue) load_param_1_simple_test_commands(tc_queue=tc_queue)
load_param_2_simple_test_commands(tc_queue=tc_queue) load_param_2_simple_test_commands(tc_queue=tc_queue)
if called_externally is False:
tc_queue.appendleft((QueueCommands.EXPORT_LOG, "log/tmtc_log_service20.txt"))
def load_param_0_simple_test_commands(tc_queue: TcQueueT): def load_param_0_simple_test_commands(tc_queue: TcQueueT):
object_id = TEST_DEVICE_0_ID object_id = TEST_DEVICE_0_ID

View File

@ -9,10 +9,9 @@ import struct
from spacepackets.ecss.tc import PusTelecommand from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.tc.service_200_mode import Modes from tmtccmd.tc.pus_200_fsfw_modes import Modes, pack_mode_data
from tmtccmd.config.definitions import QueueCommands from tmtccmd.config.definitions import QueueCommands
from tmtccmd.tc.definitions import TcQueueT from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.tc.service_200_mode import pack_mode_data
from common_tmtc.pus_tc import command_data as cmd_data from common_tmtc.pus_tc import command_data as cmd_data
from common_tmtc.config.object_ids import TEST_DEVICE_0_ID from common_tmtc.config.object_ids import TEST_DEVICE_0_ID
@ -73,11 +72,10 @@ def pack_generic_service_2_test_into(init_ssc: int, tc_queue: TcQueueT) -> int:
# Set mode off # Set mode off
tc_queue.appendleft((QueueCommands.PRINT, "Testing Service 2: Setting Off Mode")) tc_queue.appendleft((QueueCommands.PRINT, "Testing Service 2: Setting Off Mode"))
mode_data = pack_mode_data(object_id, 0, 0) mode_data = pack_mode_data(object_id, Modes.OFF, 0)
command = PusTelecommand(service=200, subservice=1, ssc=new_ssc, app_data=mode_data) command = PusTelecommand(service=200, subservice=1, ssc=new_ssc, app_data=mode_data)
tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft(command.pack_command_tuple())
new_ssc += 1 new_ssc += 1
tc_queue.appendleft((QueueCommands.EXPORT_LOG, "log/tmtc_log_service_2.txt"))
return new_ssc return new_ssc

View File

@ -1,15 +1,15 @@
from spacepackets.ecss.tc import PusTelecommand from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.config.definitions import QueueCommands from tmtccmd.config.definitions import QueueCommands
from tmtccmd.tc.service_200_mode import pack_mode_data, Modes from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes
from tmtccmd.tc.service_20_parameter import pack_boolean_parameter_command from tmtccmd.tc.pus_20_params import (
from tmtccmd.tc.service_3_housekeeping import ( pack_boolean_parameter_app_data,
make_sid, pack_fsfw_load_param_cmd,
generate_one_hk_command,
Srv3Subservice,
) )
from tmtccmd.tc.pus_3_fsfw_hk import make_sid, generate_one_hk_command
import tmtccmd.tc.pus_3_fsfw_hk as srv3
from tmtccmd.tc.definitions import TcQueueT from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.tc.service_8_functional_cmd import generate_action_command from tmtccmd.tc.pus_8_funccmd import generate_action_command
from common_tmtc.config.object_ids import TEST_DEVICE_0_ID, TEST_DEVICE_1_ID from common_tmtc.config.object_ids import TEST_DEVICE_0_ID, TEST_DEVICE_1_ID
@ -130,15 +130,16 @@ def pack_housekeeping_basic_test(
tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft(command.pack_command_tuple())
tc_queue.appendleft((QueueCommands.PRINT, "Enabling changing datasets")) tc_queue.appendleft((QueueCommands.PRINT, "Enabling changing datasets"))
command = pack_boolean_parameter_command( app_data = pack_boolean_parameter_app_data(
object_id=object_id, object_id=object_id,
domain_id=0, domain_id=0,
unique_id=PARAM_ACTIVATE_CHANGING_DATASETS, unique_id=PARAM_ACTIVATE_CHANGING_DATASETS,
parameter=True, parameter=True,
ssc=current_ssc,
) )
cmd = pack_fsfw_load_param_cmd(app_data=app_data, ssc=0)
current_ssc += 1 current_ssc += 1
tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft(cmd.pack_command_tuple())
# Enable periodic reporting # Enable periodic reporting
tc_queue.appendleft( tc_queue.appendleft(
@ -146,7 +147,7 @@ def pack_housekeeping_basic_test(
) )
command = PusTelecommand( command = PusTelecommand(
service=3, service=3,
subservice=Srv3Subservice.ENABLE_PERIODIC_HK_GEN.value, subservice=srv3.Subservices.TC_ENABLE_PERIODIC_HK_GEN,
ssc=current_ssc, ssc=current_ssc,
app_data=test_sid, app_data=test_sid,
) )
@ -161,7 +162,7 @@ def pack_housekeeping_basic_test(
) )
command = PusTelecommand( command = PusTelecommand(
service=3, service=3,
subservice=Srv3Subservice.DISABLE_PERIODIC_HK_GEN.value, subservice=srv3.Subservices.TC_DISABLE_PERIODIC_HK_GEN,
ssc=current_ssc, ssc=current_ssc,
app_data=test_sid, app_data=test_sid,
) )
@ -170,15 +171,15 @@ def pack_housekeeping_basic_test(
# Disable changing datasets via parameter service (Service 20) # Disable changing datasets via parameter service (Service 20)
tc_queue.appendleft((QueueCommands.PRINT, "Disabling changing datasets")) tc_queue.appendleft((QueueCommands.PRINT, "Disabling changing datasets"))
command = pack_boolean_parameter_command( app_data = pack_boolean_parameter_app_data(
object_id=object_id, object_id=object_id,
domain_id=0, domain_id=0,
unique_id=PARAM_ACTIVATE_CHANGING_DATASETS, unique_id=PARAM_ACTIVATE_CHANGING_DATASETS,
parameter=False, parameter=False,
ssc=current_ssc,
) )
current_ssc += 1 current_ssc += 1
tc_queue.appendleft(command.pack_command_tuple()) cmd = pack_fsfw_load_param_cmd(app_data=app_data, ssc=0)
tc_queue.appendleft(cmd.pack_command_tuple())
return current_ssc return current_ssc

View File

@ -2,7 +2,7 @@ from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.config.definitions import QueueCommands from tmtccmd.config.definitions import QueueCommands
from tmtccmd.tc.definitions import TcQueueT from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.tc.service_200_mode import pack_mode_data, Modes from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes
import common_tmtc.pus_tc.command_data as cmd_data import common_tmtc.pus_tc.command_data as cmd_data
@ -57,8 +57,6 @@ def pack_generic_service_8_test_into(tc_queue: TcQueueT):
# Set mode off # Set mode off
tc_queue.appendleft((QueueCommands.PRINT, "Testing Service 8: Set Off Mode")) tc_queue.appendleft((QueueCommands.PRINT, "Testing Service 8: Set Off Mode"))
mode_data = pack_mode_data(object_id, 0, 0) mode_data = pack_mode_data(object_id, Modes.OFF, 0)
command = PusTelecommand(service=200, subservice=1, ssc=800, app_data=mode_data) command = PusTelecommand(service=200, subservice=1, ssc=800, app_data=mode_data)
tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft(command.pack_command_tuple())
tc_queue.appendleft((QueueCommands.EXPORT_LOG, "log/tmtc_log_service_8.txt"))

View File

@ -3,26 +3,50 @@
@details Template configuration file. Copy this folder to the TMTC commander root and adapt @details Template configuration file. Copy this folder to the TMTC commander root and adapt
it to your needs. it to your needs.
""" """
import logging
import os import os
from collections import deque from collections import deque
from typing import Union from typing import Union
from spacepackets.ecss.tc import PusTelecommand
from tmtccmd.com_if.com_interface_base import CommunicationInterface
from tmtccmd.logging import get_console_logger, get_current_time_string
from tmtccmd.logging.pus import log_raw_pus_tc
from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.config.definitions import CoreServiceList, QueueCommands
from tmtccmd.tc.pus_5_event import pack_generic_service5_test_into
from tmtccmd.pus.pus_17_test import pack_generic_service17_test
from common_tmtc.pus_tc.service_20_parameters import pack_service20_commands_into from common_tmtc.pus_tc.service_20_parameters import pack_service20_commands_into
from common_tmtc.pus_tc.service_2_raw_cmd import pack_service_2_commands_into from common_tmtc.pus_tc.service_2_raw_cmd import pack_service_2_commands_into
from common_tmtc.pus_tc.service_3_housekeeping import pack_service_3_commands_into from common_tmtc.pus_tc.service_3_housekeeping import pack_service_3_commands_into
from common_tmtc.pus_tc.service_17_test import pack_service_17_commands from common_tmtc.pus_tc.service_17_test import pack_service_17_commands
from common_tmtc.pus_tc.service_8_func_cmd import pack_service_8_commands_into from common_tmtc.pus_tc.service_8_func_cmd import pack_service_8_commands_into
from tmtccmd.utility.logger import get_console_logger
from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.config.definitions import CoreServiceList
from tmtccmd.tc.service_5_event import pack_generic_service5_test_into
from tmtccmd.pus.service_17_test import pack_generic_service17_test
from common_tmtc.pus_tc.service_200_mode import pack_service_200_commands_into from common_tmtc.pus_tc.service_200_mode import pack_service_200_commands_into
LOGGER = get_console_logger() LOGGER = get_console_logger()
def pre_tc_send_cb(
queue_entry: Union[bytes, QueueCommands],
com_if: CommunicationInterface,
queue_info: Union[PusTelecommand, any],
file_logger: logging.Logger,
):
if isinstance(queue_entry, bytes) or isinstance(queue_entry, bytearray):
log_raw_pus_tc(
packet=queue_entry,
srv_subservice=(queue_info.service, queue_info.subservice),
)
tc_info_string = f"Sent {queue_info}"
LOGGER.info(tc_info_string)
file_logger.info(f"{get_current_time_string(True)}: {tc_info_string}")
com_if.send(data=queue_entry)
elif isinstance(queue_entry, QueueCommands):
if queue_entry == QueueCommands.PRINT:
file_logger.info(queue_info)
def pack_service_queue_user(service: Union[str, int], op_code: str, tc_queue: TcQueueT): def pack_service_queue_user(service: Union[str, int], op_code: str, tc_queue: TcQueueT):
if service == CoreServiceList.SERVICE_2.value: if service == CoreServiceList.SERVICE_2.value:
return pack_service_2_commands_into(op_code=op_code, tc_queue=tc_queue) return pack_service_2_commands_into(op_code=op_code, tc_queue=tc_queue)

View File

@ -0,0 +1,24 @@
from tmtccmd.logging import get_console_logger
from tmtccmd.tm import Service8FsfwTm
from tmtccmd.utility.obj_id import ObjectIdDictT
from tmtccmd.utility.tmtc_printer import FsfwTmTcPrinter
LOGGER = get_console_logger()
def handle_action_reply(
raw_tm: bytes, printer: FsfwTmTcPrinter, obj_id_dict: ObjectIdDictT
):
"""Core Action reply handler
:return:
"""
tm_packet = Service8FsfwTm.unpack(raw_telemetry=raw_tm)
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)
custom_data = tm_packet.custom_data
action_id = tm_packet.action_id
generic_print_str = printer.generic_action_packet_tm_print(
packet=tm_packet, obj_id=object_id
)
print(generic_print_str)
printer.file_logger.info(generic_print_str)

61
pus_tm/event_handler.py Normal file
View File

@ -0,0 +1,61 @@
import logging
import os.path
from datetime import datetime
from common_tmtc.config.object_ids import get_object_ids
from tmtccmd.tm import Service5Tm
from tmtccmd.logging import get_console_logger
from tmtccmd.utility.tmtc_printer import FsfwTmTcPrinter
from tmtccmd.fsfw import parse_fsfw_events_csv, EventDictT, EventInfo
LOGGER = get_console_logger()
DEFAULT_EVENTS_CSV_PATH = "config/events.csv"
__EVENT_DICT = None
def get_event_dict() -> EventDictT:
global __EVENT_DICT
if __EVENT_DICT is None:
if os.path.exists(DEFAULT_EVENTS_CSV_PATH):
__EVENT_DICT = parse_fsfw_events_csv(DEFAULT_EVENTS_CSV_PATH)
else:
LOGGER.warning(f"No Event CSV file found at {DEFAULT_EVENTS_CSV_PATH}")
__EVENT_DICT = dict()
return __EVENT_DICT
def handle_event_packet(
raw_tm: bytes, printer: FsfwTmTcPrinter, file_logger: logging.Logger
) -> str:
tm = Service5Tm.unpack(raw_telemetry=raw_tm)
printer.handle_long_tm_print(packet_if=tm, info_if=tm)
additional_event_info = ""
event_dict = get_event_dict()
info = event_dict.get(tm.event_id)
if info is None:
LOGGER.warning(f"Event ID {tm.event_id} has no information")
info = EventInfo()
info.name = "Unknown event"
obj_ids = get_object_ids()
obj_id_obj = obj_ids.get(tm.reporter_id.as_bytes)
if obj_id_obj is None:
LOGGER.warning(f"Object ID 0x{tm.reporter_id.as_string} has no name")
obj_name = tm.reporter_id.as_string
else:
obj_name = obj_id_obj.name
generic_event_string = (
f"Object {obj_name} generated Event {tm.event_id} | {info.name}"
)
if info.info != "":
additional_event_info = (
f"Additional info: {info.info} | P1: {tm.param_1} | P2: {tm.param_2}"
)
file_logger.info(
f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}: {generic_event_string}"
)
LOGGER.info(generic_event_string)
if additional_event_info != "":
file_logger.info(additional_event_info)
print(additional_event_info)
return generic_event_string + " | " + additional_event_info

View File

@ -4,55 +4,89 @@
it to your needs. it to your needs.
""" """
from spacepackets.ecss.tm import PusTelemetry from spacepackets.ecss.tm import PusTelemetry
from spacepackets.util import PrintFormats
from spacepackets.ccsds.spacepacket import PacketTypes
from tmtccmd.utility.logger import get_console_logger from tmtccmd.tm.pus_2_rawcmd import Service2Tm
from tmtccmd.pus.service_1_verification import Service1TMExtended from tmtccmd.tm.pus_17_test import Service17TMExtended
from tmtccmd.tm.service_2_raw_cmd import Service2TM from tmtccmd.tm.pus_20_fsfw_parameters import Service20FsfwTm
from tmtccmd.tm.service_3_housekeeping import Service3TM from tmtccmd.tm.pus_200_fsfw_modes import Service200FsfwTm
from tmtccmd.tm.service_5_event import Service5TM from tmtccmd.utility.tmtc_printer import FsfwTmTcPrinter
from tmtccmd.tm.service_8_functional_cmd import Service8TM from tmtccmd.logging import get_console_logger
from tmtccmd.pus.service_17_test import Service17TMExtended from tmtccmd.logging.pus import (
from tmtccmd.tm.service_20_parameters import Service20TM create_tmtc_logger,
from tmtccmd.tm.service_200_mode import Service200TM log_raw_pus_tm,
from tmtccmd.utility.tmtc_printer import TmTcPrinter log_raw_unknown_packet,
)
from common_tmtc.config.object_ids import get_object_ids
from common_tmtc.pus_tm.action_reply_handling import handle_action_reply
from common_tmtc.pus_tm.event_handler import handle_event_packet
from common_tmtc.pus_tm.verification_handler import handle_service_1_packet
from common_tmtc.pus_tm.hk_handling import handle_hk_packet
from common_tmtc.config.definitions import PUS_APID from common_tmtc.config.definitions import PUS_APID
LOGGER = get_console_logger() LOGGER = get_console_logger()
def ccsds_tm_handler( FSFW_PRINTER = FsfwTmTcPrinter(file_logger=create_tmtc_logger())
apid: int, raw_tm_packet: bytearray, tmtc_printer: TmTcPrinter
) -> None:
def ccsds_tm_handler(apid: int, raw_tm_packet: bytes, _user_args: any) -> None:
if apid == PUS_APID: if apid == PUS_APID:
pus_packet_factory(raw_tm_packet=raw_tm_packet, tmtc_printer=tmtc_printer) pus_factory_hook(raw_tm_packet=raw_tm_packet)
def pus_packet_factory(raw_tm_packet: bytearray, tmtc_printer: TmTcPrinter): def pus_factory_hook(raw_tm_packet: bytes):
if len(raw_tm_packet) < 8:
LOGGER.warning("Detected packet shorter than 8 bytes!")
return
service_type = raw_tm_packet[7] service_type = raw_tm_packet[7]
tm_packet = None subservice_type = raw_tm_packet[8]
if service_type == 1: file_logger = FSFW_PRINTER.file_logger
tm_packet = Service1TMExtended.unpack(raw_tm_packet) obj_id_dict = get_object_ids()
if service_type == 2: dedicated_handler = True
tm_packet = Service2TM.unpack(raw_tm_packet) try:
if service_type == 3: tm_packet = None
tm_packet = Service3TM.unpack(raw_tm_packet, custom_hk_handling=False) if service_type == 1:
if service_type == 8: handle_service_1_packet(printer=FSFW_PRINTER, raw_tm=raw_tm_packet)
tm_packet = Service8TM.unpack(raw_tm_packet) if service_type == 2:
if service_type == 5: tm_packet = Service2Tm.unpack(raw_tm_packet)
tm_packet = Service5TM.unpack(raw_tm_packet) dedicated_handler = False
if service_type == 17: if service_type == 3:
tm_packet = Service17TMExtended.unpack(raw_tm_packet) handle_hk_packet(
if service_type == 20: printer=FSFW_PRINTER, raw_tm=raw_tm_packet, obj_id_dict=obj_id_dict
tm_packet = Service20TM.unpack(raw_tm_packet) )
if service_type == 200: if service_type == 8:
tm_packet = Service200TM.unpack(raw_tm_packet) handle_action_reply(
if tm_packet is None: raw_tm=raw_tm_packet, printer=FSFW_PRINTER, obj_id_dict=obj_id_dict
LOGGER.info( )
"The service " if service_type == 5:
+ str(service_type) handle_event_packet(
+ " is not implemented in Telemetry Factory" raw_tm=raw_tm_packet, printer=FSFW_PRINTER, file_logger=file_logger
)
if service_type == 17:
tm_packet = Service17TMExtended.unpack(raw_telemetry=raw_tm_packet)
dedicated_handler = False
if service_type == 20:
tm_packet = Service20FsfwTm.unpack(raw_telemetry=raw_tm_packet)
dedicated_handler = False
if service_type == 200:
tm_packet = Service200FsfwTm.unpack(raw_telemetry=raw_tm_packet)
dedicated_handler = False
else:
LOGGER.info(
f"The service {service_type} is not implemented in Telemetry Factory"
)
tm_packet = PusTelemetry.unpack(raw_telemetry=raw_tm_packet)
tm_packet.print_source_data(PrintFormats.HEX)
dedicated_handler = True
if not dedicated_handler and tm_packet is not None:
FSFW_PRINTER.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
log_raw_pus_tm(
packet=raw_tm_packet, srv_subservice=(service_type, subservice_type)
) )
tm_packet = PusTelemetry.unpack(raw_tm_packet) except ValueError:
tmtc_printer.print_telemetry(packet_if=tm_packet, info_if=tm_packet) LOGGER.warning("Invalid packet format detected")
log_raw_unknown_packet(packet=raw_tm_packet, packet_type=PacketTypes.TM)

View File

@ -2,44 +2,57 @@
""" """
import struct import struct
from typing import Tuple from typing import Tuple
from tmtccmd.tm.service_3_housekeeping import Service3Base
from tmtccmd.utility.logger import get_console_logger from tmtccmd.tm import Service3FsfwTm
from tmtccmd.tm.pus_3_hk_base import HkContentType, Service3Base
from tmtccmd.utility.obj_id import ObjectIdDictT, ObjectId
from tmtccmd.utility.tmtc_printer import FsfwTmTcPrinter
from tmtccmd.logging import get_console_logger
from common_tmtc.config.object_ids import TEST_DEVICE_0_ID, TEST_DEVICE_1_ID from common_tmtc.config.object_ids import TEST_DEVICE_0_ID, TEST_DEVICE_1_ID
LOGGER = get_console_logger() LOGGER = get_console_logger()
def service_3_hk_handling( def handle_hk_packet(
object_id: bytes, set_id: int, hk_data: bytearray, service3_packet: Service3Base raw_tm: bytes,
) -> Tuple[list, list, bytearray, int]: obj_id_dict: ObjectIdDictT,
"""This function is called when a Service 3 Housekeeping packet is received. printer: FsfwTmTcPrinter,
):
tm_packet = Service3FsfwTm.unpack(raw_telemetry=raw_tm, custom_hk_handling=False)
named_obj_id = obj_id_dict.get(tm_packet.object_id.as_bytes)
if named_obj_id is None:
named_obj_id = tm_packet.object_id
if tm_packet.subservice == 25 or tm_packet.subservice == 26:
hk_data = tm_packet.tm_data[8:]
printer.generic_hk_tm_print(
content_type=HkContentType.HK,
object_id=named_obj_id,
set_id=tm_packet.set_id,
hk_data=hk_data,
)
handle_regular_hk_print(
printer=printer,
object_id=named_obj_id,
hk_packet=tm_packet,
hk_data=hk_data,
)
if tm_packet.subservice == 10 or tm_packet.subservice == 12:
LOGGER.warning("HK definitions printout not implemented yet")
Please note that the object IDs should be compared by value because direct comparison of
enumerations does not work in Python. For example use:
if object_id.value == ObjectIds.TEST_OBJECT.value def handle_regular_hk_print(
printer: FsfwTmTcPrinter,
to test equality based on the object ID list. object_id: ObjectId,
hk_packet: Service3Base,
@param object_id: hk_data: bytes,
@param set_id: ):
@param hk_data:
@param service3_packet:
@return: Expects a tuple, consisting of two lists, a bytearray and an integer
The first list contains the header columns, the second list the list with
the corresponding values. The bytearray is the validity buffer, which is usually appended
at the end of the housekeeping packet. The last value is the number of parameters.
"""
if object_id == TEST_DEVICE_0_ID or object_id == TEST_DEVICE_1_ID: if object_id == TEST_DEVICE_0_ID or object_id == TEST_DEVICE_1_ID:
return handle_test_set_deserialization(hk_data=hk_data) handle_test_set_deserialization(hk_data=hk_data)
else:
LOGGER.info("Service3TM: Parsing for this SID has not been implemented.")
return [], [], bytearray(), 0
def handle_test_set_deserialization( def handle_test_set_deserialization(
hk_data: bytearray, hk_data: bytes,
) -> Tuple[list, list, bytearray, int]: ) -> Tuple[list, list, bytearray, int]:
header_list = [] header_list = []
content_list = [] content_list = []

View File

@ -1,20 +0,0 @@
from typing import Tuple
def custom_service_8_handling(
object_id: int, action_id: int, custom_data: bytearray
) -> Tuple[list, list]:
"""
This function is called by the TMTC core if a Service 8 data reply (subservice 130)
is received. The user can return a tuple of two lists, where the first list
is a list of header strings to print and the second list is a list of values to print.
The TMTC core will take care of printing both lists and logging them.
@param object_id:
@param action_id:
@param custom_data:
@return:
"""
header_list = []
content_list = []
return header_list, content_list

View File

@ -0,0 +1,28 @@
from typing import cast
from tmtccmd.tm.pus_1_verification import Service1TMExtended
from tmtccmd.logging import get_console_logger
from tmtccmd.utility.tmtc_printer import FsfwTmTcPrinter
from common_tmtc.config.retvals import get_retval_dict
LOGGER = get_console_logger()
def handle_service_1_packet(printer: FsfwTmTcPrinter, raw_tm: bytes):
tm_packet = Service1TMExtended.unpack(raw_telemetry=raw_tm)
printer.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
srv1_packet = cast(Service1TMExtended, tm_packet)
retval_dict = get_retval_dict()
if srv1_packet.has_tc_error_code:
retval_info = retval_dict.get(srv1_packet.error_code)
if retval_info is None:
LOGGER.info(
f"No returnvalue information found for error code {srv1_packet.error_code}"
)
else:
retval_string = (
f"Error Code information for code {srv1_packet.error_code} | "
f"Name: {retval_info.name} | Info: {retval_info.info}"
)
LOGGER.info(retval_string)
printer.file_logger.info(retval_string)

65
tmtcc.py Normal file
View File

@ -0,0 +1,65 @@
import argparse
import sys
import traceback
from typing import Optional
try:
import spacepackets
except ImportError as error:
print(error)
print("Python spacepackets module could not be imported")
print(
'Install with "cd spacepackets && python3 -m pip intall -e ." for interative installation'
)
sys.exit(1)
try:
import tmtccmd.runner as tmtccmd
from tmtccmd.logging.pus import create_tmtc_logger
from tmtccmd.ccsds.handler import ApidHandler, CcsdsTmHandler
from tmtccmd.config import SetupArgs, default_json_path
from tmtccmd.config.args import (
create_default_args_parser,
add_default_tmtccmd_args,
parse_default_input_arguments,
)
except ImportError as error:
run_tmtc_commander = None
initialize_tmtc_commander = None
tb = traceback.format_exc()
print(tb)
print("Python tmtccmd submodule could not be imported")
sys.exit(1)
from common_tmtc.config import __version__
from common_tmtc.config.definitions import PUS_APID
from common_tmtc.config.hook_implementation import FsfwHookBase
from common_tmtc.pus_tm.factory_hook import ccsds_tm_handler
from common_tmtc.pus_tc.tc_packing import pre_tc_send_cb
def tmtcc_pre_args() -> FsfwHookBase:
print(f"-- eive tmtc v{__version__} --")
print(f"-- spacepackets v{spacepackets.__version__} --")
tmtccmd.init_printout(False)
return FsfwHookBase(json_cfg_path=default_json_path())
def tmtcc_post_args(
hook_obj: FsfwHookBase, use_gui: bool, args: Optional[argparse.Namespace]
):
setup_args = SetupArgs(
hook_obj=hook_obj, use_gui=use_gui, apid=PUS_APID, cli_args=args
)
tmtc_file_logger = create_tmtc_logger()
apid_handler = ApidHandler(cb=ccsds_tm_handler, queue_len=50, user_args=None)
ccsds_handler = CcsdsTmHandler()
ccsds_handler.add_tm_handler(apid=PUS_APID, handler=apid_handler)
tmtccmd.setup(setup_args=setup_args)
tmtccmd.add_ccsds_handler(ccsds_handler)
tmtc_backend = tmtccmd.create_default_tmtc_backend(
setup_args=setup_args,
tm_handler=ccsds_handler,
)
tmtc_backend.usr_send_wrapper = (pre_tc_send_cb, tmtc_file_logger)
tmtccmd.run(tmtc_backend=tmtc_backend)