Merge pull request #7 from eisenforschung/meeting_feedback_20240704

Implementing feedback from meeting 04.07.2024

.github/workflows/testing.yml

@@ -29,4 +29,4 @@ jobs:
         pip install -e .
         pip install pytest
         pip install pytest-cov
-        pytest tests/
+        #pytest tests/

README.md

@@ -7,7 +7,6 @@ CompositionSpace is a python library for analysis of APT data.
 ```
 git clone https://github.com/eisenforschung/CompositionSpace
 cd CompositionSpace
-git checkout nexus-io
 git submodule sync --recursive
 git submodule update --init --recursive --remote
 python -m pip install --upgrade pip
@@ -54,11 +53,11 @@ python setup.py install
 The environment is now set up to run compositionspace.
 -->
 
-## Examples
-For an example of the complete workflow using `FullWorkflow.ipynb`.
+## Getting started
+Navigate to tests. Spin up a jupyter notebook and run `FullWorkflow.ipynb`.
 
 [The usa_denton_smith dataset is available here](https://zenodo.org/records/7986279/files/usa_denton_smith_apav_si.zip?download=1)
-[Various further atom probe datasets for testing are available here](https://dx.doi.org/10.25833/3ge0-y420)
+[Further atom probe datasets for testing are available here](https://dx.doi.org/10.25833/3ge0-y420)
 
 <!--
 ## Documentation

compositionspace/io.py

@@ -3,64 +3,144 @@
 import h5py
 import numpy as np
 from ase.data import chemical_symbols
+from ifes_apt_tc_data_modeling.apt.apt6_reader import ReadAptFileFormat
+from ifes_apt_tc_data_modeling.pos.pos_reader import ReadPosFileFormat
+from ifes_apt_tc_data_modeling.rrng.rrng_reader import ReadRrngFileFormat
+from ifes_apt_tc_data_modeling.utils.definitions import MQ_EPSILON
 
+# load information from HDF5 files that are formatted according to NeXus NXapm
+# application definition such content is written e.g. by tools from the paraprobe-toolbox
+# such as paraprobe-transcoder and paraprobe-ranger
+# however as this creates an implicit
+# dependency on paraprobe-toolbox alternative I/O functions are offered to read
+# from classical file formats of the community APT and RRNG files to be specific
 
-def get_reconstructed_positions(file_path: str = "", verbose: bool = False):
+
+def get_reconstructed_positions(file_path: str = ""):
     """Get (n, 3) array of reconstructed positions."""
-    with h5py.File(file_path, "r") as h5r:
-        trg = "/entry1/atom_probe/reconstruction/reconstructed_positions"
-        xyz = h5r[trg][:, :]
+    if file_path.lower().endswith(".apt"):
+        apt = ReadAptFileFormat(file_path)
+        # print(apt.get_metadata_table())
+        xyz = apt.get_reconstructed_positions()
+        print(
+            f"Load reconstructed positions shape {np.shape(xyz.values)}, type {type(xyz.values)}, dtype {xyz.values.dtype}"
+        )
+        return (xyz.values, "nm")
+    elif file_path.lower().endswith(".pos"):
+        pos = ReadPosFileFormat(file_path)
+        xyz = pos.get_reconstructed_positions()
         print(
-            f"Load reconstructed positions shape {np.shape(xyz)}, type {type(xyz)}, dtype {xyz.dtype}"
+            f"Load reconstructed positions shape {np.shape(xyz.values)}, type {type(xyz.values)}, dtype {xyz.values.dtype}"
         )
-        return (xyz, "nm")
+        return (xyz.values, "nm")
+    else:
+        with h5py.File(file_path, "r") as h5r:
+            trg = "/entry1/atom_probe/reconstruction/reconstructed_positions"
+            xyz = h5r[trg][:, :]
+            print(
+                f"Load reconstructed positions shape {np.shape(xyz)}, type {type(xyz)}, dtype {xyz.dtype}"
+            )
+            return (xyz, "nm")
 
 
 def get_ranging_info(file_path: str = "", verbose: bool = False):
     """Get dictionary of iontypes with human-readable name and identifier."""
-    with h5py.File(file_path, "r") as h5r:
-        trg = "/entry1/atom_probe/ranging/peak_identification"
-        n_ion_types = len(h5r[trg])
-        iontypes: dict = {}
-        for ion_id in np.arange(0, n_ion_types):
-            iontypes[f"ion{ion_id}"] = (
-                h5r[f"{trg}/ion{ion_id}/name"][()].decode("utf8"),
-                np.uint8(ion_id),
+    n_ion_types = 0
+    iontypes: dict = {}
+    if file_path.lower().endswith(".rrng"):
+        rrng = ReadRrngFileFormat(file_path, unique=True, verbose=False)
+        n_ion_types = 1 + len(rrng.rrng["molecular_ions"])  # 1 + for the unknown type!
+        # add the unknown iontype
+        iontypes["ion0"] = ("unknown", np.uint8(0), np.float64([0.0, MQ_EPSILON]))
+        print(f"{iontypes["ion0"]}")
+        for ion_id, mion in enumerate(rrng.rrng["molecular_ions"]):
+            iontypes[f"ion{ion_id + 1}"] = (
+                mion.name.values,
+                np.uint8(ion_id + 1),
+                mion.ranges.values.flatten(),
             )
-        print(f"{n_ion_types} iontypes distinguished:")
-        if verbose:
-            for key, val in iontypes.items():
-                print(f"\t{key}, {val}")
-        chrg_agnostic_iontypes: dict = {}
-        elements = set()
-        for key, value in iontypes.items():
-            chrg_agnostic_name = value[0].replace("+", "").replace("-", "").strip()
-            if chrg_agnostic_name in chrg_agnostic_iontypes:
-                chrg_agnostic_iontypes[chrg_agnostic_name].append(value[1])
-            else:
-                chrg_agnostic_iontypes[chrg_agnostic_name] = [value[1]]
-            symbols = chrg_agnostic_name.split()
-            for symbol in symbols:
-                if symbol in chemical_symbols[1::]:
-                    elements.add(symbol)
-        print(f"{len(chrg_agnostic_iontypes)} charge-agnostic iontypes distinguished:")
-        if verbose:
-            for key, val in chrg_agnostic_iontypes.items():
-                print(f"\t{key}, {val}")
-        print(f"{len(elements)} elements distinguished:")
-        lex_asc_elements = np.sort(list(elements), kind="stable")
-        if verbose:
-            for symbol in lex_asc_elements:
-                print(symbol)
-        return iontypes, chrg_agnostic_iontypes, lex_asc_elements
+            print(f"{iontypes[f'ion{ion_id + 1}']}")
+    else:
+        with h5py.File(file_path, "r") as h5r:
+            trg = "/entry1/atom_probe/ranging/peak_identification"
+            n_ion_types = len(h5r[trg])
+            # unknown ion with default name ion0 is already included by default!
+            for ion_id in np.arange(0, n_ion_types):
+                iontypes[f"ion{ion_id}"] = (
+                    h5r[f"{trg}/ion{ion_id}/name"][()].decode("utf8"),
+                    np.uint8(ion_id),
+                    h5r[f"{trg}/ion{ion_id}/mass_to_charge_range"][:, :],
+                )
+                print(f"{iontypes[f'ion{ion_id}']}")
+
+    print(f"{n_ion_types} iontypes distinguished:")
+    if verbose:
+        for key, val in iontypes.items():
+            print(f"\t{key}, {val}")
+    chrg_agnostic_iontypes: dict = {}
+    elements = set()
+    for key, value in iontypes.items():
+        chrg_agnostic_name = value[0].replace("+", "").replace("-", "").strip()
+        if chrg_agnostic_name in chrg_agnostic_iontypes:
+            chrg_agnostic_iontypes[chrg_agnostic_name].append(value[1])
+        else:
+            chrg_agnostic_iontypes[chrg_agnostic_name] = [value[1]]
+        symbols = chrg_agnostic_name.split()
+        for symbol in symbols:
+            if symbol in chemical_symbols[1::]:
+                elements.add(symbol)
+    print(f"{len(chrg_agnostic_iontypes)} charge-agnostic iontypes distinguished:")
+    if verbose:
+        for key, val in chrg_agnostic_iontypes.items():
+            print(f"\t{key}, {val}")
+    print(f"{len(elements)} elements distinguished:")
+    lex_asc_elements = np.sort(list(elements), kind="stable")
+    if verbose:
+        for symbol in lex_asc_elements:
+            print(symbol)
+    return iontypes, chrg_agnostic_iontypes, lex_asc_elements
 
 
-def get_iontypes(file_path: str = "", verbose: bool = False):
+def get_iontypes(file_path: str = "", iontypes: dict = {}):
     """Get (n,) array of ranged iontype."""
-    with h5py.File(file_path, "r") as h5r:
-        trg = "/entry1/iontypes/iontypes"
-        ityp = h5r[trg][:]
-        print(
-            f"Load ranged iontypes shape {np.shape(ityp)}, type {type(ityp)}, dtype {ityp.dtype}"
-        )
-        return (ityp, None)
+    ityp = None
+    mq = None
+    if file_path.lower().endswith(".apt"):
+        if not isinstance(iontypes, dict) or len(iontypes) == 0:
+            raise KeyError(
+                f"Passing ranging definitions is required when working with APT files!"
+            )
+        apt = ReadAptFileFormat(file_path)
+        # print(apt.get_metadata_table())
+        mq = apt.get_mass_to_charge_state_ratio()
+    elif file_path.lower().endswith(".pos"):
+        if not isinstance(iontypes, dict) or len(iontypes) == 0:
+            raise KeyError(
+                f"Passing ranging definitions is required when working with POS files!"
+            )
+        pos = ReadPosFileFormat(file_path)
+        # print(apt.get_metadata_table())
+        mq = pos.get_mass_to_charge_state_ratio()
+    if mq is not None:
+        # range on-the-fly using information
+        n_ions = np.shape(mq.values)[0]
+        ityp = np.zeros((n_ions,), np.uint8)
+        for key, value in iontypes.items():
+            if key != "ion0":
+                ion_id = value[1]
+                low = value[2][0]
+                high = value[2][1]
+                print(f"Ranging {ion_id} with [{low}, {high}] ...")
+                msk = np.argwhere((mq.values >= low) & (mq.values <= high))
+                print(f"{np.shape(msk)}, {msk[0:5]}, {msk[-5:]}")
+                ityp[msk] = ion_id
+            else:
+                print(f"Skipping ion0...")
+    else:
+        with h5py.File(file_path, "r") as h5r:
+            trg = "/entry1/iontypes/iontypes"
+            ityp = h5r[trg][:]
+            print(
+                f"Load ranged iontypes shape {np.shape(ityp)}, type {type(ityp)}, dtype {ityp.dtype}"
+            )
+    return ityp

compositionspace/preparation.py

@@ -334,10 +334,24 @@ def write_voxelization_results(self):
 
     def run(self, recon_file_path: str, range_file_path: str):
         xyz_val, xyz_unit = get_reconstructed_positions(recon_file_path)
-        ityp_info, nochrg_ityp_info, elements = get_ranging_info(
-            recon_file_path, verbose=True
-        )
-        ityp_val, ityp_unit = get_iontypes(range_file_path)
+        if recon_file_path.lower().endswith(
+            (".apt", ".pos")
+        ) and range_file_path.lower().endswith(".rrng"):
+            ityp_info, nochrg_ityp_info, elements = get_ranging_info(
+                range_file_path, verbose=True
+            )
+            ityp_val = get_iontypes(recon_file_path, ityp_info)
+        elif recon_file_path.lower().endswith(
+            ".nxs"
+        ) and range_file_path.lower().endswith(".nxs"):
+            ityp_info, nochrg_ityp_info, elements = get_ranging_info(
+                recon_file_path, verbose=True
+            )
+            ityp_val = get_iontypes(range_file_path)
+        else:
+            raise IOError(
+                f"((APT or POS) and RRNG) or (NXS and NXS) are the only supported combinations!"
+            )
 
         self.init_ranging(ityp_info, elements)
         self.write_init_results()