Add support for Jeol epma data format (Issue #115)

rsciio/jeol/_api.py

@@ -152,6 +152,65 @@ def _read_asw(filename, **kwargs):
 
 file_reader.__doc__ %= (FILENAME_DOC, LAZY_DOC, RETURNS_DOC)
 
+def _read_epma_img(filename, **kwargs):
+    """
+    Parameters
+    ----------
+    filename : str
+        img file name
+    kwargs :
+        not used
+
+    Returns
+    -------
+    image_list : list of images(dict)
+        (Always len(image_list) == 1)
+        None if image is not an EPMA map.
+    """
+    with open(filename, "br") as fd:
+        file_magic = np.fromfile(fd, "<I", 1)[0]
+        if file_magic != 1:   #  Is this correct?
+            return None
+        fd.seek(32)
+        _w = 256
+        _h = 256
+        _count = _w * _h
+        epma_img = np.fromfile(fd, dtype=np.double, count=_count).reshape((_h, _w))
+
+    axes = [
+        {
+            "name": "y",
+            "size": _h,
+            "offset": 0,
+            "scale": 1.0,
+            "units": "pixels",
+        },
+        {
+            "name": "x",
+            "size": _w,
+            "offset": 0,
+            "scale": 1.0,
+            "units": "pixels",
+        },
+    ]
+    metadata = {
+        "General": {
+            "original_filename": os.path.basename(filename),
+            "title": os.path.basename(filename),
+        },
+        "Signal": {
+            "record_by": "image",
+            "signal_type": "EPMA",
+        },
+    }
+
+    dictionary = {
+        "data": epma_img,
+        "axes": axes,
+        "metadata": metadata,
+    }
+    return [dictionary]
+
 
 def _read_img(filename, **kwargs):
     """
@@ -170,7 +229,12 @@ def _read_img(filename, **kwargs):
 
     with open(filename, "br") as fd:
         file_magic = np.fromfile(fd, "<I", 1)[0]
-        if file_magic != 52:
+        if file_magic != 52:  # Not an EDS related image
+            # Try to read as an EPMA image
+            _epma = _read_epma_img(filename)
+            if _epma is not None:
+                return _epma
+
             _logger.warning(f"Not a valid JEOL img format '{filename}'")
             return []
 

rsciio/tests/test_jeol.py

@@ -18,7 +18,6 @@
 
 import gc
 from pathlib import Path
-import tempfile
 import zipfile
 
 import numpy as np
@@ -531,24 +530,23 @@ def test_pts_frame_shift():
         )
 
 
-def test_broken_files():
+def test_broken_files(tmp_path):
     TEST_BROKEN_FILES = ["test.asw", "test.pts", "test.img"]
-    with tempfile.TemporaryDirectory() as tmpdir:
-        for _file in TEST_BROKEN_FILES:
-            file = Path(tmpdir) / _file
-            with open(file, "w") as fd:
-                fd.write("aaaaaaaa")
-            if file.suffix == ".asw":
-                # in case of asw, valid data can not be obtained
-                with pytest.raises(ValueError, match="Not a valid JEOL asw format"):
-                    _ = hs.load(file)
-            else:
-                # just skipping broken files
-                s = hs.load(file)
-                assert s == []
-
-
-def test_seq_eds_files():
+    for _file in TEST_BROKEN_FILES:
+        file = Path(tmp_path) / _file
+        with open(file, "w") as fd:
+            fd.write("aaaaaaaa")
+        if file.suffix == ".asw":
+            # in case of asw, valid data can not be obtained
+            with pytest.raises(ValueError, match="Not a valid JEOL asw format"):
+                _ = hs.load(file)
+        else:
+            # just skipping broken files
+            s = hs.load(file)
+            assert s == []
+
+
+def test_seq_eds_files(tmp_path):
     pos0 = [0.0, 0.0, -0.000132, 0.000132]
     pos = [
         [0.0, 0.0, 0.0, 0.0],
@@ -557,86 +555,85 @@ def test_seq_eds_files():
     ]
     memo = ["", "030", "035"]
     test_file = Path(__file__).resolve().parent / "jeol_seq_eds_files.zip"
-    with tempfile.TemporaryDirectory() as tmpdir:
-        with zipfile.ZipFile(test_file, "r") as zipped:
-            zipped.extractall(tmpdir)
-
-        # test reading sequential acuired EDS spectrum
-        s = hs.load(Path(tmpdir) / "1" / "1.ASW")
-        # check if three subfiles are in file (img, eds, eds)
-        assert len(s) == 3
-        # check positional information in subfiles
-        for i, p in enumerate(pos):
-            sampleinfo = s[i].original_metadata["asw"]["SampleInfo"]["0"]
-            viewinfo = sampleinfo["ViewInfo"]["0"]
-            np.testing.assert_allclose(viewinfo["PositionMM2"], pos0)
-            viewdata_asw = viewinfo["ViewData"]
-            viewdata = s[i].original_metadata["asw_viewdata"]
-            np.testing.assert_allclose(viewdata["PositionMM2"], p)
-            np.testing.assert_allclose(
-                viewdata["PositionMM2"], viewdata_asw[i]["PositionMM2"]
-            )
-            assert viewdata["Memo"] == memo[i]
-        assert isinstance(s[1], hs.signals.EDSTEMSpectrum)
-        assert isinstance(s[2], hs.signals.EDSTEMSpectrum)
-
-        # test with broken asw file
-        fname = Path(tmpdir) / "1" / "1.ASW"
-        fname2 = Path(tmpdir) / "1" / "2.ASW"
-        with open(fname, "rb") as f:
-            data = bytearray(f.read())
-
-        # No ViewData
-        data2 = data.copy()
-        data2[0x42D] = 0x30
-        with open(fname2, "wb") as f:
-            f.write(data2)
-        dat = hs.load(fname2)
-        assert len(dat) == 0
-
-        # No ViewInfo
-        data2 = data.copy()
-        data2[0x1AD] = 0x30
-        with open(fname2, "wb") as f:
-            f.write(data2)
-        dat = hs.load(fname2)
-        assert len(dat) == 0
-
-        # No SampleInfo
-        data2 = data.copy()
-        data2[0x6E] = 0x30
-        with open(fname2, "wb") as f:
-            f.write(data2)
-        dat = hs.load(fname2)
-        assert len(dat) == 0
-
-        # test read for pseudo SEM eds/img data
-        sub_dir = Path(tmpdir) / "1" / "Sample" / "00_View002"
-        test_files = ["View002_0000000.img", "View002_0000001.eds"]
-
-        # rewrite AccV  200 kV to 20 kV to generate pseudo SEM data
-        # .img
-        with open(sub_dir / test_files[0], "rb") as f:
-            data = bytearray(f.read())
-            data[0x75BC] = 0xA0
-            data[0x75BD] = 0x41
-        with open(sub_dir / ("x" + test_files[0]), "wb") as f:
-            f.write(data)
-        s = hs.load(sub_dir / ("x" + test_files[0]))
-        assert "SEM" in s.metadata["Acquisition_instrument"]
-
-        # .eds
-        with open(sub_dir / test_files[1], "rb") as f:
-            data = bytearray(f.read())
-            data[0x4B13] = 0x34
-        with open(sub_dir / ("x" + test_files[1]), "wb") as f:
-            f.write(data)
-        s = hs.load(sub_dir / ("x" + test_files[1]))
-        assert isinstance(s, hs.signals.EDSSEMSpectrum)
-        assert "SEM" in s.metadata["Acquisition_instrument"]
-
-
-def test_frame_start_index():
+    with zipfile.ZipFile(test_file, "r") as zipped:
+        zipped.extractall(tmp_path)
+
+    # test reading sequential acuired EDS spectrum
+    s = hs.load(Path(tmp_path) / "1" / "1.ASW")
+    # check if three subfiles are in file (img, eds, eds)
+    assert len(s) == 3
+    # check positional information in subfiles
+    for i, p in enumerate(pos):
+        sampleinfo = s[i].original_metadata["asw"]["SampleInfo"]["0"]
+        viewinfo = sampleinfo["ViewInfo"]["0"]
+        np.testing.assert_allclose(viewinfo["PositionMM2"], pos0)
+        viewdata_asw = viewinfo["ViewData"]
+        viewdata = s[i].original_metadata["asw_viewdata"]
+        np.testing.assert_allclose(viewdata["PositionMM2"], p)
+        np.testing.assert_allclose(
+            viewdata["PositionMM2"], viewdata_asw[i]["PositionMM2"]
+        )
+        assert viewdata["Memo"] == memo[i]
+    assert isinstance(s[1], hs.signals.EDSTEMSpectrum)
+    assert isinstance(s[2], hs.signals.EDSTEMSpectrum)
+
+    # test with broken asw file
+    fname = Path(tmp_path) / "1" / "1.ASW"
+    fname2 = Path(tmp_path) / "1" / "2.ASW"
+    with open(fname, "rb") as f:
+        data = bytearray(f.read())
+
+    # No ViewData
+    data2 = data.copy()
+    data2[0x42D] = 0x30
+    with open(fname2, "wb") as f:
+        f.write(data2)
+    dat = hs.load(fname2)
+    assert len(dat) == 0
+
+    # No ViewInfo
+    data2 = data.copy()
+    data2[0x1AD] = 0x30
+    with open(fname2, "wb") as f:
+        f.write(data2)
+    dat = hs.load(fname2)
+    assert len(dat) == 0
+
+    # No SampleInfo
+    data2 = data.copy()
+    data2[0x6E] = 0x30
+    with open(fname2, "wb") as f:
+        f.write(data2)
+    dat = hs.load(fname2)
+    assert len(dat) == 0
+
+    # test read for pseudo SEM eds/img data
+    sub_dir = Path(tmp_path) / "1" / "Sample" / "00_View002"
+    test_files = ["View002_0000000.img", "View002_0000001.eds"]
+
+    # rewrite AccV  200 kV to 20 kV to generate pseudo SEM data
+    # .img
+    with open(sub_dir / test_files[0], "rb") as f:
+        data = bytearray(f.read())
+        data[0x75BC] = 0xA0
+        data[0x75BD] = 0x41
+    with open(sub_dir / ("x" + test_files[0]), "wb") as f:
+        f.write(data)
+    s = hs.load(sub_dir / ("x" + test_files[0]))
+    assert "SEM" in s.metadata["Acquisition_instrument"]
+
+    # .eds
+    with open(sub_dir / test_files[1], "rb") as f:
+        data = bytearray(f.read())
+        data[0x4B13] = 0x34
+    with open(sub_dir / ("x" + test_files[1]), "wb") as f:
+        f.write(data)
+    s = hs.load(sub_dir / ("x" + test_files[1]))
+    assert isinstance(s, hs.signals.EDSSEMSpectrum)
+    assert "SEM" in s.metadata["Acquisition_instrument"]
+
+
+def test_frame_start_index(tmp_path):
     file = TESTS_FILE_PATH / "Sample" / "00_View000" / TEST_FILES[7]
     frame_start_index_ref = [
         0,
@@ -697,16 +694,30 @@ def test_frame_start_index():
     assert s.data.shape == (2, 16, 16, 8)
 
     # test with pseudo "SEM" data
-    with tempfile.TemporaryDirectory() as tmpdir:
-        test_file = Path(tmpdir) / "test.pts"
-        with open(file, "rb") as f:
-            data = bytearray(f.read())
-            # AcckV = 20 kV
-            data[0x1116] = 0xA0
-            data[0x1117] = 0x41
-        with open(test_file, "wb") as f:
-            f.write(data)
-            s = hs.load(
-                test_file, downsample=[32, 32], rebin_energy=512, SI_dtype=np.int32
-            )
-        assert s.metadata["Signal"]["signal_type"] == "EDS_SEM"
+    test_file = Path(tmp_path) / "test.pts"
+    with open(file, "rb") as f:
+        data = bytearray(f.read())
+        # AcckV = 20 kV
+        data[0x1116] = 0xA0
+        data[0x1117] = 0x41
+    with open(test_file, "wb") as f:
+        f.write(data)
+        s = hs.load(test_file, downsample=[32, 32], rebin_energy=512, SI_dtype=np.int32)
+    assert s.metadata["Signal"]["signal_type"] == "EDS_SEM"
+
+
+def test_epma_map(tmp_path):
+    EPMA_HEADER = (
+        b'\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x08\x00\x00\x00'
+        b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
+    )
+    img_size = 256
+    test_file = Path(tmp_path) / "epma.map"
+    vals = np.arange(img_size * img_size, dtype=np.double)
+    with open(test_file, "wb") as f:
+        f.write(EPMA_HEADER)
+        vals.tofile(f, "")
+    s = hs.load(test_file)
+    assert s.metadata.Signal.signal_type == "EPMA"
+    for y in range(img_size):
+        assert s.data[y][10] == y * 256 + 10