'liveimage' (live stream, generated by Lambda) in get_image added

src/AsapoWorker/data_handler.py

@@ -2,10 +2,21 @@ import io
 import numpy as np
 import hdf5plugin
 import h5py
+import zlib
 from skimage.io import imread
 
 
 def get_image(data, metadata):
+    if metadata["name"] == 'liveimage':
+        image = zlib.decompress(data)
+        if metadata['meta']['frame_depth'] == 24:
+            image = np.frombuffer(image, dtype=np.uint32)
+        else:
+            image = np.frombuffer(image, dtype=np.uint16)
+
+        image = np.reshape(image, (metadata['meta']['frame_height'],
+                                   metadata['meta']['frame_width']))
+        return image
     typ = metadata["name"].rpartition(".")[-1]
     if typ == "tif":
         buf = io.BytesIO(data)

tests/test_data_handler.py

 import io
 import numpy as np
 import sys
+import zlib
 
 from AsapoWorker.data_handler import get_image, serialize_ndarray
 
@@ -26,3 +27,29 @@ def test_handle_h5():
                 'name': 'file.h5'}
     image2 = get_image(data.getvalue(), metadata)
     assert np.alltrue(image == image2)
+
+def test_livestream_24bit():
+    image = np.random.randint(0, 2**24, (516, 1556)).astype("uint32")
+    data = image.flatten()
+    data = zlib.compress(data)
+    metadata = {'name': 'liveimage',
+                'meta': {'frame_depth': 24,
+                         'frame_height': 516,
+                         'frame_width': 1556}
+                }
+
+    image2 = get_image(data, metadata)
+    assert np.alltrue(image == image2)
+
+def test_livestream_12bit():
+    image = np.random.randint(0, 2**12, (516, 1556)).astype("uint16")
+    data = image.flatten()
+    data = zlib.compress(data)
+    metadata = {'name': 'liveimage',
+                'meta': {'frame_depth': 12,
+                         'frame_height': 516,
+                         'frame_width': 1556}
+                }
+
+    image2 = get_image(data, metadata)
+    assert np.alltrue(image == image2)