implementing nxdict.py

Author: Bing Li

src/tavi/data/nxdict.py

@@ -0,0 +1,147 @@
+from datetime import datetime
+from typing import Optional
+
+
+def nxsource(spicelogs, instrument_config_params):
+    source = {
+        "attrs": {"NX_class": "NXsource", "EX_required": "true"},
+        "name": {
+            "attrs": {"type": "NX_CHAR", "EX_required": "true"},
+            "dataset": "HFIR",
+        },
+        "probe": {
+            "attrs": {"type": "NX_CHAR", "EX_required": "true"},
+            "dataset": "neutron",
+        },
+    }
+    # Effective distance from sample Distance as seen by radiation from sample.
+    # This number should be negative to signify that it is upstream of the sample.
+    # nxsource.attrs["distance"] = -0.0
+    return source
+
+
+def nxmono(spicelogs, instrument_config_params):
+    metadata = spicelogs["attrs"]
+
+    try:
+        mono = {
+            "attrs": {"NX_class": "NXcrystal", "EX_required": "true"},
+            "ei": {
+                "dataset": spicelogs["ei"],
+                "attrs": {"type": "NX_FLOAT", "EX_required": "true", "units": "meV"},
+            },
+            "type": {"dataset": metadata["monochromator"], "attrs": {"type": "NX_CHAR"}},
+            "sense": {"dataset": metadata["sense"][0], "attrs": {"type": "NX_CHAR"}},
+            "m1": {
+                "dataset": spicelogs["m1"],
+                "attrs": {"type": "NX_FLOAT", "units": "degrees"},
+            },
+            "m2": {
+                "dataset": spicelogs["m2"],
+                "attrs": {"type": "NX_FLOAT", "units": "degrees"},
+            },
+            "mfocus": {"dataset": spicelogs["mfocus"], "attrs": {"type": "NX_FLOAT"}},
+            "marc": {"dataset": spicelogs["marc"], "attrs": {"type": "NX_FLOAT"}},
+            "mtrans": {"dataset": spicelogs["mtrans"], "attrs": {"type": "NX_FLOAT"}},
+            "focal_length": {"dataset": spicelogs["focal_length"], "attrs": {"type": "NX_FLOAT"}},
+        }
+    except KeyError:
+        pass
+    return mono
+
+
+def nxcoll(spicelogs, instrument_config_params):
+    return {}
+
+
+def nxana(spicelogs, instrument_config_params):
+    return {}
+
+
+def nxdet(spicelogs, instrument_config_params):
+    return {}
+
+
+def nxmonitor(spicelogs, instrument_config_params):
+    return {}
+
+
+def nxsample(spicelogs, sample_config_params):
+    return {}
+
+
+def nxslit(spicelogs, instrument_config_params):
+    return {}
+
+
+def nxflipper(spicelogs, instrument_config_params):
+    return {}
+
+
+def _spicelogs_to_nexus(spicelogs, instrument_config_params, sample_config_params):
+    metadata = spicelogs["attrs"]
+    # TODO timezone
+    start_date_time = "{} {}".format(metadata["date"], metadata["time"])
+    start_time = datetime.strptime(start_date_time, "%m/%d/%Y %I:%M:%S %p").isoformat()
+    # if "end_time" in das_logs.attrs:  # last scan never finished
+    end_date_time = metadata["end_time"]
+    end_time = datetime.strptime(end_date_time, "%m/%d/%Y %I:%M:%S %p").isoformat()
+
+    scan_dict = {
+        "attrs": {
+            "NX_class": "NXentry",
+            "EX_required": "true",
+        },
+        "SPICElogs": spicelogs,
+        "definition": {
+            "attrs": {"EX_required": "true", "type": "NX_CHAR"},
+            "dataset": "NXtas",
+        },
+        "title": {
+            "attrs": {"EX_required": "true", "type": "NX_CHAR"},
+            "dataset": metadata["scan_title"],
+        },
+        "start_time": {
+            "attrs": {"EX_required": "true", "type": "NX_DATE_TIME"},
+            "dataset": start_time,
+        },
+        "end_time": {
+            "attrs": {"type": "NX_DATE_TIME"},
+            "dataset": end_time,
+        },
+        "instrument": {
+            "attrs": {"EX_required": "true", "NX_class": "NXinstrument"},
+            "name": {"attrs": {"type": "NX_CHAR"}, "dataset": metadata["instrument"]},
+            "source": nxsource(spicelogs, instrument_config_params),
+            "collimator": nxcoll(spicelogs, instrument_config_params),
+            "monochromator": nxmono(spicelogs, instrument_config_params),
+            "analyser": nxana(spicelogs, instrument_config_params),
+            "detector": nxdet(spicelogs, instrument_config_params),
+            "slit": nxslit(spicelogs, instrument_config_params),
+            "flipper": nxflipper(spicelogs, instrument_config_params),
+        },
+        "monitor": nxmonitor(spicelogs, instrument_config_params),
+        "sample": nxsample(spicelogs, sample_config_params),
+        "data": {},
+    }
+    return scan_dict
+
+
+def daslogs_to_nexus_dict(
+    daslogs: dict,
+    instrument_config_params: Optional[str],
+    sample_config_params: Optional[str],
+) -> dict:
+    """Format DASlogs dict into NeXus dict"""
+
+    match (das_key := tuple(daslogs.keys())[0]):
+        case "SPICElogs":
+            scan_dict = _spicelogs_to_nexus(
+                daslogs["SPICElogs"],
+                instrument_config_params,
+                sample_config_params,
+            )
+        case _:
+            raise KeyError(f"Unrecogonized DASlogs key {das_key}.")
+
+    return scan_dict

src/tavi/data/nxentry.py

@@ -1,5 +1,9 @@
+from typing import Optional
+
 import h5py
 
+from tavi.data.spice_reader import spice_data_reader
+
 
 class NexusEntry(dict):
     """Read and write NeXus data files.
@@ -58,14 +62,14 @@ def _write_recursively(items, nexus_entry):
                         dv = value["dataset"]
                         if isinstance(dv, str):
                             dv = dv.encode("utf-8")
-                        ds = nexus_entry.create_dataset(
-                            name=key,
-                            data=dv,
-                            maxshape=None,
-                        )
+                        if key in nexus_entry:
+                            ds = nexus_entry[key]
+                            ds[...] = dv
+                        else:
+                            ds = nexus_entry.create_dataset(name=key, data=dv, maxshape=None)
                         NexusEntry._write_recursively(value, ds)
                     else:
-                        grp = nexus_entry.create_group(key + "/")
+                        grp = nexus_entry.require_group(key + "/")
                         NexusEntry._write_recursively(value, grp)
 
     @staticmethod
@@ -93,42 +97,72 @@ def _read_recursively(nexus_entry, items=None):
 
         return items
 
+    @staticmethod
+    def _dict_to_nexus_entry(nexus_dict):
+        """convert nested dict to instances of the NexusEntry class"""
+        nexus_entries = {}
+        for scan_num, scan_content in nexus_dict.items():
+            content_list = []
+            for key, val in scan_content.items():
+                content_list.append((key, val))
+            nexus_entries.update({scan_num: NexusEntry(content_list)})
+        return nexus_entries
+
+    # TODO read in instrument and sample configuratio json files
     @classmethod
-    def from_spice(cls, path_to_spice_folder: str) -> dict:
+    def from_spice(
+        cls,
+        path_to_spice_folder: str,
+        scan_num: Optional[int] = None,
+        path_to_instrument_json: Optional[str] = None,
+        path_to_sample_json: Optional[str] = None,
+    ) -> dict:
         """return a NexusEntry instance from loading a SPICE file
 
         Args:
             path_to_spice_folder (str): path to a SPICE folder
+            scan_num (int): read all scans in folder if not None
+            path_to_instrument_json: Optional[str] = None,
+            path_to_sample_json: Optional[str] = None,
         """
+        # TODO validate path
 
-        nexus_dict = {}
-        return cls([(key, val) for key, val in nexus_dict.items()])
+        nexus_dict = spice_data_reader(
+            path_to_spice_folder,
+            scan_num,
+            path_to_instrument_json,
+            path_to_sample_json,
+        )
+
+        return NexusEntry._dict_to_nexus_entry(nexus_dict)
 
     @classmethod
-    def from_nexus(cls, path_to_nexus: str) -> dict:
+    def from_nexus(cls, path_to_nexus: str, scan_num: Optional[int] = None) -> dict:
         """return a NexusEntry instance from loading a NeXus file
 
         Args:
             path_to_nexus (str): path to a NeXus file with the extension .h5
+            scan_num (int): read all scans in file if not None
         """
         with h5py.File(path_to_nexus, "r") as nexus_file:
-            nexus_dict = NexusEntry._read_recursively(nexus_file)
-        return cls([(key, val) for key, val in nexus_dict.items()])
+            if scan_num is None:  # read all scans in file
+                nexus_dict = NexusEntry._read_recursively(nexus_file)
+            else:  # read one scan only
+                key = f"scan{scan_num:04}"
+                nexus_dict = {key: NexusEntry._read_recursively(nexus_file[key])}
+
+        return NexusEntry._dict_to_nexus_entry(nexus_dict)
 
-    # TODO
     def to_nexus(self, path_to_nexus: str, name="scan") -> None:
         """write a NexueEntry instance to a NeXus file
 
         Args:
             path_to_nexus (str): path to a NeXus file with the extention .h5
         """
         with h5py.File(path_to_nexus, "a") as nexus_file:
-            # TODO what if a group alreay exsit??
-            pass
-            scan_grp = nexus_file.create_group(name + "/")
+            scan_grp = nexus_file.require_group(name + "/")
             NexusEntry._write_recursively(self, scan_grp)
 
-    # TODO need to catch errors when multiple IPTS are loaded in a tavi file
     def get(self, key, ATTRS=False, default=None):
         """
         Return dataset spicified by key regardless of the hierarchy.

src/tavi/data/scan_old/spice_to_nexus.py

@@ -216,7 +216,7 @@ def _nexus_mono(nxmono, das_logs, instrument_config_params=None):
         nxmono["ei"].attrs["units"] = "meV"
 
         nxmono.create_dataset(name="type", data=das_logs.attrs["monochromator"], maxshape=None)
-        nxmono.attrs["type"] = "NX_CHAR"
+        nxmono["type"].attrs["type"] = "NX_CHAR"
 
         nxmono.create_dataset(name="m1", data=das_logs["m1"], maxshape=None)
         nxmono["m1"].attrs["type"] = "NX_FLOAT"