feat: Add Beckman Echo Cherry Pick parser

SUPPORTED_INSTRUMENT_SOFTWARE.adoc

@@ -31,7 +31,8 @@ The parsers follow maturation levels of: Recommended, Candidate Release, Working
 |Benchling Waters Empower Adapter|Recommended|BENCHLING/2023/09
 |Cytiva Unicorn|Recommended|BENCHLING/2023/09
 |Benchling Thermo Fisher Scientific Chromeleon|Recommended|BENCHLING/2023/09
-.2+|Liquid Handler|Beckman Coulter Biomek|Recommended|BENCHLING/2024/11
+.3+|Liquid Handler|Beckman Coulter Biomek|Recommended|BENCHLING/2024/11
+|Beckman Echo Cherry Pick|Recommended|BENCHLING/2024/11
 |Beckman Echo Plate Reformat|Recommended|BENCHLING/2024/11
 .3+|Multi Analyte Profiling|Bio-Rad Bio-Plex Manager|Recommended|BENCHLING/2024/09
 |Luminex INTELLIFLEX|Recommended|BENCHLING/2024/09

src/allotropy/parser_factory.py

@@ -30,6 +30,9 @@
 from allotropy.parsers.beckman_coulter_biomek.beckman_coulter_biomek_parser import (
     BeckmanCoulterBiomekParser,
 )
+from allotropy.parsers.beckman_echo_cherry_pick.beckman_echo_cherry_pick_parser import (
+    BeckmanEchoCherryPickParser,
+)
 from allotropy.parsers.beckman_echo_plate_reformat.beckman_echo_plate_reformat_parser import (
     BeckmanEchoPlateReformatParser,
 )
@@ -142,6 +145,7 @@ class Vendor(Enum):
     APPBIO_QUANTSTUDIO_DESIGNANDANALYSIS = "APPBIO_QUANTSTUDIO_DESIGNANDANALYSIS"
     BENCHLING_EMPOWER = "BENCHLING_EMPOWER"
     BECKMAN_COULTER_BIOMEK = "BECKMAN_COULTER_BIOMEK"
+    BECKMAN_ECHO_CHERRY_PICK = "BECKMAN_ECHO_CHERRY_PICK"
     BECKMAN_ECHO_PLATE_REFORMAT = "BECKMAN_ECHO_PLATE_REFORMAT"
     BMG_LABTECH_SMART_CONTROL = "BMG_LABTECH_SMART_CONTROL"
     BMG_MARS = "BMG_MARS"
@@ -244,6 +248,7 @@ def get_parser(
     Vendor.APPBIO_QUANTSTUDIO: AppBioQuantStudioParser,
     Vendor.APPBIO_QUANTSTUDIO_DESIGNANDANALYSIS: AppBioQuantStudioDesignandanalysisParser,
     Vendor.BECKMAN_COULTER_BIOMEK: BeckmanCoulterBiomekParser,
+    Vendor.BECKMAN_ECHO_CHERRY_PICK: BeckmanEchoCherryPickParser,
     Vendor.BECKMAN_ECHO_PLATE_REFORMAT: BeckmanEchoPlateReformatParser,
     Vendor.BECKMAN_PHARMSPEC: PharmSpecParser,
     Vendor.BECKMAN_VI_CELL_BLU: ViCellBluParser,

src/allotropy/parsers/beckman_echo_cherry_pick/beckman_echo_cherry_pick_parser.py

@@ -0,0 +1,37 @@
+import pandas as pd
+
+from allotropy.allotrope.models.adm.liquid_handler.benchling._2024._11.liquid_handler import (
+    Model,
+)
+from allotropy.allotrope.schema_mappers.adm.liquid_handler.benchling._2024._11.liquid_handler import (
+    Data,
+    Mapper,
+)
+from allotropy.named_file_contents import NamedFileContents
+from allotropy.parsers.beckman_echo_cherry_pick.beckman_echo_cherry_pick_reader import (
+    BeckmanEchoCherryPickReader,
+)
+from allotropy.parsers.beckman_echo_cherry_pick.beckman_echo_cherry_pick_structure import (
+    create_measurement_groups,
+    create_metadata,
+)
+from allotropy.parsers.beckman_echo_cherry_pick.constants import DISPLAY_NAME
+from allotropy.parsers.release_state import ReleaseState
+from allotropy.parsers.vendor_parser import VendorParser
+
+
+class BeckmanEchoCherryPickParser(VendorParser[Data, Model]):
+    DISPLAY_NAME = DISPLAY_NAME
+    RELEASE_STATE = ReleaseState.RECOMMENDED
+    SUPPORTED_EXTENSIONS = BeckmanEchoCherryPickReader.SUPPORTED_EXTENSIONS
+    SCHEMA_MAPPER = Mapper
+
+    def create_data(self, named_file_contents: NamedFileContents) -> Data:
+        reader = BeckmanEchoCherryPickReader(named_file_contents)
+        measurement_groups = create_measurement_groups(
+            pd.concat(reader.sections.values(), ignore_index=True), reader.header
+        )
+        return Data(
+            create_metadata(reader.header, named_file_contents.original_file_path),
+            measurement_groups,
+        )

src/allotropy/parsers/beckman_echo_cherry_pick/beckman_echo_cherry_pick_reader.py

@@ -0,0 +1,76 @@
+from io import StringIO
+import re
+
+import numpy as np
+import pandas as pd
+
+from allotropy.exceptions import AllotropeConversionError
+from allotropy.named_file_contents import NamedFileContents
+from allotropy.parsers.lines_reader import (
+    LinesReader,
+)
+from allotropy.parsers.utils.pandas import (
+    df_to_series_data,
+    read_csv,
+    SeriesData,
+)
+from allotropy.parsers.utils.values import assert_not_none
+
+
+class BeckmanEchoCherryPickReader:
+    SUPPORTED_EXTENSIONS = "csv"
+    header: SeriesData
+    sections: dict[str, pd.DataFrame]
+
+    def __init__(self, named_file_contents: NamedFileContents) -> None:
+        reader = LinesReader.create(named_file_contents)
+
+        # read header section, up to a data section delimiter
+        header_lines = [
+            line.strip() for line in reader.pop_until(r"^\[.+\]") if line.strip()
+        ]
+        if not header_lines:
+            msg = "Cannot parse data from empty header."
+            raise AllotropeConversionError(msg)
+
+        # now read multiple sections, each starting with a section title: '[Section Title]', with header row and multiple data rows.
+        sections = {}
+
+        # we expect to have at least one section, and that the current line of the reader is at the title
+        # (due to reader.pop_until(r"^\[.+\]") above)
+        while not reader.is_empty():
+            match = re.match(
+                r"^\[(.+)\]",
+                assert_not_none(reader.pop(), "Unexpected empty section"),
+            )
+            if match:
+                title = str(
+                    assert_not_none(
+                        match, f"Cannot read title section: {reader.get()}"
+                    ).groups()[0]
+                )
+                data_lines = list(
+                    reader.pop_until_empty()
+                )  # read all lines of section, including header line
+                reader.drop_empty()  # sections are separated by empty lines
+
+                sections[title] = read_csv(
+                    StringIO("\n".join(data_lines)), sep=","
+                ).replace(np.nan, None)
+            else:
+                break
+
+        # read footer section, after last tabular data section
+        footer_lines = list(reader.pop_until_empty())
+        header_lines.extend(footer_lines)
+
+        # parse header/footer to SeriesData
+        with StringIO("\n".join(header_lines)) as csv_stream:
+            raw_data = read_csv(
+                csv_stream, header=None, sep=",", skipinitialspace=True, index_col=0
+            )
+        raw_data.index = raw_data.index.str.replace("*", "")
+        header = df_to_series_data(raw_data.T.replace(np.nan, None))
+
+        self.header = header
+        self.sections = sections

src/allotropy/parsers/beckman_echo_cherry_pick/beckman_echo_cherry_pick_structure.py

@@ -0,0 +1,131 @@
+from functools import partial
+from pathlib import Path
+
+import pandas as pd
+
+from allotropy.allotrope.schema_mappers.adm.liquid_handler.benchling._2024._11.liquid_handler import (
+    Error,
+    Measurement,
+    MeasurementGroup,
+    Metadata,
+)
+from allotropy.exceptions import AllotropeConversionError
+from allotropy.parsers.beckman_echo_cherry_pick import constants
+from allotropy.parsers.constants import NOT_APPLICABLE
+from allotropy.parsers.utils.pandas import map_rows, SeriesData
+from allotropy.parsers.utils.uuids import random_uuid_str
+
+
+def create_metadata(header_footer_data: SeriesData, file_path: str) -> Metadata:
+    path = Path(file_path)
+    return Metadata(
+        file_name=path.name,
+        asm_file_identifier=path.with_suffix(".json").name,
+        unc_path=str(path),
+        data_system_instance_identifier=NOT_APPLICABLE,
+        device_type=constants.DEVICE_TYPE,
+        product_manufacturer=constants.PRODUCT_MANUFACTURER,
+        software_version=header_footer_data[str, "Application Version"],
+        model_number=header_footer_data.get(str, "Instrument Model"),
+        equipment_serial_number=header_footer_data.get(str, "Instrument Serial Number"),
+        software_name=header_footer_data[str, "Application Name"],
+        device_system_custom_info={
+            "Instrument Software Version": header_footer_data.get(
+                str, "Instrument Software Version"
+            )
+        },
+        custom_info=header_footer_data.get_unread(),
+    )
+
+
+def _create_measurement(row_data: SeriesData, run_date_time: str | None) -> Measurement:
+    def convert_echo_nl_to_ul(value: float | None) -> float | None:
+        return (
+            (value * constants.CHERRY_PICK_REPORT_VOLUME_CONVERSION_TO_UL)
+            if value is not None
+            else None
+        )
+
+    # If the Date Time Point only contains time (no date) combine with Run Date/Time to create measurement time.
+    # We detect if there is a date in the timestamp by checking for a space, and if there is no space, add the date
+    # component of
+    date_time_point = row_data[str, "Date Time Point"]
+    if " " not in date_time_point.strip():
+        if not run_date_time or " " not in run_date_time:
+            msg = "Cannot parse timestamp for measurement, 'Date Time Point' does not contain a date (time only) and there is no valid 'Run Date/Time' in the header to infer date from."
+            raise AllotropeConversionError(msg)
+        run_date_time_date = run_date_time.split(" ")
+        date_time_point = f"{run_date_time_date[0]} {date_time_point}"
+
+    return Measurement(
+        identifier=random_uuid_str(),
+        measurement_time=date_time_point,
+        sample_identifier=row_data.get(str, "Sample ID", NOT_APPLICABLE),
+        source_plate=row_data[str, "Source Plate Barcode"],
+        source_well=row_data[str, "Source Well"],
+        source_location=row_data[str, "Source Plate Name"],
+        destination_plate=row_data[str, "Destination Plate Barcode"],
+        destination_well=row_data[str, "Destination Well"],
+        destination_location=row_data[str, "Destination Plate Name"],
+        aspiration_volume=convert_echo_nl_to_ul(row_data.get(float, "Actual Volume")),
+        transfer_volume=convert_echo_nl_to_ul(row_data.get(float, "Actual Volume")),
+        injection_volume_setting=convert_echo_nl_to_ul(
+            row_data.get(float, "Transfer Volume")
+        ),
+        device_control_custom_info={
+            "sample name": row_data.get(str, "Sample Name"),
+            "destination well x offset": row_data.get(
+                float, "Destination Well X Offset"
+            ),
+            "destination well y offset": row_data.get(
+                float, "Destination Well Y Offset"
+            ),
+            "current fluid volume": row_data.get(
+                float, "Current Fluid Volume"
+            ),  # This is already in uL, so don't convert to nL
+            "intended transfer volume": convert_echo_nl_to_ul(
+                row_data.get(float, "Transfer Volume")
+            ),
+            "source labware name": row_data.get(str, "Source Plate Type"),
+            "destination labware name": row_data.get(str, "Destination Plate Type"),
+            "fluid composition": row_data.get(str, "Fluid Composition"),
+            "fluid units": row_data.get(str, "Fluid Units"),
+            "fluid type": row_data.get(str, "Fluid Type"),
+            "transfer status": row_data.get(str, "Transfer Status", NOT_APPLICABLE),
+        },
+        errors=[
+            Error(
+                error=row_data[str, "Transfer Status"],
+                feature=row_data[str, "Transfer Status"].split(": ")[0],
+            )
+        ]
+        if row_data.get(str, "Transfer Status")
+        else [],
+        custom_info=row_data.get_unread(),
+    )
+
+
+def create_measurement_groups(
+    data: pd.DataFrame, header: SeriesData
+) -> list[MeasurementGroup]:
+    run_date_time = header.get(str, "Run Date/Time")
+    create_measurement = partial(_create_measurement, run_date_time=run_date_time)
+    return [
+        MeasurementGroup(
+            analyst=header[str, "User Name"],
+            analytical_method_identifier=header.get(str, "Protocol Name"),
+            experimental_data_identifier=header.get(str, "Run ID"),
+            measurements=map_rows(data, create_measurement),
+            custom_info={
+                "Run Date/Time": run_date_time,
+                "Sample Pick List File Name": header.get(
+                    str, "Sample Pick List File Name"
+                ),
+                "Control Pick List File Name": header.get(
+                    str, "Control Pick List File Name"
+                ),
+                "Reference ID": header.get(str, "Reference ID"),
+                "Order ID": header.get(str, "Order ID"),
+            },
+        )
+    ]

src/allotropy/parsers/beckman_echo_cherry_pick/constants.py

@@ -0,0 +1,10 @@
+DISPLAY_NAME = "Beckman Echo Cherry Pick"
+DEVICE_TYPE = "liquid handler"
+PRODUCT_MANUFACTURER = "Beckman Coulter"
+SOFTWARE_NAME = "Labcyte Echo Cherry Pick"
+
+TRANSFER_RESULTS_LABEL = "DETAILS"
+TRANSFER_ERRORS_LABEL = "EXCEPTIONS"
+
+CHERRY_PICK_REPORT_VOLUME_UNITS = "nL"
+CHERRY_PICK_REPORT_VOLUME_CONVERSION_TO_UL = 0.001