refactor agilent gen5

Author: joshua-benchling

src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py

@@ -9,24 +9,26 @@
 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,
+    Gen5DataContext,
     get_concentrations,
     get_identifiers,
     get_kinetic_measurements,
     get_results_section,
     get_temperature,
     HeaderData,
     KineticData,
+    KineticResultProcessor,
     ReadData,
+    ResultProcessor,
+    SpectralResultProcessor,
+    StandardResultProcessor,
 )
 from allotropy.parsers.agilent_gen5.constants import (
     NO_MEASUREMENTS_ERROR,
-    ReadType,
 )
 from allotropy.parsers.release_state import ReleaseState
+from allotropy.parsers.utils.timestamp_parser import TimestampParser
 from allotropy.parsers.vendor_parser import VendorParser
 
 
@@ -36,85 +38,66 @@ class AgilentGen5Parser(VendorParser[Data, Model]):
     SUPPORTED_EXTENSIONS = AgilentGen5Reader.SUPPORTED_EXTENSIONS
     SCHEMA_MAPPER = Mapper
 
+    def __init__(self, timestamp_parser: TimestampParser | None = None):
+        super().__init__(timestamp_parser)
+        self._processors = [
+            SpectralResultProcessor(),
+            KineticResultProcessor(),
+            StandardResultProcessor(),
+        ]
+
     def create_data(self, named_file_contents: NamedFileContents) -> Data:
         reader = AgilentGen5Reader(named_file_contents)
+        context = self._extract_data_context(
+            reader, named_file_contents.original_file_path
+        )
 
-        if (results_section := get_results_section(reader)) is None:
-            reader.header_data.get_unread()
+        processor = self._get_processor(context)
+        measurement_groups, calculated_data = processor.process(context)
+
+        if not measurement_groups:
             raise AllotropeConversionError(NO_MEASUREMENTS_ERROR)
 
-        header_data = HeaderData.create(
-            reader.header_data, named_file_contents.original_file_path
+        return Data(
+            metadata=create_metadata(context.header_data),
+            measurement_groups=measurement_groups,
+            calculated_data=calculated_data,
         )
-        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"))
+    def _extract_data_context(
+        self, reader: AgilentGen5Reader, file_path: str
+    ) -> Gen5DataContext:
+        results_section = get_results_section(reader)
+        if not results_section:
+            reader.header_data.get_unread()
+            raise AllotropeConversionError(NO_MEASUREMENTS_ERROR)
+
+        kinetic_result = get_kinetic_measurements(reader.time_section)
         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,
-            )
+        return Gen5DataContext(
+            header_data=HeaderData.create(reader.header_data, file_path),
+            read_data=ReadData.create(reader.procedure_details),
+            kinetic_data=KineticData.create(reader.procedure_details),
+            results_section=results_section,
+            sample_identifiers=get_identifiers(reader.layout_section),
+            concentration_values=get_concentrations(reader.layout_section),
+            actual_temperature=get_temperature(reader.actual_temperature_section),
+            kinetic_measurements=kinetic_measurements,
+            kinetic_elapsed_time=kinetic_elapsed_time,
+            kinetic_errors=kinetic_errors,
+            wavelength_section=reader.wavelength_section,
+        )
 
-        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,
-            )
+    def _get_processor(self, context: Gen5DataContext) -> ResultProcessor:
+        """Get the appropriate processor for the given context."""
+        for processor in self._processors:
+            if processor.can_process(context):
+                return processor
 
-        if not measurement_groups:
-            raise AllotropeConversionError(NO_MEASUREMENTS_ERROR)
-
-        return Data(
-            metadata=create_metadata(header_data),
-            measurement_groups=measurement_groups,
-            calculated_data=calculated_data,
-        )
+        msg = "No suitable processor found for the data."
+        raise AllotropeConversionError(msg)

src/allotropy/parsers/agilent_gen5/agilent_gen5_reader.py

@@ -59,3 +59,31 @@ 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")