Added message queue to communications, moved checking to inside run() function, adjusted flight interface test to reflect new inputs

Author: HermanG05

modules/communications/communications.py

@@ -25,7 +25,6 @@ def create(
         cls,
         home_position: position_global.PositionGlobal,
         local_logger: logger.Logger,
-        worker_id: worker_enum.WorkerEnum,
     ) -> "tuple[True, Communications] | tuple[False, None]":
         """
         Logs data and forwards it.
@@ -35,14 +34,13 @@ def create(
         Returns: Success, class object.
         """
 
-        return True, Communications(cls.__create_key, home_position, local_logger, worker_id)
+        return True, Communications(cls.__create_key, home_position, local_logger)
 
     def __init__(
         self,
         class_private_create_key: object,
         home_position: position_global.PositionGlobal,
         local_logger: logger.Logger,
-        worker_id: worker_enum.WorkerEnum,
     ) -> None:
         """
         Private constructor, use create() method.
@@ -51,7 +49,6 @@ def __init__(
 
         self.__home_position = home_position
         self.__logger = local_logger
-        self.__worker_id = worker_id
 
     def run(
         self,
@@ -96,7 +93,7 @@ def run(
         for object in object_in_world_global:
 
             result, message = message_encoding_decoding.encode_position_global(
-                self.__worker_id, object
+                worker_enum.WorkerEnum.COMMUNICATIONS_WORKER, object
             )
             if not result:
                 self.__logger.warning("Conversion from PositionGlobal to bytes failed", True)

modules/communications/communications_worker.py

@@ -22,6 +22,7 @@ def communications_worker(
     home_position_queue: queue_proxy_wrapper.QueueProxyWrapper,
     input_queue: queue_proxy_wrapper.QueueProxyWrapper,
     output_queue: queue_proxy_wrapper.QueueProxyWrapper,
+    message_output_queue: queue_proxy_wrapper.QueueProxyWrapper,
     controller: worker_controller.WorkerController,
 ) -> None:
     """
@@ -33,7 +34,6 @@ def communications_worker(
     """
 
     worker_name = pathlib.Path(__file__).stem
-    worker_id = worker_enum.WorkerEnum.COMMUNICATIONS_WORKER
     process_id = os.getpid()
     result, local_logger = logger.Logger.create(f"{worker_name}_{process_id}", True)
     if not result:
@@ -54,7 +54,7 @@ def communications_worker(
 
     local_logger.info(f"Home position received: {home_position}", True)
 
-    result, comm = communications.Communications.create(home_position, local_logger, worker_id)
+    result, comm = communications.Communications.create(home_position, local_logger)
     if not result:
         local_logger.error("Worker failed to create class object", True)
         return
@@ -89,16 +89,18 @@ def communications_worker(
             continue
 
         result, metadata = metadata_encoding_decoding.encode_metadata(
-            worker_id, len(list_of_messages)
+            worker_enum.WorkerEnum.COMMUNICATIONS_WORKER, len(list_of_messages)
         )
         if not result:
             local_logger.error("Failed to encode metadata", True)
             continue
 
         output_queue.queue.put(metadata)
+        message_output_queue.queue.put(metadata)
 
         for message in list_of_messages:
 
             time.sleep(period)
 
             output_queue.queue.put(message)
+            message_output_queue.queue.put(message)

modules/flight_interface/flight_interface.py

@@ -103,9 +103,10 @@ def run(self, message: bytes) -> "tuple[bool, odometry_and_time.OdometryAndTime
 
         self.__logger.info(str(odometry_and_time_object), True)
 
-        result = self.controller.send_statustext_msg(message)
-        if not result:
-            self.__logger.error("Failed to send statustext message", True)
+        if not isinstance(message, bytes):
+            self.__logger.warning(f"Skipping unexpected input: {message}")
+        else:
+            result = self.controller.send_statustext_msg(message)
 
         return True, odometry_and_time_object
 

modules/flight_interface/flight_interface_worker.py

@@ -19,9 +19,9 @@ def flight_interface_worker(
     baud_rate: int,
     period: float,
     input_queue: queue_proxy_wrapper.QueueProxyWrapper,
+    coordinates_input_queue: queue_proxy_wrapper.QueueProxyWrapper,
     output_queue: queue_proxy_wrapper.QueueProxyWrapper,
     communications_output_queue: queue_proxy_wrapper.QueueProxyWrapper,
-    coordinates_input_queue: queue_proxy_wrapper.QueueProxyWrapper,
     controller: worker_controller.WorkerController,
 ) -> None:
     """
@@ -67,13 +67,8 @@ def flight_interface_worker(
         try:
             coordinate = coordinates_input_queue.queue.get_nowait()
         except queue.Empty:
-            local_logger.warning("No more coordinates to process")
             coordinate = None
 
-        if not isinstance(coordinate, bytes):
-            local_logger.warning(f"Skipping unexpected input: {coordinate}")
-            continue
-
         result, value = interface.run(coordinate)
         if not result:
             continue

tests/integration/test_flight_interface_worker.py

@@ -11,35 +11,12 @@
 from modules import decision_command
 from utilities.workers import queue_proxy_wrapper
 from utilities.workers import worker_controller
-from modules.common.modules import position_global
-from modules.common.modules.data_encoding import message_encoding_decoding
-from modules.common.modules.data_encoding import worker_enum
+
 
 MAVLINK_CONNECTION_ADDRESS = "tcp:localhost:14550"
 FLIGHT_INTERFACE_TIMEOUT = 10.0  # seconds
 FLIGHT_INTERFACE_BAUD_RATE = 57600  # symbol rate
 FLIGHT_INTERFACE_WORKER_PERIOD = 0.1  # seconds
-WORK_COUNT = 4
-COMMUNICATIONS_WORKER_ID = worker_enum.WorkerEnum.COMMUNICATIONS_WORKER
-
-
-def simulate_communications_worker(
-    in_queue: queue_proxy_wrapper.QueueProxyWrapper,
-    data_point: position_global.PositionGlobal,
-) -> None:
-    """
-    Encode coordinates and place into queue.
-    """
-    result, message = message_encoding_decoding.encode_position_global(
-        COMMUNICATIONS_WORKER_ID, data_point
-    )
-    assert result
-    assert message is not None
-
-    in_queue.queue.put(message)
-
-    return
-
 
 def apply_decision_test(
     in_queue: queue_proxy_wrapper.QueueProxyWrapper,
@@ -129,7 +106,6 @@ def main() -> int:
     home_position_out_queue = queue_proxy_wrapper.QueueProxyWrapper(mp_manager)
     in_queue = queue_proxy_wrapper.QueueProxyWrapper(mp_manager)
     communications_in_queue = queue_proxy_wrapper.QueueProxyWrapper(mp_manager)
-    communications_out_queue = queue_proxy_wrapper.QueueProxyWrapper(mp_manager)
 
     worker = mp.Process(
         target=flight_interface_worker.flight_interface_worker,
@@ -139,9 +115,9 @@ def main() -> int:
             FLIGHT_INTERFACE_BAUD_RATE,
             FLIGHT_INTERFACE_WORKER_PERIOD,
             in_queue,  # Added input_queue
+            communications_in_queue,
             out_queue,
             home_position_out_queue,
-            communications_in_queue,
             controller,
         ),
     )
@@ -155,28 +131,6 @@ def main() -> int:
     home_position = home_position_out_queue.queue.get()
     assert home_position is not None
 
-    data_points = [
-        position_global.PositionGlobal.create(43.471468, -80.544205, 335),
-        position_global.PositionGlobal.create(43.6629, -79.3957, 105),
-        position_global.PositionGlobal.create(43.2609, -79.9192, 100),
-        position_global.PositionGlobal.create(43.7735, -79.5019, 170),
-    ]
-
-    # Simulate communications worker
-    for i in range(0, WORK_COUNT):
-        simulate_communications_worker(communications_in_queue, home_position, data_points[i])
-
-    # Test flight interface worker sending statustext messages
-    for i in range(0, WORK_COUNT):
-        try:
-            input_data: bytes = communications_out_queue.queue.get_nowait()
-            assert input_data is not None
-        except queue.Empty:
-            print("Output queue has no more messages to process, exiting")
-            break
-
-    assert communications_out_queue.queue.empty()
-
     # Run the apply_decision tests
     test_result = apply_decision_test(in_queue, out_queue)
     if not test_result: