feat(api): allow setting runtime parameter values and CSV files in cl…

…i analysis (#16387)

# Overview

Closes AUTH-873.

This PR adds two new command line arguments to allow the setting and
override of runtime parameters in cli analysis, for both basic RTP types
(boolean, string, integer, and float) as well as CSV files.

The two new arguments are `--rtp-values` for the basic RTP values and
`--rtp-files` for CSV files. Both take similar input in the form of
serialized json strings.

For RTP values the data passed is in the form of `{variable_name:
value}` and for CSV files it is in the form of `{variable_name:
/path/to/file}`.

Because the CSV file that is used for local analysis is not a data file
stored on the robot, and therefore does not have a UUID associated for
it in the `"id"` property for the `file` property in the runtime
parameters reported we substitute an empty string. The code to find the
UUID currently assumes it to be the parent folder name, so logic was
added to check if it is a UUID and change it to an empty string if it is
not, to prevent the leaking of folder names into analysis.

## Test Plan and Hands on Testing

Tested local analysis with the following two protocols and the following
RTP arguments and made sure analysis reflected the changed values:

```
requirements = {
    "robotType": "Flex",
    "apiLevel": "2.18"
}


def add_parameters(parameters):
    parameters.add_int(
        display_name="Sample count",
        variable_name="sample_count",
        default=2,
        minimum=1,
        maximum=6,
        description="How many samples to process."
    )
    parameters.add_float(
        display_name="Pipette volume",
        variable_name="volume",
        default=20,
        choices=[
            {"display_name": "Low Volume", "value": 10.23},
            {"display_name": "Medium Volume", "value": 20.0},
            {"display_name": "High Volume", "value": 50.5},
        ],
        description="How many microliters to pipette of each sample.",
    )
    parameters.add_bool(
        display_name="Dry Run",
        variable_name="dry_run",
        default=False,
        description="Skip aspirate and dispense steps."
    )
    parameters.add_str(
        display_name="Pipette Name",
        variable_name="pipette",
        choices=[
            {"display_name": "Single channel 50µL", "value": "flex_1channel_50"},
            {"display_name": "Single channel 1000µL", "value": "flex_1channel_1000"},
            {"display_name": "Eight Channel 50µL", "value": "flex_8channel_50"},
            {"display_name": "Eight Channel 1000µL", "value": "flex_8channel_1000"},
        ],
        default="flex_1channel_50",
        description="What pipette to use during the protocol.",
    )



def run(context):
    pass
```

with `--rtp-values='{"sample_count": 3, "volume": 10.23, "dry_run":
true, "pipette": "flex_8channel_50"}'`

and

```
requirements = {
    "robotType": "Flex",
    "apiLevel": "2.20"
}


def add_parameters(parameters):
    parameters.add_str(
        display_name="Pipette Name",
        variable_name="pipette",
        choices=[
            {"display_name": "Single channel 50µL", "value": "flex_1channel_50"},
            {"display_name": "Single channel 1000µL", "value": "flex_1channel_1000"},
            {"display_name": "Eight Channel 50µL", "value": "flex_8channel_50"},
            {"display_name": "Eight Channel 1000µL", "value": "flex_8channel_1000"},
        ],
        default="flex_1channel_50",
        description="What pipette to use during the protocol.",
    )
    parameters.add_csv_file(
        display_name="CSV Data",
        variable_name="csv_data",
        description="CSV file containing labware and volume information."
    )



def run(context):
    context.params.csv_data.file
```

with `--rtp-files='{"csv_data":
"/Users/jeremyleon/Downloads/example1.csv"}'`

## Changelog

- added `--rtp-values` and `--rtp-files` optional command line arguments
for cli analysis to allow overrides for RTP values and setting a local
file to be used for CSV file protocols.

## Review requests

Should we put in some special value for `"id"` when using analysis?
Right now it's defaulting to an empty string and it is a client-only
concern, but we may want a specific value at some point. Or have the
analysis take a slightly different shape when it's local CLI analysis.

## Risk assessment

Low, only affects CLI code and new arguments do not change

api/src/opentrons/cli/analyze.py

@@ -23,8 +23,14 @@
 )
 import logging
 import sys
+import json
 
-from opentrons.protocol_engine.types import RunTimeParameter, EngineStatus
+from opentrons.protocol_engine.types import (
+    RunTimeParameter,
+    CSVRuntimeParamPaths,
+    PrimitiveRunTimeParamValuesType,
+    EngineStatus,
+)
 from opentrons.protocols.api_support.types import APIVersion
 from opentrons.protocol_reader import (
     ProtocolReader,
@@ -104,8 +110,22 @@ class _Output:
     type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"], case_sensitive=False),
     default="WARNING",
 )
+@click.option(
+    "--rtp-values",
+    help="Serialized JSON of runtime parameter variable names to values.",
+    default="{}",
+    type=str,
+)
+@click.option(
+    "--rtp-files",
+    help="Serialized JSON of runtime parameter variable names to file paths.",
+    default="{}",
+    type=str,
+)
 def analyze(
     files: Sequence[Path],
+    rtp_values: str,
+    rtp_files: str,
     json_output: Optional[IO[bytes]],
     human_json_output: Optional[IO[bytes]],
     log_output: str,
@@ -125,7 +145,7 @@ def analyze(
 
     try:
         with _capture_logs(log_output, log_level):
-            sys.exit(run(_analyze, files, outputs, check))
+            sys.exit(run(_analyze, files, rtp_values, rtp_files, outputs, check))
     except click.ClickException:
         raise
     except Exception as e:
@@ -194,6 +214,37 @@ def _get_input_files(files_and_dirs: Sequence[Path]) -> List[Path]:
     return results
 
 
+def _get_runtime_parameter_values(
+    serialized_rtp_values: str,
+) -> PrimitiveRunTimeParamValuesType:
+    rtp_values = {}
+    try:
+        for variable_name, value in json.loads(serialized_rtp_values).items():
+            if not isinstance(value, (bool, int, float, str)):
+                raise click.BadParameter(
+                    f"Runtime parameter '{value}' is not of allowed type boolean, integer, float or string",
+                    param_hint="--rtp-values",
+                )
+            rtp_values[variable_name] = value
+    except json.JSONDecodeError as error:
+        raise click.BadParameter(
+            f"JSON decode error: {error}", param_hint="--rtp-values"
+        )
+    return rtp_values
+
+
+def _get_runtime_parameter_paths(serialized_rtp_files: str) -> CSVRuntimeParamPaths:
+    try:
+        return {
+            variable_name: Path(path_string)
+            for variable_name, path_string in json.loads(serialized_rtp_files).items()
+        }
+    except json.JSONDecodeError as error:
+        raise click.BadParameter(
+            f"JSON decode error: {error}", param_hint="--rtp-files"
+        )
+
+
 R = TypeVar("R")
 
 
@@ -238,7 +289,11 @@ def _convert_exc() -> Iterator[EnumeratedError]:
         )
 
 
-async def _do_analyze(protocol_source: ProtocolSource) -> RunResult:
+async def _do_analyze(
+    protocol_source: ProtocolSource,
+    rtp_values: PrimitiveRunTimeParamValuesType,
+    rtp_paths: CSVRuntimeParamPaths,
+) -> RunResult:
 
     orchestrator = await create_simulating_orchestrator(
         robot_type=protocol_source.robot_type, protocol_config=protocol_source.config
@@ -247,8 +302,8 @@ async def _do_analyze(protocol_source: ProtocolSource) -> RunResult:
         await orchestrator.load(
             protocol_source=protocol_source,
             parse_mode=ParseMode.NORMAL,
-            run_time_param_values=None,
-            run_time_param_paths=None,
+            run_time_param_values=rtp_values,
+            run_time_param_paths=rtp_paths,
         )
     except Exception as error:
         err_id = "analysis-setup-error"
@@ -285,9 +340,16 @@ async def _do_analyze(protocol_source: ProtocolSource) -> RunResult:
 
 
 async def _analyze(
-    files_and_dirs: Sequence[Path], outputs: Sequence[_Output], check: bool
+    files_and_dirs: Sequence[Path],
+    rtp_values: str,
+    rtp_files: str,
+    outputs: Sequence[_Output],
+    check: bool,
 ) -> int:
     input_files = _get_input_files(files_and_dirs)
+    parsed_rtp_values = _get_runtime_parameter_values(rtp_values)
+    rtp_paths = _get_runtime_parameter_paths(rtp_files)
+
     try:
         protocol_source = await ProtocolReader().read_saved(
             files=input_files,
@@ -296,7 +358,7 @@ async def _analyze(
     except ProtocolFilesInvalidError as error:
         raise click.ClickException(str(error))
 
-    analysis = await _do_analyze(protocol_source)
+    analysis = await _do_analyze(protocol_source, parsed_rtp_values, rtp_paths)
     return_code = _get_return_code(analysis)
 
     if not outputs:

api/src/opentrons/protocol_api/_parameter_context.py

@@ -1,4 +1,5 @@
 """Parameter context for python protocols."""
+import uuid
 from typing import List, Optional, Union, Dict
 
 from opentrons.protocols.api_support.types import APIVersion
@@ -251,8 +252,16 @@ def initialize_csv_files(
                     f" but '{variable_name}' is not a CSV parameter."
                 )
 
-            # The parent folder in the path will be the file ID, so we can use that to resolve that here
+            # TODO(jbl 2024-09-30) Refactor this so file ID is passed as its own argument and not assumed from the path
+            # If this is running on a robot, the parent folder in the path will be the file ID
+            # If it is running locally, most likely the parent folder will not be a UUID, so instead we will change
+            # this to be an empty string
             file_id = file_path.parent.name
+            try:
+                uuid.UUID(file_id, version=4)
+            except ValueError:
+                file_id = ""
+
             file_name = file_path.name
 
             with file_path.open("rb") as fh:

api/tests/opentrons/cli/test_cli.py

@@ -28,7 +28,11 @@ class _AnalysisCLIResult:
 
 
 def _get_analysis_result(
-    protocol_files: List[Path], output_type: str, check: bool = False
+    protocol_files: List[Path],
+    output_type: str,
+    check: bool = False,
+    rtp_values: Optional[str] = None,
+    rtp_files: Optional[str] = None,
 ) -> _AnalysisCLIResult:
     """Run `protocol_files` as a single protocol through the analysis CLI.
 
@@ -41,11 +45,16 @@ def _get_analysis_result(
     with tempfile.TemporaryDirectory() as temp_dir:
         analysis_output_file = Path(temp_dir) / "analysis_output.json"
         runner = CliRunner()
-        args = [
-            output_type,
-            str(analysis_output_file),
-            *[str(p.resolve()) for p in protocol_files],
-        ]
+        args = [output_type, str(analysis_output_file)]
+
+        if rtp_values is not None:
+            args.extend(["--rtp-values", rtp_values])
+
+        if rtp_files is not None:
+            args.extend(["--rtp-files", rtp_files])
+
+        args.extend([str(p.resolve()) for p in protocol_files])
+
         if check:
             args.append("--check")
 
@@ -262,6 +271,61 @@ def test_python_error_line_numbers(
     assert error["detail"] == expected_detail
 
 
+@pytest.mark.parametrize("output", ["--json-output", "--human-json-output"])
+def test_run_time_parameter_setting(
+    tmp_path: Path,
+    output: str,
+) -> None:
+    """Test that a RTP can be set to a non default value for analysis.
+
+    Also verify that analysis result contains all static data about the protocol.
+    """
+    python_protocol_source = textwrap.dedent(
+        """\
+            requirements = {"robotType": "OT-2", "apiLevel": "2.18"}
+
+            def add_parameters(parameters):
+                parameters.add_bool(
+                    display_name="Dry Run",
+                    variable_name="dry_run",
+                    default=False,
+                )
+            def run(protocol):
+                pass
+        """
+    )
+    protocol_source_file = tmp_path / "protocol.py"
+    protocol_source_file.write_text(python_protocol_source, encoding="utf-8")
+    result = _get_analysis_result(
+        [protocol_source_file], output, rtp_values=json.dumps({"dry_run": True})
+    )
+
+    assert result.exit_code == 0
+
+    assert result.json_output is not None
+    assert result.json_output["robotType"] == "OT-2 Standard"
+    assert result.json_output["result"] == AnalysisResult.OK
+    assert result.json_output["pipettes"] == []
+    assert result.json_output["commands"]  # There should be a home command
+    assert result.json_output["labware"] == []
+    assert result.json_output["liquids"] == []
+    assert result.json_output["modules"] == []
+    assert result.json_output["config"] == {
+        "apiVersion": [2, 18],
+        "protocolType": "python",
+    }
+    assert result.json_output["files"] == [{"name": "protocol.py", "role": "main"}]
+    assert result.json_output["runTimeParameters"] == [
+        {
+            "displayName": "Dry Run",
+            "variableName": "dry_run",
+            "type": "bool",
+            "value": True,
+            "default": False,
+        }
+    ]
+
+
 @pytest.mark.parametrize("output", ["--json-output", "--human-json-output"])
 def test_run_time_parameter_error(
     tmp_path: Path,
@@ -312,6 +376,64 @@ def run(protocol):
     )
 
 
+@pytest.mark.parametrize("output", ["--json-output", "--human-json-output"])
+def test_rtp_csv_file_setting(
+    tmp_path: Path,
+    output: str,
+) -> None:
+    """Test that a CSV file can be set for analysis.
+
+    Also verify that analysis result contains all static data about the protocol.
+    """
+    python_protocol_source = textwrap.dedent(
+        """\
+            requirements = {"robotType": "OT-2", "apiLevel": "2.20"}
+
+            def add_parameters(parameters):
+                parameters.add_csv_file(
+                    display_name="CSV File",
+                    variable_name="csv_file",
+                )
+            def run(protocol):
+                protocol.params.csv_file.contents
+        """
+    )
+    protocol_source_file = tmp_path / "protocol.py"
+    protocol_source_file.write_text(python_protocol_source, encoding="utf-8")
+    csv_source_file = tmp_path / "csv_file.csv"
+    csv_source_file.write_text("a,b,c", encoding="utf-8")
+
+    result = _get_analysis_result(
+        [protocol_source_file],
+        output,
+        rtp_files=json.dumps({"csv_file": str(csv_source_file.resolve())}),
+    )
+
+    assert result.exit_code == 0
+
+    assert result.json_output is not None
+    assert result.json_output["robotType"] == "OT-2 Standard"
+    assert result.json_output["result"] == AnalysisResult.OK
+    assert result.json_output["pipettes"] == []
+    assert result.json_output["commands"]  # There should be a home command
+    assert result.json_output["labware"] == []
+    assert result.json_output["liquids"] == []
+    assert result.json_output["modules"] == []
+    assert result.json_output["config"] == {
+        "apiVersion": [2, 20],
+        "protocolType": "python",
+    }
+    assert result.json_output["files"] == [{"name": "protocol.py", "role": "main"}]
+    assert result.json_output["runTimeParameters"] == [
+        {
+            "displayName": "CSV File",
+            "variableName": "csv_file",
+            "type": "csv_file",
+            "file": {"id": "", "name": "csv_file.csv"},
+        }
+    ]
+
+
 @pytest.mark.parametrize("output", ["--json-output", "--human-json-output"])
 def test_file_required_error(
     tmp_path: Path,