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

Merge remote-tracking branch 'origin/develop' into mueller/master

This commit is contained in:
Robin Müller 2021-02-14 12:18:46 +01:00
parent c10fbe7fbe 110ec9644a
commit 55cf05918c
13 changed files with 189 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.idea/runConfigurations/tmtcclient_PDU2_Test_UDP.xml
View File

@ -13,7 +13,7 @@
<option name="ADD_SOURCE_ROOTS" value="true" />
<EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
<option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtc_client_cli.py" />
<option name="PARAMETERS" value="-m 3 -c 2 -s PDU2 --boardIP 192.168.133.10 --clientIP 192.168.133.4 -l" />
<option name="PARAMETERS" value="-m 3 -c 2 -s PDU2 --boardIP 192.168.133.10 --clientIP 192.168.133.4 -l -t 4" />
<option name="SHOW_COMMAND_LINE" value="false" />
<option name="EMULATE_TERMINAL" value="true" />
<option name="MODULE_MODE" value="false" />

4
config/tmtcc_definitions.py
View File

@ -32,7 +32,9 @@ class ServiceList(enum.Enum):
PDU1 = auto()
PDU2 = auto()
ACU = auto()
TMP1075 = auto()
TMP1075_1 = auto()
TMP1075_2 = auto()
HEATER = auto()
class SerialConfig(enum.Enum):

8
config/tmtcc_globals.py
View File

@ -154,8 +154,12 @@ def add_globals_post_args_parsing(args: argparse.Namespace):
service = ServiceList.PDU2
elif service == "acu":
service = ServiceList.ACU
elif service == "tmp1075":
service = ServiceList.TMP1075
elif service == "tmp1075_1":
service = ServiceList.TMP1075_1
elif service == "tmp1075_2":
service = ServiceList.TMP1075_2
elif service == "heater":
service = ServiceList.HEATER
else:
logger.warning("Service not known! Setting standard service 17")
service = ServiceList.SERVICE_17

4
config/tmtcc_object_ids.py
View File

@ -10,6 +10,7 @@ from typing import Dict
class ObjectIds(enum.Enum):
from enum import auto
INVALID = auto()
PUS_SERVICE_17 = auto()
TEST_DEVICE = auto()
P60DOCK_HANDLER_ID = auto()
@ -18,6 +19,7 @@ class ObjectIds(enum.Enum):
ACU_HANDLER_ID = auto()
TMP1075_1_HANDLER_ID = auto()
TMP1075_2_HANDLER_ID = auto()
HEATER = auto()
def set_object_ids(object_id_dict: Dict[ObjectIds, bytearray]):
@ -31,5 +33,7 @@ def set_object_ids(object_id_dict: Dict[ObjectIds, bytearray]):
o_ids.ACU_HANDLER_ID: bytearray([0x44, 0x00, 0x00, 0x4]),
o_ids.TMP1075_1_HANDLER_ID: bytearray([0x44, 0x00, 0x00, 0x5]),
o_ids.TMP1075_2_HANDLER_ID: bytearray([0x44, 0x00, 0x00, 0x6]),
o_ids.HEATER: bytearray([0x54, 0x00, 0x00, 0x1]),
o_ids.INVALID: bytearray([0xFF, 0xFF, 0xFF, 0xFF]),
}
)

12
gomspace/gomspace_common.py
View File

@ -14,6 +14,7 @@ class GomspaceDeviceActions:
PARAM_GET = bytearray([0x0, 0x0, 0x0, 0x00])
PARAM_SET = bytearray([0x0, 0x0, 0x0, 0xFF])
WDT_RESET = bytearray([0x0, 0x0, 0x0, 0x9])
REQUEST_HK_TABLE = bytearray([0x0, 0x0, 0x0, 0x10])
class TableIds:
@ -118,3 +119,14 @@ def pack_reboot_command(object_id: bytearray) -> bytearray:
command = bytearray()
command += object_id + action_id
return command
def pack_request_full_hk_table_command(object_id: bytearray) -> bytearray:
""" Function to generate the command to request the full housekeeping table from a gomspace
device.
@param object_id The object id of the gomspace device handler.
"""
action_id = GomspaceDeviceActions.REQUEST_HK_TABLE
command = bytearray()
command = object_id + action_id
return command

56
pus_tc/tmtcc_tc_heater.py Normal file
View File

@ -0,0 +1,56 @@
# -*- coding: utf-8 -*-
"""
@file tmtcc_tc_heater.py
@brief Command sequence to test the HeaterHandler
@author J. Meier
@date 30.01.2021
"""
from tmtc_core.core.tmtc_core_definitions import QueueCommands
from tmtc_core.pus_tc.tmtcc_pus_tc_packer import TcQueueT
from tmtc_core.pus_tc.tmtcc_pus_tc_base import PusTelecommand
class SwitchNumbers:
HEATER_0 = 0
HEATER_1 = 1
HEATER_2 = 2
HEATER_3 = 3
HEATER_4 = 4
HEATER_5 = 5
HEATER_6 = 6
HEATER_7 = 7
NUMBER_OF_SWITCHES = 8
class ActionIds:
SWITCH_HEATER = bytearray([0x0, 0x0, 0x0, 0x0])
def pack_heater_test_into(object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
tc_queue.appendleft((QueueCommands.PRINT, "Testing Heater Switching"))
heater_number = int(input("Type number of heater to switch: "))
if heater_number >= SwitchNumbers.NUMBER_OF_SWITCHES:
print("Invalid heater switch number")
return
action = int(input("Turn switch on or off? (0 - off, 1 - on): "))
if action != 0 and action != 1:
print("Invalid action defined. Must be 0 (off) or 1 (on")
debug_string = "Switching heater " + str(heater_number)
tc_queue.appendleft((QueueCommands.PRINT, debug_string))
command = pack_switch_heater_command(object_id, heater_number, action)
command = PusTelecommand(service=8, subservice=128, ssc=300, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
def pack_switch_heater_command(object_id: bytearray, switch_nr: int, switch_action: int) -> bytearray:
""" Function to generate the command switch a heater
@param object_id The object id of the HeaterHandler object.
@param switch_nr The switch number identifying the heater to switch
@param switch_action Action to perform. 0 - Sets switch off, 1 - Sets switch on.
"""
action_id = ActionIds.SWITCH_HEATER
command = object_id + action_id
command.append(switch_nr)
command.append(switch_action)
return command

6
pus_tc/tmtcc_tc_p60dock.py
View File

@ -17,19 +17,19 @@ class P60DockTestProcedure:
@details Setting all to True will run all tests.
Setting all to False will only run the tests set to True.
"""
all = True
all = False
reboot = False
read_gnd_wdt = False
gnd_wdt_reset = False
ping = False
channel_3_off = False # pdu2
read_temperature1 = False
read_temperature1 = True
read_channel_3_state = False # pdu2
read_cur_lu_lim_0 = False
channel_3_on = False # pdu2
invalid_table_id_test = False # Test to check if software properly handles invalid table ids
invalid_address_test = False # Test to check if software properly handles invalid addresses
invalid_parameter_size_test = True
invalid_parameter_size_test = False
class P60DockConfigTable:

19
pus_tc/tmtcc_tc_packer_hook.py
View File

@ -19,6 +19,7 @@ from pus_tc.tmtcc_tc_acu import pack_acu_test_into
from tmtc_core.core.tmtcc_object_id_manager import get_object_id
from config.tmtcc_object_ids import ObjectIds
from pus_tc.tmtcc_tc_tmp1075 import pack_tmp1075_test_into
from pus_tc.tmtcc_tc_heater import pack_heater_test_into
LOGGER = get_logger()
@ -32,17 +33,25 @@ def pack_service_queue_user(service: ServiceList, op_code: str, service_queue: T
object_id = get_object_id(ObjectIds.P60DOCK_HANDLER_ID)
return pack_p60dock_test_into(object_id, service_queue)
if service == ServiceList.PDU1:
object_id = get_object_id(ObjectIds.PDU1_HANDLER_ID)
return pack_pdu1_test_into(object_id, service_queue)
pdu1_object_id = get_object_id(ObjectIds.PDU1_HANDLER_ID)
p60dock_object_id = get_object_id(ObjectIds.P60DOCK_HANDLER_ID)
return pack_pdu1_test_into(pdu1_object_id, p60dock_object_id, service_queue)
if service == ServiceList.PDU2:
object_id = get_object_id(ObjectIds.PDU2_HANDLER_ID)
return pack_pdu2_test_into(object_id, service_queue)
pdu2_object_id = get_object_id(ObjectIds.PDU2_HANDLER_ID)
p60dock_object_id = get_object_id(ObjectIds.P60DOCK_HANDLER_ID)
return pack_pdu2_test_into(pdu2_object_id, p60dock_object_id, service_queue)
if service == ServiceList.ACU:
object_id = get_object_id(ObjectIds.ACU_HANDLER_ID)
return pack_acu_test_into(object_id, service_queue)
if service == ServiceList.TMP1075:
if service == ServiceList.TMP1075_1:
object_id = get_object_id(ObjectIds.TMP1075_1_HANDLER_ID)
return pack_tmp1075_test_into(object_id, service_queue)
if service == ServiceList.TMP1075_2:
object_id = get_object_id(ObjectIds.TMP1075_2_HANDLER_ID)
return pack_tmp1075_test_into(object_id, service_queue)
if service == ServiceList.HEATER:
object_id = get_object_id(ObjectIds.HEATER)
return pack_heater_test_into(object_id, service_queue)
LOGGER.warning("Invalid Service !")

31
pus_tc/tmtcc_tc_pdu1.py
View File

@ -8,50 +8,41 @@
from tmtc_core.pus_tc.tmtcc_pus_tc_packer import TcQueueT
from tmtc_core.pus_tc.tmtcc_pus_tc_base import PusTelecommand
from tmtc_core.core.tmtcc_object_id_manager import get_object_id
from tmtc_core.core.tmtc_core_definitions import QueueCommands
from pus_tc.tmtcc_tc_p60dock import P60DockConfigTable
from config.tmtcc_object_ids import ObjectIds
from gomspace.gomspace_common import *
from pus_tc.tmtcc_tc_p60dock import P60DockConfigTable
from gomspace.gomspace_pdu_definitions import *
P60DOCK_HANDLER_ID = get_object_id(ObjectIds.P60DOCK_HANDLER_ID)
class PDU1TestProcedure:
"""
@brief Use this class to define the tests to perform for the PDU2.
@details Setting all to True will run all tests.
Setting all to False will only run the tests set to True.
"""
all = True
all = False
reboot = False
ping = False
read_temperature = False
def pack_pdu1_test_into(object_id: bytearray, tc_queue: TcQueueT):
tc_queue.appendleft((QueueCommands.PRINT, "Testing PDU1"))
def pack_pdu1_test_into(pdu1_object_id: bytearray, p60dock_object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
tc_queue.appendleft(("print", "Testing PDU1"))
tc_queue.appendleft((QueueCommands.PRINT, "P60 Dock: Enabling PDU1"))
command = pack_set_param_command(
P60DOCK_HANDLER_ID, P60DockConfigTable.out_en_1.parameter_address,
P60DockConfigTable.out_en_1.parameter_size, Channel.on
)
tc_queue.appendleft(("print", "P60 Dock: Enabling PDU1"))
command = pack_set_param_command(p60dock_object_id, P60DockConfigTable.out_en_1.parameter_address,
P60DockConfigTable.out_en_1.parameter_size, Channel.on)
command = PusTelecommand(service=8, subservice=128, ssc=23, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if PDU1TestProcedure.all or PDU1TestProcedure.ping:
tc_queue.appendleft((QueueCommands.PRINT, "PDU1: Ping Test"))
tc_queue.appendleft(("print", "PDU1: Ping Test"))
ping_data = bytearray([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
command = pack_ping_command(object_id, ping_data)
command = pack_ping_command(pdu1_object_id, ping_data)
command = PusTelecommand(service=8, subservice=128, ssc=22, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if PDU1TestProcedure.all or PDU1TestProcedure.read_temperature:
tc_queue.appendleft((QueueCommands.PRINT, "PDU1: Testing temperature reading"))
command = pack_get_param_command(object_id, TableIds.hk, PDUHkTable.temperature.parameter_address,
tc_queue.appendleft(("print", "PDU1: Testing temperature reading"))
command = pack_get_param_command(pdu1_object_id, TableIds.hk, PDUHkTable.temperature.parameter_address,
PDUHkTable.temperature.parameter_size)
command = PusTelecommand(service=8, subservice=128, ssc=24, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())

59
pus_tc/tmtcc_tc_pdu2.py
View File

@ -5,18 +5,13 @@
@author J. Meier
@date 17.12.2020
"""
from tmtc_core.core.tmtc_core_definitions import QueueCommands
from tmtc_core.pus_tc.tmtcc_pus_tc_packer import TcQueueT
from tmtc_core.pus_tc.tmtcc_pus_tc_base import PusTelecommand
from tmtc_core.core.tmtcc_object_id_manager import get_object_id
from tmtc_core.core.tmtc_core_definitions import QueueCommands
from config.tmtcc_object_ids import ObjectIds
from gomspace.gomspace_common import *
from gomspace.gomspace_pdu_definitions import *
from pus_tc.tmtcc_tc_p60dock import P60DockConfigTable
P60DOCK_HANDLER_ID = get_object_id(ObjectIds.P60DOCK_HANDLER_ID)
class PDU2TestProcedure:
"""
@ -24,84 +19,86 @@ class PDU2TestProcedure:
@details Setting all to True will run all tests.
Setting all to False will only run the tests set to True.
"""
all = True
all = False
reboot = False
read_gnd_wdt = False
gnd_wdt_reset = False
ping = False
channel_2_off = False # TCS Heater
read_temperature = False
read_temperature1 = False
read_channel_2_state = False # TCS Heater
read_cur_lu_lim_0 = False # OBC
channel_2_on = False # TCS Heater
channel_2_off = False # Reaction wheels 5V
read_temperature = True
read_channel_2_state = False # Reaction wheels 5V
read_cur_lu_lim_0 = False # OBC
channel_2_on = False # Reaction wheels 5V
invalid_table_id_test = False # Test to check if software properly handles invalid table ids
invalid_address_test = False # Test to check if software properly handles invalid addresses
invalid_parameter_size_test = True
invalid_parameter_size_test = False
request_hk_table = False
def pack_pdu2_test_into(object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
def pack_pdu2_test_into(pdu2_object_id: bytearray, p60dock_object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
tc_queue.appendleft((QueueCommands.PRINT, "Testing PDU2"))
tc_queue.appendleft((QueueCommands.PRINT, "P60 Dock: Enabling PDU2"))
command = pack_set_param_command(P60DOCK_HANDLER_ID, P60DockConfigTable.out_en_3.parameter_address,
command = pack_set_param_command(p60dock_object_id, P60DockConfigTable.out_en_3.parameter_address,
P60DockConfigTable.out_en_3.parameter_size, Channel.on)
command = PusTelecommand(service=8, subservice=128, ssc=23, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if PDU2TestProcedure.all or PDU2TestProcedure.reboot:
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Reboot"))
command = pack_reboot_command(object_id)
command = pack_reboot_command(pdu2_object_id)
command = PusTelecommand(service=8, subservice=128, ssc=22, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if PDU2TestProcedure.all or PDU2TestProcedure.read_gnd_wdt:
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Reading ground watchdog timer value"))
command = pack_get_param_command(object_id, TableIds.hk, PDUHkTable.wdt_gnd_left.parameter_address,
command = pack_get_param_command(pdu2_object_id, TableIds.hk, PDUHkTable.wdt_gnd_left.parameter_address,
PDUHkTable.wdt_gnd_left.parameter_size)
command = PusTelecommand(service=8, subservice=128, ssc=20, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if PDU2TestProcedure.all or PDU2TestProcedure.gnd_wdt_reset:
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Testing ground watchdog reset"))
command = pack_gnd_wdt_reset_command(object_id)
command = pack_gnd_wdt_reset_command(pdu2_object_id)
command = PusTelecommand(service=8, subservice=128, ssc=21, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if PDU2TestProcedure.all or PDU2TestProcedure.ping:
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Ping Test"))
ping_data = bytearray([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
command = pack_ping_command(object_id, ping_data)
command = pack_ping_command(pdu2_object_id, ping_data)
command = PusTelecommand(service=8, subservice=128, ssc=22, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if PDU2TestProcedure.all or PDU2TestProcedure.channel_2_on:
tc_queue.appendleft((QueueCommands.PRINT,
"PDU2: Testing setting output channel 2 on (TCS Heater)"))
command = pack_set_param_command(object_id, PDUConfigTable.out_en_2.parameter_address,
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Testing setting output channel 2 on (TCS Heater)"))
command = pack_set_param_command(pdu2_object_id, PDUConfigTable.out_en_2.parameter_address,
PDUConfigTable.out_en_2.parameter_size, Channel.on)
command = PusTelecommand(service=8, subservice=128, ssc=23, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if PDU2TestProcedure.all or PDU2TestProcedure.read_temperature1:
if PDU2TestProcedure.all or PDU2TestProcedure.read_temperature:
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Testing temperature reading"))
command = pack_get_param_command(object_id, TableIds.hk, PDUHkTable.temperature.parameter_address,
command = pack_get_param_command(pdu2_object_id, TableIds.hk, PDUHkTable.temperature.parameter_address,
PDUHkTable.temperature.parameter_size)
command = PusTelecommand(service=8, subservice=128, ssc=24, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if PDU2TestProcedure.all or PDU2TestProcedure.read_channel_2_state:
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Reading output channel 2 state (TCS Heater)"))
command = pack_get_param_command(object_id, TableIds.config, PDUConfigTable.out_en_2.parameter_address,
command = pack_get_param_command(pdu2_object_id, TableIds.config, PDUConfigTable.out_en_2.parameter_address,
PDUConfigTable.out_en_2.parameter_size)
command = PusTelecommand(service=8, subservice=128, ssc=25, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if PDU2TestProcedure.all or PDU2TestProcedure.read_cur_lu_lim_0:
tc_queue.appendleft((QueueCommands.PRINT,
"PDU2: Reading current limit value of output channel 0 (OBC)"))
command = pack_get_param_command(object_id, TableIds.config,
PDUConfigTable.cur_lu_lim_0.parameter_address,
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Reading current limit value of output channel 0 (OBC)"))
command = pack_get_param_command(pdu2_object_id, TableIds.config, PDUConfigTable.cur_lu_lim_0.parameter_address,
PDUConfigTable.cur_lu_lim_0.parameter_size)
command = PusTelecommand(service=8, subservice=128, ssc=26, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if PDU2TestProcedure.all or PDU2TestProcedure.channel_2_off:
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Testing setting output channel 2 off"))
command = pack_set_param_command(object_id, PDUConfigTable.out_en_2.parameter_address,
command = pack_set_param_command(pdu2_object_id, PDUConfigTable.out_en_2.parameter_address,
PDUConfigTable.out_en_2.parameter_size, Channel.off)
command = PusTelecommand(service=8, subservice=128, ssc=27, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if PDU2TestProcedure.all or PDU2TestProcedure.request_hk_table:
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Requesting housekeeping table"))
command = pack_request_full_hk_table_command(pdu2_object_id)
command = PusTelecommand(service=8, subservice=128, ssc=28, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
return tc_queue

17
pus_tc/tmtcc_tc_tmp1075.py
View File

@ -5,7 +5,7 @@
@author J. Meier
@date 06.01.2021
"""
from tmtc_core.core.tmtc_core_definitions import QueueCommands
from tmtc_core.pus_tc.tmtcc_pus_tc_packer import TcQueueT
from tmtc_core.pus_tc.tmtcc_pus_tc_base import PusTelecommand
from pus_tc.tmtcc_tc_service200_mode import pack_mode_data
@ -17,10 +17,10 @@ class Tmp1075TestProcedure:
@details Setting all to True will run all tests.
Setting all to False will only run the tests set to True.
"""
all = True
all = False
start_adc_conversion = False
get_temp = False
set_mode_normal = False # Setting mode to normal starts continuous temperature reading
set_mode_normal = True # Setting mode to normal starts continuous temperature reading
set_mode_on = False # If mode is MODE_ON, temperature will only be read on command
@ -30,28 +30,29 @@ class Tmp1075ActionIds:
def pack_tmp1075_test_into(object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
tc_queue.appendleft((QueueCommands.PRINT, "Testing Tmp1075 Temperature Sensor Handler"))
tc_queue.appendleft(("print", "Testing Tmp1075 Temperature Sensor Handler with object id: 0x" +
object_id.hex()))
if Tmp1075TestProcedure.all or Tmp1075TestProcedure.start_adc_conversion:
tc_queue.appendleft((QueueCommands.PRINT, "TMP1075: Starting new temperature conversion"))
tc_queue.appendleft(("print", "TMP1075: Starting new temperature conversion"))
command = object_id + Tmp1075ActionIds.start_adc_conversion
command = PusTelecommand(service=8, subservice=128, ssc=23, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if Tmp1075TestProcedure.all or Tmp1075TestProcedure.get_temp:
tc_queue.appendleft((QueueCommands.PRINT, "TMP1075: Read temperature"))
tc_queue.appendleft(("print", "TMP1075: Read temperature"))
command = object_id + Tmp1075ActionIds.get_temp
command = PusTelecommand(service=8, subservice=128, ssc=24, app_data=command)
tc_queue.appendleft(command.pack_command_tuple())
if Tmp1075TestProcedure.set_mode_normal:
tc_queue.appendleft((QueueCommands.PRINT, "TMP1075:: Set Mode Normal"))
tc_queue.appendleft(("print", "TMP1075: Set Mode Normal"))
mode_data = pack_mode_data(object_id, 2, 0)
command = PusTelecommand(service=200, subservice=1, ssc=220, app_data=mode_data)
tc_queue.appendleft(command.pack_command_tuple())
if Tmp1075TestProcedure.set_mode_on:
tc_queue.appendleft((QueueCommands.PRINT, "TMP1075: Set Mode On"))
tc_queue.appendleft(("print", "TMP1075: Set Mode On"))
mode_data = pack_mode_data(object_id, 1, 0)
command = PusTelecommand(service=200, subservice=1, ssc=221, app_data=mode_data)
tc_queue.appendleft(command.pack_command_tuple())

38
pus_tm/tmtcc_pus_service_8_hook.py Normal file
View File

@ -0,0 +1,38 @@
from typing import Tuple
from config.tmtcc_object_ids import ObjectIds
def user_analyze_service_8_data(
object_id: ObjectIds, 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.
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
to test equality based on the object ID list.
@param object_id:
@param action_id:
@param custom_data:
@return:
"""
if object_id.value == ObjectIds.PDU2_HANDLER_ID.value:
header_list = ['PDU2 Service 8 Reply']
data_string = str()
for index in range(len(custom_data)):
data_string += str(hex(custom_data[index])) + " , "
data_string = data_string.rstrip()
data_string = data_string.rstrip(',')
data_string = data_string.rstrip()
content_list = [data_string]
else:
header_list = []
content_list = []
return header_list, content_list

5
pus_tm/tmtcc_pus_tm_factory_hook.py
View File

@ -3,7 +3,7 @@
@details Template configuration file. Copy this folder to the TMTC commander root and adapt
it to your needs.
"""
from tmtc_core.pus_tm.tmtcc_pus_service_8 import Service8TM
from tmtc_core.pus_tm.tmtcc_pus_tm_base import PusTelemetry
from tmtc_core.utility.tmtcc_logger import get_logger
@ -20,6 +20,9 @@ def tm_user_factory_hook(raw_tm_packet: bytearray) -> PusTelemetry:
return Service1TM(raw_tm_packet)
if service_type == 5:
return Service5TM(raw_tm_packet)
if service_type == 8:
service8tm = Service8TM(raw_tm_packet)
return Service8TM(raw_tm_packet)
if service_type == 17:
return Service17TM(raw_tm_packet)
LOGGER.info("The service " + str(service_type) + " is not implemented in Telemetry Factory")