feat(api, robot-server): Plate Reader CSV output functionality

api/src/opentrons/cli/analyze.py

@@ -332,6 +332,7 @@ async def _do_analyze(
                 liquids=[],
                 wells=[],
                 hasEverEnteredErrorRecovery=False,
+                files=[],
             ),
             parameters=[],
         )

api/src/opentrons/execute.py

@@ -546,6 +546,7 @@ def _create_live_context_pe(
             hardware_api=hardware_api_wrapped,
             config=_get_protocol_engine_config(),
             deck_configuration=entrypoint_util.get_deck_configuration(),
+            file_provider=None,
             error_recovery_policy=error_recovery_policy.never_recover,
             drop_tips_after_run=False,
             post_run_hardware_state=PostRunHardwareState.STAY_ENGAGED_IN_PLACE,

api/src/opentrons/protocol_api/core/engine/module_core.py

@@ -586,11 +586,13 @@ def initialize(
         )
         self._initialized_value = wavelengths
 
-    def read(self) -> Optional[Dict[int, Dict[str, float]]]:
-        """Initiate a read on the Absorbance Reader, and return the results. During Analysis, this will return None."""
+    def read(self, filename: Optional[str]) -> Optional[Dict[int, Dict[str, float]]]:
+        """Initiate a read on the Absorbance Reader, and return the results. During Analysis, this will return a measurement of zero for all wells."""
         if self._initialized_value:
             self._engine_client.execute_command(
-                cmd.absorbance_reader.ReadAbsorbanceParams(moduleId=self.module_id)
+                cmd.absorbance_reader.ReadAbsorbanceParams(
+                    moduleId=self.module_id, fileName=filename
+                )
             )
         if not self._engine_client.state.config.use_virtual_modules:
             read_result = (

api/src/opentrons/protocol_api/core/module.py

@@ -365,7 +365,7 @@ def initialize(
         """Initialize the Absorbance Reader by taking zero reading."""
 
     @abstractmethod
-    def read(self) -> Optional[Dict[int, Dict[str, float]]]:
+    def read(self, filename: Optional[str]) -> Optional[Dict[int, Dict[str, float]]]:
         """Get an absorbance reading from the Absorbance Reader."""
 
     @abstractmethod

api/src/opentrons/protocol_api/module_contexts.py

@@ -1035,6 +1035,9 @@ def initialize(
         )
 
     @requires_version(2, 21)
-    def read(self) -> Optional[Dict[int, Dict[str, float]]]:
-        """Initiate read on the Absorbance Reader. Returns a dictionary of wavelengths to dictionary of values ordered by well name."""
-        return self._core.read()
+    def read(self, filename: Optional[str]) -> Optional[Dict[int, Dict[str, float]]]:
+        """Initiate read on the Absorbance Reader. Returns a dictionary of wavelengths to dictionary of values ordered by well name.
+
+        :param filename: Optional, if a filename is provided a CSV file will be saved as a result of the read action containing measurement data.
+        """
+        return self._core.read(filename=filename)

api/src/opentrons/protocol_engine/commands/absorbance_reader/read.py

@@ -1,14 +1,22 @@
 """Command models to read absorbance."""
 from __future__ import annotations
-from typing import Optional, Dict, TYPE_CHECKING
+from datetime import datetime
+from typing import Optional, Dict, TYPE_CHECKING, List
 from typing_extensions import Literal, Type
 
 from pydantic import BaseModel, Field
 
 from ..command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
-from ...errors import CannotPerformModuleAction
+from ...errors import CannotPerformModuleAction, StorageLimitReachedError
 from ...errors.error_occurrence import ErrorOccurrence
 
+from ...resources.file_provider import (
+    PlateReaderDataTransform,
+    ReadData,
+    MAXIMUM_CSV_FILE_LIMIT,
+)
+from ...resources import FileProvider
+
 if TYPE_CHECKING:
     from opentrons.protocol_engine.state.state import StateView
     from opentrons.protocol_engine.execution import EquipmentHandler
@@ -21,6 +29,10 @@ class ReadAbsorbanceParams(BaseModel):
     """Input parameters for an absorbance reading."""
 
     moduleId: str = Field(..., description="Unique ID of the Absorbance Reader.")
+    fileName: Optional[str] = Field(
+        None,
+        description="Optional file name to use when storing the results of a measurement.",
+    )
 
 
 class ReadAbsorbanceResult(BaseModel):
@@ -29,6 +41,10 @@ class ReadAbsorbanceResult(BaseModel):
     data: Optional[Dict[int, Dict[str, float]]] = Field(
         ..., description="Absorbance data points per wavelength."
     )
+    fileIds: Optional[List[str]] = Field(
+        ...,
+        description="List of file IDs for files output as a result of a Read action.",
+    )
 
 
 class ReadAbsorbanceImpl(
@@ -40,10 +56,12 @@ def __init__(
         self,
         state_view: StateView,
         equipment: EquipmentHandler,
+        file_provider: FileProvider,
         **unused_dependencies: object,
     ) -> None:
         self._state_view = state_view
         self._equipment = equipment
+        self._file_provider = file_provider
 
     async def execute(
         self, params: ReadAbsorbanceParams
@@ -52,6 +70,18 @@ async def execute(
         abs_reader_substate = self._state_view.modules.get_absorbance_reader_substate(
             module_id=params.moduleId
         )
+        # TODO: we need to return a file ID and increase the file count even when a moduel is not attached
+        if params.fileName is not None:
+            # Validate that the amount of files we are about to generate does not put us higher than the limit
+            if (
+                self._state_view.files.get_filecount()
+                + len(abs_reader_substate.configured_wavelengths)
+                > MAXIMUM_CSV_FILE_LIMIT
+            ):
+                raise StorageLimitReachedError(
+                    message=f"Attempt to write file {params.fileName} exceeds file creation limit of {MAXIMUM_CSV_FILE_LIMIT} files."
+                )
+
         # Allow propagation of ModuleNotAttachedError.
         abs_reader = self._equipment.get_module_hardware_api(
             abs_reader_substate.module_id
@@ -63,24 +93,61 @@ async def execute(
             )
 
         if abs_reader is not None:
+            start_time = datetime.now()
             results = await abs_reader.start_measure()
+            finish_time = datetime.now()
             if abs_reader._measurement_config is not None:
                 asbsorbance_result: Dict[int, Dict[str, float]] = {}
                 sample_wavelengths = abs_reader._measurement_config.sample_wavelengths
+                transform_results = []
                 for wavelength, result in zip(sample_wavelengths, results):
                     converted_values = (
                         self._state_view.modules.convert_absorbance_reader_data_points(
                             data=result
                         )
                     )
                     asbsorbance_result[wavelength] = converted_values
+                    transform_results.append(
+                        ReadData.build(wavelength=wavelength, data=converted_values)
+                    )
+
+                # TODO (cb, 10-17-2024): FILE PROVIDER - Some day we may want to break the file provider behavior into a seperate API function.
+                # When this happens, we probably want to have the change the command results handler we utilize to track file IDs in engine.
+                # Today, the action handler for the FileStore looks for a ReadAbsorbanceResult command action, this will need to be delinked.
+
+                # Begin interfacing with the file provider if the user provided a filename
+                file_ids = []
+                if params.fileName is not None and abs_reader.serial_number is not None:
+                    # Create the Plate Reader Transform
+                    plate_read_result = PlateReaderDataTransform.build(
+                        read_results=transform_results,
+                        reference_wavelength=abs_reader_substate.reference_wavelength,
+                        start_time=start_time,
+                        finish_time=finish_time,
+                        serial_number=abs_reader.serial_number,
+                    )
+
+                    if isinstance(plate_read_result, PlateReaderDataTransform):
+                        # Write a CSV file for each of the measurements taken
+                        for measurement in plate_read_result.read_results:
+                            file_id = await self._file_provider.write_csv(
+                                write_data=plate_read_result.build_generic_csv(
+                                    filename=params.fileName,
+                                    measurement=measurement,
+                                )
+                            )
+                            file_ids.append(file_id)
+
+                # Return success data to api
                 return SuccessData(
-                    public=ReadAbsorbanceResult(data=asbsorbance_result),
+                    public=ReadAbsorbanceResult(
+                        data=asbsorbance_result, fileIds=file_ids
+                    ),
                     private=None,
                 )
 
         return SuccessData(
-            public=ReadAbsorbanceResult(data=None),
+            public=ReadAbsorbanceResult(data=None, fileIds=None),
             private=None,
         )
 

api/src/opentrons/protocol_engine/commands/command.py

@@ -274,6 +274,7 @@ def __init__(
         state_view: StateView,
         hardware_api: HardwareControlAPI,
         equipment: execution.EquipmentHandler,
+        file_provider: execution.FileProvider,
         movement: execution.MovementHandler,
         gantry_mover: execution.GantryMover,
         labware_movement: execution.LabwareMovementHandler,

api/src/opentrons/protocol_engine/create_protocol_engine.py

@@ -10,7 +10,7 @@
 from opentrons_shared_data.robot import load as load_robot
 
 from .protocol_engine import ProtocolEngine
-from .resources import DeckDataProvider, ModuleDataProvider
+from .resources import DeckDataProvider, ModuleDataProvider, FileProvider
 from .state.config import Config
 from .state.state import StateStore
 from .types import PostRunHardwareState, DeckConfigurationType
@@ -26,6 +26,7 @@ async def create_protocol_engine(
     error_recovery_policy: ErrorRecoveryPolicy,
     load_fixed_trash: bool = False,
     deck_configuration: typing.Optional[DeckConfigurationType] = None,
+    file_provider: typing.Optional[FileProvider] = None,
     notify_publishers: typing.Optional[typing.Callable[[], None]] = None,
 ) -> ProtocolEngine:
     """Create a ProtocolEngine instance.
@@ -37,6 +38,7 @@ async def create_protocol_engine(
             See documentation on `ErrorRecoveryPolicy`.
         load_fixed_trash: Automatically load fixed trash labware in engine.
         deck_configuration: The initial deck configuration the engine will be instantiated with.
+        file_provider: Provides access to robot server file writing procedures for protocol output.
         notify_publishers: Notifies robot server publishers of internal state change.
     """
     deck_data = DeckDataProvider(config.deck_type)
@@ -47,6 +49,7 @@ async def create_protocol_engine(
 
     module_calibration_offsets = ModuleDataProvider.load_module_calibrations()
     robot_definition = load_robot(config.robot_type)
+
     state_store = StateStore(
         config=config,
         deck_definition=deck_definition,
@@ -62,6 +65,7 @@ async def create_protocol_engine(
     return ProtocolEngine(
         state_store=state_store,
         hardware_api=hardware_api,
+        file_provider=file_provider,
     )
 
 
@@ -70,6 +74,7 @@ def create_protocol_engine_in_thread(
     hardware_api: HardwareControlAPI,
     config: Config,
     deck_configuration: typing.Optional[DeckConfigurationType],
+    file_provider: typing.Optional[FileProvider],
     error_recovery_policy: ErrorRecoveryPolicy,
     drop_tips_after_run: bool,
     post_run_hardware_state: PostRunHardwareState,
@@ -97,6 +102,7 @@ def create_protocol_engine_in_thread(
     with async_context_manager_in_thread(
         _protocol_engine(
             hardware_api,
+            file_provider,
             config,
             deck_configuration,
             error_recovery_policy,
@@ -114,6 +120,7 @@ def create_protocol_engine_in_thread(
 @contextlib.asynccontextmanager
 async def _protocol_engine(
     hardware_api: HardwareControlAPI,
+    file_provider: typing.Optional[FileProvider],
     config: Config,
     deck_configuration: typing.Optional[DeckConfigurationType],
     error_recovery_policy: ErrorRecoveryPolicy,
@@ -123,6 +130,7 @@ async def _protocol_engine(
 ) -> typing.AsyncGenerator[ProtocolEngine, None]:
     protocol_engine = await create_protocol_engine(
         hardware_api=hardware_api,
+        file_provider=file_provider,
         config=config,
         error_recovery_policy=error_recovery_policy,
         load_fixed_trash=load_fixed_trash,

api/src/opentrons/protocol_engine/engine_support.py

@@ -6,7 +6,8 @@
 
 
 def create_run_orchestrator(
-    hardware_api: HardwareControlAPI, protocol_engine: ProtocolEngine
+    hardware_api: HardwareControlAPI,
+    protocol_engine: ProtocolEngine,
 ) -> RunOrchestrator:
     """Create a RunOrchestrator instance."""
     return RunOrchestrator(

api/src/opentrons/protocol_engine/errors/__init__.py

@@ -72,6 +72,7 @@
     InvalidLiquidHeightFound,
     LiquidHeightUnknownError,
     InvalidWellDefinitionError,
+    StorageLimitReachedError,
 )
 
 from .error_occurrence import ErrorOccurrence, ProtocolCommandFailedError
@@ -152,4 +153,5 @@
     "InvalidLiquidHeightFound",
     "LiquidHeightUnknownError",
     "InvalidWellDefinitionError",
+    "StorageLimitReachedError",
 ]

api/src/opentrons/protocol_engine/errors/exceptions.py

@@ -1082,3 +1082,16 @@ def __init__(
     ) -> None:
         """Build an InvalidWellDefinitionError."""
         super().__init__(ErrorCodes.GENERAL_ERROR, message, details, wrapping)
+
+
+class StorageLimitReachedError(ProtocolEngineError):
+    """Raised to indicate that a file cannot be created due to storage limitations."""
+
+    def __init__(
+        self,
+        message: Optional[str] = None,
+        detail: Optional[Dict[str, str]] = None,
+        wrapping: Optional[Sequence[EnumeratedError]] = None,
+    ) -> None:
+        """Build an StorageLimitReached."""
+        super().__init__(ErrorCodes.GENERAL_ERROR, message, detail, wrapping)

api/src/opentrons/protocol_engine/execution/__init__.py

@@ -21,6 +21,7 @@
 from .hardware_stopper import HardwareStopper
 from .door_watcher import DoorWatcher
 from .status_bar import StatusBarHandler
+from ..resources.file_provider import FileProvider
 
 # .thermocycler_movement_flagger omitted from package's public interface.
 
@@ -45,4 +46,5 @@
     "DoorWatcher",
     "RailLightsHandler",
     "StatusBarHandler",
+    "FileProvider",
 ]

api/src/opentrons/protocol_engine/execution/command_executor.py

@@ -14,7 +14,7 @@
 from opentrons.protocol_engine.commands.command import SuccessData
 
 from ..state.state import StateStore
-from ..resources import ModelUtils
+from ..resources import ModelUtils, FileProvider
 from ..commands import CommandStatus
 from ..actions import (
     ActionDispatcher,
@@ -72,6 +72,7 @@ class CommandExecutor:
     def __init__(
         self,
         hardware_api: HardwareControlAPI,
+        file_provider: FileProvider,
         state_store: StateStore,
         action_dispatcher: ActionDispatcher,
         equipment: EquipmentHandler,
@@ -88,6 +89,7 @@ def __init__(
     ) -> None:
         """Initialize the CommandExecutor with access to its dependencies."""
         self._hardware_api = hardware_api
+        self._file_provider = file_provider
         self._state_store = state_store
         self._action_dispatcher = action_dispatcher
         self._equipment = equipment
@@ -116,6 +118,7 @@ async def execute(self, command_id: str) -> None:
         command_impl = queued_command._ImplementationCls(
             state_view=self._state_store,
             hardware_api=self._hardware_api,
+            file_provider=self._file_provider,
             equipment=self._equipment,
             movement=self._movement,
             gantry_mover=self._gantry_mover,

api/src/opentrons/protocol_engine/execution/create_queue_worker.py

@@ -6,6 +6,7 @@
 
 from ..state.state import StateStore
 from ..actions import ActionDispatcher
+from ..resources import FileProvider
 from .equipment import EquipmentHandler
 from .movement import MovementHandler
 from .gantry_mover import create_gantry_mover
@@ -20,6 +21,7 @@
 
 def create_queue_worker(
     hardware_api: HardwareControlAPI,
+    file_provider: FileProvider,
     state_store: StateStore,
     action_dispatcher: ActionDispatcher,
     command_generator: Callable[[], AsyncGenerator[str, None]],
@@ -28,6 +30,7 @@ def create_queue_worker(
 
     Arguments:
         hardware_api: Hardware control API to pass down to dependencies.
+        file_provider: Provides access to robot server file writing procedures for protocol output.
         state_store: StateStore to pass down to dependencies.
         action_dispatcher: ActionDispatcher to pass down to dependencies.
         error_recovery_policy: ErrorRecoveryPolicy to pass down to dependencies.
@@ -78,6 +81,7 @@ def create_queue_worker(
 
     command_executor = CommandExecutor(
         hardware_api=hardware_api,
+        file_provider=file_provider,
         state_store=state_store,
         action_dispatcher=action_dispatcher,
         equipment=equipment_handler,