First attempt to use more gpus

concert/coroutines/filters.py

@@ -51,15 +51,16 @@ def check_and_create(sl):
 
 
 @coroutine
-def queue(consumer, process_all=True, block=False):
+def queue(consumer, process_all=True, block=False, make_deepcopy=True):
     """
     queue(consumer, process_all=True, block=False)
 
     Store the incoming data in a queue and dispatch to the *consumer* in a separate thread which
     prevents the stalling on the "main" data stream. If *process_all* is True the serve loop may
     exit only when all items are sent to *consumer*. If *block* is True this coroutine blocks until
     all items in the serve loop are processed, *process_all* must be True as well for this to take
-    effect.
+    effect. If *make_deepcopy* is True, insert a deep copy of an item into the queue, otherwise just
+    a reference.
     """
     from concert.async import HAVE_GEVENT
     if cfg.ENABLE_GEVENT and HAVE_GEVENT:
@@ -73,23 +74,27 @@ def queue(consumer, process_all=True, block=False):
 
     @threaded
     def serve():
-        while serve.run or (process_all and not item_queue.empty()):
+        while True:
             item = item_queue.get()
-            consumer.send(item)
+            if item is not None:
+                consumer.send(item)
+            elif not process_all or item_queue.empty():
+                break
             item_queue.task_done()
         serve.stopped.set()
         LOG.debug("queue's serve loop stopped")
 
-    serve.run = True
     serve.stopped = Event()
     serve()
 
     try:
         while True:
             item = yield
-            item_queue.put(deepcopy(item))
+            if make_deepcopy:
+                item = deepcopy(item)
+            item_queue.put(item)
     except GeneratorExit:
-        serve.run = False
+        item_queue.put(None)
         if block:
             serve.stopped.wait()
 

concert/experiments/addons.py

@@ -2,8 +2,11 @@
 the acquired data, e.g. write images to disk, do tomographic reconstruction etc.
 """
 import logging
+import numpy as np
+from concert.async import threaded
+from concert.coroutines.base import broadcast, coroutine
 from concert.coroutines.filters import queue
-from concert.coroutines.sinks import Accumulate
+from concert.coroutines.sinks import Accumulate, Result
 
 
 LOG = logging.getLogger(__name__)
@@ -166,16 +169,120 @@ def wrapped_writer():
                 self.walker.descend(acquisition.name)
                 coro = self.walker.write()
                 if self._async:
-                    coro = queue(coro, process_all=True, block=block)
+                    coro = queue(coro, process_all=True, block=block, make_deepcopy=False)
                 return coro
             finally:
                 self.walker.ascend()
 
         return wrapped_writer
 
 
+class OnlineReconstruction(Addon):
+    def __init__(self, experiment, reco_args, process_normalization=False,
+                 consumer=None, block=False, wait_for_projections=False,
+                 walker=None, slice_directory='online-slices'):
+        from multiprocessing.pool import ThreadPool
+        from threading import Event
+        from concert.ext.ufo import GeneralBackprojectManager
+
+        self.experiment = experiment
+        self.dark_result = Result()
+        self.flat_result = Result()
+        self.manager = GeneralBackprojectManager(reco_args)
+        self.walker = walker
+        self.slice_directory = slice_directory
+        self.process_normalization = process_normalization
+        self._pool = ThreadPool(processes=2)
+        self._events = {'darks': Event(), 'flats': Event()}
+        self.consumer = consumer
+        self.block = block
+        self.wait_for_projections = wait_for_projections
+        self._consumers = {}
+        super(OnlineReconstruction, self).__init__(experiment.acquisitions)
+
+    @threaded
+    def _average_images(self, queue, im_type):
+        average = None
+        i = 0
+        while True:
+            image = queue.get()
+            if image is None:
+                if im_type == 'darks':
+                    self.manager.dark = average
+                else:
+                    self.manager.flat = average
+                self._events[im_type].set()
+                LOG.debug('%s pre-processing done', im_type)
+                break
+
+            if self.process_normalization:
+                if average is None:
+                    average = np.zeros_like(image, dtype=np.float32)
+                average = (average * i + image) / (i + 1)
+                i += 1
+            else:
+                average = image
+
+    def _create_averaging(self, im_type):
+        @coroutine
+        def create_averaging_coro():
+            try:
+                import Queue as queue_module
+            except ImportError:
+                import queue as queue_module
+
+            self._events[im_type].clear()
+            queue = queue_module.Queue()
+            self._average_images(queue, im_type)
+            try:
+                while True:
+                    image = yield
+                    queue.put(image)
+            except GeneratorExit:
+                queue.put(None)
+
+        return create_averaging_coro
+
+    def _reconstruct(self):
+        if hasattr(self.experiment, 'darks') and hasattr(self.experiment, 'flats'):
+            events = self._events.values() if self.process_normalization else None
+        else:
+            events = None
+        consumers = []
+        write_coro = None
+
+        if self.consumer:
+            consumers.append(self.consumer)
+        if self.walker:
+            self.walker.descend(self.slice_directory)
+            write_coro = self.walker.write(dsetname='slice_{:>04}.tif')
+            self.walker.ascend()
+            consumers.append(write_coro)
+        consumer = broadcast(*consumers) if consumers else None
+
+        return self.manager(consumer=consumer, block=self.block, wait_for_events=events,
+                            wait_for_projections=self.wait_for_projections)
+
+    def _attach(self):
+        if hasattr(self.experiment, 'darks') and hasattr(self.experiment, 'flats'):
+            self._consumers[self.experiment.darks] = self._create_averaging('darks')
+            self._consumers[self.experiment.flats] = self._create_averaging('flats')
+        self._consumers[self.experiment.radios] = self._reconstruct
+
+        for acq, consumer in self._consumers.items():
+            acq.consumers.append(consumer)
+
+    def _detach(self):
+        for acq, consumer in self._consumers.items():
+            acq.consumers.remove(consumer)
+
+
 class AddonError(Exception):
 
     """Addon errors."""
 
     pass
+
+
+class OnlineReconstructionError(Exception):
+    pass

concert/experiments/dummy.py

@@ -1,11 +1,76 @@
 """Dummy experiments."""
 
 import os
+import numpy as np
 from concert.experiments.base import Acquisition, Experiment
 from concert.devices.cameras.dummy import FileCamera
 from concert.progressbar import wrap_iterable
 
 
+class ImagingExperiment(Experiment):
+
+    """
+    A typical imaging experiment which consists of acquiring dark, flat and radiographic images, in
+    this case zeros or random data.
+
+    .. py:attribute:: num_darks
+
+        Number of dark images (no beam, just dark current)
+
+    .. py:attribute:: num_flats
+
+        Number of flat images (beam present, sample not)
+
+    .. py:attribute:: num_radios
+
+        Number of radiographic images
+
+    .. py:attribute:: shape
+
+        Shape of the generated images (H x W) (default: 1024 x 1024)
+
+    .. py:attribute:: random
+
+        If True, one random image is created and re-used, otherwise zeros
+
+    .. py:attribute:: dtype
+
+        Data type of the generated images (default: unsigned short)
+    """
+
+    def __init__(self, num_darks, num_flats, num_radios, shape=(1024, 1024), walker=None,
+                 random=False, dtype=np.ushort, separate_scans=True, name_fmt='scan_{:>04}'):
+        self.num_darks = num_darks
+        self.num_flats = num_flats
+        self.num_radios = num_radios
+        self.shape = shape
+        self.random = random
+        self.dtype = dtype
+        darks = Acquisition('darks', self.take_darks)
+        flats = Acquisition('flats', self.take_flats)
+        radios = Acquisition('radios', self.take_radios)
+        super(ImagingExperiment, self).__init__([darks, flats, radios], walker=walker)
+
+    def _produce_images(self, num):
+        if self.random:
+            image = np.random.normal(128., 10., size=self.shape)
+        else:
+            image = np.zeros(self.shape)
+
+        image = image.astype(self.dtype)
+        for i in wrap_iterable(range(num)):
+            yield image
+
+    def take_darks(self):
+        return self._produce_images(self.num_darks)
+
+    def take_flats(self):
+        return self._produce_images(self.num_flats)
+
+    def take_radios(self):
+        return self._produce_images(self.num_radios)
+
+
 class ImagingFileExperiment(Experiment):
 
     """