feat: Beckman Coulter Biomek Liquid Handler - add support for log headerless files

src/allotropy/parsers/beckman_coulter_biomek/beckman_coulter_biomek_reader.py

@@ -1,3 +1,6 @@
+import re
+from typing import ClassVar
+
 import pandas as pd
 
 from allotropy.named_file_contents import NamedFileContents
@@ -11,18 +14,85 @@ class BeckmanCoulterBiomekReader:
     header: SeriesData
     data: pd.DataFrame
 
+    # Fixed column names based on headerless files
+    FIXED_COLUMN_NAMES: ClassVar[list[str]] = [
+        "Time Stamp",
+        "Pod",
+        "Transfer Step",
+        "Deck Position",
+        "Labware Name",
+        "Labware Barcode",
+        "Well Index",
+        "Sample Name",
+        "Probe",
+        "Amount",
+        "Liquid Handling Technique",
+    ]
+
+    def _looks_like_data_line(self, line: str) -> bool:
+        """
+        Check if a line looks like data rather than metadata.
+        Data lines should have a date/time pattern as the first field.
+        """
+        parts = line.split(",")
+        if len(parts) < 3:
+            return False
+
+        first_part = parts[0].strip()
+        # Check if first part looks like a timestamp (contains numbers and slashes or dashes)
+        # Examples: "08/15/2025 10:59:55", "9/4/24 14:15"
+        timestamp_pattern = r"^\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\s+\d{1,2}:\d{2}"
+        return bool(re.match(timestamp_pattern, first_part))
+
     def __init__(self, named_file_contents: NamedFileContents) -> None:
         reader = CsvReader.create(named_file_contents)
 
-        header_lines = list(reader.pop_until(".*Well Index.*"))
-        header_dict = {}
-        for line in header_lines:
-            split = line.split(",")[0].split("=", maxsplit=1)
-            if len(split) != 2:
-                continue
-            header_dict[split[0].strip()] = split[1].strip()
-
-        self.header = SeriesData(pd.Series(header_dict))
-        self.data = assert_not_none(
-            reader.pop_csv_block_as_df(header="infer"), "Cannot parse empty dataset"
-        )
+        # Check if this file has a header line by looking for "Well Index"
+        has_header_line = any("Well Index" in line for line in reader.lines)
+
+        if has_header_line:
+            header_lines = list(reader.pop_until(".*Well Index.*"))
+            header_dict = {}
+            for line in header_lines:
+                split = line.split(",")[0].split("=", maxsplit=1)
+                if len(split) != 2:
+                    continue
+                header_dict[split[0].strip()] = split[1].strip()
+
+            self.header = SeriesData(pd.Series(header_dict))
+            self.data = assert_not_none(
+                reader.pop_csv_block_as_df(header="infer"), "Cannot parse empty dataset"
+            )
+        else:
+            # Handle headerless files
+            header_lines = []
+            data_lines = []
+
+            for raw_line in reader.lines:
+                line = raw_line.strip()
+                if not line:
+                    continue
+
+                if self._looks_like_data_line(line):
+                    data_lines.append(line)
+                else:
+                    header_lines.append(line)
+
+            # Parse header metadata
+            header_dict = {}
+            for line in header_lines:
+                split = line.split(",")[0].split("=", maxsplit=1)
+                if len(split) != 2:
+                    continue
+                header_dict[split[0].strip()] = split[1].strip()
+
+            self.header = SeriesData(pd.Series(header_dict))
+
+            # Create a new reader with just the data lines
+            data_reader = CsvReader(data_lines)
+            self.data = assert_not_none(
+                data_reader.pop_csv_block_as_df(
+                    header=None, names=self.FIXED_COLUMN_NAMES
+                ),
+                "Cannot parse empty dataset",
+            )

src/allotropy/parsers/beckman_coulter_biomek/beckman_coulter_biomek_structure.py

@@ -51,10 +51,14 @@ def _create_measurement(
         identifier=random_uuid_str(),
         measurement_time=dispense_data[str, "Time Stamp"],
         sample_identifier=NOT_APPLICABLE,
-        source_plate=aspiration_data[str, "Labware Barcode"],
+        source_plate=aspiration_data.get(
+            str, "Labware Barcode", validate=SeriesData.NOT_NAN
+        ),
         source_well=aspiration_data[str, "Well Index"],
         source_location=aspiration_data[str, "Deck Position"],
-        destination_plate=dispense_data[str, "Labware Barcode"],
+        destination_plate=dispense_data.get(
+            str, "Labware Barcode", validate=SeriesData.NOT_NAN
+        ),
         destination_well=dispense_data[str, "Well Index"],
         destination_location=dispense_data[str, "Deck Position"],
         aspiration_volume=aspiration_data[float, "Amount"],