refactor: Agilent gen5 - refactor adapter (#1038)

Co-authored-by: Nathan Stender <nathan.stender@benchling.com>

Author: joshua-benchling

src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py

@@ -9,22 +9,11 @@
 from allotropy.named_file_contents import NamedFileContents
 from allotropy.parsers.agilent_gen5.agilent_gen5_reader import AgilentGen5Reader
 from allotropy.parsers.agilent_gen5.agilent_gen5_structure import (
-    create_kinetic_results,
     create_metadata,
-    create_results,
-    create_spectrum_results,
-    get_concentrations,
-    get_identifiers,
-    get_kinetic_measurements,
-    get_results_section,
-    get_temperature,
-    HeaderData,
-    KineticData,
-    ReadData,
+    get_processor,
 )
 from allotropy.parsers.agilent_gen5.constants import (
     NO_MEASUREMENTS_ERROR,
-    ReadType,
 )
 from allotropy.parsers.release_state import ReleaseState
 from allotropy.parsers.vendor_parser import VendorParser
@@ -38,83 +27,16 @@ class AgilentGen5Parser(VendorParser[Data, Model]):
 
     def create_data(self, named_file_contents: NamedFileContents) -> Data:
         reader = AgilentGen5Reader(named_file_contents)
+        context = reader.extract_data_context(named_file_contents.original_file_path)
 
-        if (results_section := get_results_section(reader)) is None:
-            reader.header_data.get_unread()
-            raise AllotropeConversionError(NO_MEASUREMENTS_ERROR)
-
-        header_data = HeaderData.create(
-            reader.header_data, named_file_contents.original_file_path
-        )
-        read_data = ReadData.create(reader.sections["Procedure Details"])
-        kinetic_data = KineticData.create(reader.sections["Procedure Details"])
-
-        sample_identifiers = get_identifiers(reader.sections.get("Layout"))
-        concentration_values = get_concentrations(reader.sections.get("Layout"))
-        actual_temperature = get_temperature(reader.sections.get("Actual Temperature"))
-        kinetic_result = get_kinetic_measurements(reader.sections.get("Time"))
-        kinetic_measurements, kinetic_elapsed_time, kinetic_errors = kinetic_result or (
-            {},
-            [],
-            {},
-        )
-
-        if kinetic_data and not (kinetic_measurements and kinetic_elapsed_time):
-            msg = "Kinetic data is present in the file but no kinetic measurements data is found."
-            raise AllotropeConversionError(msg)
-
-        read_is_spectral = read_data[0].read_type == ReadType.SPECTRUM
-        if read_is_spectral and reader.sections.get("Wavelength"):
-            (
-                wavelength_measurements,
-                wavelength_calculated_data,
-            ) = create_spectrum_results(
-                header_data,
-                read_data_list=read_data,
-                wavelengths_sections=reader.sections.get("Wavelength"),
-                sample_identifiers=sample_identifiers,
-                actual_temperature=actual_temperature,
-                results_section=results_section,
-                concentration_values=concentration_values,
-            )
-
-            if not wavelength_measurements:
-                raise AllotropeConversionError(NO_MEASUREMENTS_ERROR)
-
-            return Data(
-                metadata=create_metadata(header_data),
-                measurement_groups=wavelength_measurements,
-                calculated_data=wavelength_calculated_data,
-            )
-
-        if kinetic_data:
-            measurement_groups, calculated_data = create_kinetic_results(
-                results_section,
-                header_data,
-                read_data,
-                sample_identifiers,
-                actual_temperature,
-                kinetic_data,
-                kinetic_measurements,
-                kinetic_elapsed_time,
-                kinetic_errors,
-                concentration_values,
-            )
-        else:
-            measurement_groups, calculated_data = create_results(
-                results_section,
-                header_data,
-                read_data,
-                sample_identifiers,
-                actual_temperature,
-                concentration_values,
-            )
+        processor = get_processor(context)
+        measurement_groups, calculated_data = processor.process(context)
 
         if not measurement_groups:
             raise AllotropeConversionError(NO_MEASUREMENTS_ERROR)
 
         return Data(
-            metadata=create_metadata(header_data),
+            metadata=create_metadata(context.header_data),
             measurement_groups=measurement_groups,
             calculated_data=calculated_data,
         )

src/allotropy/parsers/agilent_gen5/agilent_gen5_reader.py

@@ -2,8 +2,19 @@
 
 from allotropy.exceptions import AllotropeConversionError
 from allotropy.named_file_contents import NamedFileContents
+from allotropy.parsers.agilent_gen5.agilent_gen5_structure import (
+    Gen5DataContext,
+    get_concentrations,
+    get_identifiers,
+    get_kinetic_measurements,
+    get_temperature,
+    HeaderData,
+    KineticData,
+    ReadData,
+)
 from allotropy.parsers.agilent_gen5.constants import (
     MULTIPLATE_FILE_ERROR,
+    NO_MEASUREMENTS_ERROR,
     NO_PLATE_DATA_ERROR,
 )
 from allotropy.parsers.lines_reader import SectionLinesReader
@@ -59,3 +70,107 @@ def __init__(self, named_file_contents: NamedFileContents) -> None:
             lines = list(plate_reader.pop_until_empty())
             self.sections[lines[0].split("\t")[0].strip(":")] = lines
             plate_reader.drop_empty()
+
+    def get_required_section(self, section_name: str) -> list[str]:
+        """Get a required section, raises error if not found."""
+        section = self.sections.get(section_name)
+        if section is None:
+            msg = f"Required section '{section_name}' not found."
+            raise AllotropeConversionError(msg)
+        return section
+
+    @property
+    def procedure_details(self) -> list[str]:
+        return self.get_required_section("Procedure Details")
+
+    @property
+    def layout_section(self) -> list[str] | None:
+        return self.sections.get("Layout")
+
+    @property
+    def wavelength_section(self) -> list[str] | None:
+        return self.sections.get("Wavelength")
+
+    @property
+    def time_section(self) -> list[str] | None:
+        return self.sections.get("Time")
+
+    @property
+    def actual_temperature_section(self) -> list[str] | None:
+        return self.sections.get("Actual Temperature")
+
+    def _validate_result_sections(self, result_sections: list[list[str]]) -> None:
+        """Validates whether all the result sections dimensions are consistent."""
+        first_section = result_sections[0]
+
+        for section in result_sections[1:]:
+            if not first_section[0] == section[0] and len(first_section) == len(
+                section
+            ):
+                msg = "All result tables should have the same dimensions."
+                raise AllotropeConversionError(msg)
+
+    def get_results_section(self) -> list[str] | None:
+        """Returns a valid Results Matrix from the reader sections if found.
+
+        Checks for Results in the reader sections, if not found, creates the results matrix with all
+        sections that are correctly formatted as a results table (excluding the Layout section). If
+        no tables with results are found, returns None
+        """
+        if "Results" in self.sections:
+            return self.sections["Results"]
+
+        def is_results(section: list[str]) -> bool:
+            return (
+                len(section) > 2
+                and section[1].startswith("\t1")
+                and section[2].startswith("A\t")
+            )
+
+        result_sections = []
+        for name, section in self.sections.items():
+            if name == "Layout":
+                continue
+            if is_results(section):
+                result_sections.append(section[1:])
+
+        if result_sections:
+            self._validate_result_sections(result_sections)
+            return [
+                "Results",
+                result_sections[0][0],
+                *[
+                    section[i + 1]
+                    for i in range(len(result_sections[0]) - 1)
+                    for section in result_sections
+                ],
+            ]
+
+        return None
+
+    def extract_data_context(self, file_path: str) -> Gen5DataContext:
+        results_section = self.get_results_section()
+        if not results_section:
+            self.header_data.get_unread()
+            raise AllotropeConversionError(NO_MEASUREMENTS_ERROR)
+
+        kinetic_result = get_kinetic_measurements(self.time_section)
+        kinetic_measurements, kinetic_elapsed_time, kinetic_errors = kinetic_result or (
+            {},
+            [],
+            {},
+        )
+
+        return Gen5DataContext(
+            header_data=HeaderData.create(self.header_data, file_path),
+            read_data=ReadData.create(self.procedure_details),
+            kinetic_data=KineticData.create(self.procedure_details),
+            results_section=results_section,
+            sample_identifiers=get_identifiers(self.layout_section),
+            concentration_values=get_concentrations(self.layout_section),
+            actual_temperature=get_temperature(self.actual_temperature_section),
+            kinetic_measurements=kinetic_measurements,
+            kinetic_elapsed_time=kinetic_elapsed_time,
+            kinetic_errors=kinetic_errors,
+            wavelength_section=self.wavelength_section,
+        )