extract rtp definitions and set rtp values in runner load

api/src/opentrons/execute.py

@@ -602,7 +602,9 @@ def _run_file_non_pe(
     try:
         # TODO (spp, 2024-03-18): use true run-time param overrides once enabled
         #  for cli protocol simulation/ execution
-        execute_apiv2.run_protocol(protocol, context, run_time_param_overrides=None)
+        execute_apiv2.run_protocol(
+            protocol, context, run_time_parameters_with_overrides=None
+        )
     finally:
         context.cleanup()
 

api/src/opentrons/protocol_runner/protocol_runner.py

@@ -43,6 +43,7 @@
     RunTimeParameter,
     RunTimeParamValuesType,
 )
+from ..protocols.types import PythonProtocol
 
 
 class RunResult(NamedTuple):
@@ -193,7 +194,17 @@ async def load(
         protocol = self._protocol_file_reader.read(
             protocol_source, labware_definitions, python_parse_mode
         )
-        self._parameter_context = ParameterContext(api_version=protocol.api_level)
+        if isinstance(protocol, PythonProtocol):
+            self._parameter_context = ParameterContext(api_version=protocol.api_level)
+            run_time_parameters_with_overrides = (
+                self._protocol_executor.extract_run_parameters(
+                    protocol=protocol,
+                    parameter_context=self._parameter_context,
+                    run_time_param_overrides=run_time_param_values,
+                )
+            )
+        else:
+            run_time_parameters_with_overrides = None
         equipment_broker = None
 
         if protocol.api_level < LEGACY_PYTHON_API_VERSION_CUTOFF:
@@ -224,8 +235,7 @@ async def run_func() -> None:
             await self._protocol_executor.execute(
                 protocol=protocol,
                 context=context,
-                parameter_context=self._parameter_context,
-                run_time_param_values=run_time_param_values,
+                run_time_parameters_with_overrides=run_time_parameters_with_overrides,
             )
 
         self._task_queue.set_run_func(run_func)

api/src/opentrons/protocol_runner/python_protocol_wrappers.py

@@ -21,12 +21,14 @@
     ProtocolContext,
     ParameterContext,
     create_protocol_context,
+    Parameters,
 )
 from opentrons.protocol_api.core.engine import ENGINE_CORE_API_VERSION
 from opentrons.protocol_api.core.legacy.load_info import LoadInfo
 
 from opentrons.protocols.parse import PythonParseMode, parse
 from opentrons.protocols.execution.execute import run_protocol
+from opentrons.protocols.execution.execute_python import exec_add_parameters
 from opentrons.protocols.types import Protocol, PythonProtocol
 
 
@@ -148,10 +150,22 @@ class PythonProtocolExecutor:
     async def execute(
         protocol: Protocol,
         context: ProtocolContext,
-        parameter_context: Optional[ParameterContext],
-        run_time_param_values: Optional[RunTimeParamValuesType],
+        run_time_parameters_with_overrides: Optional[Parameters],
     ) -> None:
         """Execute a PAPIv2 protocol with a given ProtocolContext in a child thread."""
         await to_thread.run_sync(
-            run_protocol, protocol, context, parameter_context, run_time_param_values
+            run_protocol, protocol, context, run_time_parameters_with_overrides
+        )
+
+    @staticmethod
+    def extract_run_parameters(
+        protocol: PythonProtocol,
+        parameter_context: ParameterContext,
+        run_time_param_overrides: Optional[RunTimeParamValuesType],
+    ) -> Optional[Parameters]:
+        """Extract the parameters defined in the protocol, overridden with values for the run."""
+        return exec_add_parameters(
+            protocol=protocol,
+            parameter_context=parameter_context,
+            run_time_param_overrides=run_time_param_overrides,
         )

api/src/opentrons/protocols/execution/execute.py

@@ -1,8 +1,8 @@
 import logging
 from typing import Optional
 
-from opentrons.protocol_api import ProtocolContext, ParameterContext
-from opentrons.protocol_engine.types import RunTimeParamValuesType
+from opentrons.protocol_api import ProtocolContext
+from opentrons.protocol_api._parameters import Parameters
 from opentrons.protocols.execution.execute_python import run_python
 from opentrons.protocols.execution.json_dispatchers import (
     pipette_command_map,
@@ -21,27 +21,23 @@
 def run_protocol(
     protocol: Protocol,
     context: ProtocolContext,
-    parameter_context: Optional[ParameterContext] = None,
-    run_time_param_overrides: Optional[RunTimeParamValuesType] = None,
+    run_time_parameters_with_overrides: Optional[Parameters] = None,
 ) -> None:
     """Run a protocol.
 
     :param protocol: The :py:class:`.protocols.types.Protocol` to execute
     :param context: The protocol context to use.
-    :param parameter_context: The parameter context to use.
-    :param run_time_param_overrides: Any parameter values that are potentially overriding the defaults
+    :param run_time_parameters_with_overrides: Run time parameters defined in the protocol,
+        updated with the run's RTP override values.
     """
     if isinstance(protocol, PythonProtocol):
         if protocol.api_level >= APIVersion(2, 0):
             # If this is None here then we're either running simulate or execute, in any case we don't need to report
             # this in analysis which is the reason we'd pass it to this function
-            if parameter_context is None:
-                parameter_context = ParameterContext(protocol.api_level)
             run_python(
                 proto=protocol,
                 context=context,
-                parameter_context=parameter_context,
-                run_time_param_overrides=run_time_param_overrides,
+                run_time_parameters_with_overrides=run_time_parameters_with_overrides,
             )
         else:
             raise RuntimeError(f"Unsupported python API version: {protocol.api_level}")

api/src/opentrons/protocols/execution/execute_python.py

@@ -85,34 +85,57 @@ def _parse_and_set_parameters(
     return parameter_context.export_parameters_for_protocol()
 
 
-def run_python(
-    proto: PythonProtocol,
-    context: ProtocolContext,
-    parameter_context: ParameterContext,
-    run_time_param_overrides: Optional[RunTimeParamValuesType] = None,
-) -> None:
-    new_globs: Dict[Any, Any] = {}
-    exec(proto.contents, new_globs)
-    # If the protocol is written correctly, it will have defined a function
-    # like run(context: ProtocolContext). If so, that function is now in the
-    # current scope.
-
+def _get_filename(
+    protocol: PythonProtocol,
+) -> str:
     # TODO(mm, 2023-10-11): This coupling to opentrons.protocols.parse is fragile.
     # Can we get the correct filename directly from proto.contents?
-    if proto.filename and proto.filename.endswith("zip"):
+    if protocol.filename and protocol.filename.endswith("zip"):
         # The ".zip" extension needs to match what opentrons.protocols.parse recognizes as a bundle,
         # and the "protocol.ot2.py" fallback needs to match what opentrons.protocol.py sets as the
         # AST filename.
-        filename = "protocol.ot2.py"
+        return "protocol.ot2.py"
     else:
         # "<protocol>" needs to match what opentrons.protocols.parse sets as the fallback
         # AST filename.
-        filename = proto.filename or "<protocol>"
+        return protocol.filename or "<protocol>"
+
 
-    if new_globs.get("add_parameters"):
-        context._params = _parse_and_set_parameters(
-            parameter_context, run_time_param_overrides, new_globs, filename
+def exec_add_parameters(
+    protocol: PythonProtocol,
+    parameter_context: ParameterContext,
+    run_time_param_overrides: Optional[RunTimeParamValuesType],
+) -> Optional[Parameters]:
+    """Exec the add_parameters function and get the final run time parameters with overrides."""
+    new_globs: Dict[Any, Any] = {}
+    exec(protocol.contents, new_globs)
+    filename = _get_filename(protocol)
+
+    return (
+        _parse_and_set_parameters(
+            parameter_context=parameter_context,
+            run_time_param_overrides=run_time_param_overrides,
+            new_globs=new_globs,
+            filename=filename,
         )
+        if new_globs.get("add_parameters")
+        else None
+    )
+
+
+def run_python(
+    proto: PythonProtocol,
+    context: ProtocolContext,
+    run_time_parameters_with_overrides: Optional[Parameters] = None,
+) -> None:
+    new_globs: Dict[Any, Any] = {}
+    exec(proto.contents, new_globs)
+    # If the protocol is written correctly, it will have defined a function
+    # like run(context: ProtocolContext). If so, that function is now in the
+    # current scope.
+    filename = _get_filename(proto)
+    if run_time_parameters_with_overrides:
+        context._params = run_time_parameters_with_overrides
 
     try:
         _runfunc_ok(new_globs.get("run"))

api/src/opentrons/simulate.py

@@ -874,7 +874,9 @@ def _run_file_non_pe(
         try:
             # TODO (spp, 2024-03-18): use true run-time param overrides once enabled
             #  for cli protocol simulation/ execution
-            execute.run_protocol(protocol, context, run_time_param_overrides=None)
+            execute.run_protocol(
+                protocol, context, run_time_parameters_with_overrides=None
+            )
             if (
                 isinstance(protocol, PythonProtocol)
                 and protocol.api_level >= APIVersion(2, 0)

api/tests/opentrons/data/testosaur_with_rtp.py

@@ -0,0 +1,44 @@
+from opentrons import protocol_api
+
+metadata = {
+    "protocolName": "Testosaur with RTP",
+    "author": "Opentrons <[email protected]>",
+    "description": 'A variant on "Dinosaur" for testing with Run time parameters',
+    "source": "Opentrons Repository",
+    "apiLevel": "2.18",
+}
+
+
+def add_parameters(parameters: protocol_api.ParameterContext) -> None:
+    parameters.add_int(
+        display_name="Sample count",
+        variable_name="sample_count",
+        default=3,
+        minimum=1,
+        maximum=6,
+        description="How many samples to process.",
+    )
+    parameters.add_str(
+        display_name="Mount",
+        variable_name="mount",
+        choices=[
+            {"display_name": "Left Mount", "value": "left"},
+            {"display_name": "Right Mount", "value": "right"},
+        ],
+        default="left",
+        description="What mount to use.",
+    )
+
+
+def run(ctx: protocol_api.ProtocolContext) -> None:
+    tip_rack = ctx.load_labware("opentrons_96_tiprack_300ul", 8)
+    source = ctx.load_labware("nest_12_reservoir_15ml", 1)
+    dest = ctx.load_labware("corning_96_wellplate_360ul_flat", 2)
+
+    pipette = ctx.load_instrument("p300_single_gen2", ctx.params.mount, [tip_rack])  # type: ignore[attr-defined]
+
+    for i in range(ctx.params.sample_count):  # type: ignore[attr-defined]
+        pipette.pick_up_tip()
+        pipette.aspirate(50, source.wells_by_name()["A1"])
+        pipette.dispense(50, dest.wells()[i])
+        pipette.return_tip()

api/tests/opentrons/protocol_runner/smoke_tests/conftest.py

@@ -117,6 +117,58 @@ def run(ctx):
     return path
 
 
+@pytest.fixture()
+def python_protocol_file_with_run_time_params(tmp_path: Path) -> Path:
+    """Get minimal Python protocol input "file" with run time parameters."""
+    path = tmp_path / "protocol-name.py"
+    path.write_text(
+        textwrap.dedent(
+            """
+            # my protocol
+            metadata = {
+                "apiLevel": "2.18",
+            }
+            def add_parameters(params):
+                params.add_float(
+                    display_name="Aspirate volume",
+                    variable_name="aspirate_volume",
+                    default=25.5,
+                    minimum=10,
+                    maximum=50,
+                )
+                params.add_str(
+                    display_name="Mount",
+                    variable_name="mount",
+                    choices=[
+                        {"display_name": "Left Mount", "value": "left"},
+                        {"display_name": "Right Mount", "value": "right"},
+                    ],
+                    default="left",
+                )
+            def run(ctx):
+                pipette = ctx.load_instrument(
+                    instrument_name="p300_single",
+                    mount=ctx.params.mount,
+                )
+                tip_rack = ctx.load_labware(
+                    load_name="opentrons_96_tiprack_300ul",
+                    location="1",
+                )
+                reservoir = ctx.load_labware(
+                    load_name="nest_1_reservoir_195ml",
+                    location=2,
+                )
+                pipette.pick_up_tip(
+                    location=tip_rack.wells_by_name()["A1"],
+                )
+                pipette.aspirate(ctx.params.aspirate_volume, reservoir.wells()[0])
+            """
+        )
+    )
+
+    return path
+
+
 @pytest.fixture()
 def legacy_python_protocol_file(tmp_path: Path) -> Path:
     """Get an on-disk, minimal Python protocol fixture."""