feat: Add vendor auto-discovery from file contents

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.

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])
 

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,40 @@ 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):
+                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)
+
+
 def get_table_contents() -> str:
     contents = """The parsers follow maturation levels of: Recommended, Candidate Release, Working Draft.
 

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)

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)
 

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(

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

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)

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:

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:

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

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(

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(

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(

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)