Added and tested apply function to support pipeline-line operations

model_server/base/accessors.py

@@ -110,6 +110,16 @@ class GenericImageDataAccessor(ABC):
     def shape_dict(self):
         return dict(zip(('Y', 'X', 'C', 'Z'), self.data.shape))
 
+    def apply(self, func):
+        """
+        Apply func to data and return as a new in-memory accessor
+        :param func: function that receives and returns the same size np.ndarray
+        :return: InMemoryDataAccessor
+        """
+        return InMemoryDataAccessor(
+            func(self.data)
+        )
+
 class InMemoryDataAccessor(GenericImageDataAccessor):
     def __init__(self, data):
         self._data = self.conform_data(data)

tests/test_accessors.py

@@ -73,11 +73,10 @@ class TestCziImageFileAccess(unittest.TestCase):
 
     def test_write_two_channel_png(self):
         from model_server.base.process import resample_to_8bit
-        ch = 2
         cf = CziImageFileAccessor(czifile['path'])
         acc = cf.get_channels([0, 1])
         opa = output_path / f'{cf.fpath.stem}_2ch.png'
-        acc_out = InMemoryDataAccessor(resample_to_8bit(acc.data))
+        acc_out = acc.apply(resample_to_8bit)
 
         self.assertTrue(
             write_accessor_data_to_file(opa, acc_out)