Centralized Importer Class

sasdata/data_io/importer.py

@@ -1,30 +1,57 @@
+import mimetypes
+from itertools import zip_longest
 from pathlib import Path
 
 from sasdata.data import SasMeasurement
 from sasdata.data_io.importers import *
 
 Path_Type = str | Path
+Value_Type = str | int | float | list[str] | list[int] | list[float]
 
 
 class Importer:
     def __init__(self):
         pass
 
-    def import_data(self, files: Path_Type) -> (list[SasMeasurement], list[str]):
+    def import_data(self, files: list[Path_Type], configs: list[dict[str, Value_Type]] | None = None) \
+            -> (list[SasMeasurement], list[str]):
         output = []
         errors = []
-        for file in files:
+        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)
-            if measurements := self._import_from_url(file):
-                output.append(measurements)
-            else:
-                errors.append(f'File "{file}" not found.')
+            measurements, errors = self._import_from_url(file, config)
+            output.extend(measurements)
+            errors.extend(errors)
         return output, errors
 
-    def _import_from_url(self, url: Path_Type) -> list[SasMeasurement]:
+    def _import_from_url(self, url: Path_Type, config: dict[str, Value_Type] | None = None) -> (list[SasMeasurement], list[str]):
         file = Path(url)
         if not file.exists():
-            return []
-        with open(url) as f:
-            pass
-        return []
+            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
+        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)
+        except Exception as e:
+            errors.append(f'Error accessing "{file}": {e}')
+        for file in imported:
+            if hasattr(file, "errors"):
+                errors.extend(file.errors)
+        return imported, errors

sasdata/data_io/importers/import_ascii.py

@@ -2,6 +2,7 @@
 from dataclasses import dataclass, field, replace
 from enum import Enum
 from os import path
+from pathlib import Path
 
 import numpy as np
 
@@ -72,11 +73,8 @@ def columns_included(self) -> list[tuple[str, NamedUnit]]:
         ]
 
 
-# TODO: Should I make this work on a list of filenames as well?
-def guess_params_from_filename(
-    filename: str, dataset_type: DatasetType
-) -> AsciiReaderParams:
-    # Lets just assume we want all of the seaprators on. This seems to work for most files.
+def guess_params_from_filename(filename: str, dataset_type: DatasetType) -> AsciiReaderParams:
+    # Lets just assume we want all separators on. This seems to work for most files.
     separator_dict = initialise_separator_dict(True)
     with open(filename) as file:
         lines = file.readlines()
@@ -220,6 +218,6 @@ def load_data(params: AsciiReaderParams) -> list[SasMeasurement]:
     return loaded_data
 
 
-def load_data_default_params(filename: str) -> list[SasMeasurement]:
+def load_data_default_params(filename: str | Path) -> list[SasMeasurement]:
     params = guess_params_from_filename(filename, guess_dataset_type(filename))
     return load_data(params)

sasdata/data_io/importers/import_hdf5.py

@@ -1,5 +1,6 @@
 import logging
 from collections.abc import Callable
+from pathlib import Path
 
 import h5py
 import numpy as np
@@ -405,7 +406,7 @@ def parse_metadata(node : HDF5Group) -> Metadata:
                     raw=raw)
 
 
-def load_data(filename: str) -> dict[str, SasMeasurement]:
+def load_data(filename: str | Path) -> dict[str, SasMeasurement]:
     with h5py.File(filename, "r") as f:
         loaded_data: dict[str, SasMeasurement] = {}
 

sasdata/data_io/importers/import_sesans.py

@@ -5,6 +5,7 @@
 import re
 from collections import defaultdict
 from itertools import groupby
+from pathlib import Path
 
 import numpy as np
 
@@ -205,7 +206,7 @@ def parse_sesans(lines: list[str]) -> SasMeasurement:
     )
 
 
-def load_data(filename) -> SasMeasurement:
+def load_data(filename: str | Path) -> SasMeasurement:
     with open(filename) as infile:
         lines = infile.readlines()
     return parse_sesans(lines)

sasdata/data_io/importers/import_xml.py

@@ -1,5 +1,6 @@
 import logging
 from collections.abc import Callable
+from pathlib import Path
 
 import numpy as np
 from lxml import etree
@@ -282,7 +283,7 @@ def load_raw(node: etree._Element, version: str) -> MetaNode:
     return MetaNode(name=etree.QName(node).localname, attrs=attrib, contents=contents)
 
 
-def load_data(filename: str) -> dict[str, SasMeasurement]:
+def load_data(filename: str | Path) -> dict[str, SasMeasurement]:
     """Load scattering data from an XML file"""
     loaded_data: dict[str, SasMeasurement] = {}
     tree = etree.parse(filename)

test/data_io/utest_generic_file_import_class.py

@@ -8,6 +8,7 @@
 
 import numpy as np
 
+from sasdata.data_io.importer import Importer
 from sasdata.dataloader.data_info import Data1D, DataInfo, plottable_1D
 from sasdata.dataloader.filereader import FileReader
 from sasdata.dataloader.loader import Loader
@@ -103,3 +104,20 @@ def get_file_contents(self):
         self.current_datainfo = DataInfo()
         self.current_datainfo.meta_data["blah"] = self.nextline()
         self.send_to_output()
+
+
+class TestImporterClass:
+    def test_basic_import(self):
+        file_list = [
+            find("TestExtensions.xml"),
+            find("FEB18012.ASC"), # Silently Fails
+            find("Anton-Paar.pdh"),  # Silently Fails
+            find("ascii_test_1.txt"),  # Silently Fails
+            find("MAR07232_rest.h5"),  # Silently Fails
+            find("nans_in_2d_data.DAT")
+        ]
+        importer = Importer()
+        imported_data, errors = importer.import_data(file_list)
+        assert len(imported_data) == 1
+        assert imported_data[0].name == "TK49 c10_SANS"
+        assert len(errors) == 2