diff --git a/src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py b/src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py index 4efdf8e38..3bd9f7de6 100644 --- a/src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py +++ b/src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py @@ -5,6 +5,7 @@ Data, Mapper, ) +from allotropy.constants import CHARDET_ENCODING 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 ( @@ -35,6 +36,12 @@ def sniff(cls, named_file_contents: NamedFileContents) -> bool: return False def create_data(self, named_file_contents: NamedFileContents) -> Data: + if named_file_contents.encoding is None: + named_file_contents = NamedFileContents( + named_file_contents.contents, + named_file_contents.original_file_path, + CHARDET_ENCODING, + ) reader = AgilentGen5Reader(named_file_contents) context = reader.extract_data_context(named_file_contents.original_file_path) diff --git a/src/allotropy/parsers/agilent_gen5/agilent_gen5_reader.py b/src/allotropy/parsers/agilent_gen5/agilent_gen5_reader.py index e607dcfb4..862f63af0 100644 --- a/src/allotropy/parsers/agilent_gen5/agilent_gen5_reader.py +++ b/src/allotropy/parsers/agilent_gen5/agilent_gen5_reader.py @@ -20,7 +20,7 @@ MULTIPLATE_FILE_ERROR, NO_PLATE_DATA_ERROR, ) -from allotropy.parsers.lines_reader import SectionLinesReader +from allotropy.parsers.lines_reader import read_to_lines, SectionLinesReader from allotropy.parsers.utils.pandas import df_to_series_data, read_csv, SeriesData from allotropy.parsers.utils.values import assert_not_none @@ -31,7 +31,8 @@ class AgilentGen5Reader: SUPPORTED_EXTENSIONS = "txt" def __init__(self, named_file_contents: NamedFileContents) -> None: - reader = SectionLinesReader.create(named_file_contents) + lines = [line.rstrip("\t") for line in read_to_lines(named_file_contents)] + reader = SectionLinesReader(lines) plate_readers = list(reader.iter_sections("^Software Version")) if not plate_readers: @@ -75,7 +76,14 @@ def __init__(self, named_file_contents: NamedFileContents) -> None: last_read_label: str | None = None while lines := list(plate_reader.pop_until_empty()): - section_name = lines[0].split("\t")[0].strip(":") + section_name = self._get_section_name(lines[0]) + + # Check if this is a single-line section header with data in the next block + if len(lines) == 1 and section_name in ("Layout", "Results"): + plate_reader.drop_empty() + next_lines = list(plate_reader.pop_until_empty()) + if next_lines: + lines = [lines[0], *next_lines] # Check if this is a "Read X:label" section (single line, starts with "Read ") # If so, store the label and immediately read the next section @@ -87,27 +95,50 @@ def __init__(self, named_file_contents: NamedFileContents) -> None: if not lines: # No data section after the label, just continue continue - section_name = lines[0].split("\t")[0].strip(":") + section_name = self._get_section_name(lines[0]) # If this is a Time section, associate it with a read label if we have one if section_name == "Time": if last_read_label: # Format: "Read X:label" -> Time section - self.kinetic_sections[last_read_label] = lines + self.kinetic_sections[last_read_label] = self._strip_leading_tab( + lines + ) last_read_label = None elif last_section_name: # Format: "{label}" section followed by Time section # Store with just the label as key for kinetic files - self.kinetic_sections[last_section_name] = lines + self.kinetic_sections[last_section_name] = self._strip_leading_tab( + lines + ) else: # Standalone Time section (shouldn't happen, but store it anyway) - self.sections[section_name] = lines + self.sections[section_name] = self._strip_leading_tab(lines) else: - self.sections[section_name] = lines + self.sections[section_name] = self._strip_leading_tab(lines) last_section_name = section_name plate_reader.drop_empty() + @staticmethod + def _get_section_name(line: str) -> str: + fields = line.split("\t") + for field in fields: + name = field.strip('"').strip(":") + if name: + return name + return "" + + @staticmethod + def _strip_leading_tab(lines: list[str]) -> list[str]: + if len(lines) < 2: + return lines + non_header = lines[1:] if not lines[0].startswith("\t") else lines + non_empty = [line for line in non_header if line.strip()] + if non_empty and all(line.startswith("\t") for line in non_empty): + return [line[1:] if line.startswith("\t") else line for line in lines] + return lines + 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) diff --git a/src/allotropy/parsers/agilent_gen5/agilent_gen5_structure.py b/src/allotropy/parsers/agilent_gen5/agilent_gen5_structure.py index 02d4f747e..ac228294c 100644 --- a/src/allotropy/parsers/agilent_gen5/agilent_gen5_structure.py +++ b/src/allotropy/parsers/agilent_gen5/agilent_gen5_structure.py @@ -240,7 +240,7 @@ def create(cls, lines: list[str], read_mode: ReadMode) -> DeviceControlData: device_control_data = DeviceControlData() for line in lines: - strp_line = str(line.strip()) + strp_line = str(line.strip().strip('"').strip()) if strp_line == "Fluorescence Polarization": device_control_data.is_polarization = True continue