Merge pull request #29 from biometrics/stream_processing

basic support for online video processing

Merge pull request #29 from biometrics/stream_processing
basic support for online video processing
jklontz
2 parents 5e02aadf 669c2c6e
Showing 7 changed files with 855 additions and 48 deletions
app/br/CMakeLists.txt
app/br/br.cpp
sdk/openbr_plugin.cpp
sdk/openbr_plugin.h
sdk/plugins/meta.cpp
sdk/plugins/misc.cpp
sdk/plugins/stream.cpp
@@ -4,6 +4,8 @@ endif()
 add_executable(br br.cpp ${BR_RESOURCES})
 target_link_libraries(br openbr ${CMAKE_THREAD_LIBS_INIT})
+qt5_use_modules(br ${QT_DEPENDENCIES})
+
 install(TARGETS br RUNTIME DESTINATION bin)
 add_test(NAME br_initialize WORKING_DIRECTORY ${CMAKE_BINARY_DIR} COMMAND br)
@@ -18,6 +18,8 @@
 #include <stdlib.h>
 #include <string.h>
 #include <openbr.h>
+#include <openbr_plugin.h>
+#include <QtConcurrent>
 /*!
  * \defgroup cli Command Line Interface
@@ -92,9 +94,9 @@ static void check(bool condition, const char *error_message)
     }
 }
-int main(int argc, char *argv[])
+
+int fake_main(int argc, char *argv[])
 {
-    br_initialize(argc, argv);
     // Remove program name
     argv = &argv[1];
@@ -212,6 +214,22 @@ int main(int argc, char *argv[])
         }
     }
-    br_finalize();
+    QCoreApplication::exit();
     return 0;
 }
+
+
+int main(int argc, char *argv[])
+{
+    br_initialize(argc, argv);
+
+    // Run main the main process in another thread so that this one can
+    // sit in exec
+    QtConcurrent::run(fake_main, argc, argv);
+
+    QCoreApplication::exec();
+
+    br_finalize();
+
+}
+
@@ -28,6 +28,10 @@
 #include <iostream>
 #include <openbr_plugin.h>
+#ifndef BR_EMBEDDED
+#include <QApplication>
+#endif
+
 #include "version.h"
 #include "core/bee.h"
 #include "core/common.h"
@@ -797,7 +801,12 @@ void br::Context::initialize(int argc, char *argv[], const QString &amp;sdkPath)
         Globals->init(File());
     }
+#ifndef BR_EMBEDDED
+    Globals->coreApplication = QSharedPointer<QCoreApplication>(new QApplication(argc, argv));
+#else
     Globals->coreApplication = QSharedPointer<QCoreApplication>(new QCoreApplication(argc, argv));
+#endif
+
     initializeQt(sdkPath);
 #ifdef BR_DISTRIBUTED
@@ -890,6 +899,10 @@ QString br::Context::scratchPath()
 void br::Context::messageHandler(QtMsgType type, const QMessageLogContext &context, const QString &msg)
 {
+    // Something about this method is not thread safe, and will lead to crashes if qDebug
+    // statements are called from multiple threads. Unless we lock the whole thing...
+    static QMutex general_lock;
+    QMutexLocker locker(&general_lock);
     QString txt;
     switch (type) {
       case QtDebugMsg:
@@ -1113,6 +1113,7 @@ class BR_EXPORT TimeVaryingTransform : public Transform
 {
     Q_OBJECT
+public:
     virtual bool timeVarying() const { return true; }
     virtual void project(const Template &src, Template &dst) const
@@ -1127,6 +1128,23 @@ class BR_EXPORT TimeVaryingTransform : public Transform
         (void) dst; (void) src;
     }
+    // Get a compile failure if this isn't here to go along with the other
+    // projectUpdate, no idea why
+    virtual void projectUpdate(const Template & src, Template & dst)
+    {
+        (void) src; (void) dst;
+        qFatal("do something useful");
+    }
+
+    virtual void projectUpdate(const TemplateList &src, TemplateList &dst)
+    {
+        foreach (const Template & src_part, src) {
+            Template out;
+            projectUpdate(src_part, out);
+            dst.append(out);
+        }
+    }
+
 protected:
     TimeVaryingTransform(bool independent = true, bool trainable = true) : Transform(independent, trainable) {}
 };
@@ -1170,6 +1188,51 @@ protected:
     UntrainableMetaTransform() : UntrainableTransform(false) {}
 };
+/*!
+ * \brief A MetaTransform that aggregates some sub-transforms
+ */
+class BR_EXPORT CompositeTransform : public TimeVaryingTransform
+{
+    Q_OBJECT
+
+public:
+    Q_PROPERTY(QList<br::Transform*> transforms READ get_transforms WRITE set_transforms RESET reset_transforms)
+    BR_PROPERTY(QList<br::Transform*>, transforms, QList<br::Transform*>())
+
+    virtual void project(const Template &src, Template &dst) const
+    {
+        if (timeVarying()) qFatal("No const project defined for time-varying transform");
+        _project(src, dst);
+    }
+
+    virtual void project(const TemplateList &src, TemplateList &dst) const
+    {
+        if (timeVarying()) qFatal("No const project defined for time-varying transform");
+        _project(src, dst);
+    }
+
+    bool timeVarying() const { return isTimeVarying; }
+
+    void init()
+    {
+        isTimeVarying = false;
+        foreach (const br::Transform *transform, transforms) {
+            if (transform->timeVarying()) {
+                isTimeVarying = true;
+                break;
+            }
+        }
+    }
+
+protected:
+    bool isTimeVarying;
+
+    virtual void _project(const Template & src, Template & dst) const = 0;
+    virtual void _project(const TemplateList & src, TemplateList & dst) const = 0;
+
+    CompositeTransform() : TimeVaryingTransform(false) {}
+};
+
 /*! @}*/
 /*!
@@ -86,51 +86,6 @@ static void incrementStep()
 }
 /*!
- * \brief A MetaTransform that aggregates some sub-transforms
- */
-class BR_EXPORT CompositeTransform : public TimeVaryingTransform
-{
-    Q_OBJECT
-
-public:
-    Q_PROPERTY(QList<br::Transform*> transforms READ get_transforms WRITE set_transforms RESET reset_transforms)
-    BR_PROPERTY(QList<br::Transform*>, transforms, QList<br::Transform*>())
-
-    virtual void project(const Template &src, Template &dst) const
-    {
-        if (timeVarying()) qFatal("No const project defined for time-varying transform");
-        _project(src, dst);
-    }
-
-    virtual void project(const TemplateList &src, TemplateList &dst) const
-    {
-        if (timeVarying()) qFatal("No const project defined for time-varying transform");
-        _project(src, dst);
-    }
-
-    bool timeVarying() const { return isTimeVarying; }
-
-    void init()
-    {
-        isTimeVarying = false;
-        foreach (const br::Transform *transform, transforms) {
-            if (transform->timeVarying()) {
-                isTimeVarying = true;
-                break;
-            }
-        }
-    }
-
-protected:
-    bool isTimeVarying;
-
-    virtual void _project(const Template & src, Template & dst) const = 0;
-    virtual void _project(const TemplateList & src, TemplateList & dst) const = 0;
-
-    CompositeTransform() : TimeVaryingTransform(false) {}
-};
-
-/*!
  * \ingroup Transforms
  * \brief Transforms in series.
  * \author Josh Klontz \cite jklontz
@@ -284,6 +239,38 @@ BR_REGISTER(Transform, ExpandTransform)
 /*!
  * \ingroup transforms
+ * \brief It's like the opposite of ExpandTransform, but not really
+ * \author Charles Otto \cite caotto
+ *
+ * Given a set of templatelists as input, concatenate them onto a single Template
+ */
+class ContractTransform : public UntrainableMetaTransform
+{
+    Q_OBJECT
+
+    virtual void project(const TemplateList &src, TemplateList &dst) const
+    {
+        //dst = Expanded(src);
+        Template out;
+
+        foreach (const Template & t, src) {
+            out.merge(t);
+        }
+        dst.clear();
+        dst.append(out);
+    }
+
+    virtual void project(const Template & src, Template & dst) const
+    {
+        qFatal("this has gone bad");
+        (void) src; (void) dst;
+    }
+};
+
+BR_REGISTER(Transform, ContractTransform)
+
+/*!
+ * \ingroup transforms
  * \brief Transforms in parallel.
  * \author Josh Klontz \cite jklontz
  *
@@ -626,6 +613,15 @@ public:
     // Process the single elemnt templates in parallel if parallelism is enabled.
     void project(const TemplateList &src, TemplateList &dst) const
     {
+        // Little ugly, but if we own a timeVaryingTransform and this gets called
+        // cast off the const modifier and use projectUpdate. This allows us to
+        // act as a single point of entry.
+        if (transform->timeVarying())
+        {
+            DistributeTemplateTransform * non_const = (DistributeTemplateTransform *) this;
+            non_const->projectUpdate(src,dst);
+            return;
+        }
         // Pre-allocate output for each template
         QList<TemplateList> output_buffer;
         output_buffer.reserve(src.size());
@@ -655,6 +651,15 @@ public:
         for (int i=0; i<src.size(); i++) dst.append(output_buffer[i]);
     }
+    void projectUpdate(const TemplateList &src, TemplateList &dst)
+    {
+        if (!transform->timeVarying()) {
+            this->project(src, dst);
+            return;
+        }
+        this->transform->projectUpdate(src, dst);
+    }
+
     private:
@@ -19,6 +19,13 @@
 #include "core/opencvutils.h"
+#ifndef BR_EMBEDDED
+#include <QLabel>
+#include <opencv2/imgproc/imgproc.hpp>
+#include <QApplication>
+#endif
+
+
 using namespace cv;
 namespace br
@@ -90,6 +97,154 @@ int ShowTransform::counter = 0;
 BR_REGISTER(Transform, ShowTransform)
+#ifndef BR_EMBEDDED
+
+QImage toQImage(const Mat &mat)
+{
+    // Convert to 8U depth
+    Mat mat8u;
+    if (mat.depth() != CV_8U) {
+        double globalMin = std::numeric_limits<double>::max();
+        double globalMax = -std::numeric_limits<double>::max();
+
+        std::vector<Mat> mv;
+        split(mat, mv);
+        for (size_t i=0; i<mv.size(); i++) {
+            double min, max;
+            minMaxLoc(mv[i], &min, &max);
+            globalMin = std::min(globalMin, min);
+            globalMax = std::max(globalMax, max);
+        }
+        assert(globalMax >= globalMin);
+
+        double range = globalMax - globalMin;
+        if (range != 0) {
+            double scale = 255 / range;
+            convertScaleAbs(mat, mat8u, scale, -(globalMin * scale));
+        } else {
+            // Monochromatic
+            mat8u = Mat(mat.size(), CV_8UC1, Scalar((globalMin+globalMax)/2));
+        }
+    } else {
+        mat8u = mat;
+    }
+
+    // Convert to 3 channels
+    Mat mat8uc3;
+    if      (mat8u.channels() == 4) cvtColor(mat8u, mat8uc3, CV_BGRA2RGB);
+    else if (mat8u.channels() == 3) cvtColor(mat8u, mat8uc3, CV_BGR2RGB);
+    else if (mat8u.channels() == 1) cvtColor(mat8u, mat8uc3, CV_GRAY2RGB);
+
+    return QImage(mat8uc3.data, mat8uc3.cols, mat8uc3.rows, 3*mat8uc3.cols, QImage::Format_RGB888).copy();
+}
+
+// Provides slots for manipulating a QLabel, but does not inherit from QWidget.
+// Therefore, it can be moved to the main thread if not created there initially
+// since god forbid you create a QWidget subclass in not the main thread.
+class GUIProxy : public QObject
+{
+    Q_OBJECT
+
+    QLabel * window;
+
+public:
+    GUIProxy()
+    {
+        window =NULL;
+    }
+
+public slots:
+    void showImage(QPixmap pixmap)
+    {
+        window->show();
+        window->setPixmap(pixmap);
+        window->setFixedSize(pixmap.size());
+        window->update();
+    }
+
+    void createWindow()
+    {
+        delete window;
+        window = NULL;
+        window = new QLabel();
+    }
+};
+
+/*!
+ * \ingroup transforms
+ * \brief Displays templates in a GUI pop-up window using QT.
+ * \author Charles Otto \cite caotto
+ * Unlike ShowTransform, this can be used with parallelism enabled, although it
+ * is considered TimeVarying.
+ */
+class Show2Transform : public TimeVaryingTransform
+{
+    Q_OBJECT
+public:
+    Show2Transform() : TimeVaryingTransform(false, false)
+    {
+        // Create our GUI proxy
+        gui = new GUIProxy();
+        // Move it to the main thread, this means signals we send to it will
+        // be run in the main thread, which is hopefully in an event loop
+        gui->moveToThread(QApplication::instance()->thread());
+        // Connect our signals to the proxy's slots
+        connect(this, SIGNAL(needWindow()), gui, SLOT(createWindow()), Qt::BlockingQueuedConnection);
+        connect(this, SIGNAL(updateImage(QPixmap)), gui, SLOT(showImage(QPixmap)));
+    }
+
+    ~Show2Transform()
+    {
+        delete gui;
+    }
+
+    void train(const TemplateList &data) { (void) data; }
+
+    void project(const TemplateList &src, TemplateList &dst) const
+    {
+        Transform * non_const = (Show2Transform *) this;
+        non_const->projectUpdate(src,dst);
+    }
+
+    void projectUpdate(const TemplateList &src, TemplateList &dst)
+    {
+        dst = src;
+
+        if (src.empty())
+            return;
+
+        foreach (const Template & t, src) {
+            foreach(const cv::Mat & m, t) {
+                QImage qImageBuffer = toQImage(m);
+                displayBuffer.convertFromImage(qImageBuffer);
+                emit updateImage(displayBuffer);
+            }
+        }
+    }
+
+    void finalize(TemplateList & output)
+    {
+        (void) output;
+        // todo: hide ui?
+    }
+
+    void init()
+    {
+        emit needWindow();
+    }
+
+protected:
+    GUIProxy * gui;
+    QPixmap displayBuffer;
+
+signals:
+    void needWindow();
+    void updateImage(QPixmap input);
+};
+
+BR_REGISTER(Transform, Show2Transform)
+#endif
+
 /*!
  * \ingroup transforms
  * \brief Prints the template's file to stdout or stderr.
+#define QT_NO_DEBUG_OUTPUT
+
+#include <openbr_plugin.h>
+#include <QReadWriteLock>
+#include <QWaitCondition>
+#include <QThreadPool>
+
+#include "core/common.h"
+#include "core/opencvutils.h"
+#include "core/qtutils.h"
+
+#include "opencv/highgui.h"
+
+#include <iostream>
+
+using namespace cv;
+
+namespace br
+{
+
+class FrameData
+{
+public:
+    int sequenceNumber;
+    TemplateList data;
+};
+
+// A buffer shared between adjacent processing stages in a stream
+class SharedBuffer
+{
+public:
+    SharedBuffer(int _maxItems = 200) : maxItems(_maxItems) {}
+    virtual ~SharedBuffer() {}
+
+    virtual void addItem(FrameData * input)=0;
+
+    virtual FrameData * getItem()=0;
+
+    int getMaxItems() { return maxItems; }
+
+    virtual void stoppedInput() =0;
+    virtual void startInput() = 0;
+protected:
+    int maxItems;
+};
+
+// For 1 - 1 boundaries, a buffer class with a single shared buffer, a mutex
+// is used to serialize all access to the buffer.
+class SingleBuffer : public SharedBuffer
+{
+public:
+    SingleBuffer(unsigned _maxItems = 20) : SharedBuffer(_maxItems) { no_input = false; }
+
+    void stoppedInput()
+    {
+        QMutexLocker bufferLock(&bufferGuard);
+        no_input = true;
+        // Release anything waiting for input items.
+        availableInput.wakeAll();
+    }
+
+    // There will be more input
+    void startInput()
+    {
+        QMutexLocker bufferLock(&bufferGuard);
+        no_input = false;
+    }
+
+    void addItem(FrameData * input)
+    {
+        QMutexLocker bufferLock(&bufferGuard);
+
+        // If the buffer is too full, wait for space to become available
+        if (buffer.size() >= maxItems) {
+            availableOutputSpace.wait(&bufferGuard);
+        }
+
+        buffer.append(input);
+
+        // Wait for certain # of items?
+        availableInput.wakeOne();
+    }
+
+    FrameData * getItem()
+    {
+        QMutexLocker bufferLock(&bufferGuard);
+
+        if (buffer.empty()) {
+            // If no further items will come we are done here
+            if (no_input)
+                return NULL;
+            // Wait for an item
+            availableInput.wait(&bufferGuard);
+        }
+
+        // availableInput was signalled, but the buffer is still empty? We're done here.
+        if (buffer.empty())
+            return NULL;
+
+        FrameData * output = buffer.first();
+        buffer.removeFirst();
+        if (buffer.size() < maxItems / 2)
+            availableOutputSpace.wakeAll();
+        return output;
+    }
+
+private:
+    QMutex bufferGuard;
+    QWaitCondition availableInput;
+    QWaitCondition availableOutputSpace;
+    bool no_input;
+
+    QList<FrameData *> buffer;
+};
+
+// Interface for sequentially getting data from some data source.
+// Initialized off of a template, can represent a video file (stored in the template's filename)
+// or a set of images already loaded into memory stored as multiple matrices in an input template.
+class DataSource
+{
+public:
+    DataSource() {}
+    virtual ~DataSource() {}
+
+    virtual FrameData * getNext() = 0;
+    virtual void close() = 0;
+    virtual bool open(Template & input) = 0;
+    virtual bool isOpen() = 0;
+};
+
+// Read a video frame by frame using cv::VideoCapture
+class VideoDataSource : public DataSource
+{
+public:
+    VideoDataSource() {}
+
+    FrameData * getNext()
+    {
+        if (!isOpen())
+            return NULL;
+
+        FrameData * output = new FrameData();
+        output->data.append(Template(basis.file));
+        output->data.last().append(cv::Mat());
+
+        output->sequenceNumber = next_idx;
+        next_idx++;
+
+        bool res = video.read(output->data.last().last());
+        if (!res) {
+            delete output;
+            return NULL;
+        }
+        return output;
+    }
+
+    bool open(Template &input)
+    {
+        next_idx = 0;
+        basis = input;
+        video.open(input.file.name.toStdString());
+        return video.isOpened();
+    }
+
+    bool isOpen() { return video.isOpened(); }
+
+    void close() { video.release(); }
+
+private:
+    cv::VideoCapture video;
+    Template basis;
+    int next_idx;
+};
+
+// Given a template as input, return its matrices one by one on subsequent calls
+// to getNext
+class TemplateDataSource : public DataSource
+{
+public:
+    TemplateDataSource() { current_idx = INT_MAX; }
+
+    FrameData * getNext()
+    {
+        if (!isOpen())
+            return NULL;
+
+        FrameData * output = new FrameData();
+        output->data.append(basis[current_idx]);
+        current_idx++;
+
+        output->sequenceNumber = next_sequence;
+        next_sequence++;
+
+        return output;
+    }
+
+    bool open(Template &input)
+    {
+        basis = input;
+        current_idx = 0;
+        next_sequence = 0;
+        return isOpen();
+    }
+
+    bool isOpen() { return current_idx < basis.size() ; }
+
+    void close()
+    {
+        current_idx = INT_MAX;
+        basis.clear();
+    }
+
+private:
+    Template basis;
+    int current_idx;
+    int next_sequence;
+};
+
+// Given a template as input, create a VideoDataSource or a TemplateDataSource
+// depending on whether or not it looks like the input template has already
+// loaded frames into memory.
+class DataSourceManager : public DataSource
+{
+public:
+    DataSourceManager()
+    {
+        actualSource = NULL;
+    }
+
+    ~DataSourceManager()
+    {
+        close();
+    }
+
+    FrameData * getNext()
+    {
+        if (!isOpen()) return NULL;
+        return actualSource->getNext();
+    }
+
+    void close()
+    {
+        if (actualSource) {
+            actualSource->close();
+            delete actualSource;
+            actualSource = NULL;
+        }
+    }
+
+    bool open(Template & input)
+    {
+        close();
+        bool open_res = false;
+        // Input has no matrices? Its probably a video that hasn't been loaded yet
+        if (input.empty()) {
+            actualSource = new VideoDataSource();
+            open_res = actualSource->open(input);
+            qDebug("created video resource status %d", open_res);
+        }
+        else {
+            // create frame dealer
+            actualSource = new TemplateDataSource();
+            open_res = actualSource->open(input);
+        }
+        if (!isOpen()) {
+            delete actualSource;
+            actualSource = NULL;
+            return false;
+        }
+        return true;
+    }
+
+    bool isOpen() { return !actualSource ? false : actualSource->isOpen(); }
+
+protected:
+    DataSource * actualSource;
+};
+
+class ProcessingStage : public QRunnable
+{
+    friend class StreamTransform;
+public:
+    ProcessingStage()
+    {
+        setAutoDelete(false);
+    }
+
+    void markStart()
+    {
+        QMutexLocker lock(&stoppedGuard);
+        stopped = false;
+    }
+
+    void waitStop()
+    {
+        stoppedGuard.lock();
+        while (!stopped)
+        {
+            waitStopped.wait(&stoppedGuard);
+        }
+        stoppedGuard.unlock();
+    }
+
+protected:
+    void markStop()
+    {
+        QMutexLocker lock(&stoppedGuard);
+        stopped = true;
+        this->waitStopped.wakeAll();
+    }
+    QMutex stoppedGuard;
+    QWaitCondition waitStopped;
+    bool stopped;
+
+    SharedBuffer * inputBuffer;
+    SharedBuffer * outputBuffer;
+    Transform * transform;
+    int stage_id;
+
+public:
+    // We should start, and enter a wait on input data
+    void run()
+    {
+        forever
+        {
+            FrameData * currentItem = inputBuffer->getItem();
+            if (currentItem == NULL)
+                break;
+            // Project the input we got
+            transform->projectUpdate(currentItem->data);
+            // Add the result to the ouptut buffer
+            outputBuffer->addItem(currentItem);
+        }
+        markStop();
+    }
+
+};
+
+// No input buffer, instead we draw templates from some data source
+// Will be operated by the main thread for the stream
+class FirstStage : public ProcessingStage
+{
+public:
+    DataSourceManager dataSource;
+    // Start drawing frames from the datasource.
+    void run()
+    {
+        forever
+        {
+            FrameData * aFrame = dataSource.getNext();
+            if (aFrame == NULL)
+                break;
+            outputBuffer->addItem(aFrame);
+        }
+        this->markStop();
+    }
+};
+
+class LastStage : public ProcessingStage
+{
+public:
+    TemplateList getOutput()
+    {
+        return collectedOutput;
+    }
+
+private:
+    TemplateList collectedOutput;
+public:
+    void run()
+    {
+        forever
+        {
+            // Wait for input
+            FrameData * frame = inputBuffer->getItem();
+            if (frame == NULL)
+                break;
+            // Just put the item on collectedOutput
+            collectedOutput.append(frame->data);
+            delete frame;
+        }
+        this->markStop();
+    }
+};
+
+class StreamTransform : public CompositeTransform
+{
+    Q_OBJECT
+public:
+    void train(const TemplateList & data)
+    {
+        foreach(Transform * transform, transforms) {
+            transform->train(data);
+        }
+    }
+
+    bool timeVarying() const { return true; }
+
+    void project(const Template &src, Template &dst) const
+    {
+        (void) src; (void) dst;
+        qFatal("nope");
+    }
+    void project(const TemplateList & src, TemplateList & dst) const
+    {
+        (void) src; (void) dst;
+        qFatal("nope");
+    }
+
+    void projectUpdate(const Template &src, Template &dst)
+    {
+        (void) src; (void) dst;
+        qFatal("whatever");
+    }
+
+    // start processing
+    void projectUpdate(const TemplateList & src, TemplateList & dst)
+    {
+        if (src.size() != 1)
+            qFatal("Expected single template input to stream");
+
+        dst = src;
+        bool res = readStage.dataSource.open(dst[0]);
+        if (!res) {
+            qWarning("failed to stream template %s", qPrintable(dst[0].file.name));
+            return;
+        }
+
+        // Tell all buffers to expect input
+        for (int i=0; i < sharedBuffers.size(); i++) {
+            sharedBuffers[i]->startInput();
+        }
+
+        // Start our processing stages
+        for (int i=0; i < this->processingStages.size(); i++) {
+            processingStages[i]->markStart();
+            processingThreads.start(processingStages[i]);
+        }
+
+        // Start the final stage
+        collectionStage.markStart();
+        processingThreads.start(&collectionStage);
+
+        // Run the read stage ourselves
+        readStage.run();
+
+        // The read stage has stopped (since we ran the read stage).
+        // Step over the buffers, and call stoppedInput to tell the stage
+        // reading from each buffer that no more frames will be added after
+        // the current ones run out, then wait for the thread to finish.
+        for (int i =0; i < (sharedBuffers.size() - 1); i++) {
+            // Indicate that no more input will be available
+            sharedBuffers[i]->stoppedInput();
+
+            // Wait for the thread to finish.
+            this->processingStages[i]->waitStop();
+        }
+        // Wait for the collection stage to finish
+        sharedBuffers.last()->stoppedInput();
+        collectionStage.waitStop();
+
+        // dst is set to all output received by the final stage
+        dst = collectionStage.getOutput();
+    }
+
+    virtual void finalize(TemplateList & output)
+    {
+        (void) output;
+        // Not handling this yet -cao
+    }
+
+    // Create and link stages
+    void init()
+    {
+        // Set up the thread pool, 1 stage for each transform, as well as first
+        // and last stages, but the first stage is operated by the thread that
+        // calls project so the pool only needs nTransforms+1 total.
+        processingThreads.setMaxThreadCount(transforms.size() + 1);
+
+        stage_variance.reserve(transforms.size());
+        foreach (const br::Transform *transform, transforms) {
+            stage_variance.append(transform->timeVarying());
+        }
+
+        // buffer 0 -- output buffer for the read stage
+        sharedBuffers.append(new SingleBuffer());
+        readStage.outputBuffer = sharedBuffers.last();
+        readStage.stage_id = 0;
+
+        int next_stage_id = 1;
+
+        int lastBufferIdx = 0;
+        for (int i =0; i < transforms.size(); i++)
+        {
+            // Set up this stage
+            processingStages.append(new ProcessingStage());
+
+            processingStages.last()->stage_id = next_stage_id++;
+            processingStages.last()->inputBuffer = sharedBuffers[lastBufferIdx];
+            lastBufferIdx++;
+
+            sharedBuffers.append(new SingleBuffer());
+            processingStages.last()->outputBuffer = sharedBuffers.last();
+            processingStages.last()->transform = transforms[i];
+        }
+
+        collectionStage.inputBuffer = sharedBuffers.last();
+        collectionStage.stage_id = next_stage_id;
+    }
+
+    ~StreamTransform()
+    {
+        for (int i = 0; i < processingStages.size(); i++) {
+            delete processingStages[i];
+        }
+        for (int i = 0; i < sharedBuffers.size(); i++) {
+            delete sharedBuffers[i];
+        }
+
+    }
+
+protected:
+    QList<bool> stage_variance;
+
+    FirstStage readStage;
+    LastStage collectionStage;
+
+    QList<ProcessingStage *> processingStages;
+    QList<SharedBuffer *> sharedBuffers;
+
+    QThreadPool processingThreads;
+
+    void _project(const Template &src, Template &dst) const
+    {
+        (void) src; (void) dst;
+        qFatal("nope");
+    }
+    void _project(const TemplateList & src, TemplateList & dst) const
+    {
+        (void) src; (void) dst;
+        qFatal("nope");
+    }
+};
+
+BR_REGISTER(Transform, StreamTransform)
+
+
+} // namespace br
+
+#include "stream.moc"
+