service 8 reply interpretation
This commit is contained in:
parent
b4868b78e1
commit
cf5ca54fbe
@ -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" />
|
||||
|
@ -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()
|
||||
@ -33,5 +34,6 @@ def set_object_ids(object_id_dict: Dict[ObjectIds, bytearray]):
|
||||
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]),
|
||||
}
|
||||
)
|
||||
|
@ -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,7 +19,7 @@ class PDU1TestProcedure:
|
||||
@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
|
||||
|
@ -5,7 +5,7 @@
|
||||
@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 gomspace.gomspace_common import *
|
||||
@ -36,68 +36,68 @@ class PDU2TestProcedure:
|
||||
|
||||
|
||||
def pack_pdu2_test_into(pdu2_object_id: bytearray, p60dock_object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
|
||||
tc_queue.appendleft(("print", "Testing PDU2"))
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "Testing PDU2"))
|
||||
|
||||
tc_queue.appendleft(("print", "P60 Dock: Enabling PDU2"))
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "P60 Dock: Enabling PDU2"))
|
||||
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(("print", "PDU2: Reboot"))
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Reboot"))
|
||||
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(("print", "PDU2: Reading ground watchdog timer value"))
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Reading ground watchdog timer value"))
|
||||
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(("print", "PDU2: Testing ground watchdog reset"))
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Testing ground watchdog reset"))
|
||||
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(("print", "PDU2: Ping Test"))
|
||||
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(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(("print", "PDU2: Testing setting output channel 2 on (TCS Heater)"))
|
||||
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_temperature:
|
||||
tc_queue.appendleft(("print", "PDU2: Testing temperature reading"))
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Testing temperature reading"))
|
||||
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(("print", "PDU2: Reading output channel 2 state (TCS Heater)"))
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Reading output channel 2 state (TCS Heater)"))
|
||||
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(("print", "PDU2: Reading current limit value of output channel 0 (OBC)"))
|
||||
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(("print", "PDU2: Testing setting output channel 2 off"))
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Testing setting output channel 2 off"))
|
||||
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(("print", "PDU2: Requesting housekeeping 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())
|
||||
|
31
pus_tm/tmtcc_pus_service_8_hook.py
Normal file
31
pus_tm/tmtcc_pus_service_8_hook.py
Normal file
@ -0,0 +1,31 @@
|
||||
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 = ['Housekeeping Table']
|
||||
content_list = [custom_data]
|
||||
else:
|
||||
header_list = []
|
||||
content_list = []
|
||||
return header_list, content_list
|
@ -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,10 @@ 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)
|
||||
service8tm.print_source_data()
|
||||
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")
|
||||
|
Loading…
x
Reference in New Issue
Block a user