chore(api, robot-server): Manage runners via run orchestrator (#15190)

api/src/opentrons/protocol_runner/__init__.py

@@ -13,6 +13,7 @@
     AnyRunner,
 )
 from .create_simulating_runner import create_simulating_runner
+from .run_orchestrator import RunOrchestrator
 
 __all__ = [
     "AbstractRunner",
@@ -23,4 +24,5 @@
     "PythonAndLegacyRunner",
     "LiveRunner",
     "AnyRunner",
+    "RunOrchestrator",
 ]

api/src/opentrons/protocol_runner/protocol_runner.py

@@ -416,9 +416,9 @@ async def run(  # noqa: D102
 
 
 def create_protocol_runner(
-    protocol_config: Optional[Union[JsonProtocolConfig, PythonProtocolConfig]],
     protocol_engine: ProtocolEngine,
     hardware_api: HardwareControlAPI,
+    protocol_config: Union[JsonProtocolConfig, PythonProtocolConfig],
     task_queue: Optional[TaskQueue] = None,
     json_file_reader: Optional[JsonFileReader] = None,
     json_translator: Optional[JsonTranslator] = None,
@@ -427,39 +427,32 @@ def create_protocol_runner(
     python_protocol_executor: Optional[PythonProtocolExecutor] = None,
     post_run_hardware_state: PostRunHardwareState = PostRunHardwareState.HOME_AND_STAY_ENGAGED,
     drop_tips_after_run: bool = True,
-) -> AnyRunner:
+) -> Union[JsonRunner, PythonAndLegacyRunner]:
     """Create a protocol runner."""
-    if protocol_config:
-        if (
-            isinstance(protocol_config, JsonProtocolConfig)
-            and protocol_config.schema_version >= LEGACY_JSON_SCHEMA_VERSION_CUTOFF
-        ):
-            return JsonRunner(
-                protocol_engine=protocol_engine,
-                hardware_api=hardware_api,
-                json_file_reader=json_file_reader,
-                json_translator=json_translator,
-                task_queue=task_queue,
-                post_run_hardware_state=post_run_hardware_state,
-                drop_tips_after_run=drop_tips_after_run,
-            )
-        else:
-            return PythonAndLegacyRunner(
-                protocol_engine=protocol_engine,
-                hardware_api=hardware_api,
-                task_queue=task_queue,
-                python_and_legacy_file_reader=python_and_legacy_file_reader,
-                protocol_context_creator=protocol_context_creator,
-                python_protocol_executor=python_protocol_executor,
-                post_run_hardware_state=post_run_hardware_state,
-                drop_tips_after_run=drop_tips_after_run,
-            )
-
-    return LiveRunner(
-        protocol_engine=protocol_engine,
-        hardware_api=hardware_api,
-        task_queue=task_queue,
-    )
+    if (
+        isinstance(protocol_config, JsonProtocolConfig)
+        and protocol_config.schema_version >= LEGACY_JSON_SCHEMA_VERSION_CUTOFF
+    ):
+        return JsonRunner(
+            protocol_engine=protocol_engine,
+            hardware_api=hardware_api,
+            json_file_reader=json_file_reader,
+            json_translator=json_translator,
+            task_queue=task_queue,
+            post_run_hardware_state=post_run_hardware_state,
+            drop_tips_after_run=drop_tips_after_run,
+        )
+    else:
+        return PythonAndLegacyRunner(
+            protocol_engine=protocol_engine,
+            hardware_api=hardware_api,
+            task_queue=task_queue,
+            python_and_legacy_file_reader=python_and_legacy_file_reader,
+            protocol_context_creator=protocol_context_creator,
+            python_protocol_executor=python_protocol_executor,
+            post_run_hardware_state=post_run_hardware_state,
+            drop_tips_after_run=drop_tips_after_run,
+        )
 
 
 async def _yield() -> None:

api/src/opentrons/protocol_runner/run_orchestrator.py

@@ -0,0 +1,101 @@
+"""Engine/Runner provider."""
+from __future__ import annotations
+from typing import Optional, Union
+
+from . import protocol_runner, AnyRunner
+from ..hardware_control import HardwareControlAPI
+from ..protocol_engine import ProtocolEngine
+from ..protocol_engine.types import PostRunHardwareState
+from ..protocol_reader import JsonProtocolConfig, PythonProtocolConfig
+
+
+class RunOrchestrator:
+    """Provider for runners and associated protocol engine.
+
+    Build runners, manage command execution, run state and in-memory protocol engine associated to the runners.
+    """
+
+    _protocol_runner: Optional[
+        Union[protocol_runner.JsonRunner, protocol_runner.PythonAndLegacyRunner, None]
+    ]
+    _setup_runner: protocol_runner.LiveRunner
+    _fixit_runner: protocol_runner.LiveRunner
+    _hardware_api: HardwareControlAPI
+    _protocol_engine: ProtocolEngine
+
+    def __init__(
+        self,
+        protocol_engine: ProtocolEngine,
+        hardware_api: HardwareControlAPI,
+        fixit_runner: protocol_runner.LiveRunner,
+        setup_runner: protocol_runner.LiveRunner,
+        json_or_python_protocol_runner: Optional[
+            Union[protocol_runner.PythonAndLegacyRunner, protocol_runner.JsonRunner]
+        ] = None,
+        run_id: Optional[str] = None,
+    ) -> None:
+        """Initialize a run orchestrator interface.
+
+        Arguments:
+            protocol_engine: Protocol engine instance.
+            hardware_api: Hardware control API instance.
+            fixit_runner: LiveRunner for fixit commands.
+            setup_runner: LiveRunner for setup commands.
+            json_or_python_protocol_runner: JsonRunner/PythonAndLegacyRunner for protocol commands.
+            run_id: run id if any, associated to the runner/engine.
+        """
+        self.run_id = run_id
+        self._protocol_engine = protocol_engine
+        self._hardware_api = hardware_api
+        self._protocol_runner = json_or_python_protocol_runner
+        self._setup_runner = setup_runner
+        self._fixit_runner = fixit_runner
+
+    @property
+    def engine(self) -> ProtocolEngine:
+        """Get the "current" persisted ProtocolEngine."""
+        return self._protocol_engine
+
+    @property
+    def runner(self) -> AnyRunner:
+        """Get the "current" persisted ProtocolRunner."""
+        return self._protocol_runner or self._setup_runner
+
+    @classmethod
+    def build_orchestrator(
+        cls,
+        protocol_engine: ProtocolEngine,
+        hardware_api: HardwareControlAPI,
+        protocol_config: Optional[
+            Union[JsonProtocolConfig, PythonProtocolConfig]
+        ] = None,
+        post_run_hardware_state: PostRunHardwareState = PostRunHardwareState.HOME_AND_STAY_ENGAGED,
+        drop_tips_after_run: bool = True,
+        run_id: Optional[str] = None,
+    ) -> "RunOrchestrator":
+        """Build a RunOrchestrator provider."""
+        setup_runner = protocol_runner.LiveRunner(
+            protocol_engine=protocol_engine,
+            hardware_api=hardware_api,
+        )
+        fixit_runner = protocol_runner.LiveRunner(
+            protocol_engine=protocol_engine,
+            hardware_api=hardware_api,
+        )
+        json_or_python_runner = None
+        if protocol_config:
+            json_or_python_runner = protocol_runner.create_protocol_runner(
+                protocol_config=protocol_config,
+                protocol_engine=protocol_engine,
+                hardware_api=hardware_api,
+                post_run_hardware_state=post_run_hardware_state,
+                drop_tips_after_run=drop_tips_after_run,
+            )
+        return cls(
+            run_id=run_id,
+            json_or_python_protocol_runner=json_or_python_runner,
+            setup_runner=setup_runner,
+            fixit_runner=fixit_runner,
+            hardware_api=hardware_api,
+            protocol_engine=protocol_engine,
+        )

api/tests/opentrons/protocol_runner/test_protocol_runner.py

@@ -5,7 +5,7 @@
 from pytest_lazyfixture import lazy_fixture  # type: ignore[import-untyped]
 from decoy import Decoy, matchers
 from pathlib import Path
-from typing import List, cast, Optional, Union, Type
+from typing import List, cast, Union, Type
 
 from opentrons_shared_data.labware.labware_definition import LabwareDefinition
 from opentrons_shared_data.labware.dev_types import (
@@ -174,7 +174,6 @@ def live_runner_subject(
         (PythonProtocolConfig(api_version=APIVersion(2, 14)), PythonAndLegacyRunner),
         (JsonProtocolConfig(schema_version=5), PythonAndLegacyRunner),
         (PythonProtocolConfig(api_version=APIVersion(2, 13)), PythonAndLegacyRunner),
-        (None, LiveRunner),
     ],
 )
 def test_create_protocol_runner(
@@ -186,7 +185,7 @@ def test_create_protocol_runner(
     python_and_legacy_file_reader: PythonAndLegacyFileReader,
     protocol_context_creator: ProtocolContextCreator,
     python_protocol_executor: PythonProtocolExecutor,
-    config: Optional[Union[JsonProtocolConfig, PythonProtocolConfig]],
+    config: Union[JsonProtocolConfig, PythonProtocolConfig],
     runner_type: Type[AnyRunner],
 ) -> None:
     """It should return protocol runner type depending on the config."""

api/tests/opentrons/protocol_runner/test_run_orchestrator_provider.py

@@ -0,0 +1,143 @@
+"""Tests for the RunOrchestrator."""
+import pytest
+from pytest_lazyfixture import lazy_fixture  # type: ignore[import-untyped]
+from decoy import Decoy
+from typing import Union
+
+from opentrons.protocols.api_support.types import APIVersion
+from opentrons.protocol_engine import ProtocolEngine
+from opentrons.protocol_engine.types import PostRunHardwareState
+from opentrons.hardware_control import API as HardwareAPI
+from opentrons.protocol_reader import JsonProtocolConfig, PythonProtocolConfig
+from opentrons.protocol_runner.run_orchestrator import RunOrchestrator
+from opentrons import protocol_runner
+from opentrons.protocol_runner.protocol_runner import (
+    JsonRunner,
+    PythonAndLegacyRunner,
+    LiveRunner,
+)
+
+
+@pytest.fixture
+def mock_protocol_python_runner(decoy: Decoy) -> PythonAndLegacyRunner:
+    """Get a mocked out PythonAndLegacyRunner dependency."""
+    return decoy.mock(cls=PythonAndLegacyRunner)
+
+
+@pytest.fixture
+def mock_protocol_json_runner(decoy: Decoy) -> JsonRunner:
+    """Get a mocked out PythonAndLegacyRunner dependency."""
+    return decoy.mock(cls=JsonRunner)
+
+
+@pytest.fixture
+def mock_setup_runner(decoy: Decoy) -> LiveRunner:
+    """Get a mocked out LiveRunner dependency."""
+    return decoy.mock(cls=LiveRunner)
+
+
+@pytest.fixture
+def mock_fixit_runner(decoy: Decoy) -> LiveRunner:
+    """Get a mocked out LiveRunner dependency."""
+    return decoy.mock(cls=LiveRunner)
+
+
+@pytest.fixture
+def mock_protocol_engine(decoy: Decoy) -> ProtocolEngine:
+    """Get a mocked out ProtocolEngine dependency."""
+    return decoy.mock(cls=ProtocolEngine)
+
+
+@pytest.fixture
+def mock_hardware_api(decoy: Decoy) -> HardwareAPI:
+    """Get a mocked out HardwareAPI dependency."""
+    return decoy.mock(cls=HardwareAPI)
+
+
+@pytest.fixture
+def json_protocol_subject(
+    mock_protocol_engine: ProtocolEngine,
+    mock_hardware_api: HardwareAPI,
+    mock_protocol_json_runner: JsonRunner,
+    mock_fixit_runner: LiveRunner,
+    mock_setup_runner: LiveRunner,
+) -> RunOrchestrator:
+    """Get a RunOrchestrator subject with a json runner."""
+    return RunOrchestrator(
+        protocol_engine=mock_protocol_engine,
+        hardware_api=mock_hardware_api,
+        fixit_runner=mock_fixit_runner,
+        setup_runner=mock_setup_runner,
+        json_or_python_protocol_runner=mock_protocol_json_runner,
+    )
+
+
+@pytest.fixture
+def python_protocol_subject(
+    mock_protocol_engine: ProtocolEngine,
+    mock_hardware_api: HardwareAPI,
+    mock_protocol_python_runner: PythonAndLegacyRunner,
+    mock_fixit_runner: LiveRunner,
+    mock_setup_runner: LiveRunner,
+) -> RunOrchestrator:
+    """Get a RunOrchestrator subject with a python runner."""
+    return RunOrchestrator(
+        protocol_engine=mock_protocol_engine,
+        hardware_api=mock_hardware_api,
+        fixit_runner=mock_fixit_runner,
+        setup_runner=mock_setup_runner,
+        json_or_python_protocol_runner=mock_protocol_python_runner,
+    )
+
+
+@pytest.mark.parametrize(
+    "input_protocol_config, mock_protocol_runner, subject",
+    [
+        (
+            JsonProtocolConfig(schema_version=7),
+            lazy_fixture("mock_protocol_json_runner"),
+            lazy_fixture("json_protocol_subject"),
+        ),
+        (
+            PythonProtocolConfig(api_version=APIVersion(2, 14)),
+            lazy_fixture("mock_protocol_python_runner"),
+            lazy_fixture("python_protocol_subject"),
+        ),
+    ],
+)
+def test_build_run_orchestrator_provider(
+    decoy: Decoy,
+    monkeypatch: pytest.MonkeyPatch,
+    subject: RunOrchestrator,
+    mock_protocol_engine: ProtocolEngine,
+    mock_hardware_api: HardwareAPI,
+    input_protocol_config: Union[PythonProtocolConfig, JsonProtocolConfig],
+    mock_setup_runner: LiveRunner,
+    mock_fixit_runner: LiveRunner,
+    mock_protocol_runner: Union[PythonAndLegacyRunner, JsonRunner],
+) -> None:
+    """Should get a RunOrchestrator instance."""
+    mock_create_runner_func = decoy.mock(func=protocol_runner.create_protocol_runner)
+    monkeypatch.setattr(
+        protocol_runner, "create_protocol_runner", mock_create_runner_func
+    )
+
+    decoy.when(
+        mock_create_runner_func(
+            protocol_config=input_protocol_config,
+            protocol_engine=mock_protocol_engine,
+            hardware_api=mock_hardware_api,
+            post_run_hardware_state=PostRunHardwareState.HOME_AND_STAY_ENGAGED,
+            drop_tips_after_run=True,
+        )
+    ).then_return(mock_protocol_runner)
+
+    result = subject.build_orchestrator(
+        protocol_engine=mock_protocol_engine,
+        hardware_api=mock_hardware_api,
+        protocol_config=input_protocol_config,
+    )
+
+    assert isinstance(result, RunOrchestrator)
+    assert isinstance(result._setup_runner, LiveRunner)
+    assert isinstance(result._fixit_runner, LiveRunner)

robot-server/robot_server/runs/dependencies.py

@@ -27,7 +27,7 @@
 )
 
 from .run_auto_deleter import RunAutoDeleter
-from .engine_store import EngineStore, NoRunnerEnginePairError
+from .engine_store import EngineStore, NoRunnerEngineError
 from .run_store import RunStore
 from .run_data_manager import RunDataManager
 from robot_server.errors.robot_errors import (
@@ -131,7 +131,7 @@ async def get_is_okay_to_create_maintenance_run(
     """Whether a maintenance run can be created if a protocol run already exists."""
     try:
         protocol_run_state = engine_store.engine.state_view
-    except NoRunnerEnginePairError:
+    except NoRunnerEngineError:
         return True
     return (
         not protocol_run_state.commands.has_been_played()