feat(api): runtime parameters API for adding and using default parameters in protocols

api/src/opentrons/protocol_api/_parameters.py

@@ -0,0 +1,31 @@
+from typing import Dict, Optional, Any
+
+from opentrons.protocols.parameters.types import AllowedTypes, ParameterNameError
+
+
+class Parameters:
+    def __init__(self, parameters: Optional[Dict[str, AllowedTypes]] = None) -> None:
+        self._values: Dict[str, AllowedTypes] = {}
+        if parameters is not None:
+            for name, value in parameters.items():
+                self._initialize_parameter(name, value)
+
+    def _getparam(self, variable_name: str) -> Any:
+        return getattr(self, f"_{variable_name}")
+
+    def _initialize_parameter(self, variable_name: str, value: AllowedTypes) -> None:
+        if not hasattr(self, variable_name) and not hasattr(self, f"_{variable_name}"):
+            setattr(self, f"_{variable_name}", value)
+            prop = property(
+                fget=lambda s, v=variable_name: Parameters._getparam(s, v)  # type: ignore[misc]
+            )
+            setattr(Parameters, variable_name, prop)
+            self._values[variable_name] = value
+        else:
+            raise ParameterNameError(
+                f"Cannot use {variable_name} as a variable name, either duplicates another"
+                f" parameter name, Opentrons reserved function, or Python built-in"
+            )
+
+    def get_all(self) -> Dict[str, AllowedTypes]:
+        return self._values

api/src/opentrons/protocol_api/parameter_context.py

@@ -0,0 +1,169 @@
+"""Parameter context for python protocols."""
+
+from typing import List, Optional, Union
+
+from opentrons.protocols.api_support.types import APIVersion
+from opentrons.protocols.parameters import parameter_definition
+from opentrons.protocols.parameters.types import ParameterChoices
+
+from ._parameters import Parameters
+
+_ParameterDefinitionTypes = Union[
+    parameter_definition.ParameterDefinition[int],
+    parameter_definition.ParameterDefinition[bool],
+    parameter_definition.ParameterDefinition[float],
+    parameter_definition.ParameterDefinition[str],
+]
+
+
+class ParameterContext:
+    """Public context for adding parameters to a protocol."""
+
+    def __init__(self, api_version: APIVersion) -> None:
+        """Initializes a parameter context for user-set parameters."""
+        self._api_version = api_version
+        self._parameters: List[_ParameterDefinitionTypes] = []
+
+    def add_int(
+        self,
+        display_name: str,
+        variable_name: str,
+        default: int,
+        minimum: Optional[int] = None,
+        maximum: Optional[int] = None,
+        choices: Optional[List[ParameterChoices]] = None,
+        description: Optional[str] = None,
+        unit: Optional[str] = None,
+    ) -> None:
+        """Creates an integer parameter, settable within a given range or list of choices.
+
+        Arguments:
+            display_name: The display name of the int parameter as it would show up on the frontend.
+            variable_name: The variable name the int parameter will be referred to in the run context.
+            default: The default value the int parameter will be set to. This will be used in initial analysis.
+            minimum: The minimum value the int parameter can be set to (inclusive). Mutually exclusive with choices.
+            maximum: The maximum value the int parameter can be set to (inclusive). Mutually exclusive with choices.
+            choices: A list of possible choices that this parameter can be set to.
+                Mutually exclusive with minimum and maximum.
+            description: A description of the parameter as it will show up on the frontend.
+            unit: An optional unit to be appended to the end of the integer as it shown on the frontend.
+        """
+        self._parameters.append(
+            parameter_definition.create_int_parameter(
+                display_name=display_name,
+                variable_name=variable_name,
+                default=default,
+                minimum=minimum,
+                maximum=maximum,
+                choices=choices,
+                description=description,
+                unit=unit,
+            )
+        )
+
+    def add_float(
+        self,
+        display_name: str,
+        variable_name: str,
+        default: float,
+        minimum: Optional[float] = None,
+        maximum: Optional[float] = None,
+        choices: Optional[List[ParameterChoices]] = None,
+        description: Optional[str] = None,
+        unit: Optional[str] = None,
+    ) -> None:
+        """Creates a float parameter, settable within a given range or list of choices.
+
+        Arguments:
+            display_name: The display name of the float parameter as it would show up on the frontend.
+            variable_name: The variable name the float parameter will be referred to in the run context.
+            default: The default value the float parameter will be set to. This will be used in initial analysis.
+            minimum: The minimum value the float parameter can be set to (inclusive). Mutually exclusive with choices.
+            maximum: The maximum value the float parameter can be set to (inclusive). Mutually exclusive with choices.
+            choices: A list of possible choices that this parameter can be set to.
+                Mutually exclusive with minimum and maximum.
+            description: A description of the parameter as it will show up on the frontend.
+            unit: An optional unit to be appended to the end of the float as it shown on the frontend.
+        """
+        self._parameters.append(
+            parameter_definition.create_float_parameter(
+                display_name=display_name,
+                variable_name=variable_name,
+                default=default,
+                minimum=minimum,
+                maximum=maximum,
+                choices=choices,
+                description=description,
+                unit=unit,
+            )
+        )
+
+    def add_bool(
+        self,
+        display_name: str,
+        variable_name: str,
+        default: bool,
+        description: Optional[str] = None,
+    ) -> None:
+        """Creates a boolean parameter with allowable values of "On" (True) or "Off" (False).
+
+        Arguments:
+            display_name: The display name of the boolean parameter as it would show up on the frontend.
+            variable_name: The variable name the boolean parameter will be referred to in the run context.
+            default: The default value the boolean parameter will be set to. This will be used in initial analysis.
+            description: A description of the parameter as it will show up on the frontend.
+        """
+        self._parameters.append(
+            parameter_definition.create_bool_parameter(
+                display_name=display_name,
+                variable_name=variable_name,
+                default=default,
+                choices=[
+                    {"display_name": "On", "value": True},
+                    {"display_name": "Off", "value": False},
+                ],
+                description=description,
+            )
+        )
+
+    def add_str(
+        self,
+        display_name: str,
+        variable_name: str,
+        default: str,
+        choices: Optional[List[ParameterChoices]] = None,
+        description: Optional[str] = None,
+    ) -> None:
+        """Creates a string parameter, settable among given choices.
+
+        Arguments:
+            display_name: The display name of the string parameter as it would show up on the frontend.
+            variable_name: The variable name the string parameter will be referred to in the run context.
+            default: The default value the string parameter will be set to. This will be used in initial analysis.
+            choices: A list of possible choices that this parameter can be set to.
+                Mutually exclusive with minimum and maximum.
+            description: A description of the parameter as it will show up on the frontend.
+        """
+        self._parameters.append(
+            parameter_definition.create_str_parameter(
+                display_name=display_name,
+                variable_name=variable_name,
+                default=default,
+                choices=choices,
+                description=description,
+            )
+        )
+
+    def export_parameters(self) -> Parameters:
+        """Exports all parameters into a protocol run usable parameters object.
+
+        :meta private:
+
+        This is intended for Opentrons internal use only and is not a guaranteed API.
+        """
+        return Parameters(
+            parameters={
+                parameter.variable_name: parameter.value
+                for parameter in self._parameters
+            }
+        )

api/src/opentrons/protocol_api/protocol_context.py

@@ -64,6 +64,7 @@
     MagneticBlockContext,
     ModuleContext,
 )
+from ._parameters import Parameters
 
 
 logger = logging.getLogger(__name__)
@@ -167,6 +168,7 @@ def __init__(
             self._core.load_ot2_fixed_trash_bin()
 
         self._commands: List[str] = []
+        self._params: Parameters = Parameters()
         self._unsubscribe_commands: Optional[Callable[[], None]] = None
         self.clear_commands()
 
@@ -215,6 +217,10 @@ def bundled_data(self) -> Dict[str, bytes]:
         """
         return self._bundled_data
 
+    @property
+    def params(self) -> Parameters:
+        return self._params
+
     def cleanup(self) -> None:
         """Finalize and clean up the protocol context."""
         if self._unsubscribe_commands:

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

@@ -7,8 +7,12 @@
 
 from opentrons.drivers.smoothie_drivers.errors import SmoothieAlarm
 from opentrons.protocol_api import ProtocolContext
+from opentrons.protocol_api._parameters import Parameters
+from opentrons.protocol_api.parameter_context import ParameterContext
 from opentrons.protocols.execution.errors import ExceptionInProtocolError
 from opentrons.protocols.types import PythonProtocol, MalformedPythonProtocolError
+
+
 from opentrons_shared_data.errors.exceptions import ExecutionCancelledError
 
 MODULE_LOG = logging.getLogger(__name__)
@@ -29,6 +33,14 @@ def _runfunc_ok(run_func: Any):
                 )
 
 
+def _add_parameters_func_ok(add_parameters_func: Any) -> None:
+    if not callable(add_parameters_func):
+        raise SyntaxError("'add_parameters' must be a function.")
+    sig = inspect.Signature.from_callable(add_parameters_func)
+    if len(sig.parameters) != 1:
+        raise SyntaxError("Function 'add_parameters' must take exactly one argument.")
+
+
 def _find_protocol_error(tb, proto_name):
     """Return the FrameInfo for the lowest frame in the traceback from the
     protocol.
@@ -41,6 +53,34 @@ def _find_protocol_error(tb, proto_name):
         raise KeyError
 
 
+def _raise_pretty_protocol_error(exception: Exception, filename: str) -> None:
+    exc_type, exc_value, tb = sys.exc_info()
+    try:
+        frame = _find_protocol_error(tb, filename)
+    except KeyError:
+        # No pretty names, just raise it
+        raise exception
+    raise ExceptionInProtocolError(
+        exception, tb, str(exception), frame.lineno
+    ) from exception
+
+
+def _parse_and_set_parameters(
+    protocol: PythonProtocol, new_globs: Dict[Any, Any], filename: str
+) -> Parameters:
+    try:
+        _add_parameters_func_ok(new_globs.get("add_parameters"))
+    except SyntaxError as se:
+        raise MalformedPythonProtocolError(str(se))
+    parameter_context = ParameterContext(api_version=protocol.api_level)
+    new_globs["__param_context"] = parameter_context
+    try:
+        exec("add_parameters(__param_context)", new_globs)
+    except Exception as e:
+        _raise_pretty_protocol_error(exception=e, filename=filename)
+    return parameter_context.export_parameters()
+
+
 def run_python(proto: PythonProtocol, context: ProtocolContext):
     new_globs: Dict[Any, Any] = {}
     exec(proto.contents, new_globs)
@@ -60,10 +100,14 @@ def run_python(proto: PythonProtocol, context: ProtocolContext):
         # AST filename.
         filename = proto.filename or "<protocol>"
 
+    if new_globs.get("add_parameters"):
+        context._params = _parse_and_set_parameters(proto, new_globs, filename)
+
     try:
         _runfunc_ok(new_globs.get("run"))
     except SyntaxError as se:
         raise MalformedPythonProtocolError(str(se))
+
     new_globs["__context"] = context
     try:
         exec("run(__context)", new_globs)
@@ -75,10 +119,4 @@ def run_python(proto: PythonProtocol, context: ProtocolContext):
         # this is a protocol cancel and shouldn't have special logging
         raise
     except Exception as e:
-        exc_type, exc_value, tb = sys.exc_info()
-        try:
-            frame = _find_protocol_error(tb, filename)
-        except KeyError:
-            # No pretty names, just raise it
-            raise e
-        raise ExceptionInProtocolError(e, tb, str(e), frame.lineno) from e
+        _raise_pretty_protocol_error(exception=e, filename=filename)