Merge pull request #1030 from wright-group/from-solis_kinetic

`from_Solis`: import kinetic series type

Author: kameyer226

CHANGELOG.md

@@ -5,6 +5,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/).
 
 ## [Unreleased]
 
+### Added
+- `data.from_Solis`: "kinetic series" acquisition type now supported.
+
 ## [3.4.1]
 
 ### Added

WrightTools/data/_solis.py

@@ -30,7 +30,7 @@ def from_Solis(filepath, name=None, parent=None, verbose=True) -> Data:
     Parameters
     ----------
     filepath : path-like
-        Path to .txt file.
+        Path to file (should be .asc format).
         Can be either a local or remote file (http/ftp).
         Can be compressed with gz/bz2, decompression based on file name.
     name : string (optional)
@@ -61,15 +61,32 @@ def from_Solis(filepath, name=None, parent=None, verbose=True) -> Data:
     axis0 = []
     arr = []
     attrs = {}
-    while True:
-        line = f.readline().strip()[:-1]
-        if len(line) == 0:
-            break
-        else:
-            line = line.split(",")
-            line = [float(x) for x in line]
-            axis0.append(line.pop(0))
-            arr.append(line)
+
+    line0 = f.readline().strip()[:-1]
+    line0 = [float(x) for x in line0.split(",")]  # TODO: robust to space, tab, comma
+    axis0.append(line0.pop(0))
+    arr.append(line0)
+
+    def get_frames(f, arr, axis0):
+        axis0_written = False
+        while True:
+            line = f.readline().strip()[:-1]
+            if len(line) == 0:
+                break
+            else:
+                line = [float(x) for x in line.split(",")]
+                # signature of new frames is restart of axis0
+                if not axis0_written and (line[0] == axis0[0]):
+                    axis0_written = True
+                if axis0_written:
+                    line.pop(0)
+                else:
+                    axis0.append(line.pop(0))
+                arr.append(line)
+        return arr, axis0
+
+    arr, axis0 = get_frames(f, arr, axis0)
+    nframes = len(arr) // len(axis0)
 
     i = 0
     while i < 3:
@@ -94,20 +111,33 @@ def from_Solis(filepath, name=None, parent=None, verbose=True) -> Data:
         data = Data(**kwargs)
     else:
         data = parent.create_data(**kwargs)
-    arr = np.array(arr)
-    arr /= float(attrs["Exposure Time (secs)"])
-    # signal has units of Hz because time normalized
-    arr = data.create_channel(name="signal", values=arr, signed=False, units="Hz")
+
     axis0 = np.array(axis0)
     if float(attrs["Grating Groove Density (l/mm)"]) == 0:
         xname = "xindex"
         xunits = None
     else:
         xname = "wm"
         xunits = "nm"
-    data.create_variable(name=xname, values=axis0[:, None], units=xunits)
-    data.create_variable(name="yindex", values=np.arange(arr.shape[1])[None, :], units=None)
-    data.transform(data.variables[0].natural_name, "yindex")
+    axes = [xname, "yindex"]
+
+    if nframes == 1:
+        arr = np.array(arr)
+        data.create_variable(name=xname, values=axis0[:, None], units=xunits)
+        data.create_variable(name="yindex", values=np.arange(arr.shape[-1])[None, :], units=None)
+    else:
+        arr = np.array(arr).reshape(nframes, len(axis0), len(arr[0]))
+        data.create_variable(name="frame", values=np.arange(nframes)[:, None, None], units=None)
+        data.create_variable(name=xname, values=axis0[None, :, None], units=xunits)
+        data.create_variable(
+            name="yindex", values=np.arange(arr.shape[-1])[None, None, :], units=None
+        )
+        axes = ["frame"] + axes
+
+    data.transform(*axes)
+    arr /= float(attrs["Exposure Time (secs)"])
+    # signal has units of Hz because time normalized
+    data.create_channel(name="signal", values=arr, signed=False, units="Hz")
 
     for key, val in attrs.items():
         data.attrs[key] = val

WrightTools/datasets/Solis/kinetic.asc.gz

@@ -29,6 +29,16 @@ def test_xpos_ypos_fluorescence():
     data.close()
 
 
+def test_kinetic():
+    p = datasets.Solis.kinetic
+    data = wt.data.from_Solis(p)
+    assert data.shape == (10, 20, 20)
+    assert data.axis_expressions == ("frame", "xindex", "yindex")
+    assert data.units == (None, None, None)
+    data.close()
+
+
 if __name__ == "__main__":
     test_wm_ypos_fluorescence_with_filter()
     test_xpos_ypos_fluorescence()
+    test_kinetic()