fix(api): readjust hardware controller end state in engine cleanup

api/src/opentrons/execute.py

@@ -41,6 +41,7 @@
     ThreadManagedHardware,
     ThreadManager,
 )
+from opentrons.protocol_engine.types import PostRunHardwareState
 
 from opentrons.protocols import parse
 from opentrons.protocols.api_support.deck_type import (
@@ -572,7 +573,8 @@ def _create_live_context_pe(
         create_protocol_engine_in_thread(
             hardware_api=hardware_api.wrapped(),
             config=_get_protocol_engine_config(),
-            drop_tips_and_home_after=False,
+            drop_tips_after_run=False,
+            post_run_hardware_state=PostRunHardwareState.STAY_ENGAGED_IN_PLACE,
         )
     )
 

api/src/opentrons/hardware_control/api.py

@@ -490,14 +490,15 @@ async def _chained_calls() -> None:
 
         asyncio.run_coroutine_threadsafe(_chained_calls(), self._loop)
 
-    async def halt(self) -> None:
+    async def halt(self, disengage_before_stopping: bool = False) -> None:
         """Immediately stop motion, cancel execution manager and cancel running tasks.
 
         After this call, the smoothie will be in a bad state until a call to
         :py:meth:`stop`.
         """
-        await self._backend.hard_halt()
-        await self._execution_manager.cancel()
+        if disengage_before_stopping:
+            await self._backend.hard_halt()
+        await self._backend.halt()
 
     async def stop(self, home_after: bool = True) -> None:
         """
@@ -508,6 +509,7 @@ async def stop(self, home_after: bool = True) -> None:
         robot. After this call, no further recovery is necessary.
         """
         await self._backend.halt()  # calls smoothie_driver.kill()
+        await self._execution_manager.cancel()
         self._log.info("Recovering from halt")
         await self.reset()
         await self.cache_instruments()

api/src/opentrons/hardware_control/ot3api.py

@@ -743,19 +743,18 @@ async def _cancel_execution_and_running_tasks(self) -> None:
         """Cancel execution manager and all running (hardware module) tasks."""
         await self._execution_manager.cancel()
 
-    async def halt(self) -> None:
+    async def halt(self, disengage_before_stopping: bool = False) -> None:
         """Immediately disengage all present motors and clear motor and module tasks."""
-        # TODO (spp, 2023-08-22): check if disengaging motors is really required
-        await self.disengage_axes(
-            [ax for ax in Axis if self._backend.axis_is_present(ax)]
-        )
+        if disengage_before_stopping:
+            await self.disengage_axes(
+                [ax for ax in Axis if self._backend.axis_is_present(ax)]
+            )
         await self._stop_motors()
-        await self._cancel_execution_and_running_tasks()
 
     async def stop(self, home_after: bool = True) -> None:
         """Stop motion as soon as possible, reset, and optionally home."""
         await self._stop_motors()
-
+        await self._cancel_execution_and_running_tasks()
         self._log.info("Resetting OT3API")
         await self.reset()
         if home_after:

api/src/opentrons/hardware_control/protocols/motion_controller.py

@@ -8,7 +8,7 @@
 class MotionController(Protocol):
     """Protocol specifying fundamental motion controls."""
 
-    async def halt(self) -> None:
+    async def halt(self, disengage_before_stopping: bool = False) -> None:
         """Immediately stop motion.
 
         Calls to stop through the synch adapter while other calls
@@ -17,8 +17,9 @@ async def halt(self) -> None:
         smoothie. To provide actual immediate halting, call this method which
         does not require use of the loop.
 
-        After this call, the hardware will be in a bad state until a call to
-        stop
+        If disengage_before_stopping is True, the motors will disengage first and then
+        stop in place. Disengaging creates a smoother halt but requires homing after
+        in order to resume movement.
         """
         ...
 

api/src/opentrons/protocol_engine/create_protocol_engine.py

@@ -10,6 +10,7 @@
 from .protocol_engine import ProtocolEngine
 from .resources import DeckDataProvider, ModuleDataProvider
 from .state import Config, StateStore
+from .types import PostRunHardwareState
 
 
 # TODO(mm, 2023-06-16): Arguably, this not being a context manager makes us prone to forgetting to
@@ -44,7 +45,8 @@ async def create_protocol_engine(
 def create_protocol_engine_in_thread(
     hardware_api: HardwareControlAPI,
     config: Config,
-    drop_tips_and_home_after: bool,
+    drop_tips_after_run: bool,
+    post_run_hardware_state: PostRunHardwareState,
 ) -> typing.Generator[
     typing.Tuple[ProtocolEngine, asyncio.AbstractEventLoop], None, None
 ]:
@@ -66,7 +68,9 @@ def create_protocol_engine_in_thread(
     3. Joins the thread.
     """
     with async_context_manager_in_thread(
-        _protocol_engine(hardware_api, config, drop_tips_and_home_after)
+        _protocol_engine(
+            hardware_api, config, drop_tips_after_run, post_run_hardware_state
+        )
     ) as (
         protocol_engine,
         loop,
@@ -78,7 +82,8 @@ def create_protocol_engine_in_thread(
 async def _protocol_engine(
     hardware_api: HardwareControlAPI,
     config: Config,
-    drop_tips_and_home_after: bool,
+    drop_tips_after_run: bool,
+    post_run_hardware_state: PostRunHardwareState,
 ) -> typing.AsyncGenerator[ProtocolEngine, None]:
     protocol_engine = await create_protocol_engine(
         hardware_api=hardware_api,
@@ -88,4 +93,7 @@ async def _protocol_engine(
         protocol_engine.play()
         yield protocol_engine
     finally:
-        await protocol_engine.finish(drop_tips_and_home=drop_tips_and_home_after)
+        await protocol_engine.finish(
+            drop_tips_after_run=drop_tips_after_run,
+            post_run_hardware_state=post_run_hardware_state,
+        )

api/src/opentrons/protocol_engine/execution/hardware_stopper.py

@@ -7,7 +7,7 @@
 
 from ..resources.ot3_validation import ensure_ot3_hardware
 from ..state import StateStore
-from ..types import MotorAxis
+from ..types import MotorAxis, PostRunHardwareState
 from ..errors import HardwareNotSupportedError
 
 from .movement import MovementHandler
@@ -87,17 +87,27 @@ async def _drop_tip(self) -> None:
                 # should not happen during an actual run
                 log.debug(f"Pipette ID {pipette_id} no longer attached.")
 
-    async def do_halt(self) -> None:
+    async def do_halt(self, disengage_before_stopping: bool = False) -> None:
         """Issue a halt signal to the hardware API.
 
         After issuing a halt, you must call do_stop_and_recover after
         anything using the HardwareAPI has settled.
         """
-        await self._hardware_api.halt()
+        await self._hardware_api.halt(
+            disengage_before_stopping=disengage_before_stopping
+        )
 
-    async def do_stop_and_recover(self, drop_tips_and_home: bool = False) -> None:
+    async def do_stop_and_recover(
+        self,
+        post_run_hardware_state: PostRunHardwareState,
+        drop_tips_after_run: bool = False,
+    ) -> None:
         """Stop and reset the HardwareAPI, optionally dropping tips and homing."""
-        if drop_tips_and_home:
+        if drop_tips_after_run:
             await self._drop_tip()
 
-        await self._hardware_api.stop(home_after=drop_tips_and_home)
+        home_after_stop = post_run_hardware_state in (
+            PostRunHardwareState.HOME_AND_STAY_ENGAGED,
+            PostRunHardwareState.HOME_THEN_DISENGAGE,
+        )
+        await self._hardware_api.stop(home_after=home_after_stop)

api/src/opentrons/protocol_engine/protocol_engine.py

@@ -7,7 +7,6 @@
 from opentrons.hardware_control import HardwareControlAPI
 from opentrons.hardware_control.modules import AbstractModule as HardwareModuleAPI
 from opentrons.hardware_control.types import PauseType as HardwarePauseType
-
 from opentrons_shared_data.errors import (
     ErrorCodes,
     EnumeratedError,
@@ -24,6 +23,7 @@
     ModuleModel,
     Liquid,
     HexColor,
+    PostRunHardwareState,
 )
 from .execution import (
     QueueWorker,
@@ -306,8 +306,9 @@ async def wait_until_complete(self) -> None:
     async def finish(
         self,
         error: Optional[Exception] = None,
-        drop_tips_and_home: bool = True,
+        drop_tips_after_run: bool = True,
         set_run_status: bool = True,
+        post_run_hardware_state: PostRunHardwareState = PostRunHardwareState.HOME_AND_STAY_ENGAGED,
     ) -> None:
         """Gracefully finish using the ProtocolEngine, waiting for it to become idle.
 
@@ -321,19 +322,23 @@ async def finish(
 
         Arguments:
             error: An error that caused the stop, if applicable.
-            drop_tips_and_home: Whether to home and drop tips as part of cleanup.
+            drop_tips_after_run: Whether to drop tips as part of cleanup.
             set_run_status: Whether to calculate a `success` or `failure` run status.
                 If `False`, will set status to `stopped`.
+            post_run_hardware_state: The state in which to leave the gantry and motors in
+                after the run is over.
         """
         if self._state_store.commands.state.stopped_by_estop:
-            drop_tips_and_home = False
+            drop_tips_after_run = False
+            post_run_hardware_state = PostRunHardwareState.DISENGAGE_IN_PLACE
             if error is None:
                 error = EStopActivatedError(message="Estop was activated during a run")
         if error:
             if isinstance(error, EnumeratedError) and self._code_in_exception_stack(
                 error=error, code=ErrorCodes.E_STOP_ACTIVATED
             ):
-                drop_tips_and_home = False
+                drop_tips_after_run = False
+                post_run_hardware_state = PostRunHardwareState.DISENGAGE_IN_PLACE
 
             error_details: Optional[FinishErrorDetails] = FinishErrorDetails(
                 error_id=self._model_utils.generate_id(),
@@ -357,13 +362,21 @@ async def finish(
         exit_stack.push_async_callback(
             # Cleanup after hardware halt and reset the hardware controller
             self._hardware_stopper.do_stop_and_recover,
-            drop_tips_and_home=drop_tips_and_home,
+            post_run_hardware_state=post_run_hardware_state,
+            drop_tips_after_run=drop_tips_after_run,
         )
         exit_stack.callback(self._door_watcher.stop)
 
-        # Halt any movements immediately. Requires a hardware stop & reset after, in order to
-        # recover from halt and resume hardware operations.
-        exit_stack.push_async_callback(self._hardware_stopper.do_halt)
+        disengage_before_stopping = (
+            False
+            if post_run_hardware_state == PostRunHardwareState.STAY_ENGAGED_IN_PLACE
+            else True
+        )
+        # Halt any movements immediately
+        exit_stack.push_async_callback(
+            self._hardware_stopper.do_halt,
+            disengage_before_stopping=disengage_before_stopping,
+        )
         exit_stack.push_async_callback(self._queue_worker.join)  # First step.
 
         try:

api/src/opentrons/protocol_engine/types.py

@@ -629,3 +629,30 @@ class LabwareMovementStrategy(str, Enum):
     USING_GRIPPER = "usingGripper"
     MANUAL_MOVE_WITH_PAUSE = "manualMoveWithPause"
     MANUAL_MOVE_WITHOUT_PAUSE = "manualMoveWithoutPause"
+
+
+class PostRunHardwareState(Enum):
+    """State of robot gantry & motors after a stop is performed and the hardware API is reset.
+
+    HOME_AND_STAY_ENGAGED: home the gantry and keep all motors engaged. This allows the
+        robot to continue performing movement actions without re-homing
+    HOME_THEN_DISENGAGE: home the gantry and then disengage motors.
+        Reduces current consumption of the motors and prevents coil heating.
+        Re-homing is required to re-engage the motors and resume robot movement.
+    STAY_ENGAGED_IN_PLACE: do not home after the stop and keep the motors engaged.
+        Keeps gantry in the same position as prior to `stop()` execution
+        and allows the robot to execute movement commands without requiring to re-home first.
+    DISENGAGE_IN_PLACE: disengage motors and do not home the robot
+    Probable states for pipette:
+        - for 1- or 8-channel:
+            - HOME_AND_STAY_ENGAGED after protocol runs
+            - STAY_ENGAGED_IN_PLACE after maintenance runs
+        - for 96-channel:
+            - HOME_THEN_DISENGAGE after protocol runs
+            - DISENGAGE_IN_PLACE after maintenance runs
+    """
+
+    HOME_AND_STAY_ENGAGED = "homeAndStayEngaged"
+    HOME_THEN_DISENGAGE = "homeThenDisengage"
+    STAY_ENGAGED_IN_PLACE = "stayEngagedInPlace"
+    DISENGAGE_IN_PLACE = "disengageInPlace"

api/src/opentrons/protocol_runner/protocol_runner.py

@@ -35,6 +35,7 @@
     LegacyExecutor,
     LegacyLoadInfo,
 )
+from ..protocol_engine.types import PostRunHardwareState
 
 
 class RunResult(NamedTuple):
@@ -80,8 +81,9 @@ async def stop(self) -> None:
             await self._protocol_engine.stop()
         else:
             await self._protocol_engine.finish(
-                drop_tips_and_home=False,
+                drop_tips_after_run=False,
                 set_run_status=False,
+                post_run_hardware_state=PostRunHardwareState.STAY_ENGAGED_IN_PLACE,
             )
 
     @abstractmethod

api/tests/opentrons/conftest.py

@@ -25,6 +25,8 @@
 import pytest
 from decoy import Decoy
 
+from opentrons.protocol_engine.types import PostRunHardwareState
+
 try:
     import aionotify  # type: ignore[import]
 except (OSError, ModuleNotFoundError):
@@ -296,7 +298,8 @@ def _make_ot3_pe_ctx(
             use_virtual_gripper=True,
             block_on_door_open=False,
         ),
-        drop_tips_and_home_after=False,
+        drop_tips_after_run=False,
+        post_run_hardware_state=PostRunHardwareState.STAY_ENGAGED_IN_PLACE,
     ) as (
         engine,
         loop,