diff --git a/api/src/opentrons/cli/analyze.py b/api/src/opentrons/cli/analyze.py index 0b08b20e17e..f311adce402 100644 --- a/api/src/opentrons/cli/analyze.py +++ b/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: diff --git a/api/src/opentrons/protocol_api/_parameter_context.py b/api/src/opentrons/protocol_api/_parameter_context.py index 2e0e0096f44..a52aee7819b 100644 --- a/api/src/opentrons/protocol_api/_parameter_context.py +++ b/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: diff --git a/api/tests/opentrons/cli/test_cli.py b/api/tests/opentrons/cli/test_cli.py index 79d46dc1000..717e2ac6a2d 100644 --- a/api/tests/opentrons/cli/test_cli.py +++ b/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,