Compare commits
19 Commits
v1.5.0
...
fe1e6a2877
Author | SHA1 | Date | |
---|---|---|---|
fe1e6a2877 | |||
d413ec6d08 | |||
fc87e36c2e | |||
58c20e31cc | |||
06750809cb | |||
e23bc11608 | |||
bc77e6ae76 | |||
fa58c8e2c3 | |||
e3342cf474 | |||
4d2b7a6150 | |||
79e897b035 | |||
![]() |
3e466f06ef | ||
![]() |
934b79a2f1 | ||
![]() |
5fe23be649 | ||
![]() |
85f5bf2e6e | ||
7310513805 | |||
aa52ed79ea | |||
f40b70f66e | |||
![]() |
80ee42089e |
6
.gitignore
vendored
6
.gitignore
vendored
@@ -4,4 +4,8 @@ log
|
||||
.idea/*
|
||||
!.idea/runConfigurations
|
||||
|
||||
*.json
|
||||
*.json
|
||||
|
||||
/Lib
|
||||
/Scripts
|
||||
/pyvenv.cfg
|
@@ -15,6 +15,8 @@ class CustomServiceList(enum.Enum):
|
||||
ACU = "acu"
|
||||
TMP1075_1 = "tmp1075_1"
|
||||
TMP1075_2 = "tmp1075_2"
|
||||
HEATER = "heater",
|
||||
HEATER = "heater"
|
||||
IMTQ = "imtq"
|
||||
PLOC = "ploc"
|
||||
PCDU = "pcdu",
|
||||
SA_DEPLYOMENT = "sa_depl"
|
||||
|
@@ -50,6 +50,12 @@ class EiveHookObject(TmTcHookBase):
|
||||
}
|
||||
service_heater_tuple = ("Heater Device", op_code_dict_srv_heater)
|
||||
|
||||
op_code_dict_srv_imtq = {
|
||||
"0": ("IMTQ Tests All", {OpCodeDictKeys.TIMEOUT: 2.0}),
|
||||
"1": ("IMTQ perform pos X self test", {OpCodeDictKeys.TIMEOUT: 2.0}),
|
||||
}
|
||||
service_imtq_tuple = ("IMTQ Device", op_code_dict_srv_imtq)
|
||||
|
||||
service_op_code_dict[CustomServiceList.ACU.value] = service_acu_tuple
|
||||
service_op_code_dict[CustomServiceList.TMP1075_1.value] = service_tmp1075_1_tuple
|
||||
service_op_code_dict[CustomServiceList.TMP1075_2.value] = service_tmp1075_2_tuple
|
||||
@@ -58,6 +64,7 @@ class EiveHookObject(TmTcHookBase):
|
||||
service_op_code_dict[CustomServiceList.PDU1.value] = service_pdu2_tuple
|
||||
service_op_code_dict[CustomServiceList.PDU1.value] = service_pdu2_tuple
|
||||
service_op_code_dict[CustomServiceList.HEATER.value] = service_heater_tuple
|
||||
service_op_code_dict[CustomServiceList.IMTQ.value] = service_imtq_tuple
|
||||
return service_op_code_dict
|
||||
|
||||
def get_json_config_file_path(self) -> str:
|
||||
@@ -113,7 +120,7 @@ class EiveHookObject(TmTcHookBase):
|
||||
|
||||
@staticmethod
|
||||
def handle_service_3_housekeeping(
|
||||
object_id: int, set_id: int, hk_data: bytearray, service3_packet: Service3Base
|
||||
object_id: bytes, set_id: int, hk_data: bytearray, service3_packet: Service3Base
|
||||
) -> Tuple[list, list, bytearray, int]:
|
||||
from pus_tm.hk_handling import handle_user_hk_packet
|
||||
return handle_user_hk_packet(
|
||||
|
@@ -17,6 +17,8 @@ HEATER_ID = bytes([0x54, 0x00, 0x00, 0x1])
|
||||
PCDU_HANDLER_ID = bytes([0x44, 0x00, 0x10, 0x00])
|
||||
SOLAR_ARRAY_DEPLOYMENT_ID = bytes([0x44, 0x00, 0x10, 0x01])
|
||||
SYRLINKS_HANDLER = bytes([0x44, 0x00, 0x10, 0x02])
|
||||
IMTQ_HANDLER_ID = bytearray([0x44, 0x00, 0x00, 0x14])
|
||||
PLOC_ID = bytearray([0x44, 0x00, 0x00, 0x15])
|
||||
|
||||
|
||||
def get_object_ids() -> Dict[bytes, list]:
|
||||
|
@@ -6,6 +6,11 @@ class PDUConfigTable:
|
||||
out_en_1 = TableEntry(bytearray([0x00, 0x49]), TableEntry.uint8_size)
|
||||
out_en_2 = TableEntry(bytearray([0x00, 0x4A]), TableEntry.uint8_size)
|
||||
out_en_3 = TableEntry(bytearray([0x00, 0x4B]), TableEntry.uint8_size)
|
||||
out_en_4 = TableEntry(bytearray([0x00, 0x4C]), TableEntry.uint8_size)
|
||||
out_en_5 = TableEntry(bytearray([0x00, 0x4D]), TableEntry.uint8_size)
|
||||
out_en_6 = TableEntry(bytearray([0x00, 0x4E]), TableEntry.uint8_size)
|
||||
out_en_7 = TableEntry(bytearray([0x00, 0x4F]), TableEntry.uint8_size)
|
||||
out_en_8 = TableEntry(bytearray([0x00, 0x50]), TableEntry.uint8_size)
|
||||
# When channel consumes more than cur_lu_lim, channel is turned of immediately
|
||||
cur_lu_lim_0 = TableEntry(bytearray([0x00, 0xB8]), TableEntry.uint16_size)
|
||||
|
||||
|
108
pus_tc/imtq.py
Normal file
108
pus_tc/imtq.py
Normal file
@@ -0,0 +1,108 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
@file imtq.py
|
||||
@brief Tests for the ISIS IMTQ (Magnettorquer) device handler
|
||||
@author J. Meier
|
||||
@date 25.03.2021
|
||||
"""
|
||||
from tmtccmd.config.definitions import QueueCommands
|
||||
|
||||
from tmtccmd.pus_tc.packer import TcQueueT
|
||||
from tmtccmd.ecss.tc import PusTelecommand
|
||||
from tmtccmd.pus_tc.service_3_housekeeping import make_sid, generate_one_hk_command
|
||||
|
||||
|
||||
class ImtqTestProcedure:
|
||||
"""
|
||||
@brief Use this class to define the tests to perform for the IMTQ Handler.
|
||||
@details Setting all to True will run all tests.
|
||||
Setting all to False will only run the tests set to True.
|
||||
"""
|
||||
all = False
|
||||
command_dipole = False
|
||||
get_commanded_dipole = False
|
||||
positive_x_test = True
|
||||
negative_x_test = False
|
||||
positive_y_test = False
|
||||
negative_y_test = False
|
||||
positive_z_test = False
|
||||
negative_z_test = False
|
||||
|
||||
class SetIds:
|
||||
POSITIVE_X_TEST = 1
|
||||
NEGATIVE_X_TEST = 2
|
||||
POSITIVE_Y_TEST = 3
|
||||
NEGATIVE_Y_TEST = 4
|
||||
POSITIVE_Z_TEST = 5
|
||||
NEGATIVE_Z_TEST = 6
|
||||
|
||||
|
||||
class ImtqActionIds:
|
||||
start_actuation_dipole = bytearray([0x0, 0x0, 0x0, 0x02])
|
||||
get_commanded_dipole = bytearray([0x0, 0x0, 0x0, 0x03])
|
||||
perform_positive_x_test = bytearray([0x0, 0x0, 0x0, 0x07])
|
||||
perform_negative_x_test = bytearray([0x0, 0x0, 0x0, 0x08])
|
||||
perform_positive_y_test = bytearray([0x0, 0x0, 0x0, 0x09])
|
||||
perform_negative_y_test = bytearray([0x0, 0x0, 0x0, 0x0A])
|
||||
perform_positive_z_test = bytearray([0x0, 0x0, 0x0, 0x0B])
|
||||
perform_negative_z_test = bytearray([0x0, 0x0, 0x0, 0x0C])
|
||||
# Initiates the reading of the last performed self test. After sending this command the results can be downlinked
|
||||
# via the housekeeping service by using the appropriate set ids listed above.
|
||||
read_self_test_results = bytearray([0x0, 0x0, 0x0, 0x0D])
|
||||
|
||||
|
||||
def pack_imtq_test_into(object_id: bytearray, tc_queue: TcQueueT, op_code: str) -> TcQueueT:
|
||||
tc_queue.appendleft(
|
||||
(QueueCommands.PRINT,
|
||||
"Testing ISIS IMTQ handler with object id: 0x" + object_id.hex())
|
||||
)
|
||||
|
||||
if ImtqTestProcedure.all or ImtqTestProcedure.command_dipole:
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Commanding dipole"))
|
||||
command = pack_dipole_command(object_id, 2000, 2000, 2000, 10000)
|
||||
command = PusTelecommand(service=8, subservice=128, ssc=20, app_data=command)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
|
||||
if ImtqTestProcedure.all or ImtqTestProcedure.get_commanded_dipole:
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Get commanded dipole"))
|
||||
command = object_id + ImtqActionIds.get_commanded_dipole
|
||||
command = PusTelecommand(service=8, subservice=128, ssc=21, app_data=command)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
|
||||
if op_code == "0" or op_code == "1":
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Perform positive x self test"))
|
||||
command = object_id + ImtqActionIds.perform_positive_x_test
|
||||
command = PusTelecommand(service=8, subservice=128, ssc=22, app_data=command)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Initiate reading of positive x self test results"))
|
||||
command = object_id + ImtqActionIds.read_self_test_results
|
||||
command = PusTelecommand(service=8, subservice=128, ssc=23, app_data=command)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Request dataset with positive x self test results"))
|
||||
sid = make_sid(object_id, SetIds.POSITIVE_X_TEST)
|
||||
command = generate_one_hk_command(sid, 24)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
|
||||
return tc_queue
|
||||
|
||||
|
||||
def pack_dipole_command(object_id: bytearray, x_dipole: int, y_dipole: int, z_dipole: int, duration: int) -> bytearray:
|
||||
""" This function packs the command causing the ISIS IMTQ to generate a dipole.
|
||||
@param object_id The object id of the gomspace device handler.
|
||||
@param x_dipole The dipole of the x coil in 10^-4*Am^2 (max. 2000)
|
||||
@param y_dipole The dipole of the y coil in 10^-4*Am^2 (max. 2000)
|
||||
@param z_dipole The dipole of the z coil in 10^-4*Am^2 (max. 2000)
|
||||
@param duration The duration in milliseconds the dipole will be generated by the coils.
|
||||
When set to 0, the dipole will be generated until a new dipole actuation
|
||||
command is sent.
|
||||
"""
|
||||
action_id = ImtqActionIds.start_actuation_dipole
|
||||
command = bytearray()
|
||||
command = object_id + action_id
|
||||
command.extend(x_dipole.to_bytes(length=2, byteorder='big'))
|
||||
command.extend(y_dipole.to_bytes(length=2, byteorder='big'))
|
||||
command.extend(z_dipole.to_bytes(length=2, byteorder='big'))
|
||||
command.extend(duration.to_bytes(length=2, byteorder='big'))
|
||||
return command
|
@@ -24,6 +24,10 @@ class PDU1TestProcedure:
|
||||
reboot = False
|
||||
ping = False
|
||||
read_temperature = False
|
||||
turn_channel_2_on = False # Star Tracker connected to this channel (5V)
|
||||
turn_channel_2_off = False
|
||||
turn_channel_3_on = False # MTQ connected to this channel (5V)
|
||||
turn_channel_3_off = True
|
||||
|
||||
|
||||
def pack_pdu1_test_into(
|
||||
@@ -31,14 +35,6 @@ def pack_pdu1_test_into(
|
||||
):
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "Testing PDU1"))
|
||||
|
||||
tc_queue.appendleft((QueueCommands.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"))
|
||||
ping_data = bytearray([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
|
||||
@@ -53,3 +49,27 @@ def pack_pdu1_test_into(
|
||||
)
|
||||
command = PusTelecommand(service=8, subservice=128, ssc=24, app_data=command)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
if PDU1TestProcedure.all or PDU1TestProcedure.turn_channel_2_on:
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "PDU1: Turn channel 2 on (Star Tracker)"))
|
||||
command = pack_set_param_command(pdu1_object_id, PDUConfigTable.out_en_2.parameter_address,
|
||||
PDUConfigTable.out_en_2.parameter_size, Channel.on)
|
||||
command = PusTelecommand(service=8, subservice=128, ssc=30, app_data=command)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
if PDU1TestProcedure.all or PDU1TestProcedure.turn_channel_2_off:
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "PDU1: Turn channel 2 off (Star Tracker)"))
|
||||
command = pack_set_param_command(pdu1_object_id, PDUConfigTable.out_en_2.parameter_address,
|
||||
PDUConfigTable.out_en_2.parameter_size, Channel.off)
|
||||
command = PusTelecommand(service=8, subservice=128, ssc=31, app_data=command)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
if PDU1TestProcedure.all or PDU1TestProcedure.turn_channel_3_on:
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "PDU1: Turn channel 3 on (MTQ)"))
|
||||
command = pack_set_param_command(pdu1_object_id, PDUConfigTable.out_en_3.parameter_address,
|
||||
PDUConfigTable.out_en_3.parameter_size, Channel.on)
|
||||
command = PusTelecommand(service=8, subservice=128, ssc=32, app_data=command)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
if PDU1TestProcedure.all or PDU1TestProcedure.turn_channel_3_off:
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "PDU1: Turn channel 3 off (MTQ)"))
|
||||
command = pack_set_param_command(pdu1_object_id, PDUConfigTable.out_en_3.parameter_address,
|
||||
PDUConfigTable.out_en_3.parameter_size, Channel.off)
|
||||
command = PusTelecommand(service=8, subservice=128, ssc=33, app_data=command)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
|
@@ -101,4 +101,5 @@ def pack_pdu2_test_into(pdu2_object_id: bytearray, p60dock_object_id: bytearray,
|
||||
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
|
||||
|
67
pus_tc/ploc.py
Normal file
67
pus_tc/ploc.py
Normal file
@@ -0,0 +1,67 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
@file ploc.py
|
||||
@brief TMP1075 tests
|
||||
@author J. Meier
|
||||
@date 06.01.2021
|
||||
"""
|
||||
import struct
|
||||
|
||||
from tmtccmd.config.definitions import QueueCommands
|
||||
|
||||
from tmtccmd.pus_tc.packer import TcQueueT
|
||||
from tmtccmd.ecss.tc import PusTelecommand
|
||||
|
||||
|
||||
class PlocTestProcedure:
|
||||
"""
|
||||
@brief Use this class to define the tests to perform for the PLOC.
|
||||
@details Setting all to True will run all tests.
|
||||
Setting all to False will only run the tests set to True.
|
||||
"""
|
||||
all = False
|
||||
test_tc_mem_write = False
|
||||
test_tc_mem_read = True
|
||||
|
||||
|
||||
class PlocActionIds:
|
||||
tc_mem_write = bytearray([0x0, 0x0, 0x0, 0x1])
|
||||
tc_mem_read = bytearray([0x0, 0x0, 0x0, 0x2])
|
||||
|
||||
|
||||
class PlocReplyIds:
|
||||
tm_mem_read_report = 6
|
||||
|
||||
|
||||
def pack_ploc_test_into(object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
|
||||
tc_queue.appendleft(
|
||||
(QueueCommands.PRINT,
|
||||
"Testing PLOC Handler with object id: 0x" + object_id.hex())
|
||||
)
|
||||
|
||||
if PlocTestProcedure.all or PlocTestProcedure.test_tc_mem_write:
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "PLOC: TC Mem Write Test"))
|
||||
memory_address = int(input("PLOC Tc Mem Write: Type memory address: 0x"), 16)
|
||||
memory_data = int(input("PLOC Tc Mem Write: Type memory data: 0x"), 16)
|
||||
command = generate_write_mem_command(object_id, struct.pack('!I', memory_address), memory_data)
|
||||
command = PusTelecommand(service=8, subservice=128, ssc=23, app_data=command)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
|
||||
if PlocTestProcedure.all or PlocTestProcedure.test_tc_mem_read:
|
||||
tc_queue.appendleft((QueueCommands.PRINT, "PLOC: TC Mem Read Test"))
|
||||
memory_address = int(input("PLOC Tc Mem Read: Type memory address: 0x"), 16)
|
||||
command = object_id + PlocActionIds.tc_mem_read + struct.pack('!I', memory_address)
|
||||
command = PusTelecommand(service=8, subservice=128, ssc=24, app_data=command)
|
||||
tc_queue.appendleft(command.pack_command_tuple())
|
||||
|
||||
return tc_queue
|
||||
|
||||
|
||||
def generate_write_mem_command(object_id: bytearray, memory_address: bytearray, memory_data: int) -> bytearray:
|
||||
""" This function generates the command to write to a memory address within the PLOC
|
||||
@param object_id The object id of the PlocHandler
|
||||
@param memory_address The PLOC memory address where to write to.
|
||||
@param memory_data The data to write to the memory address specified by the bytearray memory_address.
|
||||
"""
|
||||
command = object_id + PlocActionIds.tc_mem_write + memory_address + struct.pack('!I', memory_data)
|
||||
return command
|
@@ -18,11 +18,13 @@ from pus_tc.p60dock import pack_p60dock_test_into
|
||||
from pus_tc.pdu2 import pack_pdu2_test_into
|
||||
from pus_tc.pdu1 import pack_pdu1_test_into
|
||||
from pus_tc.acu import pack_acu_test_into
|
||||
from pus_tc.imtq import pack_imtq_test_into
|
||||
from pus_tc.tmp1075 import pack_tmp1075_test_into
|
||||
from pus_tc.ploc import pack_ploc_test_into
|
||||
from pus_tc.heater import pack_heater_test_into
|
||||
from config.definitions import CustomServiceList
|
||||
from config.object_ids import P60_DOCK_HANDLER, PDU_1_HANDLER_ID, PDU_2_HANDLER_ID, ACU_HANDLER_ID, \
|
||||
TMP_1075_1_HANDLER_ID, TMP_1075_2_HANDLER_ID, HEATER_ID
|
||||
TMP_1075_1_HANDLER_ID, TMP_1075_2_HANDLER_ID, HEATER_ID, IMTQ_HANDLER_ID, PLOC_ID
|
||||
|
||||
|
||||
LOGGER = get_logger()
|
||||
@@ -60,6 +62,13 @@ def pack_service_queue_user(service: Union[str, int], op_code: str, service_queu
|
||||
if service == CustomServiceList.HEATER.value:
|
||||
object_id = HEATER_ID
|
||||
return pack_heater_test_into(object_id=object_id, tc_queue=service_queue)
|
||||
if service == CustomServiceList.IMTQ.value:
|
||||
object_id = IMTQ_HANDLER_ID
|
||||
return pack_imtq_test_into(object_id=object_id, tc_queue=service_queue, op_code=op_code)
|
||||
if service == CustomServiceList.PLOC.value:
|
||||
object_id = PLOC_ID
|
||||
return pack_ploc_test_into(object_id=object_id, tc_queue=service_queue)
|
||||
|
||||
LOGGER.warning("Invalid Service !")
|
||||
|
||||
|
||||
|
@@ -1,6 +1,6 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
@file tmtcc_tc_tmp1075.py
|
||||
@file tmp1075.py
|
||||
@brief TMP1075 tests
|
||||
@author J. Meier
|
||||
@date 06.01.2021
|
||||
|
@@ -12,8 +12,9 @@ from tmtccmd.utility.logger import get_logger
|
||||
LOGGER = get_logger()
|
||||
|
||||
|
||||
def handle_user_hk_packet(object_id: int, set_id: int, hk_data: bytearray,
|
||||
service3_packet: Service3Base) -> Tuple[list, list, bytearray, int]:
|
||||
def handle_user_hk_packet(
|
||||
object_id: bytes, set_id: int, hk_data: bytearray, service3_packet: Service3Base
|
||||
) -> Tuple[list, list, bytearray, int]:
|
||||
"""
|
||||
This function is called when a Service 3 Housekeeping packet is received.
|
||||
|
||||
|
@@ -10,7 +10,7 @@ from typing import Tuple
|
||||
from tmtccmd.pus_tm.service_3_housekeeping import Service3Base
|
||||
from tmtccmd.utility.logger import get_logger
|
||||
from pus_tc.syrlinks_hk_handler import SetIds
|
||||
from config.object_ids import SYRLINKS_HANDLER
|
||||
from config.object_ids import SYRLINKS_HANDLER, IMTQ_HANDLER_ID
|
||||
LOGGER = get_logger()
|
||||
|
||||
|
||||
@@ -43,6 +43,12 @@ def handle_user_hk_packet(object_id: bytes, set_id: int, hk_data: bytearray,
|
||||
else:
|
||||
LOGGER.info("Serive 3 TM: Syrlinks handler reply with unknown set id")
|
||||
return [], [], bytearray(), 0
|
||||
if object_id == IMTQ_HANDLER_ID:
|
||||
if set_id == SetIds.RX_REGISTERS_DATASET:
|
||||
return imtq_positive_x_test(hk_data)
|
||||
else:
|
||||
LOGGER.info("Serive 3 TM: Syrlinks handler reply with unknown set id")
|
||||
return [], [], bytearray(), 0
|
||||
else:
|
||||
LOGGER.info("Service 3 TM: Parsing for this SID has not been implemented.")
|
||||
return [], [], bytearray(), 0
|
||||
@@ -77,3 +83,30 @@ def handle_syrlinks_tx_registers_dataset(hk_data: bytearray) -> Tuple[list, list
|
||||
tx_agc_value = struct.unpack('!H', hk_data[2:4])
|
||||
hk_content = [tx_status, tx_waveform, tx_agc_value]
|
||||
return hk_header, hk_content, validity_buffer, 3
|
||||
|
||||
|
||||
def imtq_positive_x_test(hk_data: bytearray) -> Tuple[list, list, bytearray, int]:
|
||||
hk_header = []
|
||||
hk_content = []
|
||||
validity_buffer = bytearray()
|
||||
hk_header = ["Init Err", "Init Raw Mag X [T]", "Init Raw Mag Y [T]", "Init Raw Mag Z [T]", "Init Cal Mag X [T]",
|
||||
"Init Cal Mag Y [T]", "Init Cal Mag Z [T]", "Init Coil X Current [A]", "Init Coil Y Current [A]",
|
||||
"Init Coil Z Current [A]", "Init Coil X Temperature [°C]", "Init Coil Y Temperature [°C]",
|
||||
"Init Coil Z Temperature [°C]"]
|
||||
init_err = hk_data[0]
|
||||
init_raw_mag_x = struct.unpack('!f', hk_data[1:5])
|
||||
init_raw_mag_y = struct.unpack('!f', hk_data[5:9])
|
||||
init_raw_mag_z = struct.unpack('!f', hk_data[9:13])
|
||||
init_cal_mag_x = struct.unpack('!f', hk_data[13:17])
|
||||
init_cal_mag_y = struct.unpack('!f', hk_data[17:21])
|
||||
init_cal_mag_z = struct.unpack('!f', hk_data[21:25])
|
||||
init_coil_x_current = struct.unpack('!f', hk_data[25:29])
|
||||
init_coil_y_current = struct.unpack('!f', hk_data[29:33])
|
||||
init_coil_z_current = struct.unpack('!f', hk_data[33:37])
|
||||
init_coil_x_temperature = struct.unpack('!H', hk_data[37:39])
|
||||
init_coil_y_temperature = struct.unpack('!H', hk_data[39:41])
|
||||
init_coil_z_temperature = struct.unpack('!H', hk_data[41:43])
|
||||
hk_content = [init_err, init_raw_mag_x, init_raw_mag_y, init_raw_mag_z, init_cal_mag_x, init_cal_mag_y,
|
||||
init_cal_mag_z, init_coil_x_current, init_coil_y_current, init_coil_z_current,
|
||||
init_coil_x_temperature, init_coil_y_temperature, init_coil_z_temperature]
|
||||
return hk_header, hk_content, len(hk_header)
|
||||
|
@@ -1,5 +1,8 @@
|
||||
import struct
|
||||
from typing import Tuple
|
||||
from config.object_ids import PDU_2_HANDLER_ID
|
||||
from config.object_ids import *
|
||||
from pus_tc.imtq import ImtqActionIds
|
||||
from pus_tc.ploc import PlocReplyIds
|
||||
|
||||
|
||||
def user_analyze_service_8_data(
|
||||
@@ -25,7 +28,31 @@ def user_analyze_service_8_data(
|
||||
data_string = data_string.rstrip(',')
|
||||
data_string = data_string.rstrip()
|
||||
content_list = [data_string]
|
||||
elif object_id == IMTQ_HANDLER_ID:
|
||||
return handle_imtq_replies(action_id, custom_data)
|
||||
elif object_id == PLOC_ID:
|
||||
return handle_ploc_replies(action_id, custom_data)
|
||||
else:
|
||||
header_list = []
|
||||
content_list = []
|
||||
return header_list, content_list
|
||||
|
||||
|
||||
def handle_imtq_replies(action_id: int, custom_data: bytearray) -> Tuple[list, list]:
|
||||
header_list = []
|
||||
content_list = []
|
||||
if action_id == struct.unpack('!I', ImtqActionIds.get_commanded_dipole)[0]:
|
||||
header_list = ['Commanded X-Dipole', 'Commanded Y-Dipole', 'Commanded Z-Dipole']
|
||||
x_dipole = struct.unpack('!H', custom_data[:2])
|
||||
y_dipole = struct.unpack('!H', custom_data[2:4])
|
||||
z_dipole = struct.unpack('!H', custom_data[4:6])
|
||||
content_list = [x_dipole[0], y_dipole[0], z_dipole[0]]
|
||||
|
||||
|
||||
def handle_ploc_replies(action_id: int, custom_data: bytearray) -> Tuple[list, list]:
|
||||
header_list = []
|
||||
content_list = []
|
||||
if action_id == PlocReplyIds.tm_mem_read_report:
|
||||
header_list = ['PLOC Memory Address', 'PLOC Mem Len', 'PLOC Read Memory Data']
|
||||
content_list = [custom_data[:4], custom_data[4:6], custom_data[6:10]]
|
||||
return header_list, content_list
|
||||
|
1
tmtc_core
Submodule
1
tmtc_core
Submodule
Submodule tmtc_core added at bd46c5a852
2
tmtccmd
2
tmtccmd
Submodule tmtccmd updated: 3f39a1ffa1...39697eda49
Reference in New Issue
Block a user