Remove queue_utils

Author: andchiind

src/isar/models/communication/queues/events.py

@@ -1,6 +1,6 @@
 from collections import deque
 from queue import Empty, Queue
-from typing import TypeVar
+from typing import Optional, TypeVar
 
 from transitions import State
 
@@ -17,9 +17,23 @@ class Event(Queue[T]):
     def __init__(self) -> None:
         super().__init__(maxsize=1)
 
-    def check(self) -> T:
+    def trigger_event(self, data: T) -> None:
+        self.put(data)
+
+    def consume_event(self) -> Optional[T]:
+        try:
+            return self.get(block=False)
+        except Empty:
+            return None
+
+    def has_event(self) -> bool:
+        return (
+            self.qsize() != 0
+        )  # Queue size is not reliable, but should be sufficient for this case
+
+    def check(self) -> Optional[T]:
         if not self._qsize():
-            raise Empty
+            return None
         with self.mutex:
             queueList = list(self.queue)
             return queueList.pop()
@@ -65,7 +79,7 @@ def __init__(self) -> None:
 class StateMachineEvents:
     def __init__(self) -> None:
         self.start_mission: Event[Mission] = Event()
-        self.stop_mission: Event[str] = Event()
+        self.stop_mission: Event[bool] = Event()
         self.pause_mission: Event[bool] = Event()
         self.task_status_request: Event[str] = Event()
 

src/isar/models/communication/queues/queue_utils.py

@@ -9,7 +9,6 @@
     SharedState,
     StateMachineEvents,
 )
-from isar.models.communication.queues.queue_utils import check_for_event, trigger_event
 from isar.robot.robot_start_mission import RobotStartMissionThread
 from isar.robot.robot_status import RobotStatusThread
 from isar.robot.robot_stop_mission import RobotStopMissionThread
@@ -55,7 +54,7 @@ def stop(self) -> None:
         self.start_mission_thread = None
 
     def _start_mission_event_handler(self, event: Event[Mission]) -> None:
-        start_mission = check_for_event(event)
+        start_mission = event.consume_event()
         if start_mission is not None:
             if (
                 self.start_mission_thread is not None
@@ -74,7 +73,7 @@ def _start_mission_event_handler(self, event: Event[Mission]) -> None:
             self.start_mission_thread.start()
 
     def _task_status_request_handler(self, event: Event[str]) -> None:
-        task_id: str = check_for_event(event)
+        task_id: str = event.consume_event()
         if task_id:
             self.robot_task_status_thread = RobotTaskStatusThread(
                 self.robot_service_events,
@@ -84,8 +83,8 @@ def _task_status_request_handler(self, event: Event[str]) -> None:
             )
             self.robot_task_status_thread.start()
 
-    def _stop_mission_request_handler(self, event: Event[str]) -> None:
-        if check_for_event(event):
+    def _stop_mission_request_handler(self, event: Event[bool]) -> None:
+        if event.consume_event():
             if (
                 self.stop_mission_thread is not None
                 and self.stop_mission_thread.is_alive()
@@ -103,8 +102,8 @@ def _stop_mission_request_handler(self, event: Event[str]) -> None:
                     error_reason=ErrorReason.RobotStillStartingMissionException,
                     error_description=error_description,
                 )
-                trigger_event(
-                    self.robot_service_events.mission_failed_to_stop, error_message
+                self.robot_service_events.mission_failed_to_stop.trigger_event(
+                    error_message
                 )
                 return
             self.stop_mission_thread = RobotStopMissionThread(

src/isar/robot/robot.py

@@ -3,10 +3,6 @@
 
 from isar.config.settings import settings
 from isar.models.communication.queues.events import RobotServiceEvents
-from isar.models.communication.queues.queue_utils import (
-    trigger_event,
-    trigger_event_without_data,
-)
 from robot_interface.models.exceptions.robot_exceptions import (
     ErrorMessage,
     RobotException,
@@ -44,13 +40,13 @@ def run(self):
                     self.logger.error(
                         f"Mission is infeasible and cannot be scheduled because: {e.error_description}"
                     )
-                    trigger_event(
-                        self.robot_service_events.mission_failed,
+                    self.robot_service_events.mission_failed.trigger_event(
                         ErrorMessage(
                             error_reason=e.error_reason,
                             error_description=e.error_description,
-                        ),
+                        )
                     )
+
                     break
                 except RobotException as e:
                     retries += 1
@@ -66,25 +62,23 @@ def run(self):
                             f"{e.error_description}"
                         )
 
-                        trigger_event(
-                            self.robot_service_events.mission_failed,
+                        self.robot_service_events.mission_failed.trigger_event(
                             ErrorMessage(
                                 error_reason=e.error_reason,
                                 error_description=e.error_description,
-                            ),
+                            )
                         )
                         break
 
                     continue
 
                 started_mission = True
         except RobotInfeasibleMissionException as e:
-            trigger_event(
-                self.robot_service_events.mission_failed,
+            self.robot_service_events.mission_failed.trigger_event(
                 ErrorMessage(
                     error_reason=e.error_reason, error_description=e.error_description
                 ),
             )
 
         if started_mission:
-            trigger_event_without_data(self.robot_service_events.mission_started)
+            self.robot_service_events.mission_started.trigger_event(True)

src/isar/robot/robot_start_mission.py

@@ -4,7 +4,6 @@
 
 from isar.config.settings import settings
 from isar.models.communication.queues.events import SharedState
-from isar.models.communication.queues.queue_utils import update_shared_state
 from robot_interface.models.exceptions.robot_exceptions import RobotException
 from robot_interface.robot_interface import RobotInterface
 
@@ -49,7 +48,7 @@ def run(self):
                 self.last_robot_status_poll_time = time.time()
 
                 robot_status = self.robot.robot_status()
-                update_shared_state(self.shared_state.robot_status, robot_status)
+                self.shared_state.robot_status.update(robot_status)
             except RobotException as e:
                 self.logger.error(f"Failed to retrieve robot status: {e}")
                 continue

src/isar/robot/robot_status.py

@@ -5,10 +5,6 @@
 
 from isar.config.settings import settings
 from isar.models.communication.queues.events import RobotServiceEvents
-from isar.models.communication.queues.queue_utils import (
-    trigger_event,
-    trigger_event_without_data,
-)
 from robot_interface.models.exceptions.robot_exceptions import (
     ErrorMessage,
     RobotActionException,
@@ -51,9 +47,7 @@ def run(self) -> None:
                 time.sleep(settings.FSM_SLEEP_TIME)
                 continue
 
-            trigger_event_without_data(
-                self.robot_service_events.mission_successfully_stopped
-            )
+            self.robot_service_events.mission_successfully_stopped.trigger_event(True)
             return
 
         error_description = (
@@ -68,4 +62,4 @@ def run(self) -> None:
             error_description=error_description,
         )
 
-        trigger_event(self.robot_service_events.mission_failed_to_stop, error_message)
+        self.robot_service_events.mission_failed_to_stop.trigger_event(error_message)

src/isar/robot/robot_stop_mission.py

@@ -5,7 +5,6 @@
 
 from isar.config.settings import settings
 from isar.models.communication.queues.events import RobotServiceEvents
-from isar.models.communication.queues.queue_utils import trigger_event
 from isar.services.utilities.threaded_request import ThreadedRequest
 from robot_interface.models.exceptions.robot_exceptions import (
     ErrorMessage,
@@ -82,10 +81,7 @@ def run(self) -> None:
                 )
                 break
 
-            trigger_event(self.robot_service_events.task_status_updated, task_status)
+            self.robot_service_events.task_status_updated.trigger_event(task_status)
             return
 
-        trigger_event(
-            self.robot_service_events.task_status_failed,
-            failed_task_error,
-        )
+        self.robot_service_events.task_status_failed.trigger_event(failed_task_error)

src/isar/robot/robot_task_status.py

@@ -1,7 +1,6 @@
 import logging
 from copy import deepcopy
 from http import HTTPStatus
-from queue import Empty
 from typing import Any, List
 
 from fastapi import HTTPException
@@ -55,16 +54,16 @@ def get_state(self) -> States:
         HTTPException 500 Internal Server Error
             If the current state is not available on the queue
         """
-        try:
-            return self.shared_state.state.check()
-        except Empty:
+        current_state = self.shared_state.state.check()
+        if current_state is None:
             error_message: str = (
                 "Internal Server Error - Current state of the state machine is unknown"
             )
             self.logger.error(error_message)
             raise HTTPException(
                 status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=error_message
             )
+        return current_state
 
     def get_mission(self, mission_id: str) -> Mission:
         """Get the mission with mission_id from the current mission planner

src/isar/services/utilities/scheduling_utilities.py

@@ -15,7 +15,6 @@
     TaskSelectorStop,
 )
 from isar.models.communication.queues.events import Events, SharedState
-from isar.models.communication.queues.queue_utils import update_shared_state
 from isar.state_machine.states.await_next_mission import AwaitNextMission
 from isar.state_machine.states.blocked_protective_stop import BlockedProtectiveStop
 from isar.state_machine.states.home import Home
@@ -189,7 +188,7 @@ def iterate_current_task(self):
     def update_state(self):
         """Updates the current state of the state machine."""
         self.current_state = States(self.state)  # type: ignore
-        update_shared_state(self.shared_state.state, self.current_state)
+        self.shared_state.state.update(self.current_state)
         self._log_state_transition(self.current_state)
         self.logger.info("State: %s", self.current_state)
         self.publish_status()
@@ -208,9 +207,7 @@ def start_mission(self, mission: Mission):
         self.task_selector.initialize(tasks=self.current_mission.tasks)
 
     def send_task_status(self):
-        update_shared_state(
-            self.shared_state.state_machine_current_task, self.current_task
-        )
+        self.shared_state.state_machine_current_task.update(self.current_task)
 
     def report_task_status(self, task: Task) -> None:
         if task.status == TaskStatus.Failed:

src/isar/state_machine/state_machine.py

@@ -2,7 +2,6 @@
 
 from isar.eventhandlers.eventhandler import EventHandlerBase, EventHandlerMapping
 from isar.models.communication.queues.events import Event
-from isar.models.communication.queues.queue_utils import check_shared_state
 from robot_interface.models.mission.status import RobotStatus
 
 if TYPE_CHECKING:
@@ -15,7 +14,7 @@ def __init__(self, state_machine: "StateMachine"):
         shared_state = state_machine.shared_state
 
         def _robot_status_event_handler(event: Event[RobotStatus]):
-            robot_status: RobotStatus = check_shared_state(event)
+            robot_status: RobotStatus = event.check()
             if robot_status != RobotStatus.BlockedProtectiveStop:
                 return state_machine.robot_status_changed  # type: ignore
             return None