syrlinks commands

.idea/runConfigurations/Syrlinks.xml

@@ -1,5 +1,5 @@
 <component name="ProjectRunConfigurationManager">
-  <configuration default="false" name="Syrlinks Test UDP" type="PythonConfigurationType" factoryName="Python" folderName="Devices">
+  <configuration default="false" name="Syrlinks" type="PythonConfigurationType" factoryName="Python" folderName="Devices">
     <module name="tmtc" />
     <option name="INTERPRETER_OPTIONS" value="" />
     <option name="PARENT_ENVS" value="true" />
@@ -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 SYRLINKS -l --hk" />
+    <option name="PARAMETERS" value="-s syrlinks -l -t 6" />
     <option name="SHOW_COMMAND_LINE" value="false" />
     <option name="EMULATE_TERMINAL" value="true" />
     <option name="MODULE_MODE" value="false" />

config/definitions.py

@@ -39,3 +39,4 @@ class CustomServiceList(enum.Enum):
     CCSDS_HANDLER = 'ccsds_handler'
     PDEC_HANDLER = 'pdec_handler'
     STR_IMG_HELPER = 'str_img_helper'
+    SYRLINKS = 'syrlinks'

config/hook_implementations.py

@@ -304,6 +304,17 @@ def get_eive_service_op_code_dict(service_op_code_dict: ServiceOpCodeDictT):
     }
     service_pdec_handler_tuple = ("PDEC Handler", op_code_dict_srv_pdec_handler)
 
+    op_code_dict_srv_syrlinks_handler = {
+        "0": ("syrlinks Handler: Set TX standby", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "1": ("syrlinks Handler: Set TX modulation", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "2": ("syrlinks Handler: Set TX carrier wave", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "5": ("syrlinks Handler: Read TX status", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "6": ("syrlinks Handler: Read TX waveform", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "7": ("syrlinks Handler: Read TX AGC value high byte ", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "8": ("syrlinks Handler: Read TX AGC value low byte ", {OpCodeDictKeys.TIMEOUT: 2.0})
+    }
+    service_syrlinks_handler_tuple = ("PDEC Handler", op_code_dict_srv_syrlinks_handler)
+
     op_code_dict_srv_str_img_helper = {
         "0": ("Star Tracker Image Helper: Upload image", {OpCodeDictKeys.TIMEOUT: 2.0}),
     }
@@ -326,3 +337,4 @@ def get_eive_service_op_code_dict(service_op_code_dict: ServiceOpCodeDictT):
     service_op_code_dict[CustomServiceList.CCSDS_HANDLER.value] = service_ccsds_handler_tuple
     service_op_code_dict[CustomServiceList.PDEC_HANDLER.value] = service_pdec_handler_tuple
     service_op_code_dict[CustomServiceList.STR_IMG_HELPER.value] = service_str_img_helper_tuple
+    service_op_code_dict[CustomServiceList.SYRLINKS.value] = service_syrlinks_handler_tuple

config/object_ids.py

@@ -22,7 +22,7 @@ TMP_1075_1_HANDLER_ID = bytes([0x44, 0x42, 0x00, 0x04])
 TMP_1075_2_HANDLER_ID = bytes([0x44, 0x42, 0x00, 0x05])
 
 # Communication Object IDs
-SYRLINKS_HANDLER = bytes([0x44, 0x53, 0x00, 0xA3])
+SYRLINKS_HANDLER_ID = bytes([0x44, 0x53, 0x00, 0xA3])
 
 # ACS Object IDs
 MGM_0_HANDLER_ID = bytes([0x44, 0x12, 0x00, 0x06])

pus_tc/syrlinks_hk_handler.py

@@ -9,6 +9,7 @@
 from tmtccmd.config.definitions import QueueCommands
 from tmtccmd.tc.definitions import TcQueueT
 from tmtccmd.tc.service_3_housekeeping import make_sid, generate_one_hk_command
+from spacepackets.ecss.tc import PusTelecommand
 
 
 class SetIds:
@@ -16,13 +17,64 @@ class SetIds:
     TX_REGISTERS_DATASET = 2
 
 
-def pack_syrlinks_hk_handler_test_into(object_id: bytearray, tc_queue: TcQueueT):
+class CommandIds:
+    SET_TX_MODE_STANDBY = bytearray([0x0, 0x0, 0x0, 0x3])
+    SET_TX_MODE_MODULATION = bytearray([0x0, 0x0, 0x0, 0x4])
+    SET_TX_MODE_CW = bytearray([0x0, 0x0, 0x0, 0x5])
+    READ_TX_STATUS = bytearray([0x0, 0x0, 0x0, 0x7])
+    READ_TX_WAVEFORM = bytearray([0x0, 0x0, 0x0, 0x8])
+    READ_TX_AGC_VALUE_HIGH_BYTE = bytearray([0x0, 0x0, 0x0, 0x9])
+    READ_TX_AGC_VALUE_LOW_BYTE = bytearray([0x0, 0x0, 0x0, 0x9])
+
+
+def pack_syrlinks_command(object_id: bytearray, tc_queue: TcQueueT, op_code: str) -> TcQueueT:
+    tc_queue.appendleft(
+        (QueueCommands.PRINT,
+         "Testing PLOC memory dumper with object id: 0x" + object_id.hex())
+    )
+
+    if op_code == "0":
+        tc_queue.appendleft((QueueCommands.PRINT, "syrlinks: Set TX mode standby"))
+        command = object_id + CommandIds.SET_TX_MODE_STANDBY
+        command = PusTelecommand(service=8, subservice=128, ssc=10, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == "1":
+        tc_queue.appendleft((QueueCommands.PRINT, "syrlinks: Set TX mode modulation"))
+        command = object_id + CommandIds.SET_TX_MODE_MODULATION
+        command = PusTelecommand(service=8, subservice=128, ssc=11, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == "2":
+        tc_queue.appendleft((QueueCommands.PRINT, "syrlinks: Set TX mode CW"))
+        command = object_id + CommandIds.SET_TX_MODE_CW
+        command = PusTelecommand(service=8, subservice=128, ssc=12, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == "3":
         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())
-
+    if op_code == "4":
         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())
+    if op_code == "5":
+        tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Read TX status"))
+        command = object_id + CommandIds.READ_TX_STATUS
+        command = PusTelecommand(service=8, subservice=128, ssc=13, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == "6":
+        tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Read TX waveform"))
+        command = object_id + CommandIds.READ_TX_WAVEFORM
+        command = PusTelecommand(service=8, subservice=128, ssc=14, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == "7":
+        tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Read TX AGC value high byte"))
+        command = object_id + CommandIds.READ_TX_AGC_VALUE_HIGH_BYTE
+        command = PusTelecommand(service=8, subservice=128, ssc=15, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == "8":
+        tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Read TX AGC value low byte"))
+        command = object_id + CommandIds.READ_TX_AGC_VALUE_LOW_BYTE
+        command = PusTelecommand(service=8, subservice=128, ssc=16, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())

pus_tc/tc_packer_hook.py

@@ -27,6 +27,7 @@ from pus_tc.ploc_memory_dumper import pack_ploc_memory_dumper_cmd
 from pus_tc.ccsds_handler import pack_ccsds_handler_test
 from pus_tc.core import pack_core_commands
 from pus_tc.star_tracker import pack_star_tracker_commands
+from pus_tc.syrlinks_hk_handler import pack_syrlinks_command
 from pus_tc.gps import pack_gps_command
 from pus_tc.acs import pack_acs_command
 from pus_tc.str_img_helper import pack_str_img_helper_command
@@ -35,7 +36,7 @@ from config.object_ids import P60_DOCK_HANDLER, PDU_1_HANDLER_ID, PDU_2_HANDLER_
     ACU_HANDLER_ID, TMP_1075_1_HANDLER_ID, TMP_1075_2_HANDLER_ID, HEATER_ID, IMTQ_HANDLER_ID, \
     PLOC_MPSOC_ID, RW1_ID, RW2_ID, RW3_ID, RW4_ID, RAD_SENSOR_ID, PLOC_SUPV_ID, PLOC_UPDATER_ID, \
     STAR_TRACKER_ID, PLOC_MEMORY_DUMPER_ID, GPS_HANDLER_0_ID, GPS_HANDLER_1_ID, CCSDS_HANDLER_ID, \
-    PDEC_HANDLER_ID, STR_IMG_HELPER_ID
+    PDEC_HANDLER_ID, STR_IMG_HELPER_ID, SYRLINKS_HANDLER_ID
 
 
 LOGGER = get_console_logger()
@@ -129,6 +130,8 @@ def pack_service_queue_user(service: Union[str, int], op_code: str, service_queu
         return pack_ccsds_handler_test(object_id=CCSDS_HANDLER_ID, tc_queue=service_queue, op_code=op_code)
     if service == CustomServiceList.PDEC_HANDLER.value:
         return pack_ccsds_handler_test(object_id=PDEC_HANDLER_ID, tc_queue=service_queue, op_code=op_code)
+    if service == CustomServiceList.SYRLINKS.value:
+        return pack_syrlinks_command(object_id=SYRLINKS_HANDLER_ID, tc_queue=service_queue, op_code=op_code)
     LOGGER.warning("Invalid Service !")