diff --git a/src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py b/src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py index fe36353d64..a35cf91921 100644 --- a/src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py +++ b/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, ) diff --git a/src/allotropy/parsers/agilent_gen5/agilent_gen5_reader.py b/src/allotropy/parsers/agilent_gen5/agilent_gen5_reader.py index 7b7d0a8b0e..7d51a2f6d6 100644 --- a/src/allotropy/parsers/agilent_gen5/agilent_gen5_reader.py +++ b/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, + ) diff --git a/src/allotropy/parsers/agilent_gen5/agilent_gen5_structure.py b/src/allotropy/parsers/agilent_gen5/agilent_gen5_structure.py index 20ce3c648f..9448cc1867 100644 --- a/src/allotropy/parsers/agilent_gen5/agilent_gen5_structure.py +++ b/src/allotropy/parsers/agilent_gen5/agilent_gen5_structure.py @@ -1,12 +1,13 @@ from __future__ import annotations +from abc import ABC, abstractmethod from collections import defaultdict from dataclasses import dataclass from io import StringIO import math from pathlib import Path import re -from typing import Any +from typing import Any, cast import pandas as pd @@ -31,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, @@ -330,6 +330,33 @@ def _parse_kinetic_section(cls, line: str) -> KineticData: ) +@dataclass +class Gen5DataContext: + header_data: HeaderData + read_data: list[ReadData] + kinetic_data: KineticData | None + results_section: list[str] + sample_identifiers: dict[str, str] + concentration_values: dict[str, float | None] + actual_temperature: float | None + kinetic_measurements: dict[str, list[float | None]] + kinetic_elapsed_time: list[float] + kinetic_errors: dict[str, list[ErrorDocument]] + wavelength_section: list[str] | None + + @property + def is_spectral(self) -> bool: + return bool(self.read_data and self.read_data[0].read_type == ReadType.SPECTRUM) + + @property + def has_kinetic_data(self) -> bool: + return self.kinetic_data is not None + + @property + def has_kinetic_measurements(self) -> bool: + return bool(self.kinetic_measurements and self.kinetic_elapsed_time) + + @dataclass(frozen=True) class ReadData: read_mode: ReadMode @@ -448,66 +475,123 @@ def get_read_type(procedure_details: str) -> ReadType: def _get_measurement_labels( cls, device_control_data: DeviceControlData, read_mode: str, read_type: str ) -> tuple[list[str], dict[str, set[str]]]: + """Get measurement labels and aliases for the given read mode.""" step_label = device_control_data.step_label label_prefix = f"{step_label}:" if step_label else "" - measurement_labels = [] - # Some measurement labels may be reported in more than one format in the result rows, e.g. - # fluorescence measurements may include bandwidths, or not: 360/40,460/40 or 360,460. - label_aliases: dict[str, set[str]] = {} + if read_mode == ReadMode.ABSORBANCE: - measurement_labels = cls._get_absorbance_measurement_labels( - label_prefix, device_control_data - ) - if not measurement_labels and read_type == ReadType.SPECTRUM: - measurement_labels = [f"{step_label}:Spectrum"] - - if read_mode == ReadMode.FLUORESCENCE: - excitations: list[str] = ( - device_control_data.get_list(EXCITATION_KEY) - or device_control_data.get_list(FIXED_EXCITATION_KEY) - or device_control_data.get_list(EXCITATION_START_KEY) + return cls._get_absorbance_labels( + label_prefix, device_control_data, read_type ) - emissions: list[str] = ( - device_control_data.get_list(EMISSION_KEY) - or device_control_data.get_list(FIXED_EMISSION_KEY) - or device_control_data.get_list(EMISSION_START_KEY) + elif read_mode == ReadMode.FLUORESCENCE: + return cls._get_fluorescence_labels(label_prefix, device_control_data) + elif read_mode == ReadMode.LUMINESCENCE: + return cls._get_luminescence_labels(label_prefix, device_control_data) + else: + return [], {} + + @classmethod + def _get_absorbance_labels( + cls, label_prefix: str, device_control_data: DeviceControlData, read_type: str + ) -> tuple[list[str], dict[str, set[str]]]: + """Get measurement labels for absorbance measurements.""" + measurement_labels = cls._get_absorbance_measurement_labels( + label_prefix, device_control_data + ) + if not measurement_labels and read_type == ReadType.SPECTRUM: + measurement_labels = [f"{label_prefix.rstrip(':')}:Spectrum"] + return measurement_labels, {} + + @classmethod + def _get_fluorescence_labels( + cls, label_prefix: str, device_control_data: DeviceControlData + ) -> tuple[list[str], dict[str, set[str]]]: + """Get measurement labels for fluorescence measurements.""" + excitations, emissions = cls._get_excitation_emission_data(device_control_data) + + if not excitations or not emissions: + return ["Alpha"], {} + + measurement_labels = [ + f"{label_prefix}{excitation},{emission}" + for excitation, emission in zip(excitations, emissions, strict=True) + ] + + if device_control_data.is_polarization: + return cls._handle_polarization_labels(measurement_labels) + else: + return cls._create_fluorescence_aliases( + label_prefix, excitations, emissions ) - measurement_labels = [ - f"{label_prefix}{excitation},{emission}" - for excitation, emission in zip(excitations, emissions, strict=True) - ] - if device_control_data.is_polarization: - if len(measurement_labels) != 2: - msg = "Expected the Fluorescence Polarization read mode to contain exactly 2 filter sets." - raise AllotropeConversionError(msg) - measurement_labels = [ - f"{label} [{light_direction}]" - for label, light_direction in zip( - measurement_labels, LIGHT_DIRECTIONS, strict=True - ) - ] - label_aliases = {} - else: - label_aliases = { - f"{label_prefix}{excitation},{emission}": { - f"{label_prefix}{excitation.split('/')[0]},{emission.split('/')[0]}" - } - for excitation, emission in zip(excitations, emissions, strict=True) - } - - if not measurement_labels: - measurement_labels = ["Alpha"] - - if read_mode == ReadMode.LUMINESCENCE: - emissions = ( - device_control_data.get_list(EMISSION_KEY) - or device_control_data.get_list(FIXED_EMISSION_KEY) - or device_control_data.get_list(EMISSION_START_KEY) + @classmethod + def _get_luminescence_labels( + cls, label_prefix: str, device_control_data: DeviceControlData + ) -> tuple[list[str], dict[str, set[str]]]: + """Get measurement labels for luminescence measurements.""" + emissions = ( + device_control_data.get_list(EMISSION_KEY) + or device_control_data.get_list(FIXED_EMISSION_KEY) + or device_control_data.get_list(EMISSION_START_KEY) + ) + + measurement_labels = [] + for emission in emissions: + label = "Lum" if emission in NAN_EMISSION_EXCITATION else emission + measurement_labels.append(f"{label_prefix}{label}") + + return measurement_labels, {} + + @classmethod + def _get_excitation_emission_data( + cls, device_control_data: DeviceControlData + ) -> tuple[list[str], list[str]]: + """Extract excitation and emission data from device control data.""" + excitations = ( + device_control_data.get_list(EXCITATION_KEY) + or device_control_data.get_list(FIXED_EXCITATION_KEY) + or device_control_data.get_list(EXCITATION_START_KEY) + ) + emissions = ( + device_control_data.get_list(EMISSION_KEY) + or device_control_data.get_list(FIXED_EMISSION_KEY) + or device_control_data.get_list(EMISSION_START_KEY) + ) + return excitations, emissions + + @classmethod + def _handle_polarization_labels( + cls, measurement_labels: list[str] + ) -> tuple[list[str], dict[str, set[str]]]: + """Handle polarization measurement labels.""" + if len(measurement_labels) != 2: + msg = "Expected the Fluorescence Polarization read mode to contain exactly 2 filter sets." + raise AllotropeConversionError(msg) + + polarization_labels = [ + f"{label} [{light_direction}]" + for label, light_direction in zip( + measurement_labels, LIGHT_DIRECTIONS, strict=True ) - for emission in emissions: - label = "Lum" if emission in NAN_EMISSION_EXCITATION else emission - measurement_labels.append(f"{label_prefix}{label}") + ] + return polarization_labels, {} + + @classmethod + def _create_fluorescence_aliases( + cls, label_prefix: str, excitations: list[str], emissions: list[str] + ) -> tuple[list[str], dict[str, set[str]]]: + """Create fluorescence measurement labels and their aliases.""" + measurement_labels = [ + f"{label_prefix}{excitation},{emission}" + for excitation, emission in zip(excitations, emissions, strict=True) + ] + + label_aliases = { + f"{label_prefix}{excitation},{emission}": { + f"{label_prefix}{excitation.split('/')[0]},{emission.split('/')[0]}" + } + for excitation, emission in zip(excitations, emissions, strict=True) + } return measurement_labels, label_aliases @@ -538,117 +622,113 @@ def _get_filter_sets( device_control_data: DeviceControlData, read_mode: ReadMode, ) -> dict[str, FilterSet]: - filter_data: dict[str, FilterSet] = {} if read_mode == ReadMode.ABSORBANCE: - return filter_data - - emissions = device_control_data.get_list(EMISSION_KEY) - excitations = device_control_data.get_list(EXCITATION_KEY) - fixed_emissions = device_control_data.get_list(FIXED_EMISSION_KEY) - fixed_excitations = device_control_data.get_list(FIXED_EXCITATION_KEY) - emission_start = device_control_data.get_list(EMISSION_START_KEY) - excitation_start = device_control_data.get_list(EXCITATION_START_KEY) - mirrors = device_control_data.get_list(MIRROR_KEY) - optics = device_control_data.get_list(OPTICS_KEY) - gains = device_control_data.get_list(GAIN_KEY) + return {} + + filter_data_extractor = cls._FilterDataExtractor(device_control_data, read_mode) + filter_data = cls._create_filter_sets_for_labels( + measurement_labels, filter_data_extractor, device_control_data + ) + + for measurement_label, aliases in label_aliases.items(): + for alias in aliases: + filter_data[alias] = filter_data[measurement_label] + return filter_data + + @classmethod + def _create_filter_sets_for_labels( + cls, + measurement_labels: list[str], + extractor: _FilterDataExtractor, + device_control_data: DeviceControlData, + ) -> dict[str, FilterSet]: + gains = extractor.gains if len(measurement_labels) != len(gains): msg = f"Expected the number of measurement labels: {measurement_labels} to match the number of gains: {gains}." raise AllotropeConversionError(msg) + filter_data = {} for idx, label in enumerate(measurement_labels): - mirror = None - if mirrors and read_mode == ReadMode.FLUORESCENCE: - mirror = mirrors[idx] - - # Logic to determine which emission/excitation value to use - excitation = None - if idx < len(excitations) and excitations: - excitation = excitations[idx] - elif idx < len(fixed_excitations) and fixed_excitations: - excitation = fixed_excitations[idx] - elif idx < len(excitation_start) and excitation_start: - excitation = excitation_start[idx] - - emission = None - if idx < len(emissions) and emissions: - emission = emissions[idx] - elif idx < len(fixed_emissions) and fixed_emissions: - emission = fixed_emissions[idx] - elif idx < len(emission_start) and emission_start: - emission = emission_start[idx] - - if excitation and " nm" in excitation: - excitation = excitation.split(" nm")[0] - if emission and " nm" in emission: - emission = emission.split(" nm")[0] - - filter_data[label] = FilterSet( - emission=emission, - gain=gains[idx], - excitation=excitation, - mirror=mirror, - optics=optics[idx] if optics else None, - light_direction=( - LIGHT_DIRECTIONS[idx] - if device_control_data.is_polarization - else None - ), + filter_set = cls._create_single_filter_set( + idx, extractor, device_control_data ) - for measurement_label, aliases in label_aliases.items(): - for alias in aliases: - filter_data[alias] = filter_data[measurement_label] + filter_data[label] = filter_set return filter_data + @classmethod + def _create_single_filter_set( + cls, + idx: int, + extractor: _FilterDataExtractor, + device_control_data: DeviceControlData, + ) -> FilterSet: + excitation = cls._clean_wavelength_value(extractor.get_excitation_at_index(idx)) + emission = cls._clean_wavelength_value(extractor.get_emission_at_index(idx)) + mirror = extractor.get_mirror_at_index(idx) + optics = extractor.get_optics_at_index(idx) + gain = extractor.gains[idx] + + light_direction = ( + LIGHT_DIRECTIONS[idx] if device_control_data.is_polarization 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) + return FilterSet( + emission=emission, + gain=gain, + excitation=excitation, + mirror=mirror, + optics=optics, + light_direction=light_direction, + ) + @staticmethod + def _clean_wavelength_value(wavelength: str | None) -> str | None: + if wavelength and " nm" in wavelength: + return wavelength.split(" nm")[0] + return wavelength + + class _FilterDataExtractor: + def __init__(self, device_control_data: DeviceControlData, read_mode: ReadMode): + self.device_control_data = device_control_data + self.read_mode = read_mode + + self.emissions = device_control_data.get_list(EMISSION_KEY) + self.excitations = device_control_data.get_list(EXCITATION_KEY) + self.fixed_emissions = device_control_data.get_list(FIXED_EMISSION_KEY) + self.fixed_excitations = device_control_data.get_list(FIXED_EXCITATION_KEY) + self.emission_start = device_control_data.get_list(EMISSION_START_KEY) + self.excitation_start = device_control_data.get_list(EXCITATION_START_KEY) + self.mirrors = device_control_data.get_list(MIRROR_KEY) + self.optics = device_control_data.get_list(OPTICS_KEY) + self.gains = device_control_data.get_list(GAIN_KEY) + + def get_excitation_at_index(self, idx: int) -> str | None: + return ( + self._safe_get(self.excitations, idx) + or self._safe_get(self.fixed_excitations, idx) + or self._safe_get(self.excitation_start, idx) + ) -def get_results_section(reader: AgilentGen5Reader) -> list[str] | None: - """Returns a valid Results Matrix from the reader sections if found. + def get_emission_at_index(self, idx: int) -> str | None: + return ( + self._safe_get(self.emissions, idx) + or self._safe_get(self.fixed_emissions, idx) + or self._safe_get(self.emission_start, idx) + ) - 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") - ) + def get_mirror_at_index(self, idx: int) -> str | None: + if self.mirrors and self.read_mode == ReadMode.FLUORESCENCE: + return self._safe_get(self.mirrors, idx) + return None - 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 - ], - ] + def get_optics_at_index(self, idx: int) -> str | None: + return self._safe_get(self.optics, idx) - return None + @staticmethod + 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 get_concentrations(layout_lines: list[str] | None) -> dict[str, float | None]: @@ -1483,3 +1563,107 @@ def _convert_time_to_seconds(time_str: str | None) -> float: except ValueError: msg = f"Invalid time string: '{time_str}'." raise AllotropeConversionError(msg) from None + + +class ResultProcessor(ABC): + """Abstract base class for processing different types of Gen5 results.""" + + @classmethod + @abstractmethod + def can_process(cls, context: Gen5DataContext) -> bool: + """Check if this processor can handle the given context.""" + pass + + @abstractmethod + def process( + self, context: Gen5DataContext + ) -> tuple[list[MeasurementGroup], list[CalculatedDocument]]: + """Process the context and return measurement groups and calculated data.""" + pass + + +class SpectralResultProcessor(ResultProcessor): + """Processor for spectral measurements with wavelength data.""" + + @classmethod + def can_process(cls, context: Gen5DataContext) -> bool: + return context.is_spectral and context.wavelength_section is not None + + def process( + self, context: Gen5DataContext + ) -> tuple[list[MeasurementGroup], list[CalculatedDocument]]: + return create_spectrum_results( + context.header_data, + read_data_list=context.read_data, + wavelengths_sections=context.wavelength_section, + sample_identifiers=context.sample_identifiers, + actual_temperature=context.actual_temperature, + results_section=context.results_section, + concentration_values=context.concentration_values, + ) + + +class KineticResultProcessor(ResultProcessor): + """Processor for kinetic measurements with time series data.""" + + @classmethod + def can_process(cls, context: Gen5DataContext) -> bool: + if not context.has_kinetic_data: + return False + + if not context.has_kinetic_measurements: + msg = "Kinetic data is present in the file but no kinetic measurements data is found." + raise AllotropeConversionError(msg) + + return True + + def process( + self, context: Gen5DataContext + ) -> tuple[list[MeasurementGroup], list[CalculatedDocument]]: + return create_kinetic_results( + context.results_section, + context.header_data, + context.read_data, + context.sample_identifiers, + context.actual_temperature, + cast(KineticData, context.kinetic_data), + context.kinetic_measurements, + context.kinetic_elapsed_time, + context.kinetic_errors, + context.concentration_values, + ) + + +class StandardResultProcessor(ResultProcessor): + """Processor for standard endpoint measurements.""" + + @classmethod + def can_process(cls, _: Gen5DataContext) -> bool: + return True + + def process( + self, context: Gen5DataContext + ) -> tuple[list[MeasurementGroup], list[CalculatedDocument]]: + return create_results( + context.results_section, + context.header_data, + context.read_data, + context.sample_identifiers, + 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) diff --git a/src/allotropy/parsers/agilent_gen5_image/agilent_gen5_image_parser.py b/src/allotropy/parsers/agilent_gen5_image/agilent_gen5_image_parser.py index 10a4d4f577..3bfb60bee6 100644 --- a/src/allotropy/parsers/agilent_gen5_image/agilent_gen5_image_parser.py +++ b/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")),