Auto test valid API states

Author: andchiind

src/isar/apis/schedule/scheduling_controller.py

@@ -17,7 +17,6 @@
 )
 from isar.config.settings import robot_settings
 from isar.services.utilities.scheduling_utilities import SchedulingUtilities
-from isar.state_machine.states_enum import States
 from robot_interface.models.mission.mission import Mission
 from robot_interface.models.mission.task import TASKS
 
@@ -54,9 +53,6 @@ def start_mission(
                 detail=error_message_no_mission_definition,
             )
 
-        state: States = self.scheduling_utilities.get_state()
-        self.scheduling_utilities.verify_state_machine_ready_to_receive_mission(state)
-
         try:
             mission: Mission = to_isar_mission(
                 start_mission_definition=mission_definition
@@ -81,32 +77,12 @@ def start_mission(
     def return_home(self) -> None:
         self.logger.info("Received request to return home")
 
-        state: States = self.scheduling_utilities.get_state()
-        self.scheduling_utilities.verify_state_machine_ready_to_receive_return_home_mission(
-            state
-        )
-
         self.scheduling_utilities.return_home()
 
     @tracer.start_as_current_span("pause_mission")
     def pause_mission(self) -> ControlMissionResponse:
         self.logger.info("Received request to pause current mission")
 
-        state: States = self.scheduling_utilities.get_state()
-
-        if state not in [
-            States.Monitor,
-            States.ReturningHome,
-        ]:
-            error_message = (
-                f"Conflict - Pause command received in invalid state - State: {state}"
-            )
-            self.logger.warning(error_message)
-            raise HTTPException(
-                status_code=HTTPStatus.CONFLICT,
-                detail=error_message,
-            )
-
         pause_mission_response: ControlMissionResponse = (
             self.scheduling_utilities.pause_mission()
         )
@@ -116,15 +92,6 @@ def pause_mission(self) -> ControlMissionResponse:
     def resume_mission(self) -> ControlMissionResponse:
         self.logger.info("Received request to resume current mission")
 
-        state: States = self.scheduling_utilities.get_state()
-
-        if state not in [States.Paused, States.ReturnHomePaused]:
-            error_message = (
-                f"Conflict - Resume command received in invalid state - State: {state}"
-            )
-            self.logger.warning(error_message)
-            raise HTTPException(status_code=HTTPStatus.CONFLICT, detail=error_message)
-
         resume_mission_response: ControlMissionResponse = (
             self.scheduling_utilities.resume_mission()
         )
@@ -143,25 +110,6 @@ def stop_mission(
 
         self.logger.info("Received request to stop current mission")
 
-        state: States = self.scheduling_utilities.get_state()
-
-        if (
-            state == States.UnknownStatus
-            or state == States.Stopping
-            or state == States.Offline
-            or state == States.Home
-            or state == States.ReturningHome
-            or state == States.GoingToLockdown
-            or state == States.GoingToRecharging
-            or state == States.Recharging
-            or state == States.Maintenance
-        ):
-            error_message = (
-                f"Conflict - Stop command received in invalid state - State: {state}"
-            )
-            self.logger.warning(error_message)
-            raise HTTPException(status_code=HTTPStatus.CONFLICT, detail=error_message)
-
         stop_mission_response: ControlMissionResponse = (
             self.scheduling_utilities.stop_mission(
                 mission_id.mission_id if mission_id.mission_id else ""
@@ -173,84 +121,34 @@ def stop_mission(
     def release_intervention_needed(self) -> None:
         self.logger.info("Received request to release intervention needed state")
 
-        state: States = self.scheduling_utilities.get_state()
-
-        if state != States.InterventionNeeded:
-            error_message = f"Conflict - Release intervention needed command received in invalid state - State: {state}"
-            self.logger.warning(error_message)
-            raise HTTPException(
-                status_code=HTTPStatus.CONFLICT,
-                detail=error_message,
-            )
-
         self.scheduling_utilities.release_intervention_needed()
         self.logger.info("Released intervention needed state successfully")
 
     @tracer.start_as_current_span("lockdown")
     def lockdown(self) -> None:
         self.logger.info("Received request to lockdown robot")
 
-        state: States = self.scheduling_utilities.get_state()
-
-        if state == States.Lockdown:
-            error_message = "Conflict - Lockdown command received in lockdown state"
-            self.logger.warning(error_message)
-            raise HTTPException(
-                status_code=HTTPStatus.CONFLICT,
-                detail=error_message,
-            )
-
         self.scheduling_utilities.lock_down_robot()
         self.logger.info("Lockdown started successfully")
 
     @tracer.start_as_current_span("release_lockdown")
     def release_lockdown(self) -> None:
         self.logger.info("Received request to release robot lockdown")
 
-        state: States = self.scheduling_utilities.get_state()
-
-        if state != States.Lockdown:
-            error_message = f"Conflict - Release lockdown command received in invalid state - State: {state}"
-            self.logger.warning(error_message)
-            raise HTTPException(
-                status_code=HTTPStatus.CONFLICT,
-                detail=error_message,
-            )
-
         self.scheduling_utilities.release_robot_lockdown()
         self.logger.info("Released lockdown successfully")
 
     @tracer.start_as_current_span("maintenance_mode")
     def set_maintenance_mode(self) -> None:
         self.logger.info("Received request to set maintenance_mode")
 
-        state: States = self.scheduling_utilities.get_state()
-
-        if state == States.Maintenance or state == States.StoppingDueToMaintenance:
-            message = f"Conflict - Call to set maintenance mode was given while in state {state}."
-            self.logger.info(message)
-            raise HTTPException(
-                status_code=HTTPStatus.CONFLICT,
-                detail=message,
-            )
-
         self.scheduling_utilities.set_maintenance_mode()
         self.logger.info("Maintenance mode has been set")
 
     @tracer.start_as_current_span("release_maintenance_mode")
     def release_maintenance_mode(self) -> None:
         self.logger.info("Received request to release robot from maintenance mode")
 
-        state: States = self.scheduling_utilities.get_state()
-
-        if state != States.Maintenance:
-            message = f"Conflict - Release maintenance mode command received in invalid state - State: {state}"
-            self.logger.info(message)
-            raise HTTPException(
-                status_code=HTTPStatus.CONFLICT,
-                detail=message,
-            )
-
         self.scheduling_utilities.release_maintenance_mode()
         self.logger.info("Maintenance mode successfully released")
 

src/isar/models/events.py

@@ -105,9 +105,14 @@ class APIEvent[T1, T2]:
     api to state machine while the response is from state machine to api.
     """
 
-    def __init__(self, name: str):
+    prioritized: bool
+
+    def __init__(self, name: str, prioritized: bool = False):
         self.request: Event[T1] = Event("api-" + name + "-request")
         self.response: Event[T2] = Event("api-" + name + "-request")
+        self.prioritized = (
+            prioritized  # For when we want to try even if the statemachine is not ready
+        )
         self.lock: Lock = Lock()
 
 
@@ -130,13 +135,13 @@ def __init__(self) -> None:
             APIEvent("release_intervention_needed")
         )
         self.send_to_lockdown: APIEvent[EmptyMessage, LockdownResponse] = APIEvent(
-            "send_to_lockdown"
+            "send_to_lockdown", prioritized=True
         )
         self.release_from_lockdown: APIEvent[EmptyMessage, EmptyMessage] = APIEvent(
             "release_from_lockdown"
         )
         self.set_maintenance_mode: APIEvent[EmptyMessage, MaintenanceResponse] = (
-            APIEvent("set_maintenance_mode")
+            APIEvent("set_maintenance_mode", prioritized=True)
         )
         self.release_from_maintenance_mode: APIEvent[EmptyMessage, EmptyMessage] = (
             APIEvent("release_from_maintenance_mode")

src/isar/modules.py

@@ -39,11 +39,19 @@ class ApplicationContainer(containers.DeclarativeContainer):
         mqtt_queue=providers.Callable(events.provided.mqtt_queue),
     )
 
+    # State machine
+    state_machine = providers.Singleton(
+        StateMachine,
+        events=events,
+        mqtt_publisher=mqtt_client,
+    )
+
     # API and controllers
     authenticator = providers.Singleton(Authenticator)
     scheduling_utilities = providers.Singleton(
         SchedulingUtilities,
         events=events,
+        state_machine=state_machine,
     )
     scheduling_controller = providers.Singleton(
         SchedulingController, scheduling_utilities=scheduling_utilities
@@ -69,13 +77,6 @@ class ApplicationContainer(containers.DeclarativeContainer):
         storage_handlers_temp.append(blob_storage)
     storage_handlers = providers.List(*storage_handlers_temp)
 
-    # State machine
-    state_machine = providers.Singleton(
-        StateMachine,
-        events=events,
-        mqtt_publisher=mqtt_client,
-    )
-
     # Robot
     robot = providers.Singleton(
         RobotService,

src/isar/services/utilities/scheduling_utilities.py

@@ -11,12 +11,11 @@
     APIEvent,
     APIRequests,
     EmptyMessage,
-    Event,
     EventConflictError,
     Events,
     EventTimeoutError,
 )
-from isar.state_machine.states_enum import States
+from isar.state_machine.state_machine import StateMachine
 from robot_interface.models.mission.mission import Mission
 
 T1 = TypeVar("T1")
@@ -32,30 +31,12 @@ class SchedulingUtilities:
     def __init__(
         self,
         events: Events,
+        state_machine: StateMachine,
     ):
         self.api_events: APIRequests = events.api_requests
-        self.state_event: Event[States] = events.state
+        self.state_machine: StateMachine = state_machine
         self.logger = logging.getLogger("api")
 
-    def get_state(self) -> States:
-        """Return the current state of the state machine
-
-        Raises
-        ------
-        HTTPException 500 Internal Server Error
-            If the current state is not available on the queue
-        """
-        current_state = self.state_event.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 verify_robot_capable_of_mission(
         self, mission: Mission, robot_capabilities: list[str]
     ) -> bool:
@@ -83,45 +64,6 @@ def verify_robot_capable_of_mission(
 
         return True
 
-    def verify_state_machine_ready_to_receive_mission(self, state: States) -> bool:
-        """Verify that the state machine is ready to receive a mission
-
-        Raises
-        ------
-        HTTPException 409 Conflict
-            If state machine is not home, robot standing still, awaiting next mission
-            return home paused or returning home and therefore cannot start a new mission
-        """
-        if (
-            state == States.Home
-            or state == States.AwaitNextMission
-            or state == States.ReturningHome
-            or state == States.ReturnHomePaused
-        ):
-            return True
-
-        error_message = f"Conflict - Robot is not home, robot standing still, awaiting next mission or returning home - State: {state}"
-        self.logger.warning(error_message)
-        raise HTTPException(status_code=HTTPStatus.CONFLICT, detail=error_message)
-
-    def verify_state_machine_ready_to_receive_return_home_mission(
-        self, state: States
-    ) -> bool:
-        """Verify that the state machine is ready to receive a return home mission
-
-        Raises
-        ------
-        HTTPException 409 Conflict
-            If state machine is not home, robot standing still or awaiting next mission
-            and therefore cannot start a new return home mission
-        """
-        if state == States.Home or state == States.AwaitNextMission:
-            return True
-
-        error_message = f"Conflict - Robot is not home, robot standing still or awaiting next mission - State: {state}"
-        self.logger.warning(error_message)
-        raise HTTPException(status_code=HTTPStatus.CONFLICT, detail=error_message)
-
     def log_mission_overview(self, mission: Mission) -> None:
         """Log an overview of the tasks in a mission"""
         log_statements: list[str] = []
@@ -433,7 +375,20 @@ def release_maintenance_mode(self) -> None:
             self.logger.warning(error_message)
             raise HTTPException(status_code=HTTPStatus.CONFLICT, detail=error_message)
 
+    def _verify_valid_state(self, api_event: APIEvent) -> None:
+        if (
+            not api_event.prioritized
+            and not self.state_machine.current_state_handles_event(api_event.request)
+        ):
+            error_message = (
+                "Conflict - Robot is not in a state where it can handle the request"
+            )
+            self.logger.warning(error_message)
+            raise HTTPException(status_code=HTTPStatus.CONFLICT, detail=error_message)
+
     def _send_command(self, input: T1, api_event: APIEvent[T1, T2]) -> T2:
+        self._verify_valid_state(api_event)
+
         if not api_event.lock.acquire(blocking=False):
             raise EventConflictError("API event has already been sent")
 

src/isar/state_machine/state.py

@@ -64,6 +64,10 @@ def get_event_timer_by_name(
         )
         return filtered_timers[0] if len(filtered_timers) > 0 else None
 
+    def handles_event(self, event: Event) -> bool:
+        allowed_events: list[Event] = [m.event for m in self.event_handler_mappings]
+        return event in allowed_events
+
     def run(self) -> Transition | None:
         timers = deepcopy(self.timers)
         entered_time = time.time()