implementing ScanGroup class

src/tavi/convert_spice_to_nexus.py

@@ -33,5 +33,5 @@ def convert_spice_to_nexus(
 if __name__ == "__main__":
     path_to_spice_folder = "./test_data/exp424"
     path_to_spice_folder = "./test_data/exp815"  # empty runs in exp815
-    path_to_spice_folder = "./test_data/exp813"
+    # path_to_spice_folder = "./test_data/exp813"
     convert_spice_to_nexus(path_to_spice_folder)

src/tavi/data/nxentry.py

@@ -226,23 +226,26 @@ def to_nexus(self, path_to_nexus: str, name="scan") -> None:
             scan_grp = nexus_file.require_group(name + "/")
             NexusEntry._write_recursively(self, scan_grp)
             if "data" in scan_grp:  # create soft link for data
-                def_y = scan_grp["data"].attrs["signal"]
-                def_x = scan_grp["data"].attrs["axes"]
-                path_y = _find_val_path(def_y, nexus_file)
-                path_x = _find_val_path(def_x, nexus_file)
-
-                if (path_y is not None) and (scan_grp.get(path_y) is not None):
-                    def_y = "data/" + def_y
-                    if isinstance(scan_grp.get(def_y), h5py.Dataset):
-                        del scan_grp[def_y]
-                    scan_grp[def_y] = h5py.SoftLink(path_y)
-                    scan_grp[def_y + "/"].attrs["target"] = path_y
-                if (path_x is not None) and (scan_grp.get(path_x) is not None):
-                    def_x = "data/" + def_x
-                    if isinstance(scan_grp.get(def_x), h5py.Dataset):
-                        del scan_grp[def_x]
-                    scan_grp[def_x] = h5py.SoftLink(path_x)
-                    scan_grp[def_x + "/"].attrs["target"] = path_x
+                scan_grp_data = scan_grp["data"]
+                def_y = scan_grp_data.attrs["signal"]
+                def_x = scan_grp_data.attrs["axes"]
+                path_y = _find_val_path(def_y, scan_grp)
+                path_x = _find_val_path(def_x, scan_grp)
+
+                if path_y is not None:
+                    if scan_grp.get(path_y[1:]) is not None:  # remove the first "/"
+                        if isinstance(scan_grp_data.get(def_y), h5py.Dataset):
+                            del scan_grp_data[def_y]
+                        path_y = "/" + name + path_y
+                        scan_grp_data[def_y] = h5py.SoftLink(path_y)
+                        scan_grp_data[def_y].attrs["target"] = path_y
+                if path_x is not None:
+                    if scan_grp.get(path_x[1:]) is not None:  # remove the first "/"
+                        if isinstance(scan_grp_data.get(def_x), h5py.Dataset):
+                            del scan_grp_data[def_x]
+                        path_x = "/" + name + path_x
+                        scan_grp_data[def_x] = h5py.SoftLink(path_x)
+                        scan_grp_data[def_x].attrs["target"] = path_x
 
             # Create the ATTRIBUTES
             scan_grp.attrs["file_name"] = os.path.abspath(path_to_nexus)

src/tavi/data/plotter.py

@@ -5,6 +5,18 @@
 
 
 class Plot1D(object):
+    """1D Plot class
+
+    Attributes:
+        x (np.ndarray): x values
+        y (np.ndarray): y values
+        xerr (np.ndarray | None): x error bars
+        yerr (np.ndarray | None): y error bars
+
+
+    Methods:
+
+    """
 
     def __init__(
         self,
@@ -20,17 +32,42 @@ def __init__(
         self.yerr = yerr
 
         self.title: str = ""
-        self.xlim: Optional[tuple[float, float]] = None
-        self.ylim: Optional[tuple[float, float]] = None
         self.xlabel: Optional[str] = None
         self.ylabel: Optional[str] = None
         self.label: Optional[str] = None
-
+        # plot specifications
+        self.xlim: Optional[tuple[float, float]] = None
+        self.ylim: Optional[tuple[float, float]] = None
         self.color = "C0"
         self.fmt = "o"
         self.LOG_X = False
         self.LOG_Y = False
 
+    def make_labels(
+        self,
+        x_str: str,
+        y_str: str,
+        norm_channel: Optional[str],
+        norm_val: float,
+        scan_info,
+    ):
+        """Create axes labels, plot title and curve label"""
+        if norm_channel is not None:
+            if norm_channel == "time":
+                norm_channel_str = "seconds"
+            else:
+                norm_channel_str = norm_channel
+            if norm_val == 1:
+                self.ylabel = y_str + "/ " + norm_channel_str
+            else:
+                self.ylabel = y_str + f" / {norm_val} " + norm_channel_str
+        else:
+            self.ylabel = f"{y_str} / {scan_info.preset_value} {scan_info.preset_channel}"
+
+        self.xlabel = x_str
+        self.label = "scan " + str(scan_info.scan_num)
+        self.title = self.label + ": " + scan_info.scan_title
+
     def set_labels(self, ax):
         """Set labels, limits and legneds"""
 

src/tavi/data/scan.py

@@ -7,6 +7,7 @@
 
 from tavi.data.nxentry import NexusEntry
 from tavi.data.plotter import Plot1D
+from tavi.data.tavi import TAVI
 from tavi.sample.xtal import Xtal
 
 
@@ -69,12 +70,20 @@ class Scan(object):
     """
 
     def __init__(self, name: str, nexus_dict: NexusEntry) -> None:
+
         self.name: str = name
         self.nexus_dict: NexusEntry = nexus_dict
         self.data: dict = self.get_data()
         # always using spice convention
         self.SPICE_CONVENTION = True
 
+    @classmethod
+    def from_tavi(cls, tavi: TAVI, scan_num: int, exp_id: Optional[str] = None):
+        if exp_id is None:
+            exp_id = next(iter(tavi.data))
+        scan_name = f"scan{scan_num:04}"
+        return cls(scan_name, tavi.data[exp_id].get(scan_name))
+
     @classmethod
     def from_spice(
         cls,
@@ -163,34 +172,6 @@ def get_data(self):
             data_dict.update({name: self.nexus_dict.get(name)})
         return data_dict
 
-    def _make_labels(
-        self,
-        plot1d: Plot1D,
-        x_str: str,
-        y_str: str,
-        norm_channel: Literal["time", "monitor", "mcu", None],
-        norm_val: float,
-    ) -> Plot1D:
-        """generate labels and title"""
-        if norm_channel is not None:
-            if norm_channel == "time":
-                norm_channel_str = "seconds"
-            else:
-                norm_channel_str = norm_channel
-            if norm_val == 1:
-                plot1d.ylabel = y_str + "/ " + norm_channel_str
-            else:
-                plot1d.ylabel = y_str + f" / {norm_val} " + norm_channel_str
-        else:
-            preset_val = self.scan_info.preset_value
-            plot1d.ylabel = y_str + f" / {preset_val} " + self.scan_info.preset_channel
-
-        plot1d.xlabel = x_str
-        plot1d.label = "scan " + str(self.scan_info.scan_num)
-        plot1d.title = plot1d.label + ": " + self.scan_info.scan_title
-
-        return plot1d
-
     def _rebin_tol(
         self,
         x_raw: np.ndarray,
@@ -322,7 +303,7 @@ def generate_curve(
                     yerr = yerr / norm
 
             plot1d = Plot1D(x=x, y=y, yerr=yerr)
-            plot1d = self._make_labels(plot1d, x_str, y_str, norm_channel, norm_val)
+            plot1d.make_labels(x_str, y_str, norm_channel, norm_val, self.scan_info)
 
             return plot1d
 
@@ -332,17 +313,15 @@ def generate_curve(
         match rebin_type:
             case "tol":  # x weighted by normalization channel
                 x, y, yerr = self._rebin_tol(x_raw, y_raw, y_str, rebin_step, norm_channel, norm_val)
-                plot1d = Plot1D(x=x, y=y, yerr=yerr)
-                plot1d = self._make_labels(plot1d, x_str, y_str, norm_channel, norm_val)
-                return plot1d
             case "grid":
                 x, y, yerr = self._rebin_grid(x_raw, y_raw, y_str, rebin_step, norm_channel, norm_val)
-                plot1d = Plot1D(x=x, y=y, yerr=yerr)
-                plot1d = self._make_labels(plot1d, x_str, y_str, norm_channel, norm_val)
-                return plot1d
             case _:
                 raise ValueError('Unrecogonized rebin type. Needs to be "tol" or "grid".')
 
+        plot1d = Plot1D(x=x, y=y, yerr=yerr)
+        plot1d.make_labels(x_str, y_str, norm_channel, norm_val, self.scan_info)
+        return plot1d
+
     def plot_curve(
         self,
         x_str: Optional[str] = None,

src/tavi/data/scan_group.py

@@ -1,3 +1,5 @@
+from typing import Optional
+
 import matplotlib.pyplot as plt
 import numpy as np
 
@@ -18,16 +20,14 @@ class ScanGroup(object):
     Methods:
         generate_curve
         plot_curve
-        generate_waterfall
-        plot_waterfall
         generate_contour
         plot_contour
     """
 
     def __init__(
         self,
-        signals,
-        backgrounds=None,
+        signals: list[int],
+        backgrounds=Optional[list[int]],
         signal_axes=(None, None, None),
         background_axes=(None, None, None),
     ):
@@ -149,47 +149,47 @@ def plot_contour(self, contour_plot, cmap="turbo", vmax=100, vmin=0, ylim=None,
 
         fig.show()
 
-    def plot_waterfall(self, contour_plot, shifts=None, ylim=None, xlim=None, fmt="o"):
-        """Plot waterfall plot.
+    # def plot_waterfall(self, contour_plot, shifts=None, ylim=None, xlim=None, fmt="o"):
+    #     """Plot waterfall plot.
 
-        Note:
-            Horizontal is Y-axis, vertical is Z-axis. Stacked along X-axis.
-        """
+    #     Note:
+    #         Horizontal is Y-axis, vertical is Z-axis. Stacked along X-axis.
+    #     """
 
-        x, y, z, _, _, xlabel, ylabel, zlabel, title = contour_plot
+    #     x, y, z, _, _, xlabel, ylabel, zlabel, title = contour_plot
 
-        num = len(x[0])
+    #     num = len(x[0])
 
-        if shifts is not None:
-            if np.size(shifts) == 1:
-                shifts = (shifts,) * num
-        else:
-            shifts = (0,) * num
+    #     if shifts is not None:
+    #         if np.size(shifts) == 1:
+    #             shifts = (shifts,) * num
+    #     else:
+    #         shifts = (0,) * num
 
-        fig, ax = plt.subplots()
-        shift = 0
-        for i in range(num):
-            if np.isnan(z[:, i]).all():  # all nan
-                continue
-            else:
-                p = ax.errorbar(
-                    x=y[:, i],
-                    y=z[:, i] + shift,
-                    fmt=fmt,
-                    label=f"{xlabel}={np.round(x[0,i],3)}, shift={shift}",
-                )
-            shift += shifts[i]
+    #     fig, ax = plt.subplots()
+    #     shift = 0
+    #     for i in range(num):
+    #         if np.isnan(z[:, i]).all():  # all nan
+    #             continue
+    #         else:
+    #             p = ax.errorbar(
+    #                 x=y[:, i],
+    #                 y=z[:, i] + shift,
+    #                 fmt=fmt,
+    #                 label=f"{xlabel}={np.round(x[0,i],3)}, shift={shift}",
+    #             )
+    #         shift += shifts[i]
 
-        ax.set_title(title)
-        ax.set_xlabel(ylabel)
-        ax.set_ylabel(zlabel)
-        ax.grid(alpha=0.6)
-        ax.legend()
-        if xlim is not None:
-            ax.set_xlim(left=xlim[0], right=xlim[1])
-        if ylim is not None:
-            ax.set_ylim(bottom=ylim[0], top=ylim[1])
-        fig.show()
+    #     ax.set_title(title)
+    #     ax.set_xlabel(ylabel)
+    #     ax.set_ylabel(zlabel)
+    #     ax.grid(alpha=0.6)
+    #     ax.legend()
+    #     if xlim is not None:
+    #         ax.set_xlim(left=xlim[0], right=xlim[1])
+    #     if ylim is not None:
+    #         ax.set_ylim(bottom=ylim[0], top=ylim[1])
+    #     fig.show()
 
     # def generate_waterfall_scans(
     #     self,

src/tavi/data/tavi.py

@@ -23,14 +23,17 @@ class TAVI(object):
 
     """
 
-    def __init__(self) -> None:
+    def __init__(self, tavi_file_path: Optional[str] = None) -> None:
         """Initialization"""
         self.file_path: Optional[str] = None
         self.data: dict = {}
         self.processed_data: dict = {}
         self.fits: dict = {}
         self.plots: dict = {}
 
+        if tavi_file_path is not None:
+            self.open_file(tavi_file_path)
+
     @staticmethod
     def load_data(scan_dict: dict):
         data_dict: dict = {}
@@ -78,7 +81,7 @@ def load_data_from_oncat(self, user_credentials, ipts_info):
         pass
 
     def get_exp_id(self) -> list[str]:
-        "return a list of exp id"
+        "return a list of exp id from a tavi file"
         exp_id_list = []
         with h5py.File(self.file_path, "r") as tavi_file:
             for exp_id in tavi_file["data"].keys():
@@ -97,8 +100,8 @@ def open_file(self, tavi_file_path):
         data_dict = {}
         exp_id_list = self.get_exp_id()
         for exp_id in exp_id_list:
-            scans = NexusEntry.from_nexus(tavi_file_path, prefix=f"/data/{exp_id}")
-            data_dict.update({exp_id: scans})
+            nxentries = NexusEntry.from_nexus(tavi_file_path, prefix=f"/data/{exp_id}")
+            data_dict.update({exp_id: nxentries})
         self.data = data_dict
         # TODO load processed_data fits, and plots
 
@@ -120,8 +123,8 @@ def save(self, file_path: Optional[str] = None):
                 grp_data = root.create_group("data", track_order=True)
                 for exp_id, scans in self.data.items():
                     grp_data.create_group(name=exp_id, track_order=True)
-                    for scan_name, nxentry in scans.items():
-                        name = f"/data/{exp_id}/{scan_name}"
+                    for entry_name, nxentry in scans.items():
+                        name = f"data/{exp_id}/{entry_name}"
                         nxentry.to_nexus(file_path, name=name)
                 # TODO save pdata, fits and plots
                 grp_pdata = root.create_group("processed_data", track_order=True)

tests/test_nxentry.py

@@ -46,6 +46,7 @@ def test_to_nexus(nexus_entries):
     with h5py.File(path_to_nexus, "r") as nexus_file:
         assert str(nexus_file["scan0034"]["title"].asstr()[...]) == "scan_title_34"
         assert nexus_file["scan0034"].attrs["NX_class"] == "NXentry"
+        assert np.allclose(nexus_file["scan0034"]["instrument"]["detector"]["data"][...], [1, 2, 3])
     os.remove(path_to_nexus)