Compare commits
10 Commits
7cc06ef0e0
...
v1.3.0
Author | SHA1 | Date | |
---|---|---|---|
2e84c822c8 | |||
e993d2d2ea | |||
754edffeae | |||
10fa7658d0 | |||
52428e4e03 | |||
146cb52b66 | |||
81cd88f521 | |||
cf572c311c | |||
07381147f7 | |||
1d5fe4ebc7 |
@@ -1,4 +1,4 @@
|
|||||||
SW_NAME = "eive"
|
SW_NAME = "eive"
|
||||||
VERSION_MAJOR = 1
|
VERSION_MAJOR = 1
|
||||||
VERSION_MINOR = 2
|
VERSION_MINOR = 3
|
||||||
VERSION_SUBMINOR = 0
|
VERSION_SUBMINOR = 0
|
||||||
|
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:
|
||||||
|
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
|
Reference in New Issue
Block a user