nothing
' specimen: - type: experiment + method: experiment alias: usa_denton_smith_si identifier: service: undefined identifier: 'Si in specimen' - # is_persistent: false + is_persistent: false description: 'normal
bold
diff --git a/examples/nxapm.schema.archive.yaml b/examples/nxapm.schema.archive.yaml index 1cc7c31..5bc7a76 100644 --- a/examples/nxapm.schema.archive.yaml +++ b/examples/nxapm.schema.archive.yaml @@ -1,5 +1,5 @@ # group, field, and attribute names match to NXapm, for further details -# what each field should contain consult the respective docstring of the +# what each field should contain consult the respective docstring of the # quantity in NXapm definitions: name: 'apm' @@ -11,13 +11,13 @@ definitions: - 'nomad.datamodel.data.EntryData' m_annotations: # Here you can set your default values for the reader and nxdl. - template: + template: reader: apm nxdl: NXapm.nxdl # Listing quantities in the hide component will not show them in the ELN. # This would be useful to make the default values set in `template` fixed. # Leave the hide key even if you want to pass an empty list like in this example. - eln: + eln: hide: ['nxdl', 'reader'] sub_sections: entry: @@ -114,13 +114,13 @@ definitions: One string for each element with statements separated via a single space. The string is expected to have the following format: Symbol value unit +- stdev - + An example: B 1. +- 0.2, means composition of boron 1. at.-% +- 0.2 at.%. If a string contains only a symbol this is interpreted that the symbol specifies the matrix or remainder element for the composition table. - + If unit is omitted or named % this is interpreted as at.-%. Unit can be at% or wt% but all strings have to use either atom or weight percent but no mixtures. @@ -129,7 +129,7 @@ definitions: m_annotations: eln: component: StringEditQuantity - grain_diameter: + grain_diameter: type: np.float64 unit: micrometer description: | @@ -198,14 +198,14 @@ definitions: m_annotations: eln: quantities: - name: + name: type: str description: | - GUID which distinguishes the specimen from all others and especially + GUID which distinguishes the specimen from all others and especially the predecessor/origin from where the specimen was cut. In cases where the specimen was e.g. site-specifically cut from samples or in cases of an instrument session during which multiple - specimens are loaded, the name has to be descriptive enough to + specimens are loaded, the name has to be descriptive enough to resolve which specimen on e.g. the microtip array was taken. This field must not be used for an alias of the specimen. Instead, use short_title. @@ -216,7 +216,7 @@ definitions: # type: str # description: | # Reference to the location of or a GUID providing as many details - # as possible of the material, its microstructure, and its + # as possible of the material, its microstructure, and its # thermo-chemo-mechanical processing/preparation history. # m_annotations: # eln: @@ -526,7 +526,7 @@ definitions: control_software_program: type: str description: | - Name of the control software of the microscope + Name of the control software of the microscope used during acquisition (e.g. IVAS/APSuite). m_annotations: eln: diff --git a/pyproject.toml b/pyproject.toml index 1a1198f..2b57674 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ dependencies = [ "pint==0.17", "pynxtools>=0.7.0", "ifes_apt_tc_data_modeling>=0.2.2", + "numpy<=1.26.4", ] [project.urls] diff --git a/src/pynxtools_apm/concepts/mapping_functors.py b/src/pynxtools_apm/concepts/mapping_functors.py deleted file mode 100644 index 0df17d6..0000000 --- a/src/pynxtools_apm/concepts/mapping_functors.py +++ /dev/null @@ -1,351 +0,0 @@ -# -# Copyright The NOMAD Authors. -# -# This file is part of NOMAD. See https://nomad-lab.eu for further info. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# -"""Utilities for working with NeXus concepts encoded as Python dicts in the concepts dir.""" - -from datetime import datetime - -import flatdict as fd -import numpy as np -import pytz - -from pynxtools_apm.utils.get_file_checksum import get_sha256_of_file_content -from pynxtools_apm.utils.interpret_boolean import try_interpret_as_boolean -from pynxtools_apm.utils.string_conversions import rchop, string_to_number - - -def variadic_path_to_specific_path(path: str, instance_identifier: list): - """Transforms a variadic path to an actual path with instances.""" - if (path is not None) and (path != ""): - narguments = path.count("*") - if narguments == 0: # path is not variadic - return path - if len(instance_identifier) >= narguments: - tmp = path.split("*") - if len(tmp) == narguments + 1: - nx_specific_path = "" - for idx in range(0, narguments): - nx_specific_path += f"{tmp[idx]}{instance_identifier[idx]}" - idx += 1 - nx_specific_path += f"{tmp[-1]}" - return nx_specific_path - return None - - -def add_specific_metadata( - concept_mapping: dict, orgmeta: fd.FlatDict, identifier: list, template: dict -) -> dict: - """Map specific concept src on specific NeXus concept trg. - - concept_mapping: translation dict how trg and src are to be mapped - orgmeta: instance data of src concepts - identifier: list of identifier to resolve variadic paths - template: instance data resulting from a resolved src to trg concept mapping - """ - if "prefix_trg" in concept_mapping: - variadic_prefix_trg = concept_mapping["prefix_trg"] - elif "prefix" in concept_mapping: - variadic_prefix_trg = concept_mapping["prefix"] - else: - raise KeyError(f"Neither prefix nor prefix_trg found in concept_mapping!") - - if "prefix_src" in concept_mapping: - prefix_src = concept_mapping["prefix_src"] - else: - prefix_src = "" - - # process all mapping functors - # (in graphical programming these are also referred to as filters or nodes), i.e. - # an agent that gets some input does some (maybe abstract mapping) and returns an output - # as the mapping can be abstract we call it functor - if "use" in concept_mapping: - for entry in concept_mapping["use"]: - if isinstance(entry, tuple): - if len(entry) == 2: - if isinstance(entry[1], str) and entry[1] == "": - continue - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry[0]}", identifier - ) - template[f"{trg}"] = entry[1] - if "map" in concept_mapping: - for entry in concept_mapping["map"]: - if isinstance(entry, str): - if f"{prefix_src}{entry}" not in orgmeta: - continue - if ( - isinstance(orgmeta[f"{prefix_src}{entry}"], str) - and orgmeta[f"{prefix_src}{entry}"] == "" - ): - continue - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry}", identifier - ) - template[f"{trg}"] = orgmeta[f"{prefix_src}{entry}"] - if isinstance(entry, tuple): - if len(entry) == 2: - if isinstance(entry[0], str): - if f"{prefix_src}{entry[1]}" not in orgmeta: - continue - if (orgmeta[f"{prefix_src}{entry[1]}"], str) and orgmeta[ - f"{prefix_src}{entry[1]}" - ] == "": - continue - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry[0]}", identifier - ) - template[f"{trg}"] = orgmeta[f"{prefix_src}{entry[1]}"] - if "map_to_str" in concept_mapping: - for entry in concept_mapping["map_to_str"]: - if isinstance(entry, str): - if f"{prefix_src}{entry}" not in orgmeta: - continue - if ( - isinstance(orgmeta[f"{prefix_src}{entry}"], str) - and orgmeta[f"{prefix_src}{entry}"] == "" - ): - continue - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry}", identifier - ) - template[f"{trg}"] = orgmeta[f"{prefix_src}{entry}"] - if isinstance(entry, tuple): - if len(entry) == 2: - if all(isinstance(elem, str) for elem in entry): - if f"{prefix_src}{entry[1]}" not in orgmeta: - continue - if ( - isinstance(orgmeta[f"{prefix_src}{entry[1]}"], str) - and orgmeta[f"{prefix_src}{entry[1]}"] == "" - ): - continue - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry[0]}", identifier - ) - template[f"{trg}"] = orgmeta[f"{prefix_src}{entry[1]}"] - if "map_to_bool" in concept_mapping: - for entry in concept_mapping["map_to_bool"]: - if isinstance(entry, str): - if f"{prefix_src}{entry[0]}" not in orgmeta: - continue - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry[0]}", identifier - ) - template[f"{trg}"] = try_interpret_as_boolean( - orgmeta[f"{prefix_src}{entry[0]}"] - ) - if isinstance(entry, tuple): - if len(entry) == 2: - if all(isinstance(elem, str) for elem in entry): - if f"{prefix_src}{entry[1]}" not in orgmeta: - continue - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry[0]}", identifier - ) - template[f"{trg}"] = try_interpret_as_boolean( - orgmeta[f"{prefix_src}{entry[1]}"] - ) - if "map_to_real" in concept_mapping: - for entry in concept_mapping["map_to_real"]: - if isinstance(entry, str): - if isinstance(entry[0], str): - if f"{prefix_src}{entry[0]}" not in orgmeta: - continue - if ( - isinstance(orgmeta[f"{prefix_src}{entry[0]}"], str) - and orgmeta[f"{prefix_src}{entry[0]}"] == "" - ): - continue - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry[0]}", identifier - ) - template[f"{trg}"] = string_to_number( - orgmeta[f"{prefix_src}{entry[0]}"] - ) - if isinstance(entry, tuple): - if len(entry) == 2: - if all(isinstance(elem, str) for elem in entry): - if f"{prefix_src}{entry[1]}" not in orgmeta: - continue - if ( - isinstance(orgmeta[f"{prefix_src}{entry[0]}"], str) - and orgmeta[f"{prefix_src}{entry[1]}"] == "" - ): - continue - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry[0]}", identifier - ) - template[f"{trg}"] = string_to_number( - orgmeta[f"{prefix_src}{entry[1]}"] - ) - elif isinstance(entry[0], str) and isinstance(entry[1], list): - if not all( - ( - isinstance(value, str) - and f"{prefix_src}{value}" in orgmeta - ) - for value in entry[1] - ): - continue - if not all( - ( - isinstance(orgmeta[f"{prefix_src}{value}"], str) - and orgmeta[f"{prefix_src}{value}"] != "" - ) - for value in entry[1] - ): - continue - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry[0]}", identifier - ) - res = [] - for value in entry[1]: - res.append( - string_to_number(orgmeta[f"{prefix_src}{value}"]) - ) - template[f"{trg}"] = np.asarray(res, np.float64) - if "map_to_real_and_multiply" in concept_mapping: - for entry in concept_mapping["map_to_real_and_multiply"]: - if isinstance(entry, tuple): - if len(entry) == 3: - if ( - isinstance(entry[0], str) - and isinstance(entry[1], str) - and isinstance(entry[2], float) - ): - if f"{prefix_src}{entry[1]}" not in orgmeta: - continue - if ( - isinstance(orgmeta[f"{prefix_src}{entry[1]}"], str) - and orgmeta[f"{prefix_src}{entry[1]}"] == "" - ): - continue - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry[0]}", identifier - ) - template[f"{trg}"] = entry[2] * string_to_number( - orgmeta[f"{prefix_src}{entry[1]}"] - ) - if "map_to_real_and_join" in concept_mapping: - for entry in concept_mapping["map_to_real_and_join"]: - if isinstance(entry, tuple): - if len(entry) == 2: - if isinstance(entry[0], str) and isinstance(entry[1], list): - if not all( - ( - isinstance(value, str) - and f"{prefix_src}{value}" in orgmeta - ) - for value in entry[1] - ): - continue - if not all( - ( - isinstance(orgmeta[f"{prefix_src}{value}"], str) - and orgmeta[f"{prefix_src}{value}"] != "" - ) - for value in entry[1] - ): - continue - res = [] - for value in entry[1]: - res.append( - string_to_number(orgmeta[f"{prefix_src}{value}"]) - ) - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry[0]}", identifier - ) - template[f"{trg}"] = np.asarray(res) - # we may need to be more specific with the return datatype here, currently default python float - if "unix_to_iso8601" in concept_mapping: - for entry in concept_mapping["unix_to_iso8601"]: - if isinstance(entry, tuple): - if ( - 2 <= len(entry) <= 3 - ): # trg, src, timestamp or empty string (meaning utc) - if all(isinstance(elem, str) for elem in entry): - if f"{prefix_src}{entry[1]}" not in orgmeta: - continue - if ( - isinstance(orgmeta[f"{prefix_src}{entry[1]}"], str) - and orgmeta[f"{prefix_src}{entry[1]}"] == "" - ): - continue - tzone = "UTC" - if len(entry) == 3: - # if not isinstance(entry[2], str): - # raise TypeError(f"{tzone} needs to be of type string!") - tzone = entry[2] - if tzone not in pytz.all_timezones: - raise ValueError( - f"{tzone} is not a timezone in pytz.all_timezones!" - ) - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry[0]}", identifier - ) - template[f"{trg}"] = datetime.fromtimestamp( - int(orgmeta[f"{prefix_src}{entry[1]}"]), - tz=pytz.timezone(tzone), - ).isoformat() - if "join_str" in concept_mapping: # currently also joining empty strings - for entry in concept_mapping["join_str"]: - if isinstance(entry, tuple): - if len(entry) == 2: - if isinstance(entry[0], str) and isinstance(entry[1], list): - if not all( - ( - isinstance(value, str) - and f"{prefix_src}{value}" in orgmeta - ) - for value in entry[1] - ): - continue - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry[0]}", identifier - ) - res = [] - for value in entry[1]: - res.append(orgmeta[f"{prefix_src}{value}"]) - template[f"{trg}"] = " ".join(res) - if "sha256" in concept_mapping: - for entry in concept_mapping["sha256"]: - if isinstance(entry, tuple): - if len(entry) == 2: - if not all(isinstance(elem, str) for elem in entry): - continue - if f"{prefix_src}{entry[1]}" not in orgmeta: - continue - if orgmeta[f"{prefix_src}{entry[1]}"] == "": - continue - trg = variadic_path_to_specific_path( - f"{variadic_prefix_trg}/{entry[0]}", identifier - ) - try: - with open(orgmeta[f"{prefix_src}{entry[1]}"], "rb") as fp: - template[f"{rchop(trg, 'checksum')}checksum"] = ( - get_sha256_of_file_content(fp) - ) - template[f"{rchop(trg, 'checksum')}type"] = "file" - template[f"{rchop(trg, 'checksum')}path"] = orgmeta[ - f"{prefix_src}{entry[1]}" - ] - template[f"{rchop(trg, 'checksum')}algorithm"] = "sha256" - except (FileNotFoundError, IOError): - print( - f"File {orgmeta[f'''{prefix_src}{entry[1]}''']} not found !" - ) - return template diff --git a/src/pynxtools_apm/concepts/mapping_functors_pint.py b/src/pynxtools_apm/concepts/mapping_functors_pint.py new file mode 100644 index 0000000..73e0119 --- /dev/null +++ b/src/pynxtools_apm/concepts/mapping_functors_pint.py @@ -0,0 +1,521 @@ +# +# Copyright The NOMAD Authors. +# +# This file is part of NOMAD. See https://nomad-lab.eu for further info. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +"""Utilities for working with NeXus concepts encoded as Python dicts in the concepts dir.""" + +from datetime import datetime +from typing import Any, Dict + +import flatdict as fd +import numpy as np +import pytz + +from pynxtools_apm.utils.get_file_checksum import get_sha256_of_file_content +from pynxtools_apm.utils.interpret_boolean import try_interpret_as_boolean +from pynxtools_apm.utils.pint_custom_unit_registry import is_not_special_unit, ureg +from pynxtools_apm.utils.string_conversions import rchop + +# best practice is use np.ndarray or np.generic as magnitude within that ureg.Quantity! +MAP_TO_DTYPES: Dict[str, type] = { + "u1": np.uint8, + "i1": np.int8, + "u2": np.uint16, + "i2": np.int16, + # "f2": np.float16, not supported yet with all HDF5 h5py versions + "u4": np.uint32, + "i4": np.int32, + "f4": np.float32, + "u8": np.uint64, + "i8": np.int64, + "f8": np.float64, + "bool": bool, +} + +# general conversion workflow +# 1. Normalize src data to str, bool, or ureg.Quantity +# These ureg.Quantities should use numpy scalar or array for the dtype of the magnitude. +# Use special NeXus unit categories unitless, dimensionless, and any. +# 2. Map on specific trg path, ureg.Unit, eventually with conversions, and dtype conversion +# Later this could include endianness +# 3. Store ureg.Quantity magnitude and if non-special also correctly converted @units +# attribute + + +def var_path_to_spcfc_path(path: str, instance_identifier: list): + """Transforms a variadic path to an actual path with instances.""" + if (path is not None) and (path != ""): + nvariadic_parts = path.count("*") + if nvariadic_parts == 0: # path is not variadic + return path + if len(instance_identifier) >= nvariadic_parts: + variadic_part = path.split("*") + if len(variadic_part) == nvariadic_parts + 1: + nx_specific_path = "" + for idx in range(0, nvariadic_parts): + nx_specific_path += ( + f"{variadic_part[idx]}{instance_identifier[idx]}" + ) + idx += 1 + nx_specific_path += f"{variadic_part[-1]}" + return nx_specific_path + + +def get_case(arg): + """Identify which case an instruction from the configuration belongs to. + Each case comes with specific instructions to resolve that are detailed + in the README.md in this source code directory.""" + if isinstance(arg, str): # str + return "case_one" + elif isinstance(arg, tuple): + if len(arg) == 2: # str, str | list + if isinstance(arg[0], str): + if isinstance(arg[1], str): + return "case_two_str" + elif isinstance(arg[1], list): + return "case_two_list" + elif len(arg) == 3: # str, str | list, ureg.Unit or str, ureg.Unit, str | list + if isinstance(arg[0], str): + if isinstance(arg[1], ureg.Unit): + if isinstance(arg[2], str): + return "case_three_str" + elif isinstance(arg[2], list): + return "case_three_list" + elif (arg[2], ureg.Unit): + if isinstance(arg[1], str): + return "case_four_str" + elif isinstance(arg[1], list): + return "case_four_list" + elif len(arg) == 4: + # str, ureg.Unit, str | list, ureg.Unit + # str, ureg.Unit, str, str + # last string points to unit string for situations where e.g. tech partner + # report HV/value, HV/Unit and these two pieces of information should be + # fused into a ureg.Quantity with target ureg.Unit given as second argument + if ( + isinstance(arg[0], str) + and isinstance(arg[1], ureg.Unit) + and isinstance(arg[3], ureg.Unit) + ): + if isinstance(arg[2], str): + return "case_five_str" + elif isinstance(arg[2], list): + return "case_five_list" + elif ( + isinstance(arg[0], str) + and isinstance(arg[1], ureg.Unit) + and isinstance(arg[2], str) + and isinstance(arg[3], str) + ): + return "case_six" + + +def map_to_dtype(trg_dtype: str, value: Any) -> Any: + # can this be done more elegantly, i.e. written more compact? + # yes I already tried MAP_TO_DTYPE[trg_dtype](value) but mypy does not like it + # error: Argument 1 has incompatible type "generic | bool | int | float | complex | + # str | bytes | memoryview"; expected "str | bytes | SupportsIndex" [arg-type] + if np.shape(value) != (): + if trg_dtype in MAP_TO_DTYPES: + if trg_dtype != "bool": + return np.asarray(value, MAP_TO_DTYPES[trg_dtype]) + else: + if hasattr(value, "dtype"): + if value.dtype is bool: + return np.asarray(value, bool) + else: + raise TypeError( + f"map_to_dtype, hitting unexpected case for array bool !" + ) + else: + raise ValueError(f"map_to_dtype, hitting unexpected case for array !") + else: + if trg_dtype in MAP_TO_DTYPES: + if trg_dtype != "bool": + that_type = MAP_TO_DTYPES[trg_dtype] + return that_type(value) + else: + return try_interpret_as_boolean(value) + else: + raise ValueError(f"map_to_dtype, hitting unexpected case for scalar !") + + +def set_value(template: dict, trg: str, src_val: Any, trg_dtype: str = "") -> dict: + """Set value in the template using trg. + + src_val can be a single value, an array, or a ureg.Quantity (scalar or array) + """ + # np.issubdtype(np.uint32, np.signedinteger) + if not trg_dtype: # go with existent dtype + if isinstance(src_val, str): + # TODO this is not rigorous need to check for null-term also and str arrays + template[f"{trg}"] = src_val + # assumes I/O to HDF5 will write specific encoding, typically variable, null-term, utf8 + elif isinstance(src_val, ureg.Quantity): + if isinstance(src_val.magnitude, (np.ndarray, np.generic)) or np.isscalar( + src_val.magnitude + ): # bool case typically not expected! + template[f"{trg}"] = src_val.magnitude + if is_not_special_unit(src_val.units): + template[f"{trg}/@units"] = f"{src_val.units}" + print( + f"WARNING::Assuming writing to HDF5 will auto-convert Python types to numpy type, trg {trg} !" + ) + else: + raise TypeError( + f"ureg.Quantity magnitude should use in-build, bool, or np !" + ) + elif isinstance(src_val, list): + if all(isinstance(val, str) for val in src_val): + template[f"{trg}"] = ", ".join(src_val) + else: + raise TypeError( + f"Not List[str] {type(src_val)} found for not trg_dtype case !" + ) + elif ( + isinstance(src_val, (np.ndarray, np.generic)) + or np.isscalar(src_val) + or isinstance(src_val, bool) + ): + template[f"{trg}"] = np.asarray(src_val) + # units may be required, need to be set explicitly elsewhere in the source code! + print( + f"WARNING::Assuming writing to HDF5 will auto-convert Python types to numpy type, trg: {trg} !" + ) + else: + raise TypeError( + f"Unexpected type {type(src_val)} found for not trg_dtype case !" + ) + else: # do an explicit type conversion + # e.g. in cases when tech partner writes float32 but e.g. NeXus assumes float64 + if isinstance(src_val, (str, bool)): + template[f"{trg}"] = try_interpret_as_boolean(src_val) + elif isinstance(src_val, ureg.Quantity): + if isinstance(src_val.magnitude, (np.ndarray, np.generic)): + template[f"{trg}"] = map_to_dtype(trg_dtype, src_val.magnitude) + if is_not_special_unit(src_val.units): + template[f"{trg}/@units"] = f"{src_val.units}" + elif np.isscalar(src_val.magnitude): # bool typically not expected + template[f"{trg}"] = map_to_dtype(trg_dtype, src_val.magnitude) + if is_not_special_unit(src_val.units): + template[f"{trg}/@units"] = f"{src_val.units}" + else: + raise TypeError( + f"Unexpected type for explicit src_val.magnitude, set_value, trg {trg} !" + ) + elif isinstance(src_val, (list, np.ndarray, np.generic)): + template[f"{trg}"] = map_to_dtype(trg_dtype, np.asarray(src_val)) + # units may be required, need to be set explicitly elsewhere in the source code! + print( + f"WARNING::Assuming I/O to HDF5 will auto-convert to numpy type, trg: {trg} !" + ) + elif np.isscalar(src_val): + template[f"{trg}"] = map_to_dtype(trg_dtype, src_val) + print( + f"WARNING::Assuming I/O to HDF5 will auto-convert to numpy type, trg: {trg} !" + ) + else: + raise TypeError( + f"Unexpected type for explicit type conversion, set_value, trg {trg} !" + ) + return template + + +def use_functor( + cmds: list, mdata: fd.FlatDict, prfx_trg: str, ids: list, template: dict +) -> dict: + """Process concept mapping for simple predefined strings and pint quantities.""" + for cmd in cmds: + if isinstance(cmd, tuple): + if len(cmd) == 2: + if isinstance(cmd[0], str): + if isinstance(cmd[1], (str, ureg.Quantity, bool)): + # str, str or str, ureg or str, bool + trg = var_path_to_spcfc_path(f"{prfx_trg}/{cmd[0]}", ids) + set_value(template, trg, cmd[1]) + return template + + +def map_functor( + cmds: list, + mdata: fd.FlatDict, + prfx_src: str, + prfx_trg: str, + ids: list, + template: dict, + trg_dtype_key: str = "", +) -> dict: + """Process concept mapping, datatype and unit conversion for quantities.""" + for cmd in cmds: + case = get_case(cmd) + if case == "case_one": # str + src_val = mdata.get(f"{prfx_src}{cmd}") + if src_val is not None and src_val != "": + trg = var_path_to_spcfc_path(f"{prfx_trg}/{cmd}", ids) + set_value(template, trg, src_val, trg_dtype_key) + elif case == "case_two_str": # str, str + src_val = mdata.get(f"{prfx_src}{cmd[1]}") + if src_val is not None and src_val != "": + trg = var_path_to_spcfc_path(f"{prfx_trg}/{cmd[0]}", ids) + set_value(template, trg, src_val, trg_dtype_key) + elif case == "case_two_list": + # ignore empty list, all src paths str, all src_val have to exist of same type + if len(cmd[1]) == 0: + continue + if not all(isinstance(val, str) for val in cmd[1]): + continue + if not all(f"{prfx_src}{val}" in mdata for val in cmd[1]): + continue + src_values = [mdata[f"{prfx_src}{val}"] for val in cmd[1]] + if len(src_values) == 0: + continue + if not all(src_val is not None and src_val != "" for src_val in src_values): + continue + if not all(type(val) is type(src_values[0]) for val in src_values): + continue + trg = var_path_to_spcfc_path(f"{prfx_trg}/{cmd[0]}", ids) + set_value(template, trg, src_values, trg_dtype_key) + elif case == "case_three_str": # str, ureg.Unit, str + src_val = mdata.get(f"{prfx_src}{cmd[2]}") + if not src_val: + continue + trg = var_path_to_spcfc_path(f"{prfx_trg}/{cmd[0]}", ids) + if isinstance(src_val, ureg.Quantity): + set_value(template, trg, src_val.to(cmd[1]), trg_dtype_key) + else: + set_value( + template, trg, ureg.Quantity(src_val, cmd[1].units), trg_dtype_key + ) + elif case == "case_three_list": # str, ureg.Unit, list + if len(cmd[2]) == 0: + continue + if not all(isinstance(val, str) for val in cmd[2]): + continue + if not all(f"{prfx_src}{val}" in mdata for val in cmd[2]): + continue + src_values = [mdata[f"{prfx_src}{val}"] for val in cmd[2]] + if not all(src_val is not None and src_val != "" for src_val in src_values): + continue + if not all(type(val) is type(src_values[0]) for val in src_values): + # need to check whether content are scalars also + continue + trg = var_path_to_spcfc_path(f"{prfx_trg}/{cmd[0]}", ids) + if isinstance(src_values, ureg.Quantity): + set_value(template, trg, src_values, trg_dtype_key) + else: + # potentially a list of ureg.Quantities with different scaling + normalize = [] + for val in src_values: + if isinstance(val, ureg.Quantity): + normalize.append(val.to(cmd[1]).magnitude) + else: + raise TypeError( + "Unimplemented case for {val} in case_three_list !" + ) + set_value( + template, + trg, + ureg.Quantity(normalize, cmd[1]), + trg_dtype_key, + ) + elif case.startswith("case_four"): + # both of these cases can be avoided in an implementation when the + # src quantity is already a pint quantity instead of some + # pure python or numpy value or array respectively + raise ValueError( + f"Hitting unimplemented case_four, instead refactor implementation such" + f"that values on the src side are pint.Quantities already!" + ) + elif case == "case_five_str": + src_val = mdata.get(f"{prfx_src}{cmd[2]}") + if not src_val: + continue + trg = var_path_to_spcfc_path(f"{prfx_trg}/{cmd[0]}", ids) + if isinstance(src_val, ureg.Quantity): + set_value(template, trg, src_val.to(cmd[1]), trg_dtype_key) + else: + pint_src = ureg.Quantity(src_val, cmd[3]) + set_value(template, trg, pint_src.to(cmd[1]), trg_dtype_key) + elif case == "case_five_list": + if len(cmd[2]) == 0: + continue + if not all(isinstance(val, str) for val in cmd[2]): + continue + if not all(f"{prfx_src}{val}" in mdata for val in cmd[2]): + continue + src_values = [mdata[f"{prfx_src}{val}"] for val in cmd[2]] + if not all(src_val is not None and src_val != "" for src_val in src_values): + continue + if isinstance(src_values[0], ureg.Quantity): + raise ValueError( + f"Hit unimplemented case that src_val is ureg.Quantity" + ) + if not all(type(val) is type(src_values[0]) for val in src_values): + continue + trg = var_path_to_spcfc_path(f"{prfx_trg}/{cmd[0]}", ids) + if isinstance(src_values, ureg.Quantity): + set_value(template, trg, src_values.to(cmd[1]), trg_dtype_key) + else: + pint_src = ureg.Quantity(src_values, cmd[3]) + set_value(template, trg, pint_src.to(cmd[1]), trg_dtype_key) + elif case == "case_six": + if f"{prfx_src}{cmd[2]}" not in mdata or f"{prfx_src}{cmd[3]}" not in mdata: + continue + src_val = mdata[f"{prfx_src}{cmd[2]}"] + src_unit = mdata[f"{prfx_src}{cmd[3]}"] + if not src_val or not src_unit: + continue + trg = var_path_to_spcfc_path(f"{prfx_trg}/{cmd[0]}", ids) + if isinstance(src_val, ureg.Quantity): + set_value(template, trg, src_val.units.to(cmd[1]), trg_dtype_key) + else: + pint_src = ureg.Quantity(src_val, ureg.Unit(src_unit)) + set_value(template, trg, pint_src.to(cmd[1]), trg_dtype_key) + return template + + +def timestamp_functor( + cmds: list, + mdata: fd.FlatDict, + prfx_src: str, + prfx_trg: str, + ids: list, + template: dict, +) -> dict: + """Process concept mapping and time format conversion.""" + for cmd in cmds: + if isinstance(cmd, tuple): + if 2 <= len(cmd) <= 3: # trg, src, timestamp or empty string (meaning utc) + if all(isinstance(elem, str) for elem in cmd): + if f"{prfx_src}{cmd[1]}" not in mdata: + continue + if mdata[f"{prfx_src}{cmd[1]}"] == "": + continue + tzone = "UTC" + if len(cmd) == 3: + tzone = cmd[2] + if tzone not in pytz.all_timezones: + raise ValueError( + f"{tzone} is not a timezone in pytz.all_timezones!" + ) + var_path_to_spcfc_path + trg = var_path_to_spcfc_path(f"{prfx_trg}/{cmd[0]}", ids) + template[f"{trg}"] = datetime.fromtimestamp( + int(mdata[f"{prfx_src}{cmd[1]}"]), + tz=pytz.timezone(tzone), + ).isoformat() + return template + + +def filehash_functor( + cmds: list, + mdata: fd.FlatDict, + prfx_src: str, + prfx_trg: str, + ids: list, + template: dict, +) -> dict: + """Process concept mapping and checksums to add context from which file NeXus content was processed.""" + for cmd in cmds: + if isinstance(cmd, tuple): + if len(cmd) == 2: + if not all(isinstance(elem, str) for elem in cmd): + continue + if f"{prfx_src}{cmd[1]}" not in mdata: + continue + if mdata[f"{prfx_src}{cmd[1]}"] == "": + continue + trg = var_path_to_spcfc_path(f"{prfx_trg}/{cmd[0]}", ids) + try: + with open(mdata[f"{prfx_src}{cmd[1]}"], "rb") as fp: + template[f"{rchop(trg, 'checksum')}checksum"] = ( + get_sha256_of_file_content(fp) + ) + template[f"{rchop(trg, 'checksum')}type"] = "file" + template[f"{rchop(trg, 'checksum')}path"] = mdata[ + f"{prfx_src}{cmd[1]}" + ] + template[f"{rchop(trg, 'checksum')}algorithm"] = "sha256" + except (FileNotFoundError, IOError): + print(f"File {mdata[f'''{prfx_src}{cmd[1]}''']} not found !") + return template + + +def add_specific_metadata_pint( + cfg: dict, mdata: fd.FlatDict, ids: list, template: dict +) -> dict: + """Map specific concept src on specific NeXus concept trg. + + cfg: a configuration dictionary from configurations/*.py mapping from src to trg + mdata: instance data of src concepts + ids: list of identifier to resolve variadic template paths to specific template paths + template: dictionary where to store mapped instance data using template paths + """ + if "prefix_trg" in cfg: + prefix_trg = cfg["prefix_trg"] + else: + raise KeyError(f"prefix_trg not found in cfg!") + if "prefix_src" in cfg: + if isinstance(cfg["prefix_src"], str): + prfx_src = [cfg["prefix_src"]] + elif isinstance(cfg["prefix_src"], list) and all( + isinstance(val, str) for val in cfg["prefix_src"] + ): + prfx_src = cfg["prefix_src"] + else: + raise ValueError(f"prefix_src needs to be a str or a list[str] !") + else: + raise KeyError(f"prefix_src not found in cfg!") + + # process all mapping functors + # (in graphical programming these are also referred to as filters or nodes), + # i.e. an agent that gets some input does something (e.g. abstract mapping) and + # returns an output, given the mapping can be abstract, we call it a functor + + # https://numpy.org/doc/stable/reference/arrays.dtypes.html + for prefix_src in prfx_src: + for functor_key in cfg: + if functor_key in ["prefix_trg", "prefix_src"]: + continue + if functor_key == "use": + use_functor(cfg["use"], mdata, prefix_trg, ids, template) + if functor_key == "map": + map_functor( + cfg[functor_key], mdata, prefix_src, prefix_trg, ids, template + ) + if functor_key.startswith("map_to_"): + dtype_key = functor_key.replace("map_to_", "") + if dtype_key in MAP_TO_DTYPES: + map_functor( + cfg[functor_key], + mdata, + prefix_src, + prefix_trg, + ids, + template, + dtype_key, + ) + else: + raise KeyError(f"Unexpected dtype_key {dtype_key} !") + if functor_key == "unix_to_iso8601": + timestamp_functor( + cfg["unix_to_iso8601"], mdata, prefix_src, prefix_trg, ids, template + ) + if functor_key == "sha256": + filehash_functor( + cfg["sha256"], mdata, prefix_src, prefix_trg, ids, template + ) + return template diff --git a/src/pynxtools_apm/config/eln_cfg.py b/src/pynxtools_apm/configurations/eln_cfg.py similarity index 58% rename from src/pynxtools_apm/config/eln_cfg.py rename to src/pynxtools_apm/configurations/eln_cfg.py index 73c6f91..a3c4c0f 100644 --- a/src/pynxtools_apm/config/eln_cfg.py +++ b/src/pynxtools_apm/configurations/eln_cfg.py @@ -17,10 +17,12 @@ # """Dict mapping custom schema instances from eln_data.yaml file on concepts in NXapm.""" +from pynxtools_apm.utils.pint_custom_unit_registry import ureg + APM_ENTRY_TO_NEXUS = { "prefix_trg": "/ENTRY[entry*]", "prefix_src": "entry/", - "map_to_str": [ + "map": [ "run_number", "operation_mode", "start_time", @@ -35,33 +37,48 @@ "prefix_trg": "/ENTRY[entry*]/sample", "prefix_src": "sample/", "map": [ - ("grain_diameter", "grain_diameter/value"), - ("grain_diameter_error", "grain_diameter_error/value"), - ("heat_treatment_temperature", "heat_treatment_temperature/value"), - ("heat_treatment_temperature_error", "heat_treatment_temperature_error/value"), - ("heat_treatment_quenching_rate", "heat_treatment_quenching_rate/value"), - ( - "heat_treatment_quenching_rate_error", - "heat_treatment_quenching_rate_error/value", - ), - ], - "map_to_str": [ "alias", "description", - "type", + ("type", "method"), ("identifier/identifier", "identifier/identifier"), ("identifier/service", "identifier/service"), - ("identifier/is_persistent", "identifier/is_persistent"), - ("grain_diameter/@units", "grain_diameter/unit"), - ("grain_diameter_error/@units", "grain_diameter/unit"), - ("heat_treatment_temperature/@units", "heat_treatment_temperature/unit"), + ], + "map_to_bool": [("identifier/is_persistent", "identifier/is_persistent")], + "map_to_f8": [ + ( + "grain_diameter", + ureg.micrometer, + "grain_diameter/value", + "grain_diameter/unit", + ), ( - "heat_treatment_temperature_error/@units", + "grain_diameter_error", + ureg.micrometer, + "grain_diameter_error/value", + "grain_diameter_error/unit", + ), + ( + "heat_treatment_temperature", + ureg.degC, + "heat_treatment_temperature/value", + "heat_treatment_temperature/unit", + ), + ( + "heat_treatment_temperature_error", + ureg.degC, + "heat_treatment_temperature_error/value", "heat_treatment_temperature_error/unit", ), - ("heat_treatment_quenching_rate/@units", "heat_treatment_quenching_rate/unit"), ( - "heat_treatment_quenching_rate_error/@units", + "heat_treatment_quenching_rate", + ureg.kelvin / ureg.second, + "heat_treatment_quenching_rate/value", + "heat_treatment_quenching_rate/unit", + ), + ( + "heat_treatment_quenching_rate_error", + ureg.kelvin / ureg.second, + "heat_treatment_quenching_rate_error/value", "heat_treatment_quenching_rate_error/unit", ), ], @@ -72,20 +89,26 @@ "prefix_trg": "/ENTRY[entry*]/specimen", "prefix_src": "specimen/", "map": [ - ("initial_radius", "initial_radius/value"), - ("shank_angle", "shank_angle/value"), - ], - "map_to_bool": ["is_polycrystalline", "is_amorphous"], - "map_to_str": [ "alias", "preparation_date", "description", - "type", + ("type", "method"), ("identifier/identifier", "identifier/identifier"), ("identifier/service", "identifier/service"), + ], + "map_to_f8": [ + ( + "initial_radius", + ureg.nanometer, + "initial_radius/value", + "initial_radius/unit", + ), + ("shank_angle", ureg.degree, "shank_angle/value", "shank_angle/unit"), + ], + "map_to_bool": [ + "is_polycrystalline", + "is_amorphous", ("identifier/is_persistent", "identifier/is_persistent"), - ("initial_radius/@units", "initial_radius/unit"), - ("shank_angle/@units", "shank_angle/unit"), ], } @@ -93,38 +116,57 @@ APM_INSTRUMENT_STATIC_TO_NEXUS = { "prefix_trg": "/ENTRY[entry*]/measurement/instrument", "prefix_src": "instrument/", - "map": [("analysis_chamber/flight_path", "nominal_flight_path/value")], - "map_to_str": [ + "map": [ "status", "instrument_name", "location", - ("FABRICATION[fabrication]/vendor", "fabrication_vendor"), - ("FABRICATION[fabrication]/model", "fabrication_model"), - ("FABRICATION[fabrication]/identifier", "fabrication_identifier"), + ("fabrication/vendor", "fabrication_vendor"), + ("fabrication/model", "fabrication_model"), + ("fabrication/identifier/identifier", "fabrication_identifier"), ("reflectron/status", "reflectron_status"), ("local_electrode/name", "local_electrode_name"), ("pulser/pulse_mode", "pulser/pulse_mode"), - ("analysis_chamber/flight_path/@units", "nominal_flight_path/unit"), + ], + "map_to_f8": [ + ( + "analysis_chamber/flight_path", + ureg.meter, + "nominal_flight_path/value", + "nominal_flight_path/unit", + ) ], } APM_INSTRUMENT_DYNAMIC_TO_NEXUS = { - "prefix_trg": "/ENTRY[entry*]/measurement/event_data_apm_set/EVENT_DATA_APM[event_data_apm]/instrument", + "prefix_trg": "/ENTRY[entry*]/measurement/event_data_apm_set/event_data_apm/instrument", "prefix_src": "instrument/", "use": [("control/target_detection_rate/@units", "ions/pulse")], "map": [ + "pulser_pulse_mode", + ("control/evaporation_control", "evaporation_control"), + ], + "map_to_f8": [ ("control/target_detection_rate", "target_detection_rate"), - ("pulser/pulse_frequency", "pulser/pulse_frequency/value"), + ( + "pulser/pulse_frequency", + ureg.kilohertz, + "pulser/pulse_frequency/value", + "pulser/pulse_frequency/unit", + ), ("pulser/pulse_fraction", "pulser/pulse_fraction"), - ("analysis_chamber/chamber_pressure", "chamber_pressure/value"), - ("stage_lab/base_temperature", "base_temperature/value"), - ], - "map_to_str": [ - ("control/evaporation_control", "evaporation_control"), - ("pulser/pulse_frequency/@units", "pulser/pulse_frequency/unit"), - ("analysis_chamber/chamber_pressure/@units", "chamber_pressure/unit"), - ("stage_lab/base_temperature/@units", "base_temperature/unit"), + ( + "analysis_chamber/chamber_pressure", + ureg.bar, + "chamber_pressure/value", + "chamber_pressure/unit", + ), + ( + "stage_lab/base_temperature", + ureg.kelvin, + "base_temperature/value", + "base_temperature/unit", + ), ], } @@ -132,7 +174,7 @@ APM_RANGE_TO_NEXUS = { "prefix_trg": "/ENTRY[entry*]/atom_probe/ranging", "prefix_src": "ranging/", - "map_to_str": [ + "map": [ ("programID[program1]/program", "program"), ("programID[program1]/program/@version", "program_version"), ], @@ -142,14 +184,15 @@ APM_RECON_TO_NEXUS = { "prefix_trg": "/ENTRY[entry*]/atom_probe/reconstruction", "prefix_src": "reconstruction/", - "map": [("field_of_view", "field_of_view/value")], - "map_to_str": [ + "map": [ "protocol_name", "crystallographic_calibration", "parameter", ("programID[program1]/program", "program"), ("programID[program1]/program/@version", "program_version"), - ("field_of_view/@units", "field_of_view/unit"), + ], + "map_to_f8": [ + ("field_of_view", ureg.centimeter, "field_of_view/value", "field_of_view/unit") ], } @@ -158,8 +201,8 @@ "prefix_trg": "/ENTRY[entry*]/atom_probe", "prefix_src": "workflow/", "sha256": [ - ("raw_data/SERIALIZED[serialized]/checksum", "raw_dat_file"), - ("hit_finding/SERIALIZED[serialized]/checksum", "hit_dat_file"), + ("raw_data/serialized/checksum", "raw_dat_file"), + ("hit_finding/serialized/checksum", "hit_dat_file"), ("reconstruction/config/checksum", "recon_cfg_file"), ], } @@ -170,7 +213,8 @@ APM_USER_TO_NEXUS = { "prefix_trg": "/ENTRY[entry*]/USER[user*]", - "map_to_str": [ + "prefix_src": "", + "map": [ "name", "affiliation", "address", @@ -185,11 +229,8 @@ APM_IDENTIFIER_TO_NEXUS = { "prefix_trg": "/ENTRY[entry*]/USER[user*]", - "use": [ - ("IDENTIFIER[identifier]/is_persistent", True), - ("IDENTIFIER[identifier]/service", "orcid"), - ], - "map_to_str": [ - ("IDENTIFIER[identifier]/identifier", "orcid"), - ], + "prefix_src": "", + "use": [("identifier/service", "orcid")], + "map": [("identifier/identifier", "orcid")], + "map_to_bool": [("identifier/is_persistent", "identifier/is_persistent")], } diff --git a/src/pynxtools_apm/config/oasis_cfg.py b/src/pynxtools_apm/configurations/oasis_cfg.py similarity index 96% rename from src/pynxtools_apm/config/oasis_cfg.py rename to src/pynxtools_apm/configurations/oasis_cfg.py index ce878a1..bdab673 100644 --- a/src/pynxtools_apm/config/oasis_cfg.py +++ b/src/pynxtools_apm/configurations/oasis_cfg.py @@ -42,18 +42,20 @@ APM_OASISCONFIG_TO_NEXUS = { "prefix_trg": "/ENTRY[entry*]", + "prefix_src": "", "use": [ ( "start_time", f"{dt.datetime.now(dt.timezone.utc).isoformat().replace('+00:00', 'Z')}", ), ], - "map_to_str": [("operation_mode")], + "map": ["operation_mode"], } APM_CSYS_MCSTASLIKE_TO_NEXUS = { "prefix_trg": "/ENTRY[entry*]/coordinate_system_set/COORDINATE_SYSTEM[coordinate_system]", + "prefix_src": "", "use": [ ( "alias", @@ -86,5 +88,6 @@ APM_EXAMPLE_TO_NEXUS = { "prefix_trg": "/ENTRY[entry*]/CITE[cite*]", - "map_to_str": [("authors"), ("doi"), ("description"), ("url")], + "prefix_src": "", + "map": ["authors", "doi", "description", "url"], } diff --git a/src/pynxtools_apm/reader.py b/src/pynxtools_apm/reader.py index c33f7ff..bc133d9 100644 --- a/src/pynxtools_apm/reader.py +++ b/src/pynxtools_apm/reader.py @@ -149,7 +149,7 @@ def read( # considered non-filled in template instance data and are thus not copied over # print("Reporting state of template before passing to HDF5 writing...") - # for keyword in template.keys(): + # for keyword in template: # print(f"keyword: {keyword}, template[keyword]: {template[keyword]}") # exit(1) diff --git a/src/pynxtools_apm/utils/create_nx_default_plots.py b/src/pynxtools_apm/utils/create_nx_default_plots.py index 9a2013a..cd58ebb 100644 --- a/src/pynxtools_apm/utils/create_nx_default_plots.py +++ b/src/pynxtools_apm/utils/create_nx_default_plots.py @@ -20,10 +20,10 @@ import numpy as np from pynxtools_apm.utils.versioning import ( - NX_APM_EXEC_NAME, - NX_APM_EXEC_VERSION, MASS_SPECTRUM_DEFAULT_BINNING, NAIVE_GRID_DEFAULT_VOXEL_SIZE, + NX_APM_EXEC_NAME, + NX_APM_EXEC_VERSION, ) @@ -200,14 +200,14 @@ def apm_default_plot_generator(template: dict, entry_id: int) -> dict: trg = f"/ENTRY[entry{entry_id}]/atom_probe/mass_to_charge_conversion/mass_to_charge" if trg in template: if isinstance(template[trg], dict): - if "compress" in template[trg].keys(): + if "compress" in template[trg]: if isinstance(template[trg]["compress"], np.ndarray): has_valid_m_z = True has_valid_xyz = False trg = f"/ENTRY[entry{entry_id}]/atom_probe/reconstruction/reconstructed_positions" if trg in template: if isinstance(template[trg], dict): - if "compress" in template[trg].keys(): + if "compress" in template[trg]: if isinstance(template[trg]["compress"], np.ndarray): has_valid_xyz = True print(f"m_z, xyz: {has_valid_m_z}, {has_valid_xyz}") diff --git a/src/pynxtools_apm/utils/interpret_boolean.py b/src/pynxtools_apm/utils/interpret_boolean.py index a3f75ec..13763ec 100644 --- a/src/pynxtools_apm/utils/interpret_boolean.py +++ b/src/pynxtools_apm/utils/interpret_boolean.py @@ -29,8 +29,10 @@ } -def try_interpret_as_boolean(arg: str) -> bool: +def try_interpret_as_boolean(arg) -> bool: """Try to interpret a human string statement if boolean be strict.""" + if isinstance(arg, bool): + return arg if arg.lower() in HUMAN_BOOLEAN_STATEMENT: return HUMAN_BOOLEAN_STATEMENT[arg.lower()] raise KeyError( diff --git a/src/pynxtools_apm/utils/io_case_logic.py b/src/pynxtools_apm/utils/io_case_logic.py index 2d6bcba..55b5b14 100644 --- a/src/pynxtools_apm/utils/io_case_logic.py +++ b/src/pynxtools_apm/utils/io_case_logic.py @@ -19,12 +19,8 @@ from typing import Dict, List, Tuple -from pynxtools_apm.concepts.mapping_functors import ( - variadic_path_to_specific_path, -) -from pynxtools_apm.utils.get_file_checksum import ( - get_sha256_of_file_content, -) +from pynxtools_apm.concepts.mapping_functors_pint import var_path_to_spcfc_path +from pynxtools_apm.utils.get_file_checksum import get_sha256_of_file_content VALID_FILE_NAME_SUFFIX_RECON = [".apt", ".pos", ".epos", ".ato", ".csv", ".h5"] VALID_FILE_NAME_SUFFIX_RANGE = [ @@ -143,7 +139,7 @@ def report_workflow(self, template: dict, entry_id: int) -> dict: # populate automatically input-files used # rely on assumption made in check_validity_of_file_combination for fpath in self.reconstruction: - prfx = variadic_path_to_specific_path( + prfx = var_path_to_spcfc_path( "/ENTRY[entry*]/atom_probe/reconstruction/results", identifier ) with open(fpath, "rb") as fp: @@ -152,7 +148,7 @@ def report_workflow(self, template: dict, entry_id: int) -> dict: template[f"{prfx}/type"] = "file" template[f"{prfx}/algorithm"] = "SHA256" for fpath in self.ranging: - prfx = variadic_path_to_specific_path( + prfx = var_path_to_spcfc_path( "/ENTRY[entry*]/atom_probe/ranging/definitions", identifier ) with open(fpath, "rb") as fp: diff --git a/src/pynxtools_apm/utils/load_ranging.py b/src/pynxtools_apm/utils/load_ranging.py index 7ce759d..eb78eb6 100644 --- a/src/pynxtools_apm/utils/load_ranging.py +++ b/src/pynxtools_apm/utils/load_ranging.py @@ -269,7 +269,7 @@ def update_atom_types_ranging_definitions_based(self, template: dict) -> dict: """Update the atom_types list in the specimen based on ranging defs.""" number_of_ion_types = 1 prefix = f"/ENTRY[entry{self.meta['entry_id']}]/atom_probe/ranging/" - if f"{prefix}number_of_ion_types" in template.keys(): + if f"{prefix}number_of_ion_types" in template: number_of_ion_types = template[f"{prefix}number_of_ion_types"] print( f"Auto-detecting elements from ranging {number_of_ion_types} ion types..." @@ -283,7 +283,7 @@ def update_atom_types_ranging_definitions_based(self, template: dict) -> dict: ) for ion_id in np.arange(1, number_of_ion_types): trg = f"{prefix}ionID[ion{ion_id}]/nuclide_list" - if trg in template.keys(): + if trg in template: nuclide_list = template[trg][:, 1] # second row of NXion/nuclide_list yields atom number to decode element for atom_number in nuclide_list: diff --git a/src/pynxtools_apm/utils/oasis_config_reader.py b/src/pynxtools_apm/utils/oasis_config_reader.py index b9b1a3d..244ff9b 100644 --- a/src/pynxtools_apm/utils/oasis_config_reader.py +++ b/src/pynxtools_apm/utils/oasis_config_reader.py @@ -17,15 +17,16 @@ # """Load deployment-specific quantities.""" +import pathlib + import flatdict as fd import yaml -import pathlib -from pynxtools_apm.concepts.mapping_functors import add_specific_metadata -from pynxtools_apm.config.oasis_cfg import ( - APM_OASISCONFIG_TO_NEXUS, +from pynxtools_apm.concepts.mapping_functors_pint import add_specific_metadata_pint +from pynxtools_apm.configurations.oasis_cfg import ( APM_CSYS_MCSTASLIKE_TO_NEXUS, APM_EXAMPLE_TO_NEXUS, + APM_OASISCONFIG_TO_NEXUS, ) @@ -55,13 +56,15 @@ def __init__(self, file_path: str, entry_id: int, verbose: bool = False): def parse_various(self, template: dict) -> dict: """Copy data from configuration applying mapping functors.""" identifier = [self.entry_id] - add_specific_metadata(APM_OASISCONFIG_TO_NEXUS, self.yml, identifier, template) + add_specific_metadata_pint( + APM_OASISCONFIG_TO_NEXUS, self.yml, identifier, template + ) return template def parse_reference_frames(self, template: dict) -> dict: """Copy data from configuration applying mapping functors.""" identifier = [self.entry_id] - add_specific_metadata( + add_specific_metadata_pint( APM_CSYS_MCSTASLIKE_TO_NEXUS, self.yml, identifier, template ) return template @@ -78,7 +81,7 @@ def parse_example(self, template: dict) -> dict: if cite_dict == {}: continue identifier = [self.entry_id, cite_id] - add_specific_metadata( + add_specific_metadata_pint( APM_EXAMPLE_TO_NEXUS, fd.FlatDict(cite_dict), identifier, diff --git a/src/pynxtools_apm/utils/oasis_eln_reader.py b/src/pynxtools_apm/utils/oasis_eln_reader.py index 9988214..6005768 100644 --- a/src/pynxtools_apm/utils/oasis_eln_reader.py +++ b/src/pynxtools_apm/utils/oasis_eln_reader.py @@ -17,26 +17,26 @@ # """Wrapping multiple parsers for vendor files with NOMAD Oasis/ELN/YAML metadata.""" -import flatdict as fd -import yaml import pathlib +import flatdict as fd +import yaml from ase.data import chemical_symbols -from pynxtools_apm.config.eln_cfg import ( + +from pynxtools_apm.concepts.mapping_functors_pint import add_specific_metadata_pint +from pynxtools_apm.configurations.eln_cfg import ( APM_ENTRY_TO_NEXUS, - APM_SAMPLE_TO_NEXUS, - APM_SPECIMEN_TO_NEXUS, - APM_INSTRUMENT_STATIC_TO_NEXUS, + APM_IDENTIFIER_TO_NEXUS, APM_INSTRUMENT_DYNAMIC_TO_NEXUS, + APM_INSTRUMENT_STATIC_TO_NEXUS, APM_RANGE_TO_NEXUS, APM_RECON_TO_NEXUS, - APM_WORKFLOW_TO_NEXUS, + APM_SAMPLE_TO_NEXUS, + APM_SPECIMEN_TO_NEXUS, APM_USER_TO_NEXUS, - APM_IDENTIFIER_TO_NEXUS, + APM_WORKFLOW_TO_NEXUS, ) - from pynxtools_apm.utils.parse_composition_table import parse_composition_table -from pynxtools_apm.concepts.mapping_functors import add_specific_metadata class NxApmNomadOasisElnSchemaParser: @@ -57,33 +57,24 @@ class NxApmNomadOasisElnSchemaParser: during the verification of the template dictionary. """ - def __init__(self, file_path: str, entry_id: int, verbose: bool = False): + def __init__(self, file_path: str = "", entry_id: int = 1, verbose: bool = False): print(f"Extracting data from ELN file: {file_path}") - if ( - pathlib.Path(file_path).name.endswith("eln_data.yaml") - or pathlib.Path(file_path).name.endswith("eln_data.yml") - ) and entry_id > 0: - self.entry_id = entry_id + if pathlib.Path(file_path).name.endswith("eln_data.yaml") or pathlib.Path( + file_path + ).name.endswith("eln_data.yml"): self.file_path = file_path + self.entry_id = entry_id if entry_id > 0 else 1 + self.verbose = verbose + try: with open(self.file_path, "r", encoding="utf-8") as stream: self.yml = fd.FlatDict(yaml.safe_load(stream), delimiter="/") - if verbose: + if self.verbose: for key, val in self.yml.items(): print(f"key: {key}, value: {val}") - else: - self.entry_id = 1 - self.file_path = "" - self.yml = {} - - def parse_entry(self, template: dict) -> dict: - identifier = [self.entry_id] - add_specific_metadata(APM_ENTRY_TO_NEXUS, self.yml, identifier, template) - return template - - def parse_sample(self, template: dict) -> dict: - identifier = [self.entry_id] - add_specific_metadata(APM_SAMPLE_TO_NEXUS, self.yml, identifier, template) - return template + except (FileNotFoundError, IOError): + print(f"File {self.file_path} not found !") + self.yml = fd.FlatDict({}, delimiter="/") + return def parse_sample_composition(self, template: dict) -> dict: """Interpret human-readable ELN input to generate consistent composition table.""" @@ -126,23 +117,33 @@ def parse_sample_composition(self, template: dict) -> dict: ion_id += 1 return template - def parse_specimen(self, template: dict) -> dict: - identifier = [self.entry_id] - add_specific_metadata(APM_SPECIMEN_TO_NEXUS, self.yml, identifier, template) - return template - - def parse_instrument_static(self, template: dict) -> dict: - identifier = [self.entry_id] - add_specific_metadata( - APM_INSTRUMENT_STATIC_TO_NEXUS, self.yml, identifier, template - ) - return template - - def parse_instrument_dynamic(self, template: dict) -> dict: - identifier = [self.entry_id] - add_specific_metadata( - APM_INSTRUMENT_DYNAMIC_TO_NEXUS, self.yml, identifier, template - ) + def parse_user(self, template: dict) -> dict: + """Copy data from user section into template.""" + src = "user" + if src in self.yml: + if isinstance(self.yml[src], list): + if all(isinstance(entry, dict) for entry in self.yml[src]) is True: + user_id = 1 + # custom schema delivers a list of dictionaries... + for user_dict in self.yml[src]: + if user_dict == {}: + continue + identifier = [self.entry_id, user_id] + add_specific_metadata_pint( + APM_USER_TO_NEXUS, + user_dict, + identifier, + template, + ) + if "orcid" not in user_dict: + continue + add_specific_metadata_pint( + APM_IDENTIFIER_TO_NEXUS, + user_dict, + identifier, + template, + ) + user_id += 1 return template def parse_pulser_source(self, template: dict) -> dict: @@ -153,7 +154,7 @@ def parse_pulser_source(self, template: dict) -> dict: return template src = "instrument/pulser/laser_source" - if src in self.yml.keys(): + if src in self.yml: if isinstance(self.yml[src], list): if all(isinstance(entry, dict) for entry in self.yml[src]) is True: laser_id = 1 @@ -161,20 +162,20 @@ def parse_pulser_source(self, template: dict) -> dict: for ldct in self.yml[src]: trg_sta = ( f"/ENTRY[entry{self.entry_id}]/measurement/" - f"instrument/pulser/SOURCE[source{laser_id}]" + f"instrument/pulser/sourceID[source{laser_id}]" ) if "name" in ldct: template[f"{trg_sta}/name"] = ldct["name"] - quantities = ["wavelength"] - for qnt in quantities: - if ("value" in ldct[qnt]) and ("unit" in ldct[qnt]): + qnt = "wavelength" + if qnt in ldct: + if "value" in ldct[qnt] and "unit" in ldct[qnt]: template[f"{trg_sta}/{qnt}"] = ldct[qnt]["value"] template[f"{trg_sta}/{qnt}/@units"] = ldct[qnt]["unit"] trg_dyn = ( f"/ENTRY[entry{self.entry_id}]/measurement/" - f"event_data_apm_set/EVENT_DATA_APM[event_data_apm]/" - f"instrument/pulser/SOURCE[source{laser_id}]" + f"event_data_apm_set/event_data_apm/instrument/" + f"pulser/sourceID[source{laser_id}]" ) quantities = ["power", "pulse_energy"] for qnt in quantities: @@ -189,61 +190,21 @@ def parse_pulser_source(self, template: dict) -> dict: print("WARNING: pulse_mode != voltage but no laser details specified!") return template - def parse_user(self, template: dict) -> dict: - """Copy data from user section into template.""" - src = "user" - if src in self.yml: - if isinstance(self.yml[src], list): - if all(isinstance(entry, dict) for entry in self.yml[src]) is True: - user_id = 1 - # custom schema delivers a list of dictionaries... - for user_dict in self.yml[src]: - if user_dict == {}: - continue - identifier = [self.entry_id, user_id] - add_specific_metadata( - APM_USER_TO_NEXUS, - fd.FlatDict(user_dict), - identifier, - template, - ) - if "orcid" not in user_dict: - continue - add_specific_metadata( - APM_IDENTIFIER_TO_NEXUS, - fd.FlatDict(user_dict), - identifier, - template, - ) - user_id += 1 - return template - - def parse_range(self, template: dict) -> dict: - identifier = [self.entry_id] - add_specific_metadata(APM_RANGE_TO_NEXUS, self.yml, identifier, template) - return template - - def parse_recon(self, template: dict) -> dict: - identifier = [self.entry_id] - add_specific_metadata(APM_RECON_TO_NEXUS, self.yml, identifier, template) - return template - - def parse_workflow(self, template: dict) -> dict: - identifier = [self.entry_id] - add_specific_metadata(APM_WORKFLOW_TO_NEXUS, self.yml, identifier, template) - return template - def parse(self, template: dict) -> dict: """Copy data from self into template the appdef instance.""" - self.parse_entry(template) - self.parse_sample(template) self.parse_sample_composition(template) - self.parse_specimen(template) - self.parse_instrument_static(template) - self.parse_instrument_dynamic(template) - self.parse_pulser_source(template) self.parse_user(template) - self.parse_range(template) - self.parse_recon(template) - self.parse_workflow(template) + self.parse_pulser_source(template) + identifier = [self.entry_id] + for cfg in [ + APM_ENTRY_TO_NEXUS, + APM_SAMPLE_TO_NEXUS, + APM_SPECIMEN_TO_NEXUS, + APM_INSTRUMENT_STATIC_TO_NEXUS, + APM_INSTRUMENT_DYNAMIC_TO_NEXUS, + APM_RANGE_TO_NEXUS, + APM_RECON_TO_NEXUS, + APM_WORKFLOW_TO_NEXUS, + ]: + add_specific_metadata_pint(cfg, self.yml, identifier, template) return template diff --git a/src/pynxtools_apm/utils/pint_custom_unit_registry.py b/src/pynxtools_apm/utils/pint_custom_unit_registry.py new file mode 100644 index 0000000..42fbffd --- /dev/null +++ b/src/pynxtools_apm/utils/pint_custom_unit_registry.py @@ -0,0 +1,43 @@ +# +# Copyright The NOMAD Authors. +# +# This file is part of NOMAD. See https://nomad-lab.eu for further info. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +"""A customized unit registry for handling units with pint.""" + +import pint +from pint import UnitRegistry + +ureg = UnitRegistry() +# ureg.formatter.default_format = "D" +# https://pint.readthedocs.io/en/stable/user/formatting.html + + +# customizations for NeXus +ureg.define("nx_unitless = 1") +ureg.define("nx_dimensionless = 1") +ureg.define("nx_any = 1") + +NX_UNITLESS = ureg.Quantity(1, ureg.nx_unitless) +NX_DIMENSIONLESS = ureg.Quantity(1, ureg.nx_dimensionless) +NX_ANY = ureg.Quantity(1, ureg.nx_any) + + +def is_not_special_unit(units: pint.Unit) -> bool: + """True if not a special NeXus unit category.""" + for special_units in [NX_UNITLESS.units, NX_DIMENSIONLESS.units, NX_ANY.units]: + if units == special_units: + return False + return True diff --git a/tests/run_tests.ipynb b/tests/run_tests.ipynb index 0d278b2..735954c 100644 --- a/tests/run_tests.ipynb +++ b/tests/run_tests.ipynb @@ -2,18 +2,10 @@ "cells": [ { "cell_type": "code", - "execution_count": 21, + "execution_count": null, "id": "fc9345cb-46bf-4df3-9dac-1bde062d9020", "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "/home/kaiobach/Research/hu_hu_hu/sprint24/pynxtools-apm/pynxtools_apm/tests\n" - ] - } - ], + "outputs": [], "source": [ "import os\n", "print(os.getcwd())" @@ -29,45 +21,35 @@ }, { "cell_type": "code", - "execution_count": 22, + "execution_count": null, "id": "48f6e922-52d3-412f-8345-0641d33e676e", "metadata": {}, "outputs": [], "source": [ "tests = {\n", " # typical full tests with eln_data.yaml apm.oasis.specific.yaml Si.apt 87D_1.rng\n", - " # \"eln\": [(\"eln_data.yaml\", \"apm.oasis.specific.yaml\")],\n", - " # \"apt\": [\"Si.apt\"],\n", - " # \"csv\": [\"Annealed_CoCrNi_100.csv\"],\n", - " # \"env\": [\"ErMnO.env\"],\n", - " # \"epos\": [\"R45_04472-v03.epos\"],\n", - " # \"fig\": [\"Superalloy_MassSpec_ranged.fig.txt\"],\n", - " # \"imago\": [\"default.analysis\"],\n", - " # \"pos\": [\"ErMnO_pole.pos\"],\n", - " # \"pyccapt\": [(\"1748_Al_range_.h5\", \"1748_Al.h5\")],\n", - " # \"rng\": [\"87D_1.rng\"],\n", - " # \"rrng\": [\"VAlN_film_plan-view_700C.rrng\"]\n", + " \"eln\": [(\"eln_data.yaml\", \"apm.oasis.specific.yaml\")],\n", + " \"apt\": [\"Si.apt\"],\n", + " \"csv\": [\"Annealed_CoCrNi_100.csv\"],\n", + " \"env\": [\"ErMnO.env\"],\n", + " \"epos\": [\"R45_04472-v03.epos\"],\n", + " \"fig\": [\"Superalloy_MassSpec_ranged.fig.txt\"],\n", + " \"imago\": [\"default.analysis\"],\n", + " \"pos\": [\"ErMnO_pole.pos\"],\n", + " \"pyccapt\": [(\"1748_Al_range_.h5\", \"1748_Al.h5\")],\n", + " \"rng\": [\"87D_1.rng\"],\n", + " \"rrng\": [\"VAlN_film_plan-view_700C.rrng\"]\n", "}" ] }, { "cell_type": "code", - "execution_count": 23, + "execution_count": null, "id": "6ae5057c-2101-4792-9804-3af046003db7", "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "/home/kaiobach/Research/hu_hu_hu/sprint24/pynxtools-apm/.py3.12.4/bin/python\n" - ] - } - ], + "outputs": [], "source": [ - "# ! mkdir -p log\n", - "# ! mkdir -p prod\n", - "# ! which python" + "! mkdir -p log && mkdir -p prod && which python" ] }, { @@ -80,19 +62,10 @@ }, { "cell_type": "code", - "execution_count": 24, + "execution_count": null, "id": "809a88fe-5e4f-4429-b668-8a7956210606", "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "Running test apt/Si.apt\n", - "Ran all tests\n" - ] - } - ], + "outputs": [], "source": [ "for parser_type, list_of_tests in tests.items():\n", " for entry in list_of_tests:\n",