ENH Add crop pipeline

pims/process.py

@@ -0,0 +1,66 @@
+from __future__ import (absolute_import, division, print_function,
+                        unicode_literals)
+
+import numpy as np
+from slicerator import pipeline, Pipeline
+from numpy.lib.arraypad import _validate_lengths
+
+
+def _crop(frame, bbox):
+    return frame[bbox[0]:bbox[2], bbox[1]:bbox[3]]
+
+
+@pipeline
+class crop(Pipeline):
+    """Crop image or image-reader`reader` by `crop_width` along each dimension.
+
+    Parameters
+    ----------
+    ar : array-like of rank N
+       Input array.
+    crop_width : {sequence, int}
+       Number of values to remove from the edges of each axis.
+       ``((before_1, after_1),`` ... ``(before_N, after_N))`` specifies
+       unique crop widths at the start and end of each axis.
+       ``((before, after),)`` specifies a fixed start and end crop
+       for every axis.
+       ``(n,)`` or ``n`` for integer ``n`` is a shortcut for
+       before = after = ``n`` for all axes.
+    order : {'C', 'F', 'A', 'K'}, optional
+       control the memory layout of the copy. See ``np.copy``.
+    Returns
+    -------
+    cropped : array
+       The cropped array.
+
+    See Also
+    --------
+    Source: ``skimage.util.crop`` (v0.12.3)
+    """
+    def __init__(self, reader, crop_width, order='K'):
+        # We have to know the frame shape that is returned by the reader.
+        try:  # In case the reader is a FramesSequence, there is an attribute
+            shape = reader.frame_shape
+            first_frame = np.empty(shape, dtype=np.bool)
+        except AttributeError:
+            first_frame = reader[0]
+            shape = first_frame.shape
+        # Validate the crop widths on the first frame
+        crops = _validate_lengths(first_frame, crop_width)
+        self._crop_slices = [slice(a, shape[i] - b)
+                             for i, (a, b) in enumerate(crops)]
+        self._crop_shape = tuple([shape[i] - b - a
+                                  for i, (a, b) in enumerate(crops)])
+        self._crop_order = order
+        # We could pass _crop to proc_func. However this adds an extra copy
+        # operation. Therefore we define our own here.
+        Pipeline.__init__(self, reader, proc_func=None)
+
+    def _get(self, key):
+        ar = self._ancestor[key]
+        return np.array(ar[self._crop_slices], order=self._crop_order,
+                        copy=True)
+
+    @property
+    def frame_shape(self):
+        return self._crop_shape

pims/tests/test_process.py

@@ -0,0 +1,65 @@
+from __future__ import (absolute_import, division, print_function,
+                        unicode_literals)
+
+import six
+
+import os
+import unittest
+import nose
+import numpy as np
+from numpy.testing import assert_equal
+from pims import FramesSequence, Frame
+from pims.process import crop
+
+
+class RandomReader(FramesSequence):
+    def __init__(self, length=10, shape=(128, 128), dtype='uint8'):
+        self._len = length
+        self._dtype = dtype
+        self._shape = shape
+        data_shape = (length,) + shape
+        if np.issubdtype(self._dtype, np.float):
+            self._data = np.random.random(data_shape).astype(self._dtype)
+        else:
+            self._data = np.random.randint(0, np.iinfo(self._dtype).max,
+                                           data_shape).astype(self._dtype)
+
+    def __len__(self):
+        return self._len
+
+    @property
+    def frame_shape(self):
+        return self._shape
+
+    @property
+    def pixel_type(self):
+        return self._dtype
+
+    def get_frame(self, i):
+        return Frame(self._data[i], frame_no=i)
+
+
+class TestPipelinesCommon(object):
+    def test_on_frame(self):
+        assert_equal(self.pipeline(self.rdr[0]), self.first_frame)
+
+    def test_on_reader(self):
+        assert_equal(self.pipeline(self.rdr)[0], self.first_frame)
+
+    def test_on_random_frame(self):
+        i = np.random.randint(0, len(self.rdr))
+        assert_equal(self.pipeline(self.rdr)[i], self.pipeline(self.rdr[i]))
+
+
+class TestCrop(TestPipelinesCommon, unittest.TestCase):
+    def setUp(self):
+        self.rdr = RandomReader(length=10, shape=(32, 33))
+        self.pipeline = lambda x: crop(x, ((5, 32-26), (7, 33-27)))
+        self.first_frame = self.rdr[0][5:26, 7:27]
+
+    def test_attrs(self):
+        proc = self.pipeline(self.rdr)
+        assert_equal(self.rdr.pixel_type, proc.pixel_type)
+        assert_equal(len(self.rdr), len(proc))
+        assert_equal(self.rdr.frame_shape, (32, 33))
+        assert_equal(proc.frame_shape, (21, 20))

setup.py

@@ -22,7 +22,7 @@
     cmdclass=versioneer.get_cmdclass(),
     description="Python Image Sequence",
     author="PIMS Contributors",
-    install_requires=['slicerator>=0.9.7', 'six>=1.8', 'numpy>=1.7'],
+    install_requires=['slicerator>=0.9.8', 'six>=1.8', 'numpy>=1.7'],
     author_email="dallan@pha.jhu.edu",
     url="https://github.com/soft-matter/pims",
     packages=['pims',