Remove unecessary state class inheritance

Author: andchiind

src/isar/state_machine/state.py

@@ -1,17 +1,15 @@
 import logging
 import time
-from abc import ABC
 from collections.abc import Callable
 from copy import deepcopy
 from dataclasses import dataclass
-from typing import Any, TypeVar
+from typing import Any
 
 from isar.config.settings import settings
 from isar.models.events import EmptyMessage, Event, Events
 from isar.state_machine.states_enum import States
 
-T_state_co = TypeVar("T_state_co", bound="State", covariant=True)
-Transition = Callable[[Events], T_state_co]
+Transition = Callable[[Events], "State"]
 
 
 @dataclass
@@ -27,7 +25,7 @@ class TimeoutHandlerMapping:
     handler: Callable[[], Transition | None]
 
 
-class State(ABC):
+class State:
     def __init__(
         self,
         signal_exit_event: Event[EmptyMessage],

src/isar/state_machine/states/await_next_mission.py

@@ -16,59 +16,55 @@
 from robot_interface.models.mission.mission import Mission
 
 
-class AwaitNextMission(State):
+def AwaitNextMission(events: Events) -> State:
 
-    def __init__(self, events: Events):
-
-        event_handlers: list[EventHandlerMapping] = [
-            EventHandlerMapping[Mission](
-                event=events.api_requests.start_mission.request,
-                handler=lambda mission: Monitor.transition_and_start_mission(
-                    mission, True
-                ),
-            ),
-            EventHandlerMapping[EmptyMessage](
-                event=events.api_requests.return_home.request,
-                handler=lambda _: ReturningHome.transition_and_start_mission(True),
-            ),
-            EventHandlerMapping[str](
-                event=events.api_requests.stop_mission.request,
-                handler=lambda mission_id: Stopping.transition_and_trigger_stop_and_respond_to_API(
-                    mission_id
-                ),
-            ),
-            EventHandlerMapping[EmptyMessage](
-                event=events.api_requests.send_to_lockdown.request,
-                handler=lambda _: GoingToLockdown.transition_and_start_mission_and_report_to_api(),
+    event_handlers: list[EventHandlerMapping] = [
+        EventHandlerMapping[Mission](
+            event=events.api_requests.start_mission.request,
+            handler=lambda mission: Monitor.transition_and_start_mission(mission, True),
+        ),
+        EventHandlerMapping[EmptyMessage](
+            event=events.api_requests.return_home.request,
+            handler=lambda _: ReturningHome.transition_and_start_mission(True),
+        ),
+        EventHandlerMapping[str](
+            event=events.api_requests.stop_mission.request,
+            handler=lambda mission_id: Stopping.transition_and_trigger_stop_and_respond_to_API(
+                mission_id
             ),
-            EventHandlerMapping[EmptyMessage](
-                event=events.robot_service_events.battery_below_mission_threshold,
-                handler=lambda _: GoingToRecharging.transition_and_start_return_home(),
-            ),
-            EventHandlerMapping[EmptyMessage](
-                event=events.api_requests.set_maintenance_mode.request,
-                handler=lambda _: Maintenance.transition_and_reply_to_API(),
-            ),
-        ]
+        ),
+        EventHandlerMapping[EmptyMessage](
+            event=events.api_requests.send_to_lockdown.request,
+            handler=lambda _: GoingToLockdown.transition_and_start_mission_and_report_to_api(),
+        ),
+        EventHandlerMapping[EmptyMessage](
+            event=events.robot_service_events.battery_below_mission_threshold,
+            handler=lambda _: GoingToRecharging.transition_and_start_return_home(),
+        ),
+        EventHandlerMapping[EmptyMessage](
+            event=events.api_requests.set_maintenance_mode.request,
+            handler=lambda _: Maintenance.transition_and_reply_to_API(),
+        ),
+    ]
 
-        timers: list[TimeoutHandlerMapping] = [
-            TimeoutHandlerMapping(
-                name="should_return_home_timer",
-                timeout_in_seconds=settings.RETURN_HOME_DELAY,
-                handler=lambda: ReturningHome.transition_and_start_mission(),
-            )
-        ]
-
-        super().__init__(
-            state_name=States.AwaitNextMission,
-            signal_exit_event=events.signal_state_machine_exit,
-            event_handler_mappings=event_handlers,
-            timers=timers,
+    timers: list[TimeoutHandlerMapping] = [
+        TimeoutHandlerMapping(
+            name="should_return_home_timer",
+            timeout_in_seconds=settings.RETURN_HOME_DELAY,
+            handler=lambda: ReturningHome.transition_and_start_mission(),
         )
+    ]
+
+    return State(
+        state_name=States.AwaitNextMission,
+        signal_exit_event=events.signal_state_machine_exit,
+        event_handler_mappings=event_handlers,
+        timers=timers,
+    )
 
 
-def transition() -> Transition[AwaitNextMission]:
-    def _transition(events: Events) -> AwaitNextMission:
+def transition() -> Transition:
+    def _transition(events: Events) -> State:
         return AwaitNextMission(events)
 
     return _transition

src/isar/state_machine/states/going_to_lockdown.py

@@ -8,37 +8,33 @@
 from robot_interface.models.mission.mission import ReturnHomeMission
 
 
-class GoingToLockdown(State):
-
-    def __init__(self, events: Events):
-
-        event_handlers: list[EventHandlerMapping] = [
-            EventHandlerMapping[ErrorMessage](
-                event=events.robot_service_events.mission_failed,
-                handler=lambda _: InterventionNeeded.transition(
-                    "Lockdown mission failed"
-                ),
-            ),
-            EventHandlerMapping[EmptyMessage](
-                event=events.robot_service_events.mission_failed_to_resume,
-                handler=lambda _: InterventionNeeded.transition(
-                    "Failed to resume return to home mission"
-                ),
-            ),
-            EventHandlerMapping[EmptyMessage](
-                event=events.robot_service_events.mission_succeeded,
-                handler=lambda _: Lockdown.transition_without_responding_to_api(),
+def GoingToLockdown(events: Events) -> State:
+
+    event_handlers: list[EventHandlerMapping] = [
+        EventHandlerMapping[ErrorMessage](
+            event=events.robot_service_events.mission_failed,
+            handler=lambda _: InterventionNeeded.transition("Lockdown mission failed"),
+        ),
+        EventHandlerMapping[EmptyMessage](
+            event=events.robot_service_events.mission_failed_to_resume,
+            handler=lambda _: InterventionNeeded.transition(
+                "Failed to resume return to home mission"
             ),
-        ]
-        super().__init__(
-            state_name=States.GoingToLockdown,
-            signal_exit_event=events.signal_state_machine_exit,
-            event_handler_mappings=event_handlers,
-        )
-
-
-def transition_and_start_mission_and_report_to_api() -> Transition[GoingToLockdown]:
-    def _transition(events: Events) -> GoingToLockdown:
+        ),
+        EventHandlerMapping[EmptyMessage](
+            event=events.robot_service_events.mission_succeeded,
+            handler=lambda _: Lockdown.transition_without_responding_to_api(),
+        ),
+    ]
+    return State(
+        state_name=States.GoingToLockdown,
+        signal_exit_event=events.signal_state_machine_exit,
+        event_handler_mappings=event_handlers,
+    )
+
+
+def transition_and_start_mission_and_report_to_api() -> Transition:
+    def _transition(events: Events) -> State:
         events.api_requests.send_to_lockdown.response.trigger_event(
             LockdownResponse(lockdown_started=True)
         )
@@ -52,8 +48,8 @@ def _transition(events: Events) -> GoingToLockdown:
     return _transition
 
 
-def transition_to_existing_mission_and_report_to_api() -> Transition[GoingToLockdown]:
-    def _transition(events: Events) -> GoingToLockdown:
+def transition_to_existing_mission_and_report_to_api() -> Transition:
+    def _transition(events: Events) -> State:
         events.api_requests.send_to_lockdown.response.trigger_event(
             LockdownResponse(lockdown_started=True)
         )

src/isar/state_machine/states/going_to_recharging.py

@@ -8,35 +8,33 @@
 from robot_interface.models.mission.mission import ReturnHomeMission
 
 
-class GoingToRecharging(State):
+def GoingToRecharging(events: Events) -> State:
 
-    def __init__(self, events: Events):
-
-        event_handlers: list[EventHandlerMapping] = [
-            EventHandlerMapping[ErrorMessage](
-                event=events.robot_service_events.mission_failed,
-                handler=lambda _: InterventionNeeded.transition(
-                    "Return home to recharge failed"
-                ),
-            ),
-            EventHandlerMapping[EmptyMessage](
-                event=events.robot_service_events.mission_succeeded,
-                handler=lambda _: Recharging.transition(),
-            ),
-            EventHandlerMapping[EmptyMessage](
-                event=events.api_requests.send_to_lockdown.request,
-                handler=lambda _: GoingToLockdown.transition_to_existing_mission_and_report_to_api(),
+    event_handlers: list[EventHandlerMapping] = [
+        EventHandlerMapping[ErrorMessage](
+            event=events.robot_service_events.mission_failed,
+            handler=lambda _: InterventionNeeded.transition(
+                "Return home to recharge failed"
             ),
-        ]
-        super().__init__(
-            state_name=States.GoingToRecharging,
-            signal_exit_event=events.signal_state_machine_exit,
-            event_handler_mappings=event_handlers,
-        )
-
-
-def transition_and_start_return_home() -> Transition[GoingToRecharging]:
-    def _transition(events: Events) -> GoingToRecharging:
+        ),
+        EventHandlerMapping[EmptyMessage](
+            event=events.robot_service_events.mission_succeeded,
+            handler=lambda _: Recharging.transition(),
+        ),
+        EventHandlerMapping[EmptyMessage](
+            event=events.api_requests.send_to_lockdown.request,
+            handler=lambda _: GoingToLockdown.transition_to_existing_mission_and_report_to_api(),
+        ),
+    ]
+    return State(
+        state_name=States.GoingToRecharging,
+        signal_exit_event=events.signal_state_machine_exit,
+        event_handler_mappings=event_handlers,
+    )
+
+
+def transition_and_start_return_home() -> Transition:
+    def _transition(events: Events) -> State:
         events.robot_service_events.mission_failed.clear_event()
         events.robot_service_events.mission_succeeded.clear_event()
 
@@ -46,8 +44,8 @@ def _transition(events: Events) -> GoingToRecharging:
     return _transition
 
 
-def transition_to_existing_mission() -> Transition[GoingToRecharging]:
-    def _transition(events: Events) -> GoingToRecharging:
+def transition_to_existing_mission() -> Transition:
+    def _transition(events: Events) -> State:
         return GoingToRecharging(events)
 
     return _transition

src/isar/state_machine/states/going_to_recharging_with_mission.py

@@ -12,66 +12,64 @@
 from robot_interface.models.mission.status import MissionStatus
 
 
-class GoingToRechargingWithMission(State):
+def GoingToRechargingWithMission(events: Events, mission: AbortedMission) -> State:
 
-    def __init__(self, events: Events, mission: AbortedMission):
+    def _mission_failed_event_handler(
+        error_message: ErrorMessage,
+    ) -> Transition:
+        publish_mission_status(
+            events.mqtt_queue,
+            mission.id,
+            MissionStatus.Failed,
+            error_message,
+        )
+        return InterventionNeeded.transition("Return home to recharge failed")
 
-        def _mission_failed_event_handler(
-            error_message: ErrorMessage,
-        ) -> Transition[InterventionNeeded.InterventionNeeded]:
-            publish_mission_status(
-                events.mqtt_queue,
-                mission.id,
-                MissionStatus.Failed,
-                error_message,
+    def _stop_mission_event_handler(
+        stop_mission_id: str,
+    ) -> Transition | None:
+        if mission.id == stop_mission_id or stop_mission_id == "":
+            events.api_requests.stop_mission.response.trigger_event(
+                ControlMissionResponse(success=True)
             )
-            return InterventionNeeded.transition("Return home to recharge failed")
-
-        def _stop_mission_event_handler(
-            stop_mission_id: str,
-        ) -> Transition[GoingToRecharging.GoingToRecharging] | None:
-            if mission.id == stop_mission_id or stop_mission_id == "":
-                events.api_requests.stop_mission.response.trigger_event(
-                    ControlMissionResponse(success=True)
-                )
-                return GoingToRecharging.transition_to_existing_mission()
-            else:
-                events.api_requests.stop_mission.response.trigger_event(
-                    ControlMissionResponse(
-                        success=False, failure_reason="Mission not found"
-                    )
+            return GoingToRecharging.transition_to_existing_mission()
+        else:
+            events.api_requests.stop_mission.response.trigger_event(
+                ControlMissionResponse(
+                    success=False, failure_reason="Mission not found"
                 )
-                return None
+            )
+            return None
 
-        event_handlers: list[EventHandlerMapping] = [
-            EventHandlerMapping[ErrorMessage](
-                event=events.robot_service_events.mission_failed,
-                handler=_mission_failed_event_handler,
-            ),
-            EventHandlerMapping[EmptyMessage](
-                event=events.robot_service_events.mission_succeeded,
-                handler=lambda _: RechargingWithMission.transition(mission),
-            ),
-            EventHandlerMapping[EmptyMessage](
-                event=events.api_requests.send_to_lockdown.request,
-                handler=lambda _: GoingToLockdown.transition_to_existing_mission_and_report_to_api(),
-            ),
-            EventHandlerMapping[str](
-                event=events.api_requests.stop_mission.request,
-                handler=_stop_mission_event_handler,
-            ),
-        ]
-        super().__init__(
-            state_name=States.GoingToRechargingWithMission,
-            signal_exit_event=events.signal_state_machine_exit,
-            event_handler_mappings=event_handlers,
-        )
+    event_handlers: list[EventHandlerMapping] = [
+        EventHandlerMapping[ErrorMessage](
+            event=events.robot_service_events.mission_failed,
+            handler=_mission_failed_event_handler,
+        ),
+        EventHandlerMapping[EmptyMessage](
+            event=events.robot_service_events.mission_succeeded,
+            handler=lambda _: RechargingWithMission.transition(mission),
+        ),
+        EventHandlerMapping[EmptyMessage](
+            event=events.api_requests.send_to_lockdown.request,
+            handler=lambda _: GoingToLockdown.transition_to_existing_mission_and_report_to_api(),
+        ),
+        EventHandlerMapping[str](
+            event=events.api_requests.stop_mission.request,
+            handler=_stop_mission_event_handler,
+        ),
+    ]
+    return State(
+        state_name=States.GoingToRechargingWithMission,
+        signal_exit_event=events.signal_state_machine_exit,
+        event_handler_mappings=event_handlers,
+    )
 
 
 def transition_and_start_return_home(
     mission: AbortedMission,
-) -> Transition[GoingToRechargingWithMission]:
-    def _transition(events: Events) -> GoingToRechargingWithMission:
+) -> Transition:
+    def _transition(events: Events) -> State:
         events.robot_service_events.mission_failed.clear_event()
         events.robot_service_events.mission_succeeded.clear_event()