solve comments

Author: joshua-benchling

src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py

@@ -10,19 +10,7 @@
 from allotropy.parsers.agilent_gen5.agilent_gen5_reader import AgilentGen5Reader
 from allotropy.parsers.agilent_gen5.agilent_gen5_structure import (
     create_metadata,
-    Gen5DataContext,
-    get_concentrations,
-    get_identifiers,
-    get_kinetic_measurements,
-    get_results_section,
-    get_temperature,
-    HeaderData,
-    KineticData,
-    KineticResultProcessor,
-    ReadData,
-    ResultProcessor,
-    SpectralResultProcessor,
-    StandardResultProcessor,
+    get_processor,
 )
 from allotropy.parsers.agilent_gen5.constants import (
     NO_MEASUREMENTS_ERROR,
@@ -40,19 +28,12 @@ class AgilentGen5Parser(VendorParser[Data, Model]):
 
     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
-        )
+        context = reader.extract_data_context(named_file_contents.original_file_path)
 
-        processor = self._get_processor(context)
+        processor = get_processor(context)
         measurement_groups, calculated_data = processor.process(context)
 
         if not measurement_groups:
@@ -63,41 +44,3 @@ def create_data(self, named_file_contents: NamedFileContents) -> Data:
             measurement_groups=measurement_groups,
             calculated_data=calculated_data,
         )
-
-    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 (
-            {},
-            [],
-            {},
-        )
-
-        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,
-        )
-
-    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
-
-        msg = "No suitable processor found for the data."
-        raise AllotropeConversionError(msg)

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
@@ -87,3 +98,79 @@ def time_section(self) -> list[str] | None:
     @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,
+        )

src/allotropy/parsers/agilent_gen5/agilent_gen5_structure.py

@@ -32,7 +32,6 @@
     AllotropeConversionError,
     AllotropyParserError,
 )
-from allotropy.parsers.agilent_gen5.agilent_gen5_reader import AgilentGen5Reader
 from allotropy.parsers.agilent_gen5.constants import (
     ALPHALISA_FLUORESCENCE_FOUND,
     DATA_SOURCE_FEATURE_VALUES,
@@ -732,55 +731,6 @@ def _safe_get(data_list: list[str], idx: int) -> str | None:
             return data_list[idx] if data_list and idx < len(data_list) else None
 
 
-def _validate_result_sections(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(reader: AgilentGen5Reader) -> 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 reader.sections:
-        return reader.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 reader.sections.items():
-        if name == "Layout":
-            continue
-        if is_results(section):
-            result_sections.append(section[1:])
-
-    if result_sections:
-        _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 get_concentrations(layout_lines: list[str] | None) -> dict[str, float | None]:
     """Extract concentration/dilution values from the Layout section.
 
@@ -1618,8 +1568,9 @@ def _convert_time_to_seconds(time_str: str | None) -> float:
 class ResultProcessor(ABC):
     """Abstract base class for processing different types of Gen5 results."""
 
+    @classmethod
     @abstractmethod
-    def can_process(self, context: Gen5DataContext) -> bool:
+    def can_process(cls, context: Gen5DataContext) -> bool:
         """Check if this processor can handle the given context."""
         pass
 
@@ -1634,7 +1585,8 @@ def process(
 class SpectralResultProcessor(ResultProcessor):
     """Processor for spectral measurements with wavelength data."""
 
-    def can_process(self, context: Gen5DataContext) -> bool:
+    @classmethod
+    def can_process(cls, context: Gen5DataContext) -> bool:
         return context.is_spectral and context.wavelength_section is not None
 
     def process(
@@ -1654,7 +1606,8 @@ def process(
 class KineticResultProcessor(ResultProcessor):
     """Processor for kinetic measurements with time series data."""
 
-    def can_process(self, context: Gen5DataContext) -> bool:
+    @classmethod
+    def can_process(cls, context: Gen5DataContext) -> bool:
         if not context.has_kinetic_data:
             return False
 
@@ -1684,7 +1637,8 @@ def process(
 class StandardResultProcessor(ResultProcessor):
     """Processor for standard endpoint measurements."""
 
-    def can_process(self, _: Gen5DataContext) -> bool:
+    @classmethod
+    def can_process(cls, _: Gen5DataContext) -> bool:
         return True
 
     def process(
@@ -1698,3 +1652,18 @@ def process(
             context.actual_temperature,
             context.concentration_values,
         )
+
+
+def get_processor(
+    context: Gen5DataContext,
+) -> SpectralResultProcessor | KineticResultProcessor | StandardResultProcessor:
+    """Factory function to get the appropriate processor for the given context."""
+    if SpectralResultProcessor.can_process(context):
+        return SpectralResultProcessor()
+    if KineticResultProcessor.can_process(context):
+        return KineticResultProcessor()
+    if StandardResultProcessor.can_process(context):
+        return StandardResultProcessor()
+
+    msg = "No suitable processor found for the data."
+    raise AllotropeConversionError(msg)

src/allotropy/parsers/agilent_gen5_image/agilent_gen5_image_parser.py

@@ -9,7 +9,6 @@
 from allotropy.parsers.agilent_gen5.agilent_gen5_reader import AgilentGen5Reader
 from allotropy.parsers.agilent_gen5.agilent_gen5_structure import (
     get_identifiers,
-    get_results_section,
     HeaderData,
 )
 from allotropy.parsers.agilent_gen5_image.agilent_gen5_image_structure import (
@@ -38,7 +37,7 @@ def create_data(self, named_file_contents: NamedFileContents) -> Data:
         return Data(
             metadata=create_metadata(header_data),
             measurement_groups=create_results(
-                get_results_section(reader),
+                reader.get_results_section(),
                 header_data,
                 read_data,
                 get_identifiers(reader.sections.get("Layout")),