253 class importer

sasdata/data_io/importer.py

@@ -1,57 +1,211 @@
-import mimetypes
+"""
+Base class and content-based dispatch for all sasdata importers.
+
+Every module in :mod:`sasdata.data_io.importers` defines a child of :class:`Importer`. Each child knows
+
+* how to recognise a file it can read, from the file *contents* (:meth:`Importer.can_import`), and
+* how to read that file into SasMeasurement objects (:meth:`Importer.load_data`).
+
+Children register themselves automatically when they are defined. :meth:`Importer.find_importer` then picks the
+appropriate child for any file, so callers do not need to know the file format (or trust the file extension)::
+
+    from sasdata.data_io.importer import Importer
+
+    data = Importer.load("my_file.xml")                 # dict[str, SasMeasurement]
+    measurements, errors = Importer().import_data(["a.h5", "b.ses", "c.txt"])
+"""
+
+import importlib
+import io
 from itertools import zip_longest
 from pathlib import Path
+from typing import IO, ClassVar
 
 from sasdata.data import SasMeasurement
-from sasdata.data_io.importers import *
 
 Path_Type = str | Path
+Source_Type = Path_Type | IO[bytes]
 Value_Type = str | int | float | list[str] | list[int] | list[float]
 
+# Number of bytes read from the start of a file when sniffing its contents.
+SNIFF_SIZE = 4096
+
 
 class Importer:
-    def __init__(self):
-        pass
+    """
+    Base class for all data importers.
+
+    Child classes must implement :meth:`load_data` and :meth:`can_import`, and may set :attr:`priority` (lower values
+    are tested first) and :attr:`name`. Instantiating ``Importer`` directly gives a generic importer whose
+    :meth:`load_data` and :meth:`import_data` select the correct child importer for each file automatically.
+    """
+
+    # Human-readable name of the format handled by the importer.
+    name: ClassVar[str] = "Generic"
+    # Order in which the importer is tested when inferring the file type; lower values are tested first.
+    # Generic fallback importers (e.g. ASCII) should use a high value.
+    priority: ClassVar[int] = 100
+    # All concrete importers, populated automatically by :meth:`__init_subclass__`.
+    _registry: ClassVar[list[type["Importer"]]] = []
+
+    def __init_subclass__(cls, register: bool = True, **kwargs):
+        super().__init_subclass__(**kwargs)
+        if register and cls not in Importer._registry:
+            Importer._registry.append(cls)
+
+    def __init__(self, config: dict[str, Value_Type] | None = None):
+        self.config: dict[str, Value_Type] = config if config is not None else {}
+
+    # ------------------------------------------------------------------------------------------------------------------
+    # Interface to be implemented by every child importer
+    # ------------------------------------------------------------------------------------------------------------------
+
+    @classmethod
+    def can_import(cls, source: Source_Type) -> bool:
+        """Return True if the contents of ``source`` look like a file this importer can read. Must not raise."""
+        return False
+
+    def load_data(self, source: Source_Type) -> dict[str, SasMeasurement]:
+        """
+        Load all measurements contained in ``source``.
+
+        :param source: Path to the file to load (or an open binary file object, where supported).
+        :return: A dictionary of the measurements found in the file, keyed by a name unique within the file.
+        """
+        importer = self.find_importer(source)
+        if importer is None:
+            raise ValueError(f'Unable to determine the file type of "{source}".')
+        return importer(self.config).load_data(source)
+
+    # ------------------------------------------------------------------------------------------------------------------
+    # File-type inference
+    # ------------------------------------------------------------------------------------------------------------------
+
+    @staticmethod
+    def importers() -> list[type["Importer"]]:
+        """All registered importers, in the order in which they will be tested."""
+        # Importing the package imports every importer module, which registers the child classes.
+        importlib.import_module("sasdata.data_io.importers")
+        return sorted(Importer._registry, key=lambda importer: importer.priority)
+
+    @classmethod
+    def find_importer(cls, source: Source_Type) -> type["Importer"] | None:
+        """Infer the type of ``source`` from its contents and return the importer class able to read it."""
+        for importer in cls.importers():
+            if importer.can_import(source):
+                return importer
+        return None
+
+    @classmethod
+    def load(cls, source: Source_Type, config: dict[str, Value_Type] | None = None) -> dict[str, SasMeasurement]:
+        """Convenience method: infer the file type of ``source`` and load it."""
+        return Importer(config).load_data(source)
+
+    # ------------------------------------------------------------------------------------------------------------------
+    # Bulk import
+    # ------------------------------------------------------------------------------------------------------------------
 
     def import_data(self, files: list[Path_Type], configs: list[dict[str, Value_Type]] | None = None) \
-            -> (list[SasMeasurement], list[str]):
-        output = []
-        errors = []
+            -> tuple[list[SasMeasurement], list[str]]:
+        """
+        Import a series of files, inferring the importer for each one from its contents.
+
+        :param files: Files to import.
+        :param configs: Optional per-file configuration. If there are fewer configs than files, the last config is
+            reused for the remaining files.
+        :return: All of the measurements loaded, and a list of error messages for anything that could not be loaded.
+        """
+        output: list[SasMeasurement] = []
+        errors: list[str] = []
         if configs is None:
             configs = []
         fill_value = configs[-1] if len(configs) > 0 else None
         for file, config in zip_longest(files, configs, fillvalue=fill_value):
-            file = Path(file)
-            measurements, errors = self._import_from_url(file, config)
+            if file == fill_value:
+                # If this is reached, the most likely scenario is len(configs) > len(files) in which case, break out of the loop
+                break
+            measurements, file_errors = self._import_file(Path(file), config)
             output.extend(measurements)
-            errors.extend(errors)
+            errors.extend(file_errors)
         return output, errors
 
-    def _import_from_url(self, url: Path_Type, config: dict[str, Value_Type] | None = None) -> (list[SasMeasurement], list[str]):
-        file = Path(url)
+    def _import_file(self, file: Path, config: dict[str, Value_Type] | None = None) \
+            -> tuple[list[SasMeasurement], list[str]]:
         if not file.exists():
             return [], [f'File does not exist: "{file}"']
-        # Deprecation Warning:
-        #     mimetypes.guess_type() is soft deprecated in favor of mimetypes.guess_file_type() as of python 3.13
-        #     Once we move forward from python 3.12, this line should change
-        mime_type, encoding = mimetypes.guess_type(url)
-        imported = []
-        errors = []
-        # TODO: Need a viable config system that can pass a dictionary of value locations to each importer
+        importer = self.find_importer(file)
+        if importer is None:
+            return [], [f'Unable to determine the file type of "{file}"']
+        merged_config = {**self.config, **(config or {})}
         try:
-            match mime_type:
-                case 'application/xml' | 'text/xml':
-                    file_dict = import_xml.load_data(file)
-                    imported.extend(list(file_dict.values()))
-                case 'application/vnd.hdfgroup.hdf5':
-                    file_list = import_hdf5.load_data(file)
-                    imported.extend(file_list)
-                case _:
-                    file_list = import_ascii.load_data_default_params(file)
-                    imported.extend(file_list)
+            imported = list(importer(merged_config).load_data(file).values())
         except Exception as e:
-            errors.append(f'Error accessing "{file}": {e}')
-        for file in imported:
-            if hasattr(file, "errors"):
-                errors.extend(file.errors)
+            return [], [f'Error loading "{file}" with the {importer.name} importer: {e}']
+        errors = []
+        for measurement in imported:
+            errors.extend(getattr(measurement, "errors", []))
         return imported, errors
+
+
+# ----------------------------------------------------------------------------------------------------------------------
+# Helpers shared by the importers when sniffing file contents
+# ----------------------------------------------------------------------------------------------------------------------
+
+def read_header(source: Source_Type, size: int = SNIFF_SIZE) -> bytes:
+    """
+    Read up to ``size`` bytes from the start of ``source`` without changing the position of a file object.
+    Returns an empty bytes object if the source cannot be read.
+    """
+    try:
+        if isinstance(source, Path_Type):
+            path = Path(source)
+            if not path.is_file():
+                return b""
+            with open(path, "rb") as f:
+                return f.read(size)
+        if isinstance(source, io.IOBase) or hasattr(source, "read"):
+            position = source.tell()
+            try:
+                source.seek(0)
+                header = source.read(size)
+            finally:
+                source.seek(position)
+            return header if isinstance(header, bytes) else header.encode()
+    except (OSError, ValueError):
+        # File access or read errors will fall through here, implying the reader cannot interpret the file.
+        # An empty byte string will be returned, but there may be other cases where this try/except are bypassed, so
+        # explicitly do nothing here.
+        pass
+    return b""
+
+
+def is_text(header: bytes) -> bool:
+    """Heuristic check that a file header is text rather than binary data."""
+    if not header or b"\x00" in header:
+        return False
+    try:
+        header.decode("utf-8")
+    except UnicodeDecodeError as e:
+        # The header may have been truncated in the middle of a multibyte character.
+        if e.start < len(header) - 4:
+            try:
+                header.decode("latin-1")
+            except UnicodeDecodeError:
+                return False
+            # latin-1 decodes anything, so reject if there are many control characters.
+            controls = sum(1 for b in header if b < 32 and b not in (9, 10, 12, 13))
+            return controls / len(header) < 0.05
+    return True
+
+
+def measurements_by_name(measurements: list[SasMeasurement]) -> dict[str, SasMeasurement]:
+    """Key a list of measurements by name, adding a numeric suffix to any repeated names so none are lost."""
+    output: dict[str, SasMeasurement] = {}
+    for measurement in measurements:
+        key = measurement.name
+        index = 2
+        while key in output:
+            key = f"{measurement.name} ({index})"
+            index += 1
+        output[key] = measurement
+    return output

sasdata/data_io/importers/__init__.py

@@ -1 +1,15 @@
-__all__ = ['import_ascii', 'import_hdf5', 'import_sesans', 'import_xml']
+"""
+Importers for each of the file formats supported by sasdata.
+
+Each module defines a child of :class:`sasdata.data_io.importer.Importer`. Importing this package registers all of
+them, so :meth:`Importer.find_importer` can select the correct one from the contents of a file.
+"""
+
+from sasdata.data_io.importers.import_ascii import AsciiImporter
+from sasdata.data_io.importers.import_hdf5 import HDF5Importer
+from sasdata.data_io.importers.import_sesans import SesansImporter
+from sasdata.data_io.importers.import_xml import XmlImporter
+
+__all__ = [
+    'AsciiImporter', 'HDF5Importer', 'SesansImporter', 'XmlImporter',
+]

sasdata/data_io/importers/import_ascii.py

@@ -3,7 +3,6 @@
 from dataclasses import dataclass, field, replace
 from enum import Enum
 from os import path
-from pathlib import Path
 
 import numpy as np
 
@@ -14,6 +13,7 @@
     pairings,
 )
 from sasdata.data import SasMeasurement
+from sasdata.data_io.importer import Importer, Path_Type, Source_Type, is_text, measurements_by_name, read_header
 from sasdata.dataset_types import DatasetType, one_dim, unit_kinds
 from sasdata.default_units import get_default_unit
 from sasdata.guess import (
@@ -207,7 +207,7 @@ def merge_uncertainties(quantities: dict[str, Quantity]) -> dict[str, Quantity]:
     return new_quantities
 
 
-def load_data(params: AsciiReaderParams) -> list[SasMeasurement]:
+def load_measurements(params: AsciiReaderParams) -> list[SasMeasurement]:
     """
     This loads a series of SasMeasurement objects based on the params.
     The amount of SasMeasurement objects loaded will depend on how many
@@ -239,6 +239,47 @@ def load_data(params: AsciiReaderParams) -> list[SasMeasurement]:
     return loaded_data
 
 
-def load_data_default_params(filename: str | Path) -> list[SasMeasurement]:
-    params = guess_params_from_filename(filename, guess_dataset_type(filename))
-    return load_data(params)
+class AsciiImporter(Importer):
+    """
+    Importer for column-based ASCII files.
+
+    The importer can be given an :class:`AsciiReaderParams` object (e.g. generated by the ASCII Reader Dialog in
+    SasView) describing how to read the file(s). If no parameters are given, they are guessed from the file contents.
+    As almost any text file can be read as ASCII, this importer is the fallback used when no other importer recognises
+    a file, so it is tested last.
+    """
+
+    name = "ASCII"
+    priority = 1000
+
+    def __init__(self, config: dict | None = None, params: AsciiReaderParams | None = None):
+        super().__init__(config)
+        self.params = params
+
+    @classmethod
+    def can_import(cls, source: Source_Type) -> bool:
+        """Any text file on disk could be a column-based ASCII file."""
+        if not isinstance(source, Path_Type):
+            return False
+        return is_text(read_header(source))
+
+    def load_data(self, source: Source_Type | None = None) -> dict[str, SasMeasurement]:
+        """
+        Load ASCII data, keyed by the basename of each file.
+
+        :param source: The file to load. If parameters were given to the importer, they are used to read this file;
+            otherwise, the parameters are guessed from the file contents. If ``source`` is None, every file listed in
+            the importer's parameters is loaded.
+        """
+        if source is None:
+            if self.params is None:
+                raise ValueError("A file to load, or AsciiReaderParams listing the files to load, must be given.")
+            params = self.params
+        elif not isinstance(source, Path_Type):
+            raise TypeError("The ASCII importer can only load files from a path.")
+        elif self.params is not None:
+            params = replace(self.params, filenames=[str(source)])
+        else:
+            filename = str(source)
+            params = guess_params_from_filename(filename, guess_dataset_type(filename))
+        return measurements_by_name(load_measurements(params))