feat(robot-server): show run time parameters in pending analysis (#15229

)

Closes AUTH-282

# Overview

Follow-up to #15217 

This PR changes a *pending analysis'* `AnalysisSummary` to include the
run time parameters (with overrides).
In order to do this, we will now be parsing the RTPs before analysis
response is created. This is a very quick process and shouldn't cause
any perceptible delays in the response of neither `POST /protocols` nor
`POST /runs/.../analyses` endpoints, as long as recommended python
protocol guidelines are followed.

# Test Plan

- [ ] Upload a protocol with run time parameters, verify that the
response contains run time parameter definitions along with the values
specified in the request
- [ ] Send a `POST` request on `/protocols/{protocolId}/analyses`
endpoint with different RTP values, or by specifying the
`forceReanalysis` flag, and verify that the newly pending analysis also
contains (updated) RTPs
- [ ] Verify that pending summaries in `GET` endpoints also contain RTPs
- [ ] Verify that summaries of completed analyses don't expose RTPs,
while RTPs are available in the full analyses.

# Changelog

- Main change is the introduction of `AnalysesManager`. The analyses
manager is responsible for creating a protocol analyzer per analysis,
extracting the run time parameters from the runner and starting the
analysis.
- To facilitate this extraction of run time parameters, the
`ProtocolAnalyzer` has been divided into two functions- one that loads
the runner and other that runs the runner.
- Updated the router to use `AnalysesManager` for starting analyses
instead of directly interacting with `ProtocolAnalyzer`.

# Review requests

- Does the new architecture of `AnalysesManager` & updated
`ProtocolAnalyzer` makes sense.
- This PR doesn't expose the run time parameters in analysis summaries
of completed analyses because of two reasons- (1) as opposed to the RTPs
of a pending analysis, the completed analysis isn't guaranteed to be
cached in memory. So we would need to read the completed analysis from
database, de-serialize it and then return it. This would be made easier
when we start storing RTPs in their own database table, but we're not
there yet. (2) I'm not sure if it is needed. A completed analysis
already contains the RTPs.
Would like to get others' opinions on this.
- Usual code review

# Risk assessment

- Medium. It changes a core part of protocol analysis but the end effect
is a small change to the API. As long as tests are passing, it shouldn't
cause any problems.
- Small risk of analysis behavior change for badly written protocols as
mentioned in
[this](#15229 (comment))
comment.

robot-server/robot_server/protocols/analyses_manager.py

@@ -0,0 +1,64 @@
+"""A collaborator for managing protocol analyses."""
+from typing import Optional
+
+from opentrons.protocol_engine.types import RunTimeParamValuesType
+
+from robot_server.protocols.analysis_models import (
+    AnalysisStatus,
+    AnalysisSummary,
+)
+from robot_server.protocols.analysis_store import AnalysisStore
+from robot_server.protocols import protocol_analyzer
+from robot_server.protocols.protocol_store import ProtocolResource
+from robot_server.service.task_runner import TaskRunner
+
+
+class AnalysesManager:
+    """A Collaborator that manages and provides an interface to Protocol Analyzers."""
+
+    def __init__(self, analysis_store: AnalysisStore, task_runner: TaskRunner) -> None:
+        self._analysis_store = analysis_store
+        self._task_runner = task_runner
+
+    async def start_analysis(
+        self,
+        analysis_id: str,
+        protocol_resource: ProtocolResource,
+        run_time_param_values: Optional[RunTimeParamValuesType],
+    ) -> AnalysisSummary:
+        """Start an analysis of the given protocol resource with run time param values."""
+        analyzer = protocol_analyzer.create_protocol_analyzer(
+            analysis_store=self._analysis_store, protocol_resource=protocol_resource
+        )
+        pending = self._analysis_store.add_pending(
+            protocol_id=protocol_resource.protocol_id,
+            analysis_id=analysis_id,
+        )
+        try:
+            protocol_runner = await analyzer.load_runner(
+                run_time_param_values=run_time_param_values
+            )
+            pending.runTimeParameters = protocol_runner.run_time_parameters
+        except BaseException as error:
+            await analyzer.update_to_failed_analysis(
+                analysis_id=analysis_id,
+                protocol_robot_type=protocol_resource.source.robot_type,
+                error=error,
+                run_time_parameters=[],
+            )
+            return AnalysisSummary(
+                id=analysis_id,
+                status=AnalysisStatus.COMPLETED,
+            )
+
+        self._task_runner.run(
+            analyzer.analyze,
+            runner=protocol_runner,
+            analysis_id=analysis_id,
+            run_time_parameters=protocol_runner.run_time_parameters,
+        )
+        return AnalysisSummary(
+            id=analysis_id,
+            status=AnalysisStatus.PENDING,
+            runTimeParameters=pending.runTimeParameters,
+        )

robot-server/robot_server/protocols/analysis_models.py

@@ -57,6 +57,16 @@ class AnalysisSummary(BaseModel):
 
     id: str = Field(..., description="Unique identifier of this analysis resource")
     status: AnalysisStatus = Field(..., description="Status of the analysis")
+    runTimeParameters: Optional[List[RunTimeParameter]] = Field(
+        default=None,
+        description=(
+            "Run time parameters used during analysis."
+            " These are the parameters that are defined in the protocol, with values"
+            " specified either in the protocol creation request or reanalysis request"
+            " (whichever started this analysis), or default values from the protocol"
+            " if none are specified in the request."
+        ),
+    )
 
 
 class PendingAnalysis(BaseModel):
@@ -67,6 +77,16 @@ class PendingAnalysis(BaseModel):
         AnalysisStatus.PENDING,
         description="Status marking the analysis as pending",
     )
+    runTimeParameters: List[RunTimeParameter] = Field(
+        default_factory=list,
+        description=(
+            "Run time parameters used during analysis."
+            " These are the parameters that are defined in the protocol, with values"
+            " specified either in the protocol creation request or reanalysis request"
+            " (whichever started this analysis), or default values from the protocol"
+            " if none are specified in the request."
+        ),
+    )
 
 
 class CompletedAnalysis(BaseModel):

robot-server/robot_server/protocols/analysis_store.py

@@ -1,16 +1,13 @@
 """Protocol analysis storage."""
 from __future__ import annotations
 
-
+import sqlalchemy
 from logging import getLogger
 from typing import Dict, List, Optional
-
-from opentrons.protocol_engine.types import RunTimeParameter
 from typing_extensions import Final
-from opentrons_shared_data.robot.dev_types import RobotType
-
-import sqlalchemy
 
+from opentrons_shared_data.robot.dev_types import RobotType
+from opentrons.protocol_engine.types import RunTimeParameter, RunTimeParamValuesType
 from opentrons.protocol_engine import (
     Command,
     ErrorOccurrence,
@@ -19,8 +16,6 @@
     LoadedModule,
     Liquid,
 )
-from opentrons.protocol_engine.types import RunTimeParamValuesType
-
 from .analysis_models import (
     AnalysisSummary,
     ProtocolAnalysis,
@@ -116,7 +111,11 @@ def __init__(
             current_analyzer_version=_CURRENT_ANALYZER_VERSION,
         )
 
-    def add_pending(self, protocol_id: str, analysis_id: str) -> AnalysisSummary:
+    def add_pending(
+        self,
+        protocol_id: str,
+        analysis_id: str,
+    ) -> PendingAnalysis:
         """Add a new pending analysis to the store.
 
         Args:
@@ -129,10 +128,10 @@ def add_pending(self, protocol_id: str, analysis_id: str) -> AnalysisSummary:
         Returns:
             A summary of the just-added analysis.
         """
-        new_pending_analysis = self._pending_store.add(
-            protocol_id=protocol_id, analysis_id=analysis_id
+        return self._pending_store.add(
+            protocol_id=protocol_id,
+            analysis_id=analysis_id,
         )
-        return _summarize_pending(pending_analysis=new_pending_analysis)
 
     async def update(
         self,
@@ -243,6 +242,9 @@ def get_summaries_by_protocol(self, protocol_id: str) -> List[AnalysisSummary]:
         completed_analysis_ids = self._completed_store.get_ids_by_protocol(
             protocol_id=protocol_id
         )
+        # TODO (spp, 2024-06-05): populate runTimeParameters in the completed analysis summaries once
+        #  we start saving RTPs to their own table. Currently, fetching RTPs from a
+        #  completed analysis requires de-serializing the full analysis resource.
         completed_analysis_summaries = [
             AnalysisSummary.construct(id=analysis_id, status=AnalysisStatus.COMPLETED)
             for analysis_id in completed_analysis_ids
@@ -316,6 +318,11 @@ async def matching_rtp_values_in_analysis(
         parameters in the analysis use default values.
         """
         if analysis_summary.status == AnalysisStatus.PENDING:
+            # TODO: extract defaults and values from pending analysis now that they're available
+            #   If the pending analysis RTPs match the current RTPs, do nothing(?).
+            #   If the pending analysis RTPs DO NOT match the current RTPs, raise the
+            #   AnalysisIsPending error. Eventually, we might allow either canceling the
+            #   pending analysis or starting another analysis when there's already a pending one.
             raise AnalysisIsPendingError(analysis_summary.id)
 
         rtp_values_and_defaults_in_last_analysis = (
@@ -371,13 +378,19 @@ def __init__(self) -> None:
         self._analysis_ids_by_protocol_id: Dict[str, str] = {}
         self._protocol_ids_by_analysis_id: Dict[str, str] = {}
 
-    def add(self, protocol_id: str, analysis_id: str) -> PendingAnalysis:
+    def add(
+        self,
+        protocol_id: str,
+        analysis_id: str,
+    ) -> PendingAnalysis:
         """Add a new pending analysis and associate it with the given protocol."""
         assert (
             protocol_id not in self._analysis_ids_by_protocol_id
         ), "Protocol must not already have a pending analysis."
 
-        new_pending_analysis = PendingAnalysis.construct(id=analysis_id)
+        new_pending_analysis = PendingAnalysis.construct(
+            id=analysis_id,
+        )
 
         self._analyses_by_id[analysis_id] = new_pending_analysis
         self._analysis_ids_by_protocol_id[protocol_id] = analysis_id

robot-server/robot_server/protocols/dependencies.py

@@ -17,18 +17,19 @@
     AppStateAccessor,
     get_app_state,
 )
+from robot_server.service.task_runner import TaskRunner, get_task_runner
 from robot_server.deletion_planner import ProtocolDeletionPlanner
 from robot_server.persistence.fastapi_dependencies import (
     get_sql_engine,
     get_active_persistence_directory,
 )
 from robot_server.settings import get_settings
+from .analyses_manager import AnalysesManager
 
 from .protocol_auto_deleter import ProtocolAutoDeleter
 from .protocol_store import (
     ProtocolStore,
 )
-from .protocol_analyzer import ProtocolAnalyzer
 from .analysis_store import AnalysisStore
 
 
@@ -41,6 +42,7 @@
 
 _analysis_store_accessor = AppStateAccessor[AnalysisStore]("analysis_store")
 
+_analyses_manager_accessor = AppStateAccessor[AnalysesManager]("analyses_manager")
 _protocol_directory_init_lock = AsyncLock()
 _protocol_directory_accessor = AppStateAccessor[Path]("protocol_directory")
 
@@ -109,13 +111,21 @@ async def get_analysis_store(
     return analysis_store
 
 
-async def get_protocol_analyzer(
+async def get_analyses_manager(
+    app_state: AppState = Depends(get_app_state),
     analysis_store: AnalysisStore = Depends(get_analysis_store),
-) -> ProtocolAnalyzer:
-    """Construct a ProtocolAnalyzer for a single request."""
-    return ProtocolAnalyzer(
-        analysis_store=analysis_store,
-    )
+    task_runner: TaskRunner = Depends(get_task_runner),
+) -> AnalysesManager:
+    """Get a singleton AnalysesManager to keep track of analyzers."""
+    analyses_manager = _analyses_manager_accessor.get_from(app_state)
+
+    if analyses_manager is None:
+        analyses_manager = AnalysesManager(
+            analysis_store=analysis_store, task_runner=task_runner
+        )
+        _analyses_manager_accessor.set_on(app_state, analyses_manager)
+
+    return analyses_manager
 
 
 async def get_protocol_auto_deleter(