refactor(robot-server, api): genericize FileProvider (#19755)

Closes EXEC-1956

This PR genericizes the FileProvider interface to pave the way for writing more than CSV file types to disk. To that end, FileProvider now takes raw byte data, a MIME type, and command_metadata, which is used to generate the final filename.

FileProviderWrapper is renamed to FileProviderExecutor and now handles the entirety of file naming and writing the file to the correct location. The vision is that shortly, FileProviderExecutor will need to ingest robot-server specific metadata alongside the command_metadata for building the filename, and this step moves us in that direction.

Author: mjhuff

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

@@ -14,7 +14,9 @@
 from ...resources.file_provider import (
     PlateReaderData,
     ReadData,
-    MAXIMUM_CSV_FILE_LIMIT,
+    MAXIMUM_FILE_LIMIT,
+    MimeType,
+    ReadCmdFileNameMetadata,
 )
 from ...resources import FileProvider
 from ...state import update_types
@@ -102,10 +104,10 @@ async def execute(  # noqa: C901
             if (
                 self._state_view.files.get_filecount()
                 + len(abs_reader_substate.configured_wavelengths)
-                > MAXIMUM_CSV_FILE_LIMIT
+                > MAXIMUM_FILE_LIMIT
             ):
                 raise StorageLimitReachedError(
-                    message=f"Attempt to write file {params.fileName} exceeds file creation limit of {MAXIMUM_CSV_FILE_LIMIT} files."
+                    message=f"Attempt to write file {params.fileName} exceeds file creation limit of {MAXIMUM_FILE_LIMIT} files."
                 )
 
         asbsorbance_result: Dict[int, Dict[str, float]] = {}
@@ -174,11 +176,16 @@ async def execute(  # noqa: C901
             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,
-                        )
+                    csv_bytes = plate_read_result.build_csv_bytes(
+                        measurement=measurement,
+                    )
+                    file_id = await self._file_provider.write_file(
+                        data=csv_bytes,
+                        mime_type=MimeType.TEXT_CSV,
+                        command_metadata=ReadCmdFileNameMetadata.model_construct(
+                            base_filename=params.fileName,
+                            wavelength=measurement.wavelength,
+                        ),
                     )
                     file_ids.append(file_id)
 

api/src/opentrons/protocol_engine/resources/file_provider.py

@@ -1,36 +1,51 @@
 """File interaction resource provider."""
 from datetime import datetime
+from enum import Enum
+from io import StringIO
+import csv
 from typing import List, Optional, Callable, Awaitable, Dict
 from pydantic import BaseModel
 from ..errors import StorageLimitReachedError
 
 
-MAXIMUM_CSV_FILE_LIMIT = 400
+MAXIMUM_FILE_LIMIT = 400
 
 
-class GenericCsvTransform:
-    """Generic CSV File Type data for rows of data to be seperated by a delimeter."""
+class MimeType(str, Enum):
+    """File mime types."""
 
-    filename: str
-    rows: List[List[str]]
-    delimiter: str = ","
+    TEXT_CSV = "text/csv"
+
+
+class ReadCmdFileNameMetadata(BaseModel):
+    """Data from a plate reader `read` command used to build the finalized file name."""
+
+    base_filename: str
+    wavelength: int
+
+
+CommandFileNameMetadata = ReadCmdFileNameMetadata | None
+
+
+class FileData:
+    """File data container for writing to a file."""
+
+    data: bytes
+    mime_type: MimeType
+    command_metadata: CommandFileNameMetadata
 
     @staticmethod
     def build(
-        filename: str, rows: List[List[str]], delimiter: str = ","
-    ) -> "GenericCsvTransform":
-        """Build a Generic CSV datatype class."""
-        if "." in filename and not filename.endswith(".csv"):
-            raise ValueError(
-                f"Provided filename {filename} invalid. Only CSV file format is accepted."
-            )
-        elif "." not in filename:
-            filename = f"{filename}.csv"
-        csv = GenericCsvTransform()
-        csv.filename = filename
-        csv.rows = rows
-        csv.delimiter = delimiter
-        return csv
+        data: bytes,
+        mime_type: MimeType,
+        command_metadata: CommandFileNameMetadata = None,
+    ) -> "FileData":
+        """Build a generic file data class."""
+        file_data = FileData()
+        file_data.data = data
+        file_data.mime_type = mime_type
+        file_data.command_metadata = command_metadata
+        return file_data
 
 
 class ReadData(BaseModel):
@@ -41,29 +56,24 @@ class ReadData(BaseModel):
 
 
 class PlateReaderData(BaseModel):
-    """Data from a Opentrons Plate Reader Read. Can be converted to CSV template format."""
+    """Data from an Opentrons Plate Reader Read. Can be converted to CSV format."""
 
     read_results: List[ReadData]
     reference_wavelength: Optional[int] = None
     start_time: datetime
     finish_time: datetime
     serial_number: str
 
-    def build_generic_csv(  # noqa: C901
-        self, filename: str, measurement: ReadData
-    ) -> GenericCsvTransform:
-        """Builds a CSV compatible object containing Plate Reader Measurements.
-
-        This will also automatically reformat the provided filename to include the wavelength of those measurements.
-        """
+    def build_csv_bytes(self, measurement: ReadData) -> bytes:  # noqa: C901
+        """Builds CSV data as bytes containing Plate Reader Measurements."""
         plate_alpharows = ["A", "B", "C", "D", "E", "F", "G", "H"]
         rows = []
 
         rows.append(["", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12"])
         for i in range(8):
             row = [plate_alpharows[i]]
             for j in range(12):
-                row.append(str(measurement.data[f"{plate_alpharows[i]}{j+1}"]))
+                row.append(str(measurement.data[f"{plate_alpharows[i]}{j + 1}"]))
             rows.append(row)
         for i in range(3):
             rows.append([])
@@ -116,46 +126,50 @@ def build_generic_csv(  # noqa: C901
             ["Measurement finished at", self.finish_time.strftime("%m %d %H:%M:%S %Y")]
         )
 
-        # Ensure the filename adheres to ruleset contains the wavelength for a given measurement
-        if filename.endswith(".csv"):
-            filename = filename[:-4]
-        filename = filename + str(measurement.wavelength) + "nm.csv"
+        output = StringIO()
+        writer = csv.writer(output, delimiter=",")
+        writer.writerows(rows)
+        csv_bytes = output.getvalue().encode("utf-8")
 
-        return GenericCsvTransform.build(
-            filename=filename,
-            rows=rows,
-            delimiter=",",
-        )
+        return csv_bytes
 
 
 class FileProvider:
     """Provider class to wrap file read write interactions to the data files directory in the engine."""
 
     def __init__(
         self,
-        data_files_write_csv_callback: Optional[
-            Callable[[GenericCsvTransform], Awaitable[str]]
-        ] = None,
+        data_files_write_file_cb: Optional[Callable[[FileData], Awaitable[str]]] = None,
         data_files_filecount: Optional[Callable[[], Awaitable[int]]] = None,
     ) -> None:
         """Initialize the interface callbacks of the File Provider for data file handling within the Protocol Engine.
 
         Params:
-            data_files_write_csv_callback: Callback to write a CSV file to the data files directory and add it to the database.
+            data_files_write_file_callback: Callback to write a file to the data files directory and add it to the database.
             data_files_filecount: Callback to check the amount of data files already present in the data files directory.
         """
-        self._data_files_write_csv_callback = data_files_write_csv_callback
+        self._data_files_write_file_cb = data_files_write_file_cb
         self._data_files_filecount = data_files_filecount
 
-    async def write_csv(self, write_data: GenericCsvTransform) -> str:
-        """Writes the provided CSV object to a file in the Data Files directory. Returns the File ID of the file created."""
+    async def write_file(
+        self,
+        data: bytes,
+        mime_type: MimeType,
+        command_metadata: CommandFileNameMetadata = None,
+    ) -> str:
+        """Writes arbitrary data to a file in the Data Files directory. Returns the File ID of the file created."""
         if self._data_files_filecount is not None:
             file_count = await self._data_files_filecount()
-            if file_count >= MAXIMUM_CSV_FILE_LIMIT:
+            if file_count >= MAXIMUM_FILE_LIMIT:
                 raise StorageLimitReachedError(
-                    f"Not enough space to store file {write_data.filename}."
+                    f"Not enough space to store file. Maximum file limit of {MAXIMUM_FILE_LIMIT} reached."
+                )
+            if self._data_files_write_file_cb is not None:
+                file_data = FileData.build(
+                    data=data,
+                    mime_type=mime_type,
+                    command_metadata=command_metadata,
                 )
-            if self._data_files_write_csv_callback is not None:
-                return await self._data_files_write_csv_callback(write_data)
+                return await self._data_files_write_file_cb(file_data)
         # If we are in an analysis or simulation state, return an empty file ID
         return ""

robot-server/robot_server/file_provider/fastapi_dependencies.py

@@ -4,7 +4,7 @@
 
 import fastapi
 
-from robot_server.file_provider.provider import FileProviderWrapper
+from robot_server.file_provider.provider import FileProviderExecutor
 from robot_server.data_files.dependencies import (
     get_data_files_directory,
     get_data_files_store,
@@ -13,27 +13,27 @@
 from opentrons.protocol_engine.resources.file_provider import FileProvider
 
 
-async def get_file_provider_wrapper(
+async def get_file_provider_executor(
     data_files_directory: Annotated[Path, fastapi.Depends(get_data_files_directory)],
     data_files_store: Annotated[DataFilesStore, fastapi.Depends(get_data_files_store)],
-) -> FileProviderWrapper:
-    """Return the server's singleton `FileProviderWrapper` which provides the engine related callbacks for FileProvider."""
-    file_provider_wrapper = FileProviderWrapper(
+) -> FileProviderExecutor:
+    """Return the server's singleton `FileProviderExecutor` which provides the engine related callbacks for FileProvider."""
+    file_provider_wrapper = FileProviderExecutor(
         data_files_directory=data_files_directory, data_files_store=data_files_store
     )
 
     return file_provider_wrapper
 
 
 async def get_file_provider(
-    file_provider_wrapper: Annotated[
-        FileProviderWrapper, fastapi.Depends(get_file_provider_wrapper)
+    file_provider_executor: Annotated[
+        FileProviderExecutor, fastapi.Depends(get_file_provider_executor)
     ],
 ) -> FileProvider:
     """Return the engine `FileProvider` which accepts callbacks from FileProviderWrapper."""
     file_provider = FileProvider(
-        data_files_write_csv_callback=file_provider_wrapper.write_csv_callback,
-        data_files_filecount=file_provider_wrapper.csv_filecount_callback,
+        data_files_write_file_cb=file_provider_executor.write_file_cb,
+        data_files_filecount=file_provider_executor.filecount_cb,
     )
 
     return file_provider

robot-server/robot_server/file_provider/provider.py

@@ -1,7 +1,6 @@
-"""Wrapper to provide the callbacks utilized by the Protocol Engine File Provider."""
+"""Executor for Protocol Engine File Provider callbacks."""
 import os
 import asyncio
-import csv
 from pathlib import Path
 from typing import Annotated
 from fastapi import Depends
@@ -15,65 +14,85 @@
     DataFilesStore,
     DataFileInfo,
 )
-from opentrons.protocol_engine.resources.file_provider import GenericCsvTransform
+from opentrons.protocol_engine.resources.file_provider import (
+    FileData,
+    ReadCmdFileNameMetadata,
+)
 
 
-class FileProviderWrapper:
-    """Wrapper to provide File Read and Write capabilities to Protocol Engine."""
+class FileProviderExecutor:
+    """Executes file operations for the Protocol Engine File Provider."""
 
     def __init__(
         self,
         data_files_directory: Annotated[Path, Depends(get_data_files_directory)],
         data_files_store: Annotated[DataFilesStore, Depends(get_data_files_store)],
     ) -> None:
-        """Provides callbacks for data file manipulation for the Protocol Engine's File Provider class.
+        """Initialize the file provider executor.
 
         Params:
-            data_files_directory: The directory to store engine-create files in during a protocol run.
+            data_files_directory: The directory to store engine-created files in during a protocol run.
             data_files_store: The data files store utilized for database interaction when creating files.
         """
         self._data_files_directory = data_files_directory
         self._data_files_store = data_files_store
 
-        # dta file store is not generally safe for concurrent access.
+        # data file store is not generally safe for concurrent access.
         self._lock = asyncio.Lock()
 
-    async def write_csv_callback(
+    async def write_file_cb(
         self,
-        csv_data: GenericCsvTransform,
+        file_data: FileData,
     ) -> str:
-        """Write the provided data transform to a CSV file. Returns the File ID of the created file."""
+        """Write the provided file data to disk. Returns the File ID of the created file."""
         async with self._lock:
             file_id = await get_unique_id()
-            os.makedirs(
-                os.path.dirname(
-                    self._data_files_directory / file_id / csv_data.filename
-                ),
-                exist_ok=True,
+            final_filename = self._format_filename(file_data, file_id)
+            final_filepath = self._format_filepath(
+                filename=final_filename, file_id=file_id, file_data=file_data
             )
-            with open(
-                file=self._data_files_directory / file_id / csv_data.filename,
-                mode="w",
-                newline="",
-            ) as csvfile:
-                writer = csv.writer(csvfile, delimiter=csv_data.delimiter)
-                writer.writerows(csv_data.rows)
+
+            os.makedirs(os.path.dirname(final_filepath), exist_ok=True)
+
+            with open(file=final_filepath, mode="wb") as f:
+                f.write(file_data.data)
 
             created_at = await get_current_time()
-            # TODO (cb, 10-14-24): Engine created files do not currently get a file_hash, unlike explicitly uploaded files. Do they need one?
             file_info = DataFileInfo(
                 id=file_id,
-                name=csv_data.filename,
+                name=final_filename,
                 file_hash="",
                 created_at=created_at,
                 source=DataFileSource.GENERATED,
             )
             await self._data_files_store.insert(file_info)
             return file_id
 
-    async def csv_filecount_callback(self) -> int:
+    async def filecount_cb(self) -> int:
         """Return the current count of generated files stored within the data files directory."""
         data_file_usage_info = self._data_files_store.get_usage_info(
             DataFileSource.GENERATED
         )
         return len(data_file_usage_info)
+
+    def _format_filename(self, file_data: FileData, file_id: str) -> str:
+        """Build the finalized filename."""
+        if isinstance(file_data.command_metadata, ReadCmdFileNameMetadata):
+            metadata = file_data.command_metadata
+            base_name = metadata.base_filename
+
+            if base_name.endswith(".csv"):
+                base_name = base_name[:-4]
+
+            return base_name + str(metadata.wavelength) + "nm.csv"
+        else:
+            return f"{file_id}.dat"
+
+    def _format_filepath(
+        self, filename: str, file_id: str, file_data: FileData
+    ) -> Path:
+        """Given a finalized filename, return the full filepath for the filename."""
+        if isinstance(file_data.command_metadata, ReadCmdFileNameMetadata):
+            return self._data_files_directory / file_id / filename
+        else:
+            return self._data_files_directory / filename