From b442e81c4491a0eff7a80eaa9e7b4bd1f166d8b2 Mon Sep 17 00:00:00 2001 From: Nathan Stender Date: Thu, 23 Apr 2026 13:47:56 -0400 Subject: [PATCH 1/2] feat: Add vendor auto-discovery from file contents Add the ability to automatically detect the correct vendor/parser for a file without requiring the caller to specify it. This adds `vendor_from_file()` and `vendor_from_io()` public API functions, makes `vendor_type` optional in all existing API functions, and implements `sniff()` classmethods on all 53 parsers. Co-Authored-By: Claude Opus 4.6 --- src/allotropy/exceptions.py | 4 + src/allotropy/parser_factory.py | 17 +++ .../agilent_gen5/agilent_gen5_parser.py | 12 ++ .../agilent_gen5_image_parser.py | 12 ++ .../agilent_openlab_cds_parser.py | 4 + .../agilent_tapestation_analysis_parser.py | 13 ++ .../appbio_absolute_q_parser.py | 31 +++++ .../appbio_quantstudio_parser.py | 29 +++++ ...io_quantstudio_designandanalysis_parser.py | 14 +++ .../bd_biosciences_facsdiva_parser.py | 16 +++ .../beckman_coulter_biomek_parser.py | 17 +++ .../beckman_echo_cherry_pick_parser.py | 17 +++ .../beckman_echo_plate_reformat_parser.py | 17 +++ .../beckman_pharmspec_parser.py | 19 +++ .../beckman_vi_cell_blu/vi_cell_blu_parser.py | 21 ++++ .../beckman_vi_cell_xr/vi_cell_xr_parser.py | 68 +++++++++++ .../benchling_chromeleon_parser.py | 10 ++ .../benchling_empower_parser.py | 15 +++ .../biorad_bioplex_manager_parser.py | 19 +++ .../bmg_labtech_smart_control_parser.py | 17 +++ .../parsers/bmg_mars/bmg_mars_parser.py | 25 ++++ .../parsers/cfxmaestro/cfxmaestro_parser.py | 15 +++ .../chemometec_nc_view_parser.py | 16 +++ .../nucleoview_parser.py | 18 +++ .../ctl_immunospot/ctl_immunospot_parser.py | 12 ++ .../cytiva_biacore_insight_parser.py | 14 +++ .../cytiva_biacore_t200_control_parser.py | 4 + .../cytiva_biacore_t200_evaluation_parser.py | 4 + .../cytiva_unicorn/cytiva_unicorn_parser.py | 11 ++ .../example_weyland_yutani_parser.py | 4 + src/allotropy/parsers/flowjo/flowjo_parser.py | 4 + .../luminex_intelliflex_parser.py | 27 +++++ .../luminex_xponent/luminex_xponent_parser.py | 27 +++++ .../mabtech_apex/mabtech_apex_parser.py | 17 +++ .../methodical_mind/methodical_mind_parser.py | 14 +++ .../moldev_softmax_pro/softmax_pro_parser.py | 21 ++++ .../msd_workbench/msd_workbench_parser.py | 32 +++++ .../novabio_flex2/novabio_flex2_parser.py | 15 +++ .../perkin_elmer_envision_parser.py | 15 +++ .../qiacuity_dpcr/qiacuity_dpcr_parser.py | 17 +++ .../parsers/revvity_kaleido/kaleido_parser.py | 16 +++ .../revvity_matrix/revvity_matrix_parser.py | 48 ++++++++ .../roche_cedex_bioht_parser.py | 14 +++ .../roche_cedex_hires_parser.py | 29 +++++ .../tecan_magellan/tecan_magellan_parser.py | 28 +++++ .../thermo_fisher_genesys30_parser.py | 23 ++++ .../thermo_fisher_genesys_on_board_parser.py | 27 +++++ .../nanodrop_8000_parser.py | 18 +++ .../nanodrop_eight_parser.py | 20 ++++ .../thermo_fisher_nanodrop_one_parser.py | 46 ++++++++ .../thermo_fisher_qubit4_parser.py | 36 ++++++ .../thermo_fisher_qubit_flex_parser.py | 32 +++++ .../thermo_fisher_visionlite_parser.py | 27 +++++ .../thermo_skanit/thermo_skanit_parser.py | 26 ++++ .../unchained_labs_lunatic_stunner_parser.py | 35 ++++++ src/allotropy/parsers/vendor_parser.py | 4 + src/allotropy/to_allotrope.py | 46 +++++--- tests/discover_vendor_test.py | 111 ++++++++++++++++++ 58 files changed, 1255 insertions(+), 15 deletions(-) create mode 100644 tests/discover_vendor_test.py diff --git a/src/allotropy/exceptions.py b/src/allotropy/exceptions.py index e11689a858..736d003cf5 100644 --- a/src/allotropy/exceptions.py +++ b/src/allotropy/exceptions.py @@ -33,6 +33,10 @@ class AllotropeConversionError(AllotropyError): ... +class AllotropeVendorNotFoundError(AllotropyError): + ... + + def list_values(values: Collection[Any] | type[Enum]) -> list[str]: return sorted([str(v.value if isinstance(v, Enum) else v) for v in values]) diff --git a/src/allotropy/parser_factory.py b/src/allotropy/parser_factory.py index 0cca7e6057..f526d861ad 100644 --- a/src/allotropy/parser_factory.py +++ b/src/allotropy/parser_factory.py @@ -5,6 +5,8 @@ from typing import Any from allotropy.allotrope.path_util import ROOT_DIR +from allotropy.exceptions import AllotropeVendorNotFoundError +from allotropy.named_file_contents import NamedFileContents from allotropy.parsers.agilent_gen5.agilent_gen5_parser import AgilentGen5Parser from allotropy.parsers.agilent_gen5_image.agilent_gen5_image_parser import ( AgilentGen5ImageParser, @@ -296,6 +298,21 @@ def get_parser( } +def discover_vendor(named_file_contents: NamedFileContents) -> Vendor: + extension = named_file_contents.extension + candidates = [v for v in Vendor if extension in v.supported_extensions] + for vendor in candidates: + parser_cls = _VENDOR_TO_PARSER[vendor] + named_file_contents.contents.seek(0) + try: + if parser_cls.sniff(named_file_contents): + return vendor + except Exception: # noqa: S112 + continue + msg = f"No vendor could be identified for file with extension '.{extension}'." + raise AllotropeVendorNotFoundError(msg) + + def get_table_contents() -> str: contents = """The parsers follow maturation levels of: Recommended, Candidate Release, Working Draft. diff --git a/src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py b/src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py index 8feaeb2348..bc24b43d8c 100644 --- a/src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py +++ b/src/allotropy/parsers/agilent_gen5/agilent_gen5_parser.py @@ -21,6 +21,18 @@ class AgilentGen5Parser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = AgilentGen5Reader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read() + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + return any(line.startswith("Software Version") for line in lines[:5]) + except Exception: + return False + 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) 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 3bfb60bee6..dcd734fee4 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 @@ -26,6 +26,18 @@ class AgilentGen5ImageParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = AgilentGen5Reader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read() + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + return any(line.startswith("Software Version") for line in lines[:5]) + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = AgilentGen5Reader(named_file_contents) diff --git a/src/allotropy/parsers/agilent_openlab_cds/agilent_openlab_cds_parser.py b/src/allotropy/parsers/agilent_openlab_cds/agilent_openlab_cds_parser.py index fbc64a6d7d..b27412e59f 100644 --- a/src/allotropy/parsers/agilent_openlab_cds/agilent_openlab_cds_parser.py +++ b/src/allotropy/parsers/agilent_openlab_cds/agilent_openlab_cds_parser.py @@ -25,6 +25,10 @@ class AgilentOpenLabCDSParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "rslt" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, _named_file_contents: NamedFileContents) -> bool: + return True + def create_data(self, named_file_contents: NamedFileContents) -> Data: structured_data = decode_data(named_file_contents.get_bytes_stream()) return Data( diff --git a/src/allotropy/parsers/agilent_tapestation_analysis/agilent_tapestation_analysis_parser.py b/src/allotropy/parsers/agilent_tapestation_analysis/agilent_tapestation_analysis_parser.py index 9f2799c62e..caff66d6f8 100644 --- a/src/allotropy/parsers/agilent_tapestation_analysis/agilent_tapestation_analysis_parser.py +++ b/src/allotropy/parsers/agilent_tapestation_analysis/agilent_tapestation_analysis_parser.py @@ -23,6 +23,19 @@ class AgilentTapestationAnalysisParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "xml" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + tree = ET.parse(named_file_contents.contents) # noqa: S314 + root = tree.getroot() + child_tags = { + child.tag.split("}")[-1] if "}" in child.tag else child.tag + for child in root + } + return "FileInformation" in child_tags and "ScreenTapes" in child_tags + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: try: root_element = ET.parse( # noqa: S314 diff --git a/src/allotropy/parsers/appbio_absolute_q/appbio_absolute_q_parser.py b/src/allotropy/parsers/appbio_absolute_q/appbio_absolute_q_parser.py index cda1e68065..ecc3c840f1 100644 --- a/src/allotropy/parsers/appbio_absolute_q/appbio_absolute_q_parser.py +++ b/src/allotropy/parsers/appbio_absolute_q/appbio_absolute_q_parser.py @@ -1,5 +1,7 @@ from __future__ import annotations +import zipfile + from allotropy.allotrope.models.adm.pcr.benchling._2023._09.dpcr import Model from allotropy.allotrope.schema_mappers.adm.pcr.BENCHLING._2023._09.dpcr import ( Data, @@ -26,6 +28,35 @@ class AppbioAbsoluteQParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = AppbioAbsoluteQReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + if named_file_contents.extension == "zip": + named_file_contents.contents.seek(0) + with zipfile.ZipFile(named_file_contents.get_bytes_stream()) as zf: + return any(name.endswith("_summary.csv") for name in zf.namelist()) + named_file_contents.contents.seek(0) + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if len(lines) < 2: + return False + all_cols: set[str] = set() + for line in lines[:2]: + all_cols.update(c.strip() for c in line.split(",")) + has_instrument_plate = "Instrument" in all_cols and "Plate" in all_cols + if has_instrument_plate: + return bool(all_cols & {"Dye", "Target", "Channels"}) or any( + "_target" in c for c in all_cols + ) + return ( + "Run name" in all_cols and "Target" in all_cols and "Total" in all_cols + ) + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = AppbioAbsoluteQReader(named_file_contents) wells = Well.create_wells(reader.data) diff --git a/src/allotropy/parsers/appbio_quantstudio/appbio_quantstudio_parser.py b/src/allotropy/parsers/appbio_quantstudio/appbio_quantstudio_parser.py index 12d0b6518d..186103eaae 100644 --- a/src/allotropy/parsers/appbio_quantstudio/appbio_quantstudio_parser.py +++ b/src/allotropy/parsers/appbio_quantstudio/appbio_quantstudio_parser.py @@ -1,3 +1,5 @@ +import openpyxl + from allotropy.allotrope.models.adm.pcr.rec._2024._09.qpcr import Model from allotropy.allotrope.schema_mappers.adm.pcr.rec._2024._09.qpcr import Data, Mapper from allotropy.named_file_contents import NamedFileContents @@ -31,6 +33,33 @@ class AppBioQuantStudioParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = AppBioQuantStudioReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + if named_file_contents.extension == "txt": + named_file_contents.contents.seek(0) + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") + if isinstance(raw, bytes) + else raw + ) + lines = text.splitlines() + has_star_keys = any( + line.startswith("* ") and " = " in line for line in lines[:10] + ) + has_brackets = any(line.startswith("[") for line in lines) + return has_star_keys or has_brackets + # xlsx: QuantStudio XLSX files have sheet names wrapped in brackets + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + sheet_names = wb.sheetnames + wb.close() + return any(name.startswith("[") for name in sheet_names) + except Exception: + return False + def parse_data( self, reader: AppBioQuantStudioReader, original_file_path: str ) -> Data: diff --git a/src/allotropy/parsers/appbio_quantstudio_designandanalysis/appbio_quantstudio_designandanalysis_parser.py b/src/allotropy/parsers/appbio_quantstudio_designandanalysis/appbio_quantstudio_designandanalysis_parser.py index fc117f40fe..841f7ccbf8 100644 --- a/src/allotropy/parsers/appbio_quantstudio_designandanalysis/appbio_quantstudio_designandanalysis_parser.py +++ b/src/allotropy/parsers/appbio_quantstudio_designandanalysis/appbio_quantstudio_designandanalysis_parser.py @@ -1,3 +1,5 @@ +import openpyxl + from allotropy.allotrope.models.adm.pcr.rec._2024._09.qpcr import Model from allotropy.allotrope.schema_mappers.adm.pcr.rec._2024._09.qpcr import ( Data, @@ -30,6 +32,18 @@ class AppBioQuantStudioDesignandanalysisParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = DesignQuantstudioReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + sheet_names = set(wb.sheetnames) + wb.close() + return "Results" in sheet_names + except Exception: + return False + def parse_rt_pcr( self, reader: DesignQuantstudioReader, original_file_path: str ) -> Data: diff --git a/src/allotropy/parsers/bd_biosciences_facsdiva/bd_biosciences_facsdiva_parser.py b/src/allotropy/parsers/bd_biosciences_facsdiva/bd_biosciences_facsdiva_parser.py index 1a470a3433..9c76c276c0 100644 --- a/src/allotropy/parsers/bd_biosciences_facsdiva/bd_biosciences_facsdiva_parser.py +++ b/src/allotropy/parsers/bd_biosciences_facsdiva/bd_biosciences_facsdiva_parser.py @@ -25,6 +25,22 @@ class BDFACSDivaParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "xml" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + tree = ET.parse(named_file_contents.contents) # noqa: S314 + root = tree.getroot() + root_tag = root.tag.split("}")[-1] if "}" in root.tag else root.tag + if root_tag == "bdfacs": + return True + child_tags = { + child.tag.split("}")[-1] if "}" in child.tag else child.tag + for child in root + } + return "experiment" in child_tags or "specimen" in child_tags + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: try: root_element_et = ET.parse( # noqa: S314 diff --git a/src/allotropy/parsers/beckman_coulter_biomek/beckman_coulter_biomek_parser.py b/src/allotropy/parsers/beckman_coulter_biomek/beckman_coulter_biomek_parser.py index 2ce0f65ff2..f123f99752 100644 --- a/src/allotropy/parsers/beckman_coulter_biomek/beckman_coulter_biomek_parser.py +++ b/src/allotropy/parsers/beckman_coulter_biomek/beckman_coulter_biomek_parser.py @@ -24,6 +24,23 @@ class BeckmanCoulterBiomekParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = BeckmanCoulterBiomekReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if not lines: + return False + first_line = lines[0] + if "Well Index" in first_line: + return True + return first_line.startswith("Method = ") + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = BeckmanCoulterBiomekReader(named_file_contents) return Data( diff --git a/src/allotropy/parsers/beckman_echo_cherry_pick/beckman_echo_cherry_pick_parser.py b/src/allotropy/parsers/beckman_echo_cherry_pick/beckman_echo_cherry_pick_parser.py index 8e33c221b9..31bade6250 100644 --- a/src/allotropy/parsers/beckman_echo_cherry_pick/beckman_echo_cherry_pick_parser.py +++ b/src/allotropy/parsers/beckman_echo_cherry_pick/beckman_echo_cherry_pick_parser.py @@ -1,3 +1,5 @@ +import re + import pandas as pd from allotropy.allotrope.models.adm.liquid_handler.benchling._2024._11.liquid_handler import ( @@ -26,6 +28,21 @@ class BeckmanEchoCherryPickParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = BeckmanEchoCherryPickReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + for line in lines[:20]: + if re.match(r"^\[.+\]", line): + return True + return False + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = BeckmanEchoCherryPickReader(named_file_contents) measurement_groups = create_measurement_groups( diff --git a/src/allotropy/parsers/beckman_echo_plate_reformat/beckman_echo_plate_reformat_parser.py b/src/allotropy/parsers/beckman_echo_plate_reformat/beckman_echo_plate_reformat_parser.py index daf6a3b13f..23bee7f21d 100644 --- a/src/allotropy/parsers/beckman_echo_plate_reformat/beckman_echo_plate_reformat_parser.py +++ b/src/allotropy/parsers/beckman_echo_plate_reformat/beckman_echo_plate_reformat_parser.py @@ -1,3 +1,5 @@ +import re + import pandas as pd from allotropy.allotrope.models.adm.liquid_handler.benchling._2024._11.liquid_handler import ( @@ -26,6 +28,21 @@ class BeckmanEchoPlateReformatParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = BeckmanEchoPlateReformatReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + for line in lines[:20]: + if re.match(r"^\[.+\]", line): + return True + return False + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = BeckmanEchoPlateReformatReader(named_file_contents) measurement_groups = create_measurement_groups( diff --git a/src/allotropy/parsers/beckman_pharmspec/beckman_pharmspec_parser.py b/src/allotropy/parsers/beckman_pharmspec/beckman_pharmspec_parser.py index acfb3a64da..6e0cb11159 100644 --- a/src/allotropy/parsers/beckman_pharmspec/beckman_pharmspec_parser.py +++ b/src/allotropy/parsers/beckman_pharmspec/beckman_pharmspec_parser.py @@ -1,3 +1,5 @@ +import openpyxl + from allotropy.allotrope.models.adm.solution_analyzer.rec._2024._09.solution_analyzer import ( Model, ) @@ -26,6 +28,23 @@ class PharmSpecParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = BeckmanPharmspecReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + ws = wb[wb.sheetnames[0]] + for row in ws.iter_rows(max_row=20, values_only=True): + for cell in row: + if isinstance(cell, str) and "Particle" in cell: + wb.close() + return True + wb.close() + return False + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = BeckmanPharmspecReader(named_file_contents) distributions = Distribution.create_distributions(reader.data) diff --git a/src/allotropy/parsers/beckman_vi_cell_blu/vi_cell_blu_parser.py b/src/allotropy/parsers/beckman_vi_cell_blu/vi_cell_blu_parser.py index 549321e7e4..707d1d71f3 100644 --- a/src/allotropy/parsers/beckman_vi_cell_blu/vi_cell_blu_parser.py +++ b/src/allotropy/parsers/beckman_vi_cell_blu/vi_cell_blu_parser.py @@ -24,6 +24,27 @@ class ViCellBluParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = ViCellBluReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read(8192) + if isinstance(raw, bytes): + if raw[:2] == b"\xff\xfe": + text = raw[2:].decode("utf-16-le", errors="replace") + elif raw[:2] == b"\xfe\xff": + text = raw[2:].decode("utf-16-be", errors="replace") + else: + text = raw.decode("utf-8", errors="replace") + else: + text = raw + lines = text.splitlines() + if not lines: + return False + header = lines[0] + return "Viability (%)" in header and "Viable (x10^6) cells/mL" in header + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: return Data( create_metadata(named_file_contents.original_file_path), diff --git a/src/allotropy/parsers/beckman_vi_cell_xr/vi_cell_xr_parser.py b/src/allotropy/parsers/beckman_vi_cell_xr/vi_cell_xr_parser.py index 05f11a8694..6fce6a221b 100644 --- a/src/allotropy/parsers/beckman_vi_cell_xr/vi_cell_xr_parser.py +++ b/src/allotropy/parsers/beckman_vi_cell_xr/vi_cell_xr_parser.py @@ -1,5 +1,10 @@ from __future__ import annotations +from xml.etree import ElementTree +import zipfile + +import openpyxl + from allotropy.allotrope.models.adm.cell_counting.rec._2024._09.cell_counting import ( Model, ) @@ -17,6 +22,8 @@ from allotropy.parsers.release_state import ReleaseState from allotropy.parsers.vendor_parser import VendorParser +_VI_CELL_MARKERS = ("Vi-CELL", "Vi-Cell") + class ViCellXRParser(VendorParser[Data, Model]): DISPLAY_NAME = "Beckman Coulter Vi-Cell XR" @@ -24,6 +31,67 @@ class ViCellXRParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "txt,xls,xlsx" SCHEMA_MAPPER = Mapper + @classmethod + def _check_xlsx_via_zip(cls, named_file_contents: NamedFileContents) -> bool: + """Fallback for xlsx files that openpyxl cannot open (e.g. corrupt styles).""" + stream = named_file_contents.get_bytes_stream() + with zipfile.ZipFile(stream) as zf: + if "xl/sharedStrings.xml" not in zf.namelist(): + return False + with zf.open("xl/sharedStrings.xml") as ss: + tree = ElementTree.parse(ss) # noqa: S314 + for elem in tree.iter(): + if elem.text and any( + marker in elem.text for marker in _VI_CELL_MARKERS + ): + return True + return False + + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + if named_file_contents.extension == "txt": + named_file_contents.contents.seek(0) + for raw_line in named_file_contents.contents: + text = ( + raw_line.decode("utf-8") + if isinstance(raw_line, bytes) + else raw_line + ) + text = text.strip() + if text: + named_file_contents.contents.seek(0) + return any(marker in text for marker in _VI_CELL_MARKERS) + named_file_contents.contents.seek(0) + return False + if named_file_contents.extension == "xlsx": + try: + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + ws = wb[wb.sheetnames[0]] + first_row = next(ws.iter_rows(max_row=1, values_only=True), None) + wb.close() + if first_row is None: + return False + first_cell = str(first_row[0]) if first_row[0] is not None else "" + return any(marker in first_cell for marker in _VI_CELL_MARKERS) + except Exception: + # Fallback: read xlsx as zip for files with corrupt styles + named_file_contents.contents.seek(0) + return cls._check_xlsx_via_zip(named_file_contents) + # xls: cannot use openpyxl, check using xlrd-style heuristic + # Read first few bytes to check it's an Excel file, then trust extension + named_file_contents.contents.seek(0) + header = named_file_contents.contents.read(8) + named_file_contents.contents.seek(0) + if isinstance(header, str): + header = header.encode("utf-8") + # OLE2 compound document magic number (xls files) + return header[:8] == b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader_data = create_reader_data(named_file_contents) if not reader_data.data: diff --git a/src/allotropy/parsers/benchling_chromeleon/benchling_chromeleon_parser.py b/src/allotropy/parsers/benchling_chromeleon/benchling_chromeleon_parser.py index c747aa5cd3..a99ec7491e 100644 --- a/src/allotropy/parsers/benchling_chromeleon/benchling_chromeleon_parser.py +++ b/src/allotropy/parsers/benchling_chromeleon/benchling_chromeleon_parser.py @@ -1,3 +1,5 @@ +import json + from allotropy.allotrope.models.adm.liquid_chromatography.benchling._2023._09.liquid_chromatography import ( Model, ) @@ -24,6 +26,14 @@ class BenchlingChromeleonParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = BenchlingChromeleonReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + data = json.load(named_file_contents.contents) + return isinstance(data, dict) and isinstance(data.get("injections"), list) + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = BenchlingChromeleonReader(named_file_contents) return Data( diff --git a/src/allotropy/parsers/benchling_empower/benchling_empower_parser.py b/src/allotropy/parsers/benchling_empower/benchling_empower_parser.py index 371d56785f..06a42d5e16 100644 --- a/src/allotropy/parsers/benchling_empower/benchling_empower_parser.py +++ b/src/allotropy/parsers/benchling_empower/benchling_empower_parser.py @@ -1,3 +1,5 @@ +import json + from allotropy.allotrope.models.adm.liquid_chromatography.benchling._2023._09.liquid_chromatography import ( Model, ) @@ -24,6 +26,19 @@ class BenchlingEmpowerParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = BenchlingEmpowerReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + data = json.load(named_file_contents.contents) + return ( + isinstance(data, dict) + and "values" in data + and isinstance(data["values"], dict) + and "fields" in data["values"] + ) + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = BenchlingEmpowerReader(named_file_contents) diff --git a/src/allotropy/parsers/biorad_bioplex_manager/biorad_bioplex_manager_parser.py b/src/allotropy/parsers/biorad_bioplex_manager/biorad_bioplex_manager_parser.py index 0a6df745e6..3e8ff6b508 100644 --- a/src/allotropy/parsers/biorad_bioplex_manager/biorad_bioplex_manager_parser.py +++ b/src/allotropy/parsers/biorad_bioplex_manager/biorad_bioplex_manager_parser.py @@ -1,5 +1,7 @@ from __future__ import annotations +import xml.etree.ElementTree as ET # noqa: N817 + from allotropy.allotrope.models.adm.multi_analyte_profiling.benchling._2024._09.multi_analyte_profiling import ( Model, ) @@ -28,6 +30,23 @@ class BioradBioplexParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "xml" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + tree = ET.parse(named_file_contents.contents) # noqa: S314 + root = tree.getroot() + child_tags = { + child.tag.split("}")[-1] if "}" in child.tag else child.tag + for child in root + } + return ( + "Samples" in child_tags + and "Wells" in child_tags + and "PlateDimensions" in child_tags + ) + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = BioradBioplexReader(named_file_contents) diff --git a/src/allotropy/parsers/bmg_labtech_smart_control/bmg_labtech_smart_control_parser.py b/src/allotropy/parsers/bmg_labtech_smart_control/bmg_labtech_smart_control_parser.py index 459d3c5058..8f3e5d69ac 100644 --- a/src/allotropy/parsers/bmg_labtech_smart_control/bmg_labtech_smart_control_parser.py +++ b/src/allotropy/parsers/bmg_labtech_smart_control/bmg_labtech_smart_control_parser.py @@ -1,5 +1,7 @@ from functools import partial +import openpyxl + from allotropy.allotrope.models.adm.plate_reader.rec._2024._06.plate_reader import ( Model, ) @@ -28,6 +30,21 @@ class BmgLabtechSmartControlParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = BmgLabtechSmartControlReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + sheet_names = {name.lower() for name in wb.sheetnames} + wb.close() + return ( + "protocol information" in sheet_names + and "microplate end point" in sheet_names + ) + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = BmgLabtechSmartControlReader(named_file_contents) measurement_groups = map_rows( diff --git a/src/allotropy/parsers/bmg_mars/bmg_mars_parser.py b/src/allotropy/parsers/bmg_mars/bmg_mars_parser.py index b36817ef99..db22db6984 100644 --- a/src/allotropy/parsers/bmg_mars/bmg_mars_parser.py +++ b/src/allotropy/parsers/bmg_mars/bmg_mars_parser.py @@ -1,5 +1,7 @@ from __future__ import annotations +import re + from allotropy.allotrope.models.adm.plate_reader.rec._2024._06.plate_reader import ( Model, ) @@ -24,6 +26,29 @@ class BmgMarsParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "csv" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read() + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + has_key_value = False + has_raw_data = False + for line in lines[:30]: + if re.match(r"^.+:\s+.+", line): + has_key_value = True + break + for line in lines: + stripped = line.strip() + if stripped == "Raw Data" or stripped.startswith("Raw Data"): + has_raw_data = True + break + return has_key_value and has_raw_data + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = BmgMarsReader(named_file_contents) header = Header.create(reader.header, reader.header_content) diff --git a/src/allotropy/parsers/cfxmaestro/cfxmaestro_parser.py b/src/allotropy/parsers/cfxmaestro/cfxmaestro_parser.py index f902044524..c9c30266f9 100644 --- a/src/allotropy/parsers/cfxmaestro/cfxmaestro_parser.py +++ b/src/allotropy/parsers/cfxmaestro/cfxmaestro_parser.py @@ -19,6 +19,21 @@ class CfxmaestroParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = CFXMaestroReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if not lines: + return False + header = lines[0] + return "Well" in header and "Fluor" in header + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = CFXMaestroReader(named_file_contents) return Data( diff --git a/src/allotropy/parsers/chemometec_nc_view/chemometec_nc_view_parser.py b/src/allotropy/parsers/chemometec_nc_view/chemometec_nc_view_parser.py index 6f79b992cc..75feb14fba 100644 --- a/src/allotropy/parsers/chemometec_nc_view/chemometec_nc_view_parser.py +++ b/src/allotropy/parsers/chemometec_nc_view/chemometec_nc_view_parser.py @@ -25,6 +25,22 @@ class ChemometecNcViewParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = ChemometecNcViewReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + named_file_contents.contents.seek(0) + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if not lines: + return False + header = lines[0] + return "INSTRUMENT" in header and "VIABILITY" in header + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = ChemometecNcViewReader(named_file_contents) return Data( diff --git a/src/allotropy/parsers/chemometec_nucleoview/nucleoview_parser.py b/src/allotropy/parsers/chemometec_nucleoview/nucleoview_parser.py index d58ddc806b..dc6b527644 100644 --- a/src/allotropy/parsers/chemometec_nucleoview/nucleoview_parser.py +++ b/src/allotropy/parsers/chemometec_nucleoview/nucleoview_parser.py @@ -22,6 +22,24 @@ class ChemometecNucleoviewParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = NucleoviewReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + named_file_contents.contents.seek(0) + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + if "Viability (%)" not in text: + return False + lines = text.splitlines() + for line in lines[:5]: + if ":\t" in line or ";" in line: + return True + return False + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: df = NucleoviewReader.read(named_file_contents.contents) data_groups = map_rows(df, create_measurement_groups) diff --git a/src/allotropy/parsers/ctl_immunospot/ctl_immunospot_parser.py b/src/allotropy/parsers/ctl_immunospot/ctl_immunospot_parser.py index b28ed624d2..ca9eba9f73 100644 --- a/src/allotropy/parsers/ctl_immunospot/ctl_immunospot_parser.py +++ b/src/allotropy/parsers/ctl_immunospot/ctl_immunospot_parser.py @@ -21,6 +21,18 @@ class CtlImmunospotParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = CtlImmunospotReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read() + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + return any("ImmunoSpot" in line for line in lines[:20]) + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = CtlImmunospotReader(named_file_contents) return Data( diff --git a/src/allotropy/parsers/cytiva_biacore_insight/cytiva_biacore_insight_parser.py b/src/allotropy/parsers/cytiva_biacore_insight/cytiva_biacore_insight_parser.py index c5c449b9d5..b0c3021a29 100644 --- a/src/allotropy/parsers/cytiva_biacore_insight/cytiva_biacore_insight_parser.py +++ b/src/allotropy/parsers/cytiva_biacore_insight/cytiva_biacore_insight_parser.py @@ -1,3 +1,5 @@ +import openpyxl + from allotropy.allotrope.models.adm.binding_affinity_analyzer.wd._2024._12.binding_affinity_analyzer import ( Model, ) @@ -28,6 +30,18 @@ class CytivaBiacoreInsightParser(VendorParser[MapperData, Model]): SUPPORTED_EXTENSIONS = CytivaBiacoreInsightReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + sheet_names = set(wb.sheetnames) + wb.close() + return "Properties" in sheet_names and "Report point table" in sheet_names + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> MapperData: reader = CytivaBiacoreInsightReader.create(named_file_contents) data = Data.create(reader) diff --git a/src/allotropy/parsers/cytiva_biacore_t200_control/cytiva_biacore_t200_control_parser.py b/src/allotropy/parsers/cytiva_biacore_t200_control/cytiva_biacore_t200_control_parser.py index b05dc8ec0d..337f0f03b6 100644 --- a/src/allotropy/parsers/cytiva_biacore_t200_control/cytiva_biacore_t200_control_parser.py +++ b/src/allotropy/parsers/cytiva_biacore_t200_control/cytiva_biacore_t200_control_parser.py @@ -29,6 +29,10 @@ class CytivaBiacoreT200ControlParser(VendorParser[MapperData, Model]): SUPPORTED_EXTENSIONS = "blr" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, _named_file_contents: NamedFileContents) -> bool: + return True + def create_data(self, named_file_contents: NamedFileContents) -> MapperData: base_data = DictData(decode_data(named_file_contents)) data = Data.create(base_data) diff --git a/src/allotropy/parsers/cytiva_biacore_t200_evaluation/cytiva_biacore_t200_evaluation_parser.py b/src/allotropy/parsers/cytiva_biacore_t200_evaluation/cytiva_biacore_t200_evaluation_parser.py index b14356f32f..f5c0e335f3 100644 --- a/src/allotropy/parsers/cytiva_biacore_t200_evaluation/cytiva_biacore_t200_evaluation_parser.py +++ b/src/allotropy/parsers/cytiva_biacore_t200_evaluation/cytiva_biacore_t200_evaluation_parser.py @@ -20,6 +20,10 @@ class CytivaBiacoreT200EvaluationParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "bme" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, _named_file_contents: NamedFileContents) -> bool: + return True + def create_data(self, named_file_contents: NamedFileContents) -> Data: metadata, groups = _create_data(named_file_contents) return Data(metadata=metadata, measurement_groups=groups) diff --git a/src/allotropy/parsers/cytiva_unicorn/cytiva_unicorn_parser.py b/src/allotropy/parsers/cytiva_unicorn/cytiva_unicorn_parser.py index f133160330..8ac8dcbcd9 100644 --- a/src/allotropy/parsers/cytiva_unicorn/cytiva_unicorn_parser.py +++ b/src/allotropy/parsers/cytiva_unicorn/cytiva_unicorn_parser.py @@ -1,3 +1,5 @@ +import zipfile + from allotropy.allotrope.models.adm.liquid_chromatography.benchling._2023._09.liquid_chromatography import ( Model, ) @@ -26,6 +28,15 @@ class CytivaUnicornParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "zip" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + named_file_contents.contents.seek(0) + with zipfile.ZipFile(named_file_contents.get_bytes_stream()) as zf: + return any(name.endswith(".zip") for name in zf.namelist()) + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: handler = UnicornZipHandler(named_file_contents.get_bytes_stream()) results = handler.get_results() diff --git a/src/allotropy/parsers/example_weyland_yutani/example_weyland_yutani_parser.py b/src/allotropy/parsers/example_weyland_yutani/example_weyland_yutani_parser.py index 7381dd222e..68f19aa5f0 100644 --- a/src/allotropy/parsers/example_weyland_yutani/example_weyland_yutani_parser.py +++ b/src/allotropy/parsers/example_weyland_yutani/example_weyland_yutani_parser.py @@ -27,6 +27,10 @@ class ExampleWeylandYutaniParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = ExampleWeylandYutaniReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: # noqa: ARG003 + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = ExampleWeylandYutaniReader(named_file_contents) basic_assay_info = BasicAssayInfo.create(reader.bottom) diff --git a/src/allotropy/parsers/flowjo/flowjo_parser.py b/src/allotropy/parsers/flowjo/flowjo_parser.py index 30543dfd12..d5aec89218 100644 --- a/src/allotropy/parsers/flowjo/flowjo_parser.py +++ b/src/allotropy/parsers/flowjo/flowjo_parser.py @@ -61,6 +61,10 @@ class FlowjoParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "wsp" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, _named_file_contents: NamedFileContents) -> bool: + return True + def create_data(self, named_file_contents: NamedFileContents) -> Data: try: root_element_et = ET.parse( # noqa: S314 diff --git a/src/allotropy/parsers/luminex_intelliflex/luminex_intelliflex_parser.py b/src/allotropy/parsers/luminex_intelliflex/luminex_intelliflex_parser.py index 5850ec56fc..2c5b18d22d 100644 --- a/src/allotropy/parsers/luminex_intelliflex/luminex_intelliflex_parser.py +++ b/src/allotropy/parsers/luminex_intelliflex/luminex_intelliflex_parser.py @@ -27,6 +27,33 @@ class LuminexIntelliflexParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = LuminexXponentReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if not lines: + return False + first_line = lines[0] + if ( + "INSTRUMENT TYPE" in first_line + and "WELL LOCATION" in first_line + and "SAMPLE ID" in first_line + ): + return True + for line in lines: + stripped = line.strip() + if stripped: + return stripped.startswith("Program,") or stripped.startswith( + '"Program",' + ) + return False + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = LuminexXponentReader(named_file_contents) data = XponentData.create(reader) diff --git a/src/allotropy/parsers/luminex_xponent/luminex_xponent_parser.py b/src/allotropy/parsers/luminex_xponent/luminex_xponent_parser.py index f2e38f04f0..bdfaf44f46 100644 --- a/src/allotropy/parsers/luminex_xponent/luminex_xponent_parser.py +++ b/src/allotropy/parsers/luminex_xponent/luminex_xponent_parser.py @@ -24,6 +24,33 @@ class LuminexXponentParser(VendorParser[MapperData, Model]): SUPPORTED_EXTENSIONS = LuminexXponentReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if not lines: + return False + first_line = lines[0] + if ( + "INSTRUMENT TYPE" in first_line + and "WELL LOCATION" in first_line + and "SAMPLE ID" in first_line + ): + return True + for line in lines: + stripped = line.strip() + if stripped: + return stripped.startswith("Program,") or stripped.startswith( + '"Program",' + ) + return False + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> MapperData: reader = LuminexXponentReader(named_file_contents) data = Data.create(reader) diff --git a/src/allotropy/parsers/mabtech_apex/mabtech_apex_parser.py b/src/allotropy/parsers/mabtech_apex/mabtech_apex_parser.py index f6f3c5ab96..19d5e441dd 100644 --- a/src/allotropy/parsers/mabtech_apex/mabtech_apex_parser.py +++ b/src/allotropy/parsers/mabtech_apex/mabtech_apex_parser.py @@ -1,5 +1,7 @@ from collections import defaultdict +import openpyxl + from allotropy.allotrope.models.adm.plate_reader.benchling._2023._09.plate_reader import ( Model, ) @@ -26,6 +28,21 @@ class MabtechApexParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = MabtechApexReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + sheet_names = set(wb.sheetnames) + wb.close() + has_plate_info = ( + "Plate Information" in sheet_names or "Plate Info" in sheet_names + ) + return has_plate_info and "Plate Database" in sheet_names + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = MabtechApexReader.create(named_file_contents) diff --git a/src/allotropy/parsers/methodical_mind/methodical_mind_parser.py b/src/allotropy/parsers/methodical_mind/methodical_mind_parser.py index 38cbce4427..58ac8ca1e3 100644 --- a/src/allotropy/parsers/methodical_mind/methodical_mind_parser.py +++ b/src/allotropy/parsers/methodical_mind/methodical_mind_parser.py @@ -1,5 +1,7 @@ from __future__ import annotations +import re + from allotropy.allotrope.models.adm.plate_reader.rec._2024._06.plate_reader import ( Model, ) @@ -27,6 +29,18 @@ class MethodicalMindParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "txt" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read() + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + return any(re.match(r"^=+Data=+$", line) for line in lines[:50]) + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = MethodicalMindReader(named_file_contents) return Data( diff --git a/src/allotropy/parsers/moldev_softmax_pro/softmax_pro_parser.py b/src/allotropy/parsers/moldev_softmax_pro/softmax_pro_parser.py index 0c37fb1b45..3cd946d6e7 100644 --- a/src/allotropy/parsers/moldev_softmax_pro/softmax_pro_parser.py +++ b/src/allotropy/parsers/moldev_softmax_pro/softmax_pro_parser.py @@ -1,3 +1,5 @@ +import re + from allotropy.allotrope.models.adm.plate_reader.rec._2025._03.plate_reader import ( Model, ) @@ -25,6 +27,25 @@ class SoftmaxproParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "txt" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read(8192) + if isinstance(raw, bytes): + if raw[:2] == b"\xff\xfe": + text = raw[2:].decode("utf-16-le", errors="replace") + else: + text = raw.decode("utf-8", errors="replace") + else: + text = raw + lines = text.splitlines() + for line in lines: + if line.strip(): + return bool(re.match(r"^##BLOCKS=\s*\d+", line)) + return False + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: lines = read_to_lines(named_file_contents) reader = CsvReader(lines) diff --git a/src/allotropy/parsers/msd_workbench/msd_workbench_parser.py b/src/allotropy/parsers/msd_workbench/msd_workbench_parser.py index 84c2e50c08..f6579df8b2 100644 --- a/src/allotropy/parsers/msd_workbench/msd_workbench_parser.py +++ b/src/allotropy/parsers/msd_workbench/msd_workbench_parser.py @@ -1,5 +1,9 @@ from __future__ import annotations +import re + +import openpyxl + from allotropy.allotrope.models.adm.plate_reader.rec._2024._06.plate_reader import ( Model, ) @@ -38,6 +42,34 @@ class MSDWorkbenchParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = MSDWorkbenchReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + if named_file_contents.extension == "xlsx": + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + sheet_names = set(wb.sheetnames) + wb.close() + return "Workbench data" in sheet_names + named_file_contents.contents.seek(0) + raw = named_file_contents.contents.read(4096) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if not lines: + return False + first_line = lines[0].strip() + if named_file_contents.extension == "csv": + return first_line.startswith("Plate_") + return bool( + re.match(r"^FileName\s*:\t", first_line) + or re.match(r"^Plate\s*#", first_line) + ) + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: if named_file_contents.extension == "txt": return self._process_methodical_mind(named_file_contents) diff --git a/src/allotropy/parsers/novabio_flex2/novabio_flex2_parser.py b/src/allotropy/parsers/novabio_flex2/novabio_flex2_parser.py index e5b0a28c46..3d5b588e81 100644 --- a/src/allotropy/parsers/novabio_flex2/novabio_flex2_parser.py +++ b/src/allotropy/parsers/novabio_flex2/novabio_flex2_parser.py @@ -26,6 +26,21 @@ class NovaBioFlexParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "csv" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if not lines: + return False + header = lines[0] + return "Date & Time" in header + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: try: data = read_csv( diff --git a/src/allotropy/parsers/perkin_elmer_envision/perkin_elmer_envision_parser.py b/src/allotropy/parsers/perkin_elmer_envision/perkin_elmer_envision_parser.py index 293ff81fac..333524073b 100644 --- a/src/allotropy/parsers/perkin_elmer_envision/perkin_elmer_envision_parser.py +++ b/src/allotropy/parsers/perkin_elmer_envision/perkin_elmer_envision_parser.py @@ -25,6 +25,21 @@ class PerkinElmerEnvisionParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "csv" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + for line in lines[:20]: + if line.startswith("Plate information"): + return True + return False + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: lines = read_to_lines(named_file_contents) reader = CsvReader(lines) diff --git a/src/allotropy/parsers/qiacuity_dpcr/qiacuity_dpcr_parser.py b/src/allotropy/parsers/qiacuity_dpcr/qiacuity_dpcr_parser.py index f4bd4a7b5c..ba67ed65cd 100644 --- a/src/allotropy/parsers/qiacuity_dpcr/qiacuity_dpcr_parser.py +++ b/src/allotropy/parsers/qiacuity_dpcr/qiacuity_dpcr_parser.py @@ -27,6 +27,23 @@ class QiacuitydPCRParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = QiacuitydPCRReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if len(lines) < 2: + return False + if not lines[0].strip().lstrip("\ufeff").startswith("sep="): + return False + header = lines[1] if len(lines) > 1 else "" + return "Partitions" in header or "Sample/NTC" in header + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = QiacuitydPCRReader(named_file_contents) # Assign stable measurement identifiers per row for data source linkage diff --git a/src/allotropy/parsers/revvity_kaleido/kaleido_parser.py b/src/allotropy/parsers/revvity_kaleido/kaleido_parser.py index 2b2782bbda..f70e86e8c8 100644 --- a/src/allotropy/parsers/revvity_kaleido/kaleido_parser.py +++ b/src/allotropy/parsers/revvity_kaleido/kaleido_parser.py @@ -22,6 +22,22 @@ class KaleidoParser(VendorParser[Data, Model]): SCHEMA_MAPPER = Mapper SUPPORTED_EXTENSIONS = "csv" + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read() + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + for line in reversed(lines): + stripped = line.strip() + if stripped: + return "Exported with Kaleido" in stripped + return False + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: data = create_data(CsvReader(read_to_lines(named_file_contents))) return Data( diff --git a/src/allotropy/parsers/revvity_matrix/revvity_matrix_parser.py b/src/allotropy/parsers/revvity_matrix/revvity_matrix_parser.py index dc2e0b81d8..886c629077 100644 --- a/src/allotropy/parsers/revvity_matrix/revvity_matrix_parser.py +++ b/src/allotropy/parsers/revvity_matrix/revvity_matrix_parser.py @@ -1,5 +1,7 @@ from functools import partial +import openpyxl + from allotropy.allotrope.models.adm.cell_counting.rec._2024._09.cell_counting import ( Model, ) @@ -20,6 +22,9 @@ from allotropy.parsers.utils.pandas import map_rows from allotropy.parsers.vendor_parser import VendorParser +_REVVITY_CSV_COLUMNS = {"Row", "Column"} +_REVVITY_XLSX_COLUMNS = {"Well Name", "Live Count", "Viability"} + class RevvityMatrixParser(VendorParser[Data, Model]): DISPLAY_NAME = DISPLAY_NAME @@ -27,6 +32,49 @@ class RevvityMatrixParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = RevvityMatrixReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + if named_file_contents.extension == "xlsx": + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + ws = wb[wb.sheetnames[0]] + # Check for header row with Well Name, or metadata rows like "Plate Name" + for row in ws.iter_rows(max_row=15, values_only=True): + cells = [str(c) for c in row if c is not None] + if any("Well Name" in c for c in cells) and any( + "Live Count" in c or "Live Concentration" in c for c in cells + ): + wb.close() + return True + # CSV-style: first row has "Row" and "Column" + if "Row" in cells and "Column" in cells: + wb.close() + return True + wb.close() + return False + named_file_contents.contents.seek(0) + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if not lines: + return False + first = lines[0] + # Metadata-style CSV: starts with "Plate Name,..." + if first.startswith("Plate Name,"): + return True + # Flat CSV: header row has "Well Name" and cell counting columns + if "Well Name" in first and ( + "Live Count" in first or "Live Cells/mL" in first + ): + return True + return False + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = RevvityMatrixReader(named_file_contents) return Data( diff --git a/src/allotropy/parsers/roche_cedex_bioht/roche_cedex_bioht_parser.py b/src/allotropy/parsers/roche_cedex_bioht/roche_cedex_bioht_parser.py index c78226697b..38eff979bc 100644 --- a/src/allotropy/parsers/roche_cedex_bioht/roche_cedex_bioht_parser.py +++ b/src/allotropy/parsers/roche_cedex_bioht/roche_cedex_bioht_parser.py @@ -25,6 +25,20 @@ class RocheCedexBiohtParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = RocheCedexBiohtReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if not lines: + return False + return "#ARC-FILE#" in lines[0] + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = RocheCedexBiohtReader(named_file_contents.contents) diff --git a/src/allotropy/parsers/roche_cedex_hires/roche_cedex_hires_parser.py b/src/allotropy/parsers/roche_cedex_hires/roche_cedex_hires_parser.py index d54af8ee5d..a3da9b74d3 100644 --- a/src/allotropy/parsers/roche_cedex_hires/roche_cedex_hires_parser.py +++ b/src/allotropy/parsers/roche_cedex_hires/roche_cedex_hires_parser.py @@ -1,3 +1,5 @@ +import openpyxl + from allotropy.allotrope.models.adm.cell_counting.rec._2024._09.cell_counting import ( Model, ) @@ -25,6 +27,33 @@ class RocheCedexHiResParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = RocheCedexHiResReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + if named_file_contents.extension == "xlsx": + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + ws = wb[wb.sheetnames[0]] + first_row = next(ws.iter_rows(max_row=1, values_only=True), None) + wb.close() + if first_row is None: + return False + cells = [str(c) for c in first_row if c is not None] + return any("Cedex ID" in c or "identifer" in c for c in cells) + named_file_contents.contents.seek(0) + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if not lines: + return False + header = lines[0] + return "Cedex ID" in header or "identifer" in header + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = RocheCedexHiResReader(named_file_contents) return Data( diff --git a/src/allotropy/parsers/tecan_magellan/tecan_magellan_parser.py b/src/allotropy/parsers/tecan_magellan/tecan_magellan_parser.py index 2c4390aadd..b39b55ad78 100644 --- a/src/allotropy/parsers/tecan_magellan/tecan_magellan_parser.py +++ b/src/allotropy/parsers/tecan_magellan/tecan_magellan_parser.py @@ -1,5 +1,7 @@ from functools import partial +import openpyxl + from allotropy.allotrope.models.adm.plate_reader.rec._2024._06.plate_reader import ( Model, ) @@ -28,6 +30,32 @@ class TecanMagellanParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = TecanMagellanReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + if len(wb.sheetnames) != 1: + wb.close() + return False + ws = wb[wb.sheetnames[0]] + found_well_positions = False + found_date_of_measurement = False + for row in ws.iter_rows(max_row=50, values_only=True): + for cell in row: + if isinstance(cell, str): + if "Well positions" in cell: + found_well_positions = True + if "Date of measurement" in cell: + found_date_of_measurement = True + if found_well_positions or found_date_of_measurement: + break + wb.close() + return found_well_positions or found_date_of_measurement + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = TecanMagellanReader(named_file_contents) metadata = MagellanMetadata.create(reader.metadata_lines) diff --git a/src/allotropy/parsers/thermo_fisher_genesys30/thermo_fisher_genesys30_parser.py b/src/allotropy/parsers/thermo_fisher_genesys30/thermo_fisher_genesys30_parser.py index d827d47c4e..86cd431f2e 100644 --- a/src/allotropy/parsers/thermo_fisher_genesys30/thermo_fisher_genesys30_parser.py +++ b/src/allotropy/parsers/thermo_fisher_genesys30/thermo_fisher_genesys30_parser.py @@ -24,6 +24,29 @@ class ThermoFisherGenesys30Parser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = ThermoFisherGenesys30Reader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + named_file_contents.contents.seek(0) + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if not lines: + return False + first_line = lines[0] + sep = "\t" if "\t" in first_line else "," + parts = first_line.split(sep) + if len(parts) < 2 or parts[0].strip() != "Scan": + return False + return any( + p.strip().lower() == "mode" + for p in (lines[1].split(sep) if len(lines) > 1 else []) + ) + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = ThermoFisherGenesys30Reader(named_file_contents) return Data( diff --git a/src/allotropy/parsers/thermo_fisher_genesys_on_board/thermo_fisher_genesys_on_board_parser.py b/src/allotropy/parsers/thermo_fisher_genesys_on_board/thermo_fisher_genesys_on_board_parser.py index 246b7dd238..1f2ad275a2 100644 --- a/src/allotropy/parsers/thermo_fisher_genesys_on_board/thermo_fisher_genesys_on_board_parser.py +++ b/src/allotropy/parsers/thermo_fisher_genesys_on_board/thermo_fisher_genesys_on_board_parser.py @@ -26,6 +26,33 @@ class ThermoFisherGenesysOnBoardParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = ThermoFisherVisionliteReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + named_file_contents.contents.seek(0) + raw = named_file_contents.contents.read(8192) + if isinstance(raw, bytes): + if raw[:2] == b"\xff\xfe": + text = raw[2:].decode("utf-16-le", errors="replace") + elif raw[:2] == b"\xfe\xff": + text = raw[2:].decode("utf-16-be", errors="replace") + else: + text = raw.decode("utf-8", errors="replace") + else: + text = raw + lines = text.splitlines() + if not lines: + return False + first = lines[0].lower().strip() + if first.startswith("sample name") or first.startswith("well position"): + return True + # Scan/kinetic format: second line starts with "nm," + if len(lines) > 1 and lines[1].strip().startswith("nm,"): + return True + return False + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: return VisionLiteData.create( ThermoFisherVisionliteReader(named_file_contents), diff --git a/src/allotropy/parsers/thermo_fisher_nanodrop_8000/nanodrop_8000_parser.py b/src/allotropy/parsers/thermo_fisher_nanodrop_8000/nanodrop_8000_parser.py index da19c8a7f1..e9216dbebf 100644 --- a/src/allotropy/parsers/thermo_fisher_nanodrop_8000/nanodrop_8000_parser.py +++ b/src/allotropy/parsers/thermo_fisher_nanodrop_8000/nanodrop_8000_parser.py @@ -25,6 +25,24 @@ class Nanodrop8000Parser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = Nanodrop8000Reader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read() + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + for line in lines[:5]: + fields = line.split("\t") + if len(fields) > 1 and any( + f.strip().lower() == "plate id" for f in fields + ): + return True + return False + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = Nanodrop8000Reader(named_file_contents) rows = map_rows(reader.data, SpectroscopyRow.create) diff --git a/src/allotropy/parsers/thermo_fisher_nanodrop_eight/nanodrop_eight_parser.py b/src/allotropy/parsers/thermo_fisher_nanodrop_eight/nanodrop_eight_parser.py index 59382c4326..4322c4d1a1 100644 --- a/src/allotropy/parsers/thermo_fisher_nanodrop_eight/nanodrop_eight_parser.py +++ b/src/allotropy/parsers/thermo_fisher_nanodrop_eight/nanodrop_eight_parser.py @@ -1,4 +1,5 @@ from functools import partial +import re from allotropy.allotrope.models.adm.spectrophotometry.benchling._2023._12.spectrophotometry import ( Model, @@ -28,6 +29,25 @@ class NanodropEightParser(VendorParser[Data, Model]): SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + nanodrop_headers = {"application", "serial number", "user name"} + found = set() + for line in lines[:5]: + if re.match(r"^[^\t]+:\t", line): + key = line.split(":\t")[0].strip().lower() + if key in nanodrop_headers: + found.add(key) + return len(found) >= 2 + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = NanodropEightReader(named_file_contents) rows = map_rows( diff --git a/src/allotropy/parsers/thermo_fisher_nanodrop_one/thermo_fisher_nanodrop_one_parser.py b/src/allotropy/parsers/thermo_fisher_nanodrop_one/thermo_fisher_nanodrop_one_parser.py index 98cfa8b948..2393e275b6 100644 --- a/src/allotropy/parsers/thermo_fisher_nanodrop_one/thermo_fisher_nanodrop_one_parser.py +++ b/src/allotropy/parsers/thermo_fisher_nanodrop_one/thermo_fisher_nanodrop_one_parser.py @@ -1,5 +1,6 @@ from pathlib import Path +import openpyxl import pandas as pd from allotropy.allotrope.models.adm.spectrophotometry.benchling._2023._12.spectrophotometry import ( @@ -19,6 +20,16 @@ from allotropy.parsers.utils.pandas import read_csv, read_multisheet_excel from allotropy.parsers.vendor_parser import VendorParser +_NANODROP_EXPERIMENT_PREFIXES = ("Nucleic Acid", "Protein A280", "Protein & Label") +_NANODROP_SHEET_PREFIXES = ( + "Nucleic Acid", + "Protein A280", + "Protein & Label", + "dsDNA", + "ssDNA", + "RNA", +) + class ThermoFisherNanodropOneParser(VendorParser[Data, Model]): DISPLAY_NAME = DISPLAY_NAME @@ -26,6 +37,41 @@ class ThermoFisherNanodropOneParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "csv,xlsx" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + if named_file_contents.extension == "xlsx": + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + sheet_name = wb.sheetnames[0] + # Check sheet name for known experiment type prefixes + if any( + sheet_name.startswith(prefix) for prefix in _NANODROP_SHEET_PREFIXES + ): + wb.close() + return True + # Also check first row for NanodropOne-specific columns + ws = wb[sheet_name] + first_row = next(ws.iter_rows(max_row=1, values_only=True), None) + wb.close() + if first_row is None: + return False + cells = [str(c) for c in first_row if c is not None] + return "A260/A280" in cells and any("Nucleic Acid" in c for c in cells) + named_file_contents.contents.seek(0) + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if not lines: + return False + header = lines[0] + return any(prefix in header for prefix in _NANODROP_EXPERIMENT_PREFIXES) + except Exception: + return False + def read_data( self, named_file_contents: NamedFileContents ) -> tuple[pd.DataFrame, str]: diff --git a/src/allotropy/parsers/thermo_fisher_qubit4/thermo_fisher_qubit4_parser.py b/src/allotropy/parsers/thermo_fisher_qubit4/thermo_fisher_qubit4_parser.py index 1da7a58b72..7d8c8b65bb 100644 --- a/src/allotropy/parsers/thermo_fisher_qubit4/thermo_fisher_qubit4_parser.py +++ b/src/allotropy/parsers/thermo_fisher_qubit4/thermo_fisher_qubit4_parser.py @@ -1,3 +1,5 @@ +import openpyxl + from allotropy.allotrope.models.adm.spectrophotometry.benchling._2023._12.spectrophotometry import ( Model, ) @@ -17,6 +19,9 @@ from allotropy.parsers.utils.pandas import map_rows from allotropy.parsers.vendor_parser import VendorParser +_QUBIT4_XLSX_RFU_COLUMNS = {"Green RFU", "Far Red RFU"} +_QUBIT4_CSV_RFU_COLUMNS = {"Green RFU", "Far Red RFU"} + class ThermoFisherQubit4Parser(VendorParser[Data, Model]): DISPLAY_NAME = "Thermo Fisher Scientific Qubit 4" @@ -24,6 +29,37 @@ class ThermoFisherQubit4Parser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = ThermoFisherQubit4Reader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + if named_file_contents.extension == "xlsx": + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + ws = wb[wb.sheetnames[0]] + first_row = next(ws.iter_rows(max_row=1, values_only=True), None) + wb.close() + if first_row is None: + return False + cells = [str(c) for c in first_row if c is not None] + return "Test Date" in cells and any( + col in cells for col in _QUBIT4_XLSX_RFU_COLUMNS + ) + named_file_contents.contents.seek(0) + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if not lines: + return False + header = lines[0] + return "Test Date" in header and any( + col in header for col in _QUBIT4_CSV_RFU_COLUMNS + ) + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: return Data( create_metadata(named_file_contents.original_file_path), diff --git a/src/allotropy/parsers/thermo_fisher_qubit_flex/thermo_fisher_qubit_flex_parser.py b/src/allotropy/parsers/thermo_fisher_qubit_flex/thermo_fisher_qubit_flex_parser.py index 99a891b799..c981bb2d9a 100644 --- a/src/allotropy/parsers/thermo_fisher_qubit_flex/thermo_fisher_qubit_flex_parser.py +++ b/src/allotropy/parsers/thermo_fisher_qubit_flex/thermo_fisher_qubit_flex_parser.py @@ -1,3 +1,5 @@ +import openpyxl + from allotropy.allotrope.models.adm.spectrophotometry.benchling._2023._12.spectrophotometry import ( Model, ) @@ -26,6 +28,36 @@ class ThermoFisherQubitFlexParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = ThermoFisherQubit4Reader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + if named_file_contents.extension == "xlsx": + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + ws = wb[wb.sheetnames[0]] + first_row = next(ws.iter_rows(max_row=1, values_only=True), None) + wb.close() + if first_row is None: + return False + cells = [str(c) for c in first_row if c is not None] + return "Test Date" in cells and "Sample RFU" in cells + named_file_contents.contents.seek(0) + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + lines = text.splitlines() + if not lines: + return False + # Skip sep= line if present + header = lines[1] if lines[0].strip().startswith("sep=") else lines[0] + return "Test Date" in header and ( + "Sample RFU" in header or "Run ID" in header + ) + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: df = ThermoFisherQubitFlexReader.read(named_file_contents) return create_data(df, named_file_contents.original_file_path) diff --git a/src/allotropy/parsers/thermo_fisher_visionlite/thermo_fisher_visionlite_parser.py b/src/allotropy/parsers/thermo_fisher_visionlite/thermo_fisher_visionlite_parser.py index 814ae7d68c..16fcbebcb5 100644 --- a/src/allotropy/parsers/thermo_fisher_visionlite/thermo_fisher_visionlite_parser.py +++ b/src/allotropy/parsers/thermo_fisher_visionlite/thermo_fisher_visionlite_parser.py @@ -23,6 +23,33 @@ class ThermoFisherVisionliteParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = ThermoFisherVisionliteReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + named_file_contents.contents.seek(0) + raw = named_file_contents.contents.read(8192) + if isinstance(raw, bytes): + if raw[:2] == b"\xff\xfe": + text = raw[2:].decode("utf-16-le", errors="replace") + elif raw[:2] == b"\xfe\xff": + text = raw[2:].decode("utf-16-be", errors="replace") + else: + text = raw.decode("utf-8", errors="replace") + else: + text = raw + lines = text.splitlines() + if not lines: + return False + first = lines[0].lower().strip() + if first.startswith("sample name") or first.startswith("well position"): + return True + # Scan/kinetic format: second line starts with "nm," + if len(lines) > 1 and lines[1].strip().startswith("nm,"): + return True + return False + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: return VisionLiteData.create( ThermoFisherVisionliteReader(named_file_contents), diff --git a/src/allotropy/parsers/thermo_skanit/thermo_skanit_parser.py b/src/allotropy/parsers/thermo_skanit/thermo_skanit_parser.py index 2cef70d849..343bd34cf3 100644 --- a/src/allotropy/parsers/thermo_skanit/thermo_skanit_parser.py +++ b/src/allotropy/parsers/thermo_skanit/thermo_skanit_parser.py @@ -1,3 +1,5 @@ +import openpyxl + from allotropy.allotrope.models.adm.plate_reader.rec._2025._03.plate_reader import ( Model, ) @@ -18,6 +20,30 @@ class ThermoSkanItParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = "xlsx" SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + sheet_names = set(wb.sheetnames) + # Full export with session/instrument sheets + if ( + "Session information" in sheet_names + and "Instrument information" in sheet_names + ): + wb.close() + return True + # Simplified export: first cell is "Measurement results" + ws = wb[wb.sheetnames[0]] + first_row = next(ws.iter_rows(max_row=1, values_only=True), None) + wb.close() + if first_row and first_row[0] == "Measurement results": + return True + return False + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: contents = read_multisheet_excel( named_file_contents.contents, engine="calamine" diff --git a/src/allotropy/parsers/unchained_labs_lunatic_stunner/unchained_labs_lunatic_stunner_parser.py b/src/allotropy/parsers/unchained_labs_lunatic_stunner/unchained_labs_lunatic_stunner_parser.py index 12d9162ecb..19ace6b56e 100644 --- a/src/allotropy/parsers/unchained_labs_lunatic_stunner/unchained_labs_lunatic_stunner_parser.py +++ b/src/allotropy/parsers/unchained_labs_lunatic_stunner/unchained_labs_lunatic_stunner_parser.py @@ -1,5 +1,7 @@ from __future__ import annotations +import openpyxl + from allotropy.allotrope.models.adm.plate_reader.rec._2025._03.plate_reader import ( Model, ) @@ -25,6 +27,39 @@ class UnchainedLabsLunaticStunnerParser(VendorParser[Data, Model]): SUPPORTED_EXTENSIONS = UnchainedLabsLunaticReader.SUPPORTED_EXTENSIONS SCHEMA_MAPPER = Mapper + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: + try: + if named_file_contents.extension == "xlsx": + named_file_contents.contents.seek(0) + wb = openpyxl.load_workbook( + named_file_contents.get_bytes_stream(), read_only=True + ) + ws = wb[wb.sheetnames[0]] + found = False + for row in ws.iter_rows(max_row=50, values_only=True): + for cell in row: + if isinstance(cell, str) and "sample name" in cell.lower(): + found = True + break + if found: + break + wb.close() + return found + named_file_contents.contents.seek(0) + raw = named_file_contents.contents.read(8192) + text = ( + raw.decode("utf-8", errors="replace") if isinstance(raw, bytes) else raw + ) + # Check the first portion of text as a block rather than per-line, + # because CSV headers may contain quoted fields with embedded newlines + block = text[:4096].lower() + return "sample name" in block and ( + "plate id" in block or "plate type" in block + ) + except Exception: + return False + def create_data(self, named_file_contents: NamedFileContents) -> Data: reader = UnchainedLabsLunaticReader(named_file_contents) measurement_groups, calculated_data = create_measurement_groups( diff --git a/src/allotropy/parsers/vendor_parser.py b/src/allotropy/parsers/vendor_parser.py index 68d2c0bea6..67a8dd23f3 100644 --- a/src/allotropy/parsers/vendor_parser.py +++ b/src/allotropy/parsers/vendor_parser.py @@ -33,6 +33,10 @@ def __init__(self, timestamp_parser: TimestampParser | None = None): def _get_mapper(self) -> SchemaMapper[Data, Model]: return self.SCHEMA_MAPPER(self.asm_converter_name, self._get_date_time) + @classmethod + def sniff(cls, named_file_contents: NamedFileContents) -> bool: # noqa: ARG003 + return False + @abstractmethod def create_data(self, named_file_contents: NamedFileContents) -> Data: raise NotImplementedError diff --git a/src/allotropy/to_allotrope.py b/src/allotropy/to_allotrope.py index b2f5d4b9c8..b79743d5cd 100644 --- a/src/allotropy/to_allotrope.py +++ b/src/allotropy/to_allotrope.py @@ -6,7 +6,7 @@ from allotropy.allotrope.allotrope import serialize_and_validate_allotrope from allotropy.exceptions import AllotropeConversionError from allotropy.named_file_contents import NamedFileContents -from allotropy.parser_factory import Vendor +from allotropy.parser_factory import discover_vendor, Vendor from allotropy.parsers.utils.locale_context import set_locale_context from allotropy.types import IOType @@ -16,7 +16,7 @@ def allotrope_from_io( contents: IOType, filepath: str, - vendor_type: VendorType, + vendor_type: VendorType | None = None, default_timezone: tzinfo | None = None, encoding: str | None = None, locale: str | None = None, @@ -30,23 +30,27 @@ def allotrope_from_io( def allotrope_model_from_io( contents: IOType, filepath: str, - vendor_type: VendorType, + vendor_type: VendorType | None = None, default_timezone: tzinfo | None = None, encoding: str | None = None, locale: str | None = None, ) -> Any: - try: - vendor = Vendor(vendor_type) - except ValueError as e: - msg = f"Failed to create parser, unregistered vendor: {vendor_type}." - raise AllotropeConversionError(msg) from e named_file_contents = NamedFileContents(contents, filepath, encoding) - if named_file_contents.extension not in vendor.supported_extensions: - msg = f"Unsupported file extension '{named_file_contents.extension}' for parser '{vendor.display_name}', expected one of '{vendor.supported_extensions}'." - raise AllotropeConversionError(msg) - parser = vendor.get_parser(default_timezone=default_timezone) - # Set locale context for parsing + if vendor_type is None: + vendor = discover_vendor(named_file_contents) + named_file_contents.contents.seek(0) + else: + try: + vendor = Vendor(vendor_type) + except ValueError as e: + msg = f"Failed to create parser, unregistered vendor: {vendor_type}." + raise AllotropeConversionError(msg) from e + if named_file_contents.extension not in vendor.supported_extensions: + msg = f"Unsupported file extension '{named_file_contents.extension}' for parser '{vendor.display_name}', expected one of '{vendor.supported_extensions}'." + raise AllotropeConversionError(msg) + + parser = vendor.get_parser(default_timezone=default_timezone) if locale: with set_locale_context(locale): return parser.to_allotrope(named_file_contents) @@ -56,7 +60,7 @@ def allotrope_model_from_io( def allotrope_from_file( filepath: str, - vendor_type: VendorType, + vendor_type: VendorType | None = None, default_timezone: tzinfo | None = None, encoding: str | None = None, locale: str | None = None, @@ -69,7 +73,7 @@ def allotrope_from_file( def allotrope_model_from_file( filepath: str, - vendor_type: VendorType, + vendor_type: VendorType | None = None, default_timezone: tzinfo | None = None, encoding: str | None = None, locale: str | None = None, @@ -98,3 +102,15 @@ def allotrope_model_from_file( msg = f"File not found: {filepath}." raise AllotropeConversionError(msg) from e + + +def vendor_from_file(filepath: str, encoding: str | None = None) -> Vendor: + with open(filepath, "rb") as f: + return vendor_from_io(f, filepath, encoding) + + +def vendor_from_io( + contents: IOType, filepath: str, encoding: str | None = None +) -> Vendor: + named_file_contents = NamedFileContents(contents, filepath, encoding) + return discover_vendor(named_file_contents) diff --git a/tests/discover_vendor_test.py b/tests/discover_vendor_test.py new file mode 100644 index 0000000000..bca1424bc0 --- /dev/null +++ b/tests/discover_vendor_test.py @@ -0,0 +1,111 @@ +from pathlib import Path + +import pytest + +from allotropy.parser_factory import Vendor +from allotropy.to_allotrope import vendor_from_file + +TESTS_DIR = Path(__file__).parent / "parsers" + +DIR_TO_VENDOR: dict[str, Vendor] = { + "biorad_bioplex_manager": Vendor.BIORAD_BIOPLEX, + "unchained_labs_lunatic_stunner": Vendor.UNCHAINED_LABS_LUNATIC, +} + +INDISTINGUISHABLE: dict[Vendor, set[Vendor]] = { + Vendor.AGILENT_GEN5: {Vendor.AGILENT_GEN5, Vendor.AGILENT_GEN5_IMAGE}, + Vendor.AGILENT_GEN5_IMAGE: {Vendor.AGILENT_GEN5, Vendor.AGILENT_GEN5_IMAGE}, + Vendor.APPBIO_QUANTSTUDIO: { + Vendor.APPBIO_QUANTSTUDIO, + Vendor.APPBIO_QUANTSTUDIO_DESIGNANDANALYSIS, + }, + Vendor.APPBIO_QUANTSTUDIO_DESIGNANDANALYSIS: { + Vendor.APPBIO_QUANTSTUDIO, + Vendor.APPBIO_QUANTSTUDIO_DESIGNANDANALYSIS, + }, + Vendor.LUMINEX_XPONENT: {Vendor.LUMINEX_XPONENT, Vendor.LUMINEX_INTELLIFLEX}, + Vendor.LUMINEX_INTELLIFLEX: {Vendor.LUMINEX_XPONENT, Vendor.LUMINEX_INTELLIFLEX}, + Vendor.THERMO_FISHER_VISIONLITE: { + Vendor.THERMO_FISHER_VISIONLITE, + Vendor.THERMO_FISHER_GENESYS_ON_BOARD, + }, + Vendor.THERMO_FISHER_GENESYS_ON_BOARD: { + Vendor.THERMO_FISHER_VISIONLITE, + Vendor.THERMO_FISHER_GENESYS_ON_BOARD, + }, + Vendor.BECKMAN_ECHO_CHERRY_PICK: { + Vendor.BECKMAN_ECHO_CHERRY_PICK, + Vendor.BECKMAN_ECHO_PLATE_REFORMAT, + }, + Vendor.BECKMAN_ECHO_PLATE_REFORMAT: { + Vendor.BECKMAN_ECHO_CHERRY_PICK, + Vendor.BECKMAN_ECHO_PLATE_REFORMAT, + }, + Vendor.MSD_WORKBENCH: { + Vendor.MSD_WORKBENCH, + Vendor.METHODICAL_MIND, + }, + Vendor.METHODICAL_MIND: { + Vendor.MSD_WORKBENCH, + Vendor.METHODICAL_MIND, + }, +} + +EXCLUDE_KEYWORDS = {"error", "exclude", "invalid"} + +SKIP_DIRS = {"utils", "example_weyland_yutani"} + + +def _expected_vendor(dir_name: str) -> Vendor: + if dir_name in DIR_TO_VENDOR: + return DIR_TO_VENDOR[dir_name] + return Vendor[dir_name.upper()] + + +def _is_valid_testcase(path: Path) -> bool: + if not path.is_file(): + return False + if str(path.stem).startswith("."): + return False + if "__pycache__" in str(path): + return False + if path.suffix.lower() in (".pyc", ".py", ".json", ".parquet"): + return False + if path.parts[-2] == "input": + return True + return all(keyword not in str(path).lower() for keyword in EXCLUDE_KEYWORDS) + + +def _get_discovery_test_cases() -> list[tuple[str, Path, Vendor]]: + cases = [] + for vendor_dir in sorted(TESTS_DIR.iterdir()): + if not vendor_dir.is_dir() or vendor_dir.name in SKIP_DIRS: + continue + testdata_dir = vendor_dir / "testdata" + if not testdata_dir.exists(): + continue + expected = _expected_vendor(vendor_dir.name) + for path in sorted(testdata_dir.rglob("*")): + if _is_valid_testcase(path): + test_id = f"{vendor_dir.name}/{path.relative_to(testdata_dir)}" + cases.append((test_id, path, expected)) + return cases + + +@pytest.mark.parametrize( + ("test_id", "test_file", "expected_vendor"), + _get_discovery_test_cases(), + ids=lambda x: x if isinstance(x, str) else "", +) +def test_discover_vendor( + test_id: str, test_file: Path, expected_vendor: Vendor +) -> None: + result = vendor_from_file(str(test_file)) + if expected_vendor in INDISTINGUISHABLE: + assert ( + result in INDISTINGUISHABLE[expected_vendor] + ), f"For {test_id}: expected one of {INDISTINGUISHABLE[expected_vendor]}, got {result}" + else: + assert ( + result == expected_vendor + ), f"For {test_id}: expected {expected_vendor}, got {result}" From 2b1c8cbb9aeb09f02d75581a160fe3683fadc932 Mon Sep 17 00:00:00 2001 From: Nathan Stender Date: Thu, 23 Apr 2026 14:09:17 -0400 Subject: [PATCH 2/2] refactor: Add try-parse fallback and multi-match disambiguation to discover_vendor When multiple parsers sniff True for the same file (e.g. Gen5/Gen5Image), try parsing with each to disambiguate. If no sniff matches at all, try parsing with every extension-compatible candidate as a last resort. Also adds vendor auto-discovery guidance to CLAUDE.md for new parser development. Co-Authored-By: Claude Opus 4.6 --- CLAUDE.md | 19 +++++++++++++++++++ src/allotropy/parser_factory.py | 21 ++++++++++++++++++++- 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/CLAUDE.md b/CLAUDE.md index 1b49e66725..107aa5f770 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -69,6 +69,25 @@ Generated models use these naming patterns: Quantity value types use abbreviated unit names: `TQuantityValueDegC`, `TQuantityValueMAU`, `TQuantityValueMmHg`, `TQuantityValueMicroLPermin`, `TQuantityValueM`, `TQuantityValueNM`, `TQuantityValueM1s1`, `TQuantityValueS1`, `TQuantityValueRU`, `TQuantityValueS`. +## Vendor Auto-Discovery + +Each parser has a `sniff(cls, named_file_contents) -> bool` classmethod used by `discover_vendor()` to auto-detect the correct parser for a file. The discovery logic in `parser_factory.py`: + +1. Filters candidates by file extension +2. Calls `sniff()` on each — collects all matches +3. Single match: returns immediately +4. Multiple matches: tries `create_data()` on each to disambiguate +5. No sniff matches: tries `create_data()` on all candidates as fallback +6. If sniff matched but parse failed: trusts the sniff result + +**When adding a new parser:** +- Implement `sniff()` — check file headers, sheet names, XML root tags, etc. Keep it lightweight. +- If the file extension is unique to this parser (e.g. `.blr`, `.rslt`), `return True` is fine. +- For shared extensions (`.csv`, `.txt`, `.xlsx`), check distinctive content patterns. +- **Check sniff logic of related parsers** that share the same extension — your new sniff must not false-positive on their test files, and theirs must not match yours. +- Run `hatch run test_all.py3.10:pytest tests/discover_vendor_test.py -x -q` to verify. +- Handle encoding: many files use UTF-16 LE (BOM `\xff\xfe`). Decode appropriately before text checks. + ## Code Rules - **No runtime imports.** All imports must be at module level. Never put `import` or `from ... import` inside a function body to work around circular dependencies — fix the dependency instead. diff --git a/src/allotropy/parser_factory.py b/src/allotropy/parser_factory.py index f526d861ad..0f49547a80 100644 --- a/src/allotropy/parser_factory.py +++ b/src/allotropy/parser_factory.py @@ -301,14 +301,33 @@ def get_parser( def discover_vendor(named_file_contents: NamedFileContents) -> Vendor: extension = named_file_contents.extension candidates = [v for v in Vendor if extension in v.supported_extensions] + # Pass 1: collect all sniff matches. + sniffed: list[Vendor] = [] for vendor in candidates: parser_cls = _VENDOR_TO_PARSER[vendor] named_file_contents.contents.seek(0) try: if parser_cls.sniff(named_file_contents): - return vendor + sniffed.append(vendor) except Exception: # noqa: S112 continue + # Exactly one sniff match — return without try-parse. + if len(sniffed) == 1: + return sniffed[0] + # Pass 2: multiple sniff matches — try parsing to disambiguate. + # No sniff matches — try parsing all candidates as fallback. + try_parse = sniffed or candidates + for vendor in try_parse: + named_file_contents.contents.seek(0) + try: + vendor.get_parser().create_data(named_file_contents) + return vendor + except Exception: # noqa: S112 + continue + # If sniff matched but try-parse failed (e.g. encoding issues unrelated to + # discovery), trust the sniff result rather than raising. + if sniffed: + return sniffed[0] msg = f"No vendor could be identified for file with extension '.{extension}'." raise AllotropeVendorNotFoundError(msg)