Compare commits
28 Commits
7cc06ef0e0
...
06750809cb
Author | SHA1 | Date | |
---|---|---|---|
06750809cb | |||
e23bc11608 | |||
bc77e6ae76 | |||
fa58c8e2c3 | |||
e3342cf474 | |||
4d2b7a6150 | |||
986f88e290 | |||
3b5d92f908 | |||
684ae7a9c4 | |||
79e897b035 | |||
![]() |
3e466f06ef | ||
![]() |
934b79a2f1 | ||
![]() |
5fe23be649 | ||
![]() |
85f5bf2e6e | ||
7310513805 | |||
aa52ed79ea | |||
2e84c822c8 | |||
e993d2d2ea | |||
f40b70f66e | |||
![]() |
80ee42089e | ||
754edffeae | |||
10fa7658d0 | |||
52428e4e03 | |||
146cb52b66 | |||
81cd88f521 | |||
cf572c311c | |||
07381147f7 | |||
1d5fe4ebc7 |
@@ -15,3 +15,5 @@ class CustomServiceList(enum.Enum):
|
|||||||
TMP1075_1 = "tmp1075_1"
|
TMP1075_1 = "tmp1075_1"
|
||||||
TMP1075_2 = "tmp1075_2"
|
TMP1075_2 = "tmp1075_2"
|
||||||
HEATER = "heater"
|
HEATER = "heater"
|
||||||
|
IMTQ = "imtq"
|
||||||
|
PLOC = "ploc"
|
||||||
|
@@ -18,6 +18,8 @@ TMP_1075_2_HANDLER_ID = bytearray([0x44, 0x00, 0x00, 0x6])
|
|||||||
HEATER_ID = bytearray([0x54, 0x00, 0x00, 0x1])
|
HEATER_ID = bytearray([0x54, 0x00, 0x00, 0x1])
|
||||||
PCDU_HANDLER_ID = bytearray([0x44, 0x00, 0x10, 0x00])
|
PCDU_HANDLER_ID = bytearray([0x44, 0x00, 0x10, 0x00])
|
||||||
SOLAR_ARRAY_DEPLOYMENT_ID = bytearray([0x44, 0x00, 0x10, 0x01])
|
SOLAR_ARRAY_DEPLOYMENT_ID = bytearray([0x44, 0x00, 0x10, 0x01])
|
||||||
|
IMTQ_HANDLER_ID = bytearray([0x44, 0x00, 0x00, 0x14])
|
||||||
|
PLOC_ID = bytearray([0x44, 0x00, 0x00, 0x15])
|
||||||
|
|
||||||
|
|
||||||
class ObjIdIds(enum.IntEnum):
|
class ObjIdIds(enum.IntEnum):
|
||||||
@@ -32,6 +34,8 @@ class ObjIdIds(enum.IntEnum):
|
|||||||
TMP1075_2_HANDLER_ID = 8
|
TMP1075_2_HANDLER_ID = 8
|
||||||
HEATER_ID = 9
|
HEATER_ID = 9
|
||||||
SOLAR_ARRAY_DEPLOYMENT_ID = 10
|
SOLAR_ARRAY_DEPLOYMENT_ID = 10
|
||||||
|
IMTQ_HANDLER_ID = 11
|
||||||
|
PLOC_ID = 12
|
||||||
|
|
||||||
|
|
||||||
def set_object_ids() -> Dict[int, bytearray]:
|
def set_object_ids() -> Dict[int, bytearray]:
|
||||||
@@ -47,5 +51,7 @@ def set_object_ids() -> Dict[int, bytearray]:
|
|||||||
ObjIdIds.HEATER_ID: HEATER_ID,
|
ObjIdIds.HEATER_ID: HEATER_ID,
|
||||||
ObjIdIds.PCDU_HANDLER_ID: PCDU_HANDLER_ID,
|
ObjIdIds.PCDU_HANDLER_ID: PCDU_HANDLER_ID,
|
||||||
ObjIdIds.SOLAR_ARRAY_DEPLOYMENT_ID: SOLAR_ARRAY_DEPLOYMENT_ID,
|
ObjIdIds.SOLAR_ARRAY_DEPLOYMENT_ID: SOLAR_ARRAY_DEPLOYMENT_ID,
|
||||||
|
ObjIdIds.IMTQ_HANDLER_ID: IMTQ_HANDLER_ID,
|
||||||
|
ObjIdIds.PLOC_ID: PLOC_ID
|
||||||
})
|
})
|
||||||
return object_id_dict
|
return object_id_dict
|
||||||
|
@@ -1,4 +1,4 @@
|
|||||||
SW_NAME = "eive"
|
SW_NAME = "eive"
|
||||||
VERSION_MAJOR = 1
|
VERSION_MAJOR = 1
|
||||||
VERSION_MINOR = 2
|
VERSION_MINOR = 4
|
||||||
VERSION_SUBMINOR = 0
|
VERSION_SUBMINOR = 0
|
||||||
|
@@ -6,6 +6,11 @@ class PDUConfigTable:
|
|||||||
out_en_1 = TableEntry(bytearray([0x00, 0x49]), TableEntry.uint8_size)
|
out_en_1 = TableEntry(bytearray([0x00, 0x49]), TableEntry.uint8_size)
|
||||||
out_en_2 = TableEntry(bytearray([0x00, 0x4A]), 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_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
|
# When channel consumes more than cur_lu_lim, channel is turned of immediately
|
||||||
cur_lu_lim_0 = TableEntry(bytearray([0x00, 0xB8]), TableEntry.uint16_size)
|
cur_lu_lim_0 = TableEntry(bytearray([0x00, 0xB8]), TableEntry.uint16_size)
|
||||||
|
|
||||||
|
66
pus_tc/imtq.py
Normal file
66
pus_tc/imtq.py
Normal file
@@ -0,0 +1,66 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
"""
|
||||||
|
@file imtq.py
|
||||||
|
@brief Tests for the ISIS IMTQ (Magnettorquer) device handler
|
||||||
|
@author J. Meier
|
||||||
|
@date 25.03.2021
|
||||||
|
"""
|
||||||
|
from tmtccmd.core.definitions import QueueCommands
|
||||||
|
|
||||||
|
from tmtccmd.pus_tc.packer import TcQueueT
|
||||||
|
from tmtccmd.ecss.tc import PusTelecommand
|
||||||
|
|
||||||
|
|
||||||
|
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 = True
|
||||||
|
get_commanded_dipole = True
|
||||||
|
|
||||||
|
|
||||||
|
class ImtqActionIds:
|
||||||
|
start_actuation_dipole = bytearray([0x0, 0x0, 0x0, 0x02])
|
||||||
|
get_commanded_dipole = bytearray([0x0, 0x0, 0x0, 0x03])
|
||||||
|
|
||||||
|
|
||||||
|
def pack_imtq_test_into(object_id: bytearray, tc_queue: TcQueueT) -> 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=20, app_data=command)
|
||||||
|
tc_queue.appendleft(command.pack_command_tuple())
|
||||||
|
|
||||||
|
|
||||||
|
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
|
reboot = False
|
||||||
ping = False
|
ping = False
|
||||||
read_temperature = 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(
|
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, "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:
|
if PDU1TestProcedure.all or PDU1TestProcedure.ping:
|
||||||
tc_queue.appendleft((QueueCommands.PRINT, "PDU1: Ping Test"))
|
tc_queue.appendleft((QueueCommands.PRINT, "PDU1: Ping Test"))
|
||||||
ping_data = bytearray([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
|
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)
|
command = PusTelecommand(service=8, subservice=128, ssc=24, app_data=command)
|
||||||
tc_queue.appendleft(command.pack_command_tuple())
|
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 = pack_request_full_hk_table_command(pdu2_object_id)
|
||||||
command = PusTelecommand(service=8, subservice=128, ssc=28, app_data=command)
|
command = PusTelecommand(service=8, subservice=128, ssc=28, app_data=command)
|
||||||
tc_queue.appendleft(command.pack_command_tuple())
|
tc_queue.appendleft(command.pack_command_tuple())
|
||||||
|
|
||||||
return tc_queue
|
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.core.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
|
@@ -19,7 +19,9 @@ from pus_tc.p60dock import pack_p60dock_test_into
|
|||||||
from pus_tc.pdu2 import pack_pdu2_test_into
|
from pus_tc.pdu2 import pack_pdu2_test_into
|
||||||
from pus_tc.pdu1 import pack_pdu1_test_into
|
from pus_tc.pdu1 import pack_pdu1_test_into
|
||||||
from pus_tc.acu import pack_acu_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.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 pus_tc.heater import pack_heater_test_into
|
||||||
from config.definitions import CustomServiceList
|
from config.definitions import CustomServiceList
|
||||||
from config.object_ids import ObjIdIds
|
from config.object_ids import ObjIdIds
|
||||||
@@ -54,8 +56,14 @@ def pack_service_queue_user(service: Union[str, int], op_code: str, service_queu
|
|||||||
object_id = get_object_id(ObjIdIds.TMP1075_2_HANDLER_ID)
|
object_id = get_object_id(ObjIdIds.TMP1075_2_HANDLER_ID)
|
||||||
return pack_tmp1075_test_into(object_id, service_queue)
|
return pack_tmp1075_test_into(object_id, service_queue)
|
||||||
if service == CustomServiceList.HEATER.value:
|
if service == CustomServiceList.HEATER.value:
|
||||||
object_id = get_object_id(ObjIdIds.HEATER)
|
object_id = get_object_id(ObjIdIds.HEATER_ID)
|
||||||
return pack_heater_test_into(object_id, service_queue)
|
return pack_heater_test_into(object_id, service_queue)
|
||||||
|
if service == CustomServiceList.IMTQ.value:
|
||||||
|
object_id = get_object_id(ObjIdIds.IMTQ_HANDLER_ID)
|
||||||
|
return pack_imtq_test_into(object_id, service_queue)
|
||||||
|
if service == CustomServiceList.PLOC.value:
|
||||||
|
object_id = get_object_id(ObjIdIds.PLOC_ID)
|
||||||
|
return pack_ploc_test_into(object_id, service_queue)
|
||||||
LOGGER.warning("Invalid Service !")
|
LOGGER.warning("Invalid Service !")
|
||||||
|
|
||||||
|
|
||||||
|
@@ -1,6 +1,6 @@
|
|||||||
# -*- coding: utf-8 -*-
|
# -*- coding: utf-8 -*-
|
||||||
"""
|
"""
|
||||||
@file tmtcc_tc_tmp1075.py
|
@file tmp1075.py
|
||||||
@brief TMP1075 tests
|
@brief TMP1075 tests
|
||||||
@author J. Meier
|
@author J. Meier
|
||||||
@date 06.01.2021
|
@date 06.01.2021
|
||||||
|
22
pus_tc/tmtcc_tc_solar_array_deployment.py
Normal file
22
pus_tc/tmtcc_tc_solar_array_deployment.py
Normal file
@@ -0,0 +1,22 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
"""
|
||||||
|
@file tmtcc_tc_solar_array_deployment.py
|
||||||
|
@brief The test function in this file simply returns a command which triggers the solar array deployment.
|
||||||
|
@author J. Meier
|
||||||
|
@date 15.02.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 ActionIds:
|
||||||
|
DEPLOY_SOLAR_ARRAYS = bytearray([0x0, 0x0, 0x0, 0x5])
|
||||||
|
|
||||||
|
|
||||||
|
def pack_solar_array_deployment_test_into(object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
|
||||||
|
tc_queue.appendleft((QueueCommands.PRINT, "Testing S/A Deployment"))
|
||||||
|
|
||||||
|
command = object_id + ActionIds.DEPLOY_SOLAR_ARRAYS
|
||||||
|
command = PusTelecommand(service=8, subservice=128, ssc=200, app_data=command)
|
||||||
|
tc_queue.appendleft(command.pack_command_tuple())
|
31
pus_tc/tmtcc_tc_syrlinks_hk_handler.py
Normal file
31
pus_tc/tmtcc_tc_syrlinks_hk_handler.py
Normal file
@@ -0,0 +1,31 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
"""
|
||||||
|
@file tmtcc_tc_syrlinks_hk_handler.py
|
||||||
|
@brief Syrlinks Hk Handler tests
|
||||||
|
@author J. Meier
|
||||||
|
@date 13.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.pus_tc.tmtcc_tc_service_3_housekeeping import *
|
||||||
|
|
||||||
|
|
||||||
|
class SetIds:
|
||||||
|
RX_REGISTERS_DATASET = 1
|
||||||
|
TX_REGISTERS_DATASET = 2
|
||||||
|
|
||||||
|
|
||||||
|
def pack_syrlinks_hk_handler_test_into(object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
|
||||||
|
tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Get RX Registers"))
|
||||||
|
sid = make_sid(object_id, SetIds.RX_REGISTERS_DATASET)
|
||||||
|
command = generate_one_hk_command(sid, 200)
|
||||||
|
tc_queue.appendleft(command.pack_command_tuple())
|
||||||
|
|
||||||
|
tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Get TX Registers"))
|
||||||
|
sid = make_sid(object_id, SetIds.TX_REGISTERS_DATASET)
|
||||||
|
command = generate_one_hk_command(sid, 201)
|
||||||
|
tc_queue.appendleft(command.pack_command_tuple())
|
||||||
|
|
||||||
|
return tc_queue
|
@@ -8,6 +8,7 @@ from tmtccmd.ecss.tm import PusTelemetry
|
|||||||
from tmtccmd.utility.tmtcc_logger import get_logger
|
from tmtccmd.utility.tmtcc_logger import get_logger
|
||||||
|
|
||||||
from tmtccmd.pus_tm.service_1_verification import Service1TM
|
from tmtccmd.pus_tm.service_1_verification import Service1TM
|
||||||
|
from tmtccmd.pus_tm.service_3_base import Service3Base
|
||||||
from tmtccmd.pus_tm.service_5_event import Service5TM
|
from tmtccmd.pus_tm.service_5_event import Service5TM
|
||||||
from tmtccmd.pus_tm.service_17_test import Service17TM
|
from tmtccmd.pus_tm.service_17_test import Service17TM
|
||||||
|
|
||||||
@@ -18,6 +19,8 @@ def tm_user_factory_hook(raw_tm_packet: bytearray) -> PusTelemetry:
|
|||||||
service_type = raw_tm_packet[7]
|
service_type = raw_tm_packet[7]
|
||||||
if service_type == 1:
|
if service_type == 1:
|
||||||
return Service1TM(raw_tm_packet)
|
return Service1TM(raw_tm_packet)
|
||||||
|
if service_type == 3:
|
||||||
|
return Service3Base(raw_tm_packet)
|
||||||
if service_type == 5:
|
if service_type == 5:
|
||||||
return Service5TM(raw_tm_packet)
|
return Service5TM(raw_tm_packet)
|
||||||
if service_type == 8:
|
if service_type == 8:
|
||||||
|
@@ -1,5 +1,8 @@
|
|||||||
|
import struct
|
||||||
from typing import Tuple
|
from typing import Tuple
|
||||||
from config.object_ids import ObjIdIds
|
from config.object_ids import ObjIdIds
|
||||||
|
from pus_tc.imtq import ImtqActionIds
|
||||||
|
from pus_tc.ploc import PlocReplyIds
|
||||||
|
|
||||||
|
|
||||||
def user_analyze_service_8_data(
|
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(',')
|
||||||
data_string = data_string.rstrip()
|
data_string = data_string.rstrip()
|
||||||
content_list = [data_string]
|
content_list = [data_string]
|
||||||
|
elif object_id == ObjIdIds.IMTQ_HANDLER_ID:
|
||||||
|
return handle_imtq_replies(action_id, custom_data)
|
||||||
|
elif object_id == ObjIdIds.PLOC_ID:
|
||||||
|
return handle_ploc_replies(action_id, custom_data)
|
||||||
else:
|
else:
|
||||||
header_list = []
|
header_list = []
|
||||||
content_list = []
|
content_list = []
|
||||||
return 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
|
||||||
|
81
pus_tm/tmtcc_pus_hk_handling_hook.py
Normal file
81
pus_tm/tmtcc_pus_hk_handling_hook.py
Normal file
@@ -0,0 +1,81 @@
|
|||||||
|
"""
|
||||||
|
@brief This file transfers control of housekeeping handling (PUS service 3) to the
|
||||||
|
developer
|
||||||
|
@details Template configuration file. Copy this folder to the TMTC commander root and adapt
|
||||||
|
it to your needs.
|
||||||
|
"""
|
||||||
|
import struct
|
||||||
|
from typing import Tuple
|
||||||
|
|
||||||
|
from config.tmtcc_object_ids import ObjectIds
|
||||||
|
from tmtc_core.pus_tm.tmtcc_tm_service3_base import Service3Base
|
||||||
|
from tmtc_core.utility.tmtcc_logger import get_logger
|
||||||
|
from config.tmtcc_object_ids import ObjectIds
|
||||||
|
from pus_tc.tmtcc_tc_syrlinks_hk_handler import SetIds
|
||||||
|
|
||||||
|
LOGGER = get_logger()
|
||||||
|
|
||||||
|
|
||||||
|
def handle_user_hk_packet(object_id: ObjectIds, 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.
|
||||||
|
|
||||||
|
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 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 == ObjectIds.SYRLINKS_HK_HANDLER:
|
||||||
|
if set_id == SetIds.RX_REGISTERS_DATASET:
|
||||||
|
return handle_syrlinks_rx_registers_dataset(hk_data)
|
||||||
|
elif set_id == SetIds.TX_REGISTERS_DATASET:
|
||||||
|
return handle_syrlinks_tx_registers_dataset(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
|
||||||
|
|
||||||
|
|
||||||
|
def handle_syrlinks_rx_registers_dataset(hk_data: bytearray) -> Tuple[list, list, bytearray, int]:
|
||||||
|
hk_header = []
|
||||||
|
hk_content = []
|
||||||
|
validity_buffer = bytearray()
|
||||||
|
hk_header = ["RX Status", "RX Sensitivity", "RX Frequency Shift", "RX IQ Power", "RX AGC Value", "RX Demod Eb",
|
||||||
|
"RX Demod N0", "RX Datarate"]
|
||||||
|
rx_status = hk_data[0]
|
||||||
|
rx_sensitivity = struct.unpack('!I', hk_data[1:5])
|
||||||
|
rx_frequency_shift = struct.unpack('!I', hk_data[5:9])
|
||||||
|
rx_iq_power = struct.unpack('!H', hk_data[9:11])
|
||||||
|
rx_agc_value = struct.unpack('!H', hk_data[11:13])
|
||||||
|
rx_demod_eb = struct.unpack('!I', hk_data[13:17])
|
||||||
|
rx_demod_n0 = struct.unpack('!I', hk_data[17:21])
|
||||||
|
rx_data_rate = hk_data[21]
|
||||||
|
hk_content = [rx_status, rx_sensitivity, rx_frequency_shift, rx_iq_power, rx_agc_value, rx_demod_eb, rx_demod_n0,
|
||||||
|
rx_data_rate]
|
||||||
|
return hk_header, hk_content, validity_buffer, 8
|
||||||
|
|
||||||
|
|
||||||
|
def handle_syrlinks_tx_registers_dataset(hk_data: bytearray) -> Tuple[list, list, bytearray, int]:
|
||||||
|
hk_header = []
|
||||||
|
hk_content = []
|
||||||
|
validity_buffer = bytearray()
|
||||||
|
hk_header = ["TX Status", "TX Waveform", "TX AGC value"]
|
||||||
|
tx_status = hk_data[0]
|
||||||
|
tx_waveform = hk_data[1]
|
||||||
|
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
|
1
tmtc_core
Submodule
1
tmtc_core
Submodule
Submodule tmtc_core added at bd46c5a852
2
tmtccmd
2
tmtccmd
Submodule tmtccmd updated: fcf182c5e5...1773f62856
Reference in New Issue
Block a user