feat(api): allow setting runtime parameter values and CSV files in cli analysis

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,31 @@ 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():
+            assert isinstance(
+                value, (bool, int, float, str)
+            ), f"Value '{value}' is not of allowed type boolean, integer, float or string"
+            rtp_values[variable_name] = value
+    except Exception as error:
+        raise click.ClickException(f"Could not parse runtime parameter values: {error}")
+    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 Exception as error:
+        raise click.ClickException(f"Could not parse runtime parameter files: {error}")
+
+
 R = TypeVar("R")
 
 
@@ -238,7 +283,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 +296,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 +334,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 +352,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='{"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.file
+        """
+    )
+    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='{"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,