put() method for downstream processing

api_test.py

@@ -77,7 +77,7 @@ assert(isinstance(data, np.ndarray))
 assert(data.shape == data_a[1].shape)
 assert(np.array_equal(data, data_a[1]))
 
-# test slicing in first dimension
+# test view
 
 stream_sl = stream_a[1:4]
 data_sl = data_a[1:4]
@@ -117,6 +117,13 @@ for t in tests:
     assert(sd.ndim == dd.ndim)
     assert(np.array_equal(sd.load(), dd))
 
+# test put method
+
+stream = Stream('test')
+outlet = stream.filter()
+stream.put(data_a[0])
+
+
 ###########
 
 #fo = StreamFile('api_test.h5', 'w')

bob/io/stream/stream.py

@@ -18,12 +18,16 @@ class Stream:
     def __init__(self, name=None, parent=None):
         self.name = name
         self.parent = parent
+        self.child = None
+        if isinstance(parent, Stream):
+            parent.child = self
         self.reset()
 
     # reset stream
     def reset(self):
         self._loaded = None
         self._data = None
+        self._shape = None
         self.__bounding_box = StreamArray(self)
         self.__image_points = StreamArray(self)
 
@@ -63,14 +67,34 @@ class Stream:
     # shape property
     @property
     def shape(self):
+        # if parent is file return dataset shape
         if isinstance(self.parent, StreamFile):
             return self.parent.get_stream_shape(self.name)
-        else:
+        # if has parent return parent shape
+        elif self.parent is not None:
             return self.parent.shape
+        # if parent is None return internal _shape
+        else:
+            return self._shape
 
+    # recursively set shape (if not yet defined)
     @shape.setter
     def shape(self, value):
-        raise Exception("not yet implemented")
+        # can only set shape if undefined
+        if self.shape is not None:
+            raise Exception("shape is already set")
+        # set dataset dimension (should this be allowed?)
+        if isinstance(self.parent, StreamFile):
+            parent.set_dataset_shape(name=self.name, shape=value)
+        # set recursively
+        elif self.parent is not None:
+            self.parent.shape = value
+        # end of recursion 
+        else: 
+            if isinstance(value, tuple):
+                self._shape = value
+            else:
+                raise(Exception('shape must be a tuple of int or None'))
 
     # ndim property
     @property
@@ -89,7 +113,7 @@ class Stream:
     def timestamps(self, value):
         raise Exception("not yet implemented")
 
-    # camera
+    # camera property
     @property
     def camera(self):
         if isinstance(self.parent, StreamFile):
@@ -101,7 +125,7 @@ class Stream:
     def camera(self, value):
         raise Exception("not yet implemented")
 
-    # bounding_box
+    # bounding_box property
     @property
     def bounding_box(self):
         if isinstance(self.parent, StreamFile):
@@ -109,7 +133,7 @@ class Stream:
         else:
             return self.parent.bounding_box
 
-    # image points
+    # image points property
     @property
     def image_points(self):
         if isinstance(self.parent, StreamFile):
@@ -162,6 +186,20 @@ class Stream:
         self._loaded = indices
         return self._data
 
+    # put one frame downstream 
+    # TODO define behaviour across all states
+    def put(self, data, timestamp=None):
+        # set _shape to keep track of size of frames
+        # TODO shape property (and ndim property)
+        if self._shape is None:
+            self._shape = tuple([None, *data.shape])
+        # check if same shape
+        elif data.shape != self._shape[1:]:
+            raise(Exception('data must be the same shape as previous frames, use reset() to clear shape.'))
+        # put downstream
+        if self.child is not None:
+            self.child.put(data, timestamp)
+
     # filters
     def get_available_filters(self):
         return Stream.filters
@@ -186,6 +224,7 @@ def stream_filter(name):
 # Stream Filters
 
 ### Filter general class ###
+@stream_filter('filter')
 class StreamFilter(Stream):
     def __init__(self, name, parent):
         super().__init__(name=name, parent=parent)
@@ -212,6 +251,23 @@ class StreamFilter(Stream):
         self._loaded = indices
         return self._data
 
+    # put one frame downstream 
+    # TODO define behaviour across all states
+    def put(self, data, timestamp=None):
+        # process data
+        # TODO check shape
+        # -1 implies frame is the latest
+        indices = [-1]
+        data = np.stack([data])
+        data = self.process(data, indices)
+        data = data[0]
+        # buffer data
+        self._loaded = indices
+        self._data = data
+        # put downstream
+        if self.child is not None:
+            child.put(data, timestamp)
+
 ################################################################################
 ### astype ###
 # TODO make that work more as numpy...
@@ -306,6 +362,7 @@ class StreamView(StreamFilter):
                     __ndim -= 1
         return __ndim
 
+    # load one or several frames
     def load(self, index=None):
         parent_indices = get_index_list(self.frame_view, self.parent.shape[0])
         view_indices = get_index_list(index, self.shape[0])