feat(api): add a stateless command to control the status bar

api/src/opentrons/protocol_engine/commands/__init__.py

@@ -239,6 +239,15 @@
     BlowOutInPlace,
 )
 
+from .set_status_bar import (
+    SetStatusBar,
+    SetStatusBarParams,
+    SetStatusBarCreate,
+    SetStatusBarResult,
+    SetStatusBarImplementation,
+    SetStatusBarCommandType,
+)
+
 __all__ = [
     # command type unions
     "Command",
@@ -400,6 +409,13 @@
     "BlowOutInPlaceCreate",
     "BlowOutInPlaceImplementation",
     "BlowOutInPlace",
+    # set status bar command models
+    "SetStatusBar",
+    "SetStatusBarParams",
+    "SetStatusBarCreate",
+    "SetStatusBarResult",
+    "SetStatusBarImplementation",
+    "SetStatusBarCommandType",
     # load liquid command models
     "LoadLiquid",
     "LoadLiquidCreate",

api/src/opentrons/protocol_engine/commands/command.py

@@ -169,6 +169,7 @@ def __init__(
         tip_handler: execution.TipHandler,
         run_control: execution.RunControlHandler,
         rail_lights: execution.RailLightsHandler,
+        status_bar: execution.StatusBarHandler,
     ) -> None:
         """Initialize the command implementation with execution handlers."""
         pass

api/src/opentrons/protocol_engine/commands/command_unions.py

@@ -209,6 +209,14 @@
     BlowOutInPlaceResult,
 )
 
+from .set_status_bar import (
+    SetStatusBar,
+    SetStatusBarParams,
+    SetStatusBarCreate,
+    SetStatusBarResult,
+    SetStatusBarCommandType,
+)
+
 Command = Union[
     Aspirate,
     AspirateInPlace,
@@ -235,6 +243,7 @@
     SavePosition,
     SetRailLights,
     TouchTip,
+    SetStatusBar,
     heater_shaker.WaitForTemperature,
     heater_shaker.SetTargetTemperature,
     heater_shaker.DeactivateHeater,
@@ -288,6 +297,7 @@
     SavePositionParams,
     SetRailLightsParams,
     TouchTipParams,
+    SetStatusBarParams,
     heater_shaker.WaitForTemperatureParams,
     heater_shaker.SetTargetTemperatureParams,
     heater_shaker.DeactivateHeaterParams,
@@ -342,6 +352,7 @@
     SavePositionCommandType,
     SetRailLightsCommandType,
     TouchTipCommandType,
+    SetStatusBarCommandType,
     heater_shaker.WaitForTemperatureCommandType,
     heater_shaker.SetTargetTemperatureCommandType,
     heater_shaker.DeactivateHeaterCommandType,
@@ -395,6 +406,7 @@
     SavePositionCreate,
     SetRailLightsCreate,
     TouchTipCreate,
+    SetStatusBarCreate,
     heater_shaker.WaitForTemperatureCreate,
     heater_shaker.SetTargetTemperatureCreate,
     heater_shaker.DeactivateHeaterCreate,
@@ -448,6 +460,7 @@
     SavePositionResult,
     SetRailLightsResult,
     TouchTipResult,
+    SetStatusBarResult,
     heater_shaker.WaitForTemperatureResult,
     heater_shaker.SetTargetTemperatureResult,
     heater_shaker.DeactivateHeaterResult,

api/src/opentrons/protocol_engine/commands/set_status_bar.py

@@ -0,0 +1,82 @@
+"""setStatusBar command request, result, and implementation models."""
+from __future__ import annotations
+from pydantic import BaseModel, Field
+from typing import TYPE_CHECKING, Optional, Type
+from typing_extensions import Literal
+import enum
+
+from opentrons.hardware_control.types import StatusBarState
+from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate
+
+if TYPE_CHECKING:
+    from ..execution import StatusBarHandler
+
+SetStatusBarCommandType = Literal["setStatusBar"]
+
+
+class StatusBarAnimation(enum.Enum):
+    """Status Bar animation options."""
+
+    IDLE = "idle"
+    CONFIRM = "confirm"
+    UPDATING = "updating"
+    DISCO = "disco"
+    OFF = "off"
+
+
+def _animation_to_status_bar_state(animation: StatusBarAnimation) -> StatusBarState:
+    return {
+        StatusBarAnimation.IDLE: StatusBarState.IDLE,
+        StatusBarAnimation.CONFIRM: StatusBarState.CONFIRMATION,
+        StatusBarAnimation.UPDATING: StatusBarState.UPDATING,
+        StatusBarAnimation.DISCO: StatusBarState.DISCO,
+        StatusBarAnimation.OFF: StatusBarState.OFF,
+    }[animation]
+
+
+class SetStatusBarParams(BaseModel):
+    """Payload required to set the status bar to run an animation."""
+
+    animation: StatusBarAnimation = Field(
+        ...,
+        description="The animation that should be executed on the status bar.",
+    )
+
+
+class SetStatusBarResult(BaseModel):
+    """Result data from the execution of a SetStatusBar command."""
+
+
+class SetStatusBarImplementation(
+    AbstractCommandImpl[SetStatusBarParams, SetStatusBarResult]
+):
+    """setStatusBar command implementation."""
+
+    def __init__(self, status_bar: StatusBarHandler, **kwargs: object) -> None:
+        self._status_bar = status_bar
+
+    async def execute(self, params: SetStatusBarParams) -> SetStatusBarResult:
+        """Execute the setStatusBar command."""
+        if not self._status_bar.status_bar_should_not_be_changed():
+            state = _animation_to_status_bar_state(params.animation)
+            await self._status_bar.set_status_bar(state)
+        return SetStatusBarResult()
+
+
+class SetStatusBar(BaseCommand[SetStatusBarParams, SetStatusBarResult]):
+    """setStatusBar command model."""
+
+    commandType: SetStatusBarCommandType = "setStatusBar"
+    params: SetStatusBarParams
+    result: Optional[SetStatusBarResult]
+
+    _ImplementationCls: Type[SetStatusBarImplementation] = SetStatusBarImplementation
+
+
+class SetStatusBarCreate(BaseCommandCreate[SetStatusBarParams]):
+    """setStatusBar command request model."""
+
+    commandType: SetStatusBarCommandType = "setStatusBar"
+    params: SetStatusBarParams
+
+    _CommandCls: Type[SetStatusBar] = SetStatusBar

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

@@ -18,6 +18,7 @@
 from .run_control import RunControlHandler
 from .hardware_stopper import HardwareStopper
 from .door_watcher import DoorWatcher
+from .status_bar import StatusBarHandler
 
 # .thermocycler_movement_flagger omitted from package's public interface.
 
@@ -39,4 +40,5 @@
     "HardwareStopper",
     "DoorWatcher",
     "RailLightsHandler",
+    "StatusBarHandler",
 ]

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

@@ -18,6 +18,7 @@
 from .tip_handler import TipHandler
 from .run_control import RunControlHandler
 from .rail_lights import RailLightsHandler
+from .status_bar import StatusBarHandler
 
 
 log = getLogger(__name__)
@@ -43,6 +44,7 @@ def __init__(
         tip_handler: TipHandler,
         run_control: RunControlHandler,
         rail_lights: RailLightsHandler,
+        status_bar: StatusBarHandler,
         model_utils: Optional[ModelUtils] = None,
     ) -> None:
         """Initialize the CommandExecutor with access to its dependencies."""
@@ -58,6 +60,7 @@ def __init__(
         self._run_control = run_control
         self._rail_lights = rail_lights
         self._model_utils = model_utils or ModelUtils()
+        self._status_bar = status_bar
 
     async def execute(self, command_id: str) -> None:
         """Run a given command's execution procedure.
@@ -78,6 +81,7 @@ async def execute(self, command_id: str) -> None:
             tip_handler=self._tip_handler,
             run_control=self._run_control,
             rail_lights=self._rail_lights,
+            status_bar=self._status_bar,
         )
 
         started_at = self._model_utils.get_timestamp()

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

@@ -13,6 +13,7 @@
 from .run_control import RunControlHandler
 from .command_executor import CommandExecutor
 from .queue_worker import QueueWorker
+from .status_bar import StatusBarHandler
 
 
 def create_queue_worker(
@@ -69,6 +70,8 @@ def create_queue_worker(
         hardware_api=hardware_api,
     )
 
+    status_bar_handler = StatusBarHandler(hardware_api=hardware_api)
+
     command_executor = CommandExecutor(
         hardware_api=hardware_api,
         state_store=state_store,
@@ -81,6 +84,7 @@ def create_queue_worker(
         tip_handler=tip_handler,
         run_control=run_control_handler,
         rail_lights=rail_lights_handler,
+        status_bar=status_bar_handler,
     )
 
     return QueueWorker(

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

@@ -0,0 +1,34 @@
+"""Status Bar command handling."""
+
+from opentrons.hardware_control import HardwareControlAPI
+from opentrons.hardware_control.types import StatusBarState
+
+
+class StatusBarHandler:
+    """Handle interaction with the status bar."""
+
+    _hardware_api: HardwareControlAPI
+
+    def __init__(
+        self,
+        hardware_api: HardwareControlAPI,
+    ) -> None:
+        """Initialize a StatusBarHandler instance."""
+        self._hardware_api = hardware_api
+
+    async def set_status_bar(
+        self,
+        status: StatusBarState,
+    ) -> None:
+        """Set the status bar."""
+        await self._hardware_api.set_status_bar_state(state=status)
+
+    def status_bar_should_not_be_changed(self) -> bool:
+        """Checks whether the status bar is seemingly busy."""
+        state = self._hardware_api.get_status_bar_state()
+
+        return state not in [
+            StatusBarState.IDLE,
+            StatusBarState.UPDATING,
+            StatusBarState.OFF,
+        ]

api/tests/opentrons/protocol_engine/commands/conftest.py

@@ -11,6 +11,7 @@
     RunControlHandler,
     RailLightsHandler,
     LabwareMovementHandler,
+    StatusBarHandler,
 )
 from opentrons.protocol_engine.state import StateView
 
@@ -55,3 +56,9 @@ def run_control(decoy: Decoy) -> RunControlHandler:
 def rail_lights(decoy: Decoy) -> RailLightsHandler:
     """Get a mocked out RailLightsHandler."""
     return decoy.mock(cls=RailLightsHandler)
+
+
+@pytest.fixture
+def status_bar(decoy: Decoy) -> StatusBarHandler:
+    """Get a mocked out StatusBarHandler."""
+    return decoy.mock(cls=StatusBarHandler)

api/tests/opentrons/protocol_engine/commands/test_set_status_bar.py

@@ -0,0 +1,67 @@
+"""Test setStatusBar commands."""
+
+import pytest
+from decoy import Decoy
+
+from opentrons.protocol_engine.commands.set_status_bar import (
+    SetStatusBarParams,
+    SetStatusBarResult,
+    SetStatusBarImplementation,
+    StatusBarAnimation,
+)
+
+from opentrons.hardware_control.types import StatusBarState
+from opentrons.protocol_engine.execution.status_bar import StatusBarHandler
+
+
+@pytest.fixture
+def subject(
+    status_bar: StatusBarHandler,
+) -> SetStatusBarImplementation:
+    """Returns subject under test."""
+    return SetStatusBarImplementation(status_bar=status_bar)
+
+
+async def test_status_bar_busy(
+    decoy: Decoy,
+    status_bar: StatusBarHandler,
+    subject: SetStatusBarImplementation,
+) -> None:
+    """Test when the status bar is busy."""
+    decoy.when(status_bar.status_bar_should_not_be_changed()).then_return(True)
+
+    data = SetStatusBarParams(animation=StatusBarAnimation.OFF)
+
+    result = await subject.execute(params=data)
+
+    assert result == SetStatusBarResult()
+
+    decoy.verify(await status_bar.set_status_bar(status=StatusBarState.OFF), times=0)
+
+
+@pytest.mark.parametrize(
+    argnames=["animation", "expected_state"],
+    argvalues=[
+        [StatusBarAnimation.CONFIRM, StatusBarState.CONFIRMATION],
+        [StatusBarAnimation.DISCO, StatusBarState.DISCO],
+        [StatusBarAnimation.OFF, StatusBarState.OFF],
+        [StatusBarAnimation.IDLE, StatusBarState.IDLE],
+        [StatusBarAnimation.UPDATING, StatusBarState.UPDATING],
+    ],
+)
+async def test_set_status_bar_animation(
+    decoy: Decoy,
+    status_bar: StatusBarHandler,
+    subject: SetStatusBarImplementation,
+    animation: StatusBarAnimation,
+    expected_state: StatusBarState,
+) -> None:
+    """Test when status bar is NOT busy."""
+    decoy.when(status_bar.status_bar_should_not_be_changed()).then_return(False)
+
+    data = SetStatusBarParams(animation=animation)
+
+    result = await subject.execute(params=data)
+    assert result == SetStatusBarResult()
+
+    decoy.verify(await status_bar.set_status_bar(status=expected_state), times=1)