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,20 @@ 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]) -> 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."""
+        wavelengths = self._engine_client.state.modules.get_absorbance_reader_substate(
+            self.module_id
+        ).configured_wavelengths
+        if wavelengths is None:
+            raise CannotPerformModuleAction(
+                "Cannot perform Read action on Absorbance Reader without calling `.initialize(...)` first."
+            )
         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 = (
@@ -603,7 +612,17 @@ def read(self) -> Optional[Dict[int, Dict[str, float]]]:
             raise CannotPerformModuleAction(
                 "Absorbance Reader failed to return expected read result."
             )
-        return None
+
+        # When using virtual modules, return all zeroes
+        virtual_asbsorbance_result: Dict[int, Dict[str, float]] = {}
+        for wavelength in wavelengths:
+            converted_values = (
+                self._engine_client.state.modules.convert_absorbance_reader_data_points(
+                    data=[0] * 96
+                )
+            )
+            virtual_asbsorbance_result[wavelength] = converted_values
+        return virtual_asbsorbance_result
 
     def close_lid(
         self,

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]) -> 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,17 @@ 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, export_filename: Optional[str]) -> 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 export_filename: Optional, if a filename is provided a CSV file will be saved
+             as a result of the read action containing measurement data. The filename will
+             be modified to include the wavelength used during measurement. If multiple
+             measurements are taken, then a file will be generated for each wavelength provided.
+
+        Example: If `export_filename="my_data"` and wavelengths 450 and 531 are used during
+        measurement, the output files will be "my_data_450.csv" and "my_data_531.csv".
+        """
+        return self._core.read(filename=export_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 (
+    PlateReaderData,
+    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,18 +56,21 @@ 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(
+    async def execute(  # noqa: C901
         self, params: ReadAbsorbanceParams
     ) -> SuccessData[ReadAbsorbanceResult, None]:
         """Initiate an absorbance measurement."""
         abs_reader_substate = self._state_view.modules.get_absorbance_reader_substate(
             module_id=params.moduleId
         )
+
         # Allow propagation of ModuleNotAttachedError.
         abs_reader = self._equipment.get_module_hardware_api(
             abs_reader_substate.module_id
@@ -62,10 +81,29 @@ async def execute(
                 "Cannot perform Read action on Absorbance Reader without calling `.initialize(...)` first."
             )
 
+        # 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
+            and abs_reader_substate.configured_wavelengths 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."
+                )
+
+        asbsorbance_result: Dict[int, Dict[str, float]] = {}
+        transform_results = []
+        # Handle the measurement and begin building data for return
         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
                 for wavelength, result in zip(sample_wavelengths, results):
                     converted_values = (
@@ -74,13 +112,67 @@ async def execute(
                         )
                     )
                     asbsorbance_result[wavelength] = converted_values
+                    transform_results.append(
+                        ReadData.construct(wavelength=wavelength, data=converted_values)
+                    )
+        # Handle the virtual module case for data creation (all zeroes)
+        elif self._state_view.config.use_virtual_modules:
+            start_time = finish_time = datetime.now()
+            if abs_reader_substate.configured_wavelengths is not None:
+                for wavelength in abs_reader_substate.configured_wavelengths:
+                    converted_values = (
+                        self._state_view.modules.convert_absorbance_reader_data_points(
+                            data=[0] * 96
+                        )
+                    )
+                    asbsorbance_result[wavelength] = converted_values
+                    transform_results.append(
+                        ReadData.construct(wavelength=wavelength, data=converted_values)
+                    )
+            else:
+                raise CannotPerformModuleAction(
+                    "Plate Reader data cannot be requested with a module that has not been initialized."
+                )
+
+        # 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 will 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:
+            # Create the Plate Reader Transform
+            plate_read_result = PlateReaderData.construct(
+                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 (abs_reader is not None and abs_reader.serial_number is not None)
+                else "VIRTUAL_SERIAL",
+            )
+
+            if isinstance(plate_read_result, PlateReaderData):
+                # 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=asbsorbance_result, fileIds=file_ids),
             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

@@ -75,6 +75,7 @@
     IncompleteWellDefinitionError,
     OperationLocationNotInWellError,
     InvalidDispenseVolumeError,
+    StorageLimitReachedError,
 )
 
 from .error_occurrence import ErrorOccurrence, ProtocolCommandFailedError
@@ -158,4 +159,5 @@
     "IncompleteWellDefinitionError",
     "OperationLocationNotInWellError",
     "InvalidDispenseVolumeError",
+    "StorageLimitReachedError",
 ]

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

@@ -1108,3 +1108,16 @@ def __init__(
     ) -> None:
         """Build an OperationLocationNotInWellError."""
         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",
 ]