implemeting nxdict group

Author: Bing Li

src/tavi/data/nxdict.py

@@ -2,6 +2,30 @@
 from typing import Optional
 
 
+def _add_dataset_entry(
+    entry_dict: dict,
+    key: str,
+    daslogs: dict,
+    daslogs_key: Optional[str] = None,
+    **kwargs,
+):
+    key = key if daslogs_key is None else daslogs_key
+    try:
+        val = daslogs[key]
+        entry_dict.update({key: {"dataset": val}})
+        if not kwargs:
+            return entry_dict
+        attr_dict = {}
+        for k, v in kwargs.items():
+            attr_dict.update({k: v})
+        entry_dict[key].update({"attrs": attr_dict})
+        return entry_dict
+
+    except KeyError:
+        print(f"Variable {key} cannot be found in DAS logs.")
+        return entry_dict
+
+
 def nxsource(spicelogs, instrument_config_params):
     source = {
         "attrs": {"NX_class": "NXsource", "EX_required": "true"},
@@ -21,49 +45,74 @@ def nxsource(spicelogs, instrument_config_params):
 
 
 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
+    mono = {"attrs": {"NX_class": "NXcrystal", "EX_required": "true"}}
+    mono = _add_dataset_entry(mono, key="ei", daslogs=spicelogs, type="NX_FLOAT", EX_required="true", unit="meV")
+    mono.update({"type": {"dataset": spicelogs["attrs"]["monochromator"], "attrs": {"type": "NX_CHAR"}}})
+    mono.update({"sense": {"dataset": spicelogs["attrs"]["sense"][0], "attrs": {"type": "NX_CHAR"}}})
+    mono = _add_dataset_entry(mono, key="m1", daslogs=spicelogs, type="NX_FLOAT", unit="degrees")
+    mono = _add_dataset_entry(mono, key="m2", daslogs=spicelogs, type="NX_FLOAT", unit="degrees")
+    mono = _add_dataset_entry(mono, key="mfocus", daslogs=spicelogs, type="NX_FLOAT")
+    mono = _add_dataset_entry(mono, key="marc", daslogs=spicelogs, type="NX_FLOAT")
+    mono = _add_dataset_entry(mono, key="mtrans", daslogs=spicelogs, type="NX_FLOAT")
+    mono = _add_dataset_entry(mono, key="focal_length", daslogs=spicelogs, type="NX_FLOAT")
+
     return mono
 
 
 def nxcoll(spicelogs, instrument_config_params):
-    return {}
+    pass
 
 
 def nxana(spicelogs, instrument_config_params):
-    return {}
+    ana = {"attrs": {"NX_class": "NXcrystal", "EX_required": "true"}}
+    ana = _add_dataset_entry(ana, key="ef", daslogs=spicelogs, type="NX_FLOAT", EX_required="true", unit="meV")
+    ana.update({"type": {"dataset": spicelogs["attrs"]["analyzer"], "attrs": {"type": "NX_CHAR"}}})
+    ana.update({"sense": {"dataset": spicelogs["attrs"]["sense"][2], "attrs": {"type": "NX_CHAR"}}})
+    ana = _add_dataset_entry(ana, key="a1", daslogs=spicelogs, type="NX_FLOAT", unit="degrees")
+    ana = _add_dataset_entry(ana, key="a2", daslogs=spicelogs, type="NX_FLOAT", unit="degrees")
+    ana = _add_dataset_entry(ana, key="afocus", daslogs=spicelogs, type="NX_FLOAT")
+    for i in range(8):
+        ana = _add_dataset_entry(ana, key=f"qm{i+1}", daslogs=spicelogs, type="NX_FLOAT")
+        ana = _add_dataset_entry(ana, key=f"xm{i+1}", daslogs=spicelogs, type="NX_FLOAT")
+    return ana
 
 
 def nxdet(spicelogs, instrument_config_params):
-    return {}
+    det = {"attrs": {"NX_class": "NXdetector", "EX_required": "true"}}
+    det = _add_dataset_entry(det, key="detector", daslogs=spicelogs, type="NX_INT", EX_required="true", unit="counts")
+    # polar_angle
+    return det
 
 
 def nxmonitor(spicelogs, instrument_config_params):
-    return {}
+    monitor = {"attrs": {"NX_class": "NXmonitor", "EX_required": "true"}}
+    preset_type = spicelogs["attrs"]["preset_type"]
+    match preset_type:
+        case "countfile":  # polarization data
+            print("Countfile preset type is not supported.")
+
+        case "normal":
+            preset_channel = spicelogs["attrs"]["preset_channel"]
+            monitor.update({"mode": {"dataset": preset_channel, "attrs": {"type": "NX_CHAR", "EX_required": "true"}}})
+            monitor.update(
+                {
+                    "preset": {
+                        "dataset": spicelogs["attrs"]["preset_value"],
+                        "attrs": {"type": "NX_FLOAT", "EX_required": "true"},
+                    }
+                }
+            )
+            monitor = _add_dataset_entry(monitor, key="time", daslogs=spicelogs, type="NX_FLOAT", units="seconds")
+            monitor = _add_dataset_entry(monitor, key="monitor", daslogs=spicelogs, type="NX_INT", units="counts")
+            monitor = _add_dataset_entry(monitor, key="mcu", daslogs=spicelogs, type="NX_FLOAT")
+            monitor_data_dict = monitor[preset_channel]
+            monitor.update({"data": monitor_data_dict})
+
+        case _:
+            print(f"Unrecogonized preset type {preset_type}.")
+
+    return monitor
 
 
 def nxsample(spicelogs, sample_config_params):

src/tavi/data/nxentry.py

@@ -1,6 +1,7 @@
 from typing import Optional
 
 import h5py
+import numpy as np
 
 from tavi.data.spice_reader import spice_data_reader
 
@@ -50,6 +51,27 @@ def _write_recursively(items, nexus_entry):
         string encoded with utf-8
         """
 
+        def format_dataset(value):
+            """format if type is given in attributes"""
+            dv = value["dataset"]
+
+            if attr := value.get("attrs"):
+                attr_type = attr.get("type")
+                match attr_type:
+                    case "NX_CHAR":
+                        dv = dv.encode("utf-8")
+                    case "NX_FLOAT":
+                        dv = np.array(dv).astype("float")
+                    case "NX_INT":
+                        dv = np.array(dv).astype("int")
+                    case "NX_DATE_TIME":
+                        pass
+                    case _:
+                        if isinstance(dv, str):
+                            dv = dv.encode("utf-8")
+            print(dv)
+            return dv
+
         for key, value in items.items():
             if key == "attrs":
                 for attr_key, attr_value in value.items():
@@ -59,13 +81,11 @@ def _write_recursively(items, nexus_entry):
             else:
                 if isinstance(value, dict):
                     if "dataset" in value.keys():
-                        dv = value["dataset"]
-                        if isinstance(dv, str):
-                            dv = dv.encode("utf-8")
-                        if key in nexus_entry:
+                        dv = format_dataset(value)
+                        if key in nexus_entry:  # dataset exists
                             ds = nexus_entry[key]
                             ds[...] = dv
-                        else:
+                        else:  # create dataset
                             ds = nexus_entry.create_dataset(name=key, data=dv, maxshape=None)
                         NexusEntry._write_recursively(value, ds)
                     else:

src/tavi/data/scan_old/spice_to_nexus.py

@@ -514,7 +514,7 @@ def _nexus_coll(nxcoll, das_logs, instrument_config_params=None):
         div_x = [float(v) for v in list(das_logs.attrs["collimation"].split("-"))]
         nxcoll.create_dataset(name="divergence_x", data=div_x, maxshape=None)
         nxcoll["divergence_x"].attrs["type"] = "NX_ANGLE"
-        nxcoll["divergence_x"].attrs["units"] = "minutes of aarc"
+        nxcoll["divergence_x"].attrs["units"] = "minutes of arc"
 
     except KeyError:
         pass