Init commit

tmtc/pus_tc/service_3_housekeeping.py

@@ -0,0 +1,162 @@
+from tmtccmd.core.definitions import QueueCommands
+from tmtccmd.core.object_id_manager import get_object_id
+from tmtccmd.pus_tc.service_200_mode import pack_mode_data
+from tmtccmd.pus_tc.service_20_parameter import pack_boolean_parameter_command
+from tmtccmd.pus_tc.service_3_housekeeping import make_sid, generate_one_hk_command, \
+    Srv3Subservice
+from tmtccmd.ecss.tc import PusTelecommand
+from tmtccmd.pus_tc.definitions import TcQueueT
+from tmtccmd.pus_tc.service_8_functional_cmd import generate_action_command
+
+from config.object_ids import ObjectIds
+
+# Object IDs
+TEST_DEVICE_0_ID = bytearray()
+TEST_DEVICE_1_ID = bytearray()
+
+# Set IDs
+TEST_SET_ID = 0
+
+# Action IDs
+TEST_NOTIFICATION_ACTION_ID = 3
+
+# Parameters
+PARAM_ACTIVATE_CHANGING_DATASETS = 4
+
+
+def pack_service_3_commands_into(tc_queue: TcQueueT, op_code: str):
+    global TEST_DEVICE_0_ID, TEST_DEVICE_1_ID
+    if TEST_DEVICE_0_ID == bytearray() or TEST_DEVICE_1_ID == bytearray():
+        TEST_DEVICE_0_ID = get_object_id(ObjectIds.TEST_DEVICE_0)
+        TEST_DEVICE_1_ID = get_object_id(ObjectIds.TEST_DEVICE_1)
+    current_ssc = 3000
+    # TODO: Import this from config instead
+    device_idx = 0
+    if device_idx == 0:
+        object_id = TEST_DEVICE_0_ID
+    else:
+        object_id = TEST_DEVICE_1_ID
+
+    if op_code == "0":
+        # This will pack all the tests
+        pack_service_3_test_info(tc_queue=tc_queue, init_ssc=current_ssc, object_id=object_id,
+                                 device_idx=device_idx)
+    elif op_code == "1":
+        # Extremely simple, generate one HK packet
+        pack_gen_one_hk_command(tc_queue=tc_queue, device_idx=device_idx, init_ssc=current_ssc,
+                                object_id=object_id)
+    elif op_code == "2":
+        # Housekeeping basic test
+        pack_housekeeping_basic_test(tc_queue=tc_queue, object_id=object_id, init_ssc=current_ssc)
+    elif op_code == "3":
+        # Notification demo
+        pack_notification_basic_test(tc_queue=tc_queue, object_id=object_id, init_ssc=current_ssc)
+
+
+def pack_service_3_test_info(tc_queue: TcQueueT, device_idx: int, object_id: bytearray,
+                             init_ssc: int):
+    tc_queue.appendleft((QueueCommands.PRINT, "Service 3 (Housekeeping Service): All tests"))
+    current_ssc = init_ssc
+
+    current_ssc += pack_gen_one_hk_command(
+        tc_queue=tc_queue, device_idx=device_idx, object_id=object_id, init_ssc=current_ssc
+    )
+    current_ssc += pack_housekeeping_basic_test(
+       tc_queue=tc_queue, object_id=object_id, init_ssc=current_ssc
+    )
+    current_ssc += pack_notification_basic_test(
+        tc_queue=tc_queue, object_id=object_id, init_ssc=current_ssc, enable_normal_mode=False
+    )
+
+
+def pack_gen_one_hk_command(
+        tc_queue: TcQueueT, device_idx: int, init_ssc: int, object_id: bytearray
+) -> int:
+    test_sid = make_sid(object_id=object_id, set_id=TEST_SET_ID)
+    tc_queue.appendleft(
+        (QueueCommands.PRINT, f"Service 3 Test: Generate one test set packet for "
+                              f"test device {device_idx}")
+    )
+    command = generate_one_hk_command(ssc=init_ssc, sid=test_sid)
+    init_ssc += 1
+    tc_queue.appendleft(command.pack_command_tuple())
+    return init_ssc
+
+
+def pack_housekeeping_basic_test(
+        tc_queue: TcQueueT, object_id: bytearray, init_ssc: int, enable_normal_mode: bool = True
+) -> int:
+    """
+    This basic test will request one HK packet, then it will enable periodic packets and listen
+    to the periodic packets for a few seconds. After that, HK packets will be disabled again.
+    """
+    test_sid = make_sid(object_id=object_id, set_id=TEST_SET_ID)
+    current_ssc = init_ssc
+    # Enable changing datasets via parameter service (Service 20)
+    tc_queue.appendleft((QueueCommands.PRINT, "Service 3 Test: Performing basic HK tests"))
+
+    if enable_normal_mode:
+        # Set mode normal so that sets are changed/read regularly
+        tc_queue.appendleft((QueueCommands.PRINT, "Service 3 Test: Set Normal Mode"))
+        mode_data = pack_mode_data(object_id, 2, 0)
+        command = PusTelecommand(service=200, subservice=1, ssc=current_ssc, app_data=mode_data)
+        current_ssc += 1
+        tc_queue.appendleft(command.pack_command_tuple())
+
+    tc_queue.appendleft((QueueCommands.PRINT, "Enabling changing datasets"))
+    command = pack_boolean_parameter_command(
+        object_id=object_id, domain_id=0, unique_id=PARAM_ACTIVATE_CHANGING_DATASETS,
+        parameter=True, ssc=current_ssc
+    )
+    current_ssc += 1
+    tc_queue.appendleft(command.pack_command_tuple())
+
+    # Enable periodic reporting
+    tc_queue.appendleft((QueueCommands.PRINT,
+                         "Enabling periodic thermal sensor packet generation: "))
+    command = PusTelecommand(service=3, subservice=Srv3Subservice.ENABLE_PERIODIC_HK_GEN.value,
+                             ssc=current_ssc, app_data=test_sid)
+    current_ssc += 1
+    tc_queue.appendleft(command.pack_command_tuple())
+
+    tc_queue.appendleft((QueueCommands.WAIT, 2.0))
+
+    # Disable periodic reporting
+    tc_queue.appendleft((QueueCommands.PRINT,
+                         "Disabling periodic thermal sensor packet generation: "))
+    command = PusTelecommand(service=3, subservice=Srv3Subservice.DISABLE_PERIODIC_HK_GEN.value,
+                             ssc=current_ssc, app_data=test_sid)
+    current_ssc += 1
+    tc_queue.appendleft(command.pack_command_tuple())
+
+    # Disable changing datasets via parameter service (Service 20)
+    tc_queue.appendleft((QueueCommands.PRINT, "Disabling changing datasets"))
+    command = pack_boolean_parameter_command(
+        object_id=object_id, domain_id=0, unique_id=PARAM_ACTIVATE_CHANGING_DATASETS,
+        parameter=False, ssc=current_ssc
+    )
+    current_ssc += 1
+    tc_queue.appendleft(command.pack_command_tuple())
+    return current_ssc
+
+
+def pack_notification_basic_test(tc_queue: TcQueueT, object_id: bytearray, init_ssc: int,
+                                 enable_normal_mode: bool = True) -> int:
+    current_ssc = init_ssc
+    tc_queue.appendleft((QueueCommands.PRINT, "Service 3 Test: Performing notification tests"))
+
+    if enable_normal_mode:
+        # Set mode normal so that sets are changed/read regularly
+        tc_queue.appendleft((QueueCommands.PRINT, "Service 3 Test: Set Normal Mode"))
+        mode_data = pack_mode_data(object_id, 2, 0)
+        command = PusTelecommand(service=200, subservice=1, ssc=current_ssc, app_data=mode_data)
+        current_ssc += 1
+        tc_queue.appendleft(command.pack_command_tuple())
+
+    tc_queue.appendleft((QueueCommands.PRINT, "Triggering notification"))
+    command = generate_action_command(
+        object_id=object_id, action_id=TEST_NOTIFICATION_ACTION_ID, ssc=current_ssc
+    )
+    tc_queue.appendleft(command.pack_command_tuple())
+    current_ssc += 1
+    return current_ssc