Replace pdal::PipelineExecutor with a new pdal::python::PipelineExecutor

Author: gsakkis

pdal/PyPipeline.cpp

@@ -39,18 +39,82 @@
 #include <dlfcn.h>
 #endif
 
-#include <Python.h>
-
-#include <pdal/Stage.hpp>
-#include <pdal/pdal_features.hpp>
-
 namespace pdal
 {
 namespace python
 {
 
+PipelineExecutor::PipelineExecutor(
+    std::string const& json, std::vector<std::shared_ptr<Array>> arrays, int level)
+{
+    if (level < 0 || level > 8)
+        throw pdal_error("log level must be between 0 and 8!");
+
+    LogPtr log(Log::makeLog("pypipeline", &m_logStream));
+    log->setLevel(static_cast<pdal::LogLevel>(level));
+    m_manager.setLog(log);
+
+    std::stringstream strm;
+    strm << json;
+    m_manager.readPipeline(strm);
+
+    addArrayReaders(arrays);
+}
+
+
+point_count_t PipelineExecutor::execute()
+{
+    point_count_t count = m_manager.execute();
+    m_executed = true;
+    return count;
+}
+
+
+const PointViewSet& PipelineExecutor::views() const
+{
+    if (!m_executed)
+        throw pdal_error("Pipeline has not been executed!");
+
+    return m_manager.views();
+}
+
+
+std::string PipelineExecutor::getPipeline() const
+{
+    if (!m_executed)
+        throw pdal_error("Pipeline has not been executed!");
+
+    std::stringstream strm;
+    pdal::PipelineWriter::writePipeline(m_manager.getStage(), strm);
+    return strm.str();
+}
+
+
+std::string PipelineExecutor::getMetadata() const
+{
+    if (!m_executed)
+        throw pdal_error("Pipeline has not been executed!");
+
+    std::stringstream strm;
+    MetadataNode root = m_manager.getMetadata().clone("metadata");
+    pdal::Utils::toJSON(root, strm);
+    return strm.str();
+}
+
+
+std::string PipelineExecutor::getSchema() const
+{
+    if (!m_executed)
+        throw pdal_error("Pipeline has not been executed!");
+
+    std::stringstream strm;
+    MetadataNode root = pointTable().layout()->toMetadata().clone("schema");
+    pdal::Utils::toJSON(root, strm);
+    return strm.str();
+}
+
 
-void addArrayReaders(PipelineExecutor* executor, std::vector<std::shared_ptr<Array>> arrays)
+void PipelineExecutor::addArrayReaders(std::vector<std::shared_ptr<Array>> arrays)
 {
     // Make the symbols in pdal_base global so that they're accessible
     // to PDAL plugins.  Python dlopen's this extension with RTLD_LOCAL,
@@ -65,8 +129,7 @@ void addArrayReaders(PipelineExecutor* executor, std::vector<std::shared_ptr<Arr
     if (arrays.empty())
         return;
 
-    PipelineManager& manager = executor->getManager();
-    std::vector<Stage *> roots = manager.roots();
+    std::vector<Stage *> roots = m_manager.roots();
     if (roots.size() != 1)
         throw pdal_error("Filter pipeline must contain a single root stage.");
 
@@ -81,7 +144,7 @@ void addArrayReaders(PipelineExecutor* executor, std::vector<std::shared_ptr<Arr
             MemoryViewReader::Order::ColumnMajor);
         options.add("shape", MemoryViewReader::Shape(array->shape()));
 
-        Stage& s = manager.makeReader("", "readers.memoryview", options);
+        Stage& s = m_manager.makeReader("", "readers.memoryview", options);
         MemoryViewReader& r = dynamic_cast<MemoryViewReader &>(s);
         for (auto f : array->fields())
             r.pushField(f);
@@ -101,7 +164,7 @@ void addArrayReaders(PipelineExecutor* executor, std::vector<std::shared_ptr<Arr
         roots[0]->setInput(r);
     }
 
-    manager.validateStageOptions();
+    m_manager.validateStageOptions();
 }
 
 

pdal/PyPipeline.hpp

@@ -34,21 +34,45 @@
 
 #pragma once
 
-#include <pdal/PipelineExecutor.hpp>
-
+#include <pdal/PipelineManager.hpp>
 #include <numpy/arrayobject.h>
 
 namespace pdal
 {
 namespace python
 {
 
-class Array;
-
 PyObject* buildNumpyDescriptor(PointLayoutPtr layout);
-void addArrayReaders(PipelineExecutor* executor, std::vector<std::shared_ptr<Array>> arrays);
 PyArrayObject* viewToNumpyArray(PointViewPtr view);
 PyArrayObject* meshToNumpyArray(const TriangularMesh* mesh);
 
+class Array;
+
+class PDAL_DLL PipelineExecutor {
+public:
+    PipelineExecutor(std::string const& json, std::vector<std::shared_ptr<Array>> arrays, int level);
+    virtual ~PipelineExecutor() = default;
+
+    point_count_t execute();
+
+    bool executed() const { return m_executed; }
+    const PointViewSet& views() const;
+    std::string getPipeline() const;
+    std::string getMetadata() const;
+    std::string getSchema() const;
+    std::string getLog() const { return m_logStream.str(); }
+
+protected:
+    virtual ConstPointTableRef pointTable() const { return m_manager.pointTable(); }
+
+    pdal::PipelineManager m_manager;
+    bool m_executed = false;
+
+private:
+    void addArrayReaders(std::vector<std::shared_ptr<Array>> arrays);
+
+    std::stringstream m_logStream;
+};
+
 } // namespace python
 } // namespace pdal

pdal/StreamableExecutor.cpp

@@ -180,9 +180,17 @@ char *PythonPointTable::getPoint(PointId idx)
 
 // StreamableExecutor
 
-StreamableExecutor::StreamableExecutor(std::string const& json, point_count_t chunkSize, int prefetch) :
-    PipelineExecutor(json), m_table(chunkSize, prefetch)
-{}
+StreamableExecutor::StreamableExecutor(std::string const& json,
+                                       std::vector<std::shared_ptr<Array>> arrays,
+                                       int level,
+                                       point_count_t chunkSize,
+                                       int prefetch)
+    : PipelineExecutor(json, arrays, level)
+    , m_table(chunkSize, prefetch)
+{
+    if (!m_manager.pipelineStreamable())
+        throw pdal_error("Pipeline is not streamable");
+}
 
 StreamableExecutor::~StreamableExecutor()
 {
@@ -227,16 +235,5 @@ void StreamableExecutor::done()
     m_executed = true;
 }
 
-std::string StreamableExecutor::getSchema() const
-{
-    if (!m_executed)
-        throw pdal_error("Pipeline has not been executed!");
-
-    std::stringstream strm;
-    MetadataNode root = m_table.layout()->toMetadata().clone("schema");
-    pdal::Utils::toJSON(root, strm);
-    return strm.str();
-}
-
 } // namespace python
 } // namespace pdal

pdal/StreamableExecutor.hpp

@@ -34,18 +34,10 @@
 
 #pragma once
 
-#include <mutex>
 #include <condition_variable>
-#include <queue>
-#include <memory>
 #include <thread>
-#include <sstream>
 
-#include <numpy/arrayobject.h>
-
-#include <pdal/PipelineExecutor.hpp>
-#include <pdal/PipelineManager.hpp>
-#include <pdal/PointTable.hpp>
+#include "PyPipeline.hpp"
 
 namespace pdal
 {
@@ -85,7 +77,11 @@ class PythonPointTable : public StreamPointTable
 class StreamableExecutor : public PipelineExecutor
 {
 public:
-    StreamableExecutor(std::string const& json, point_count_t chunkSize, int prefetch);
+    StreamableExecutor(std::string const& json,
+                       std::vector<std::shared_ptr<Array>> arrays,
+                       int level,
+                       point_count_t chunkSize,
+                       int prefetch);
     ~StreamableExecutor();
 
     std::string getSchema() const;
@@ -94,6 +90,7 @@ class StreamableExecutor : public PipelineExecutor
 
 private:
     void done();
+    ConstPointTableRef pointTable() const { return m_table; }
 
     PythonPointTable m_table;
     std::unique_ptr<std::thread> m_thread;

pdal/libpdalpython.pyx

@@ -86,29 +86,24 @@ cdef extern from "pdal/PointView.hpp" namespace "pdal":
     ctypedef cpp_set[PointViewPtr] PointViewSet
 
 
-cdef extern from "pdal/PipelineManager.hpp" namespace "pdal":
-    cdef cppclass PipelineManager:
-        const PointViewSet& views() const
-        bool pipelineStreamable() const
-
+cdef extern from "PyPipeline.hpp" namespace "pdal::python":
+    np.PyArrayObject* viewToNumpyArray(PointViewPtr) except +
+    np.PyArrayObject* meshToNumpyArray(const TriangularMesh*) except +
 
-cdef extern from "pdal/PipelineExecutor.hpp" namespace "pdal":
     cdef cppclass PipelineExecutor:
-        PipelineExecutor(string) except +
-        const PipelineManager& getManagerConst() except +
-        bool executed() except +
-        void read() except +
+        PipelineExecutor(string, vector[shared_ptr[Array]], int) except +
         int execute() except +
+        bool executed() except +
         string getPipeline() except +
         string getMetadata() except +
         string getSchema() except +
         string getLog() except +
-        void setLogLevel(int) except +
+        const PointViewSet& views() except +
 
 
 cdef extern from "StreamableExecutor.hpp" namespace "pdal::python":
     cdef cppclass StreamableExecutor(PipelineExecutor):
-        StreamableExecutor(string, int, int) except +
+        StreamableExecutor(string, vector[shared_ptr[Array]], int, int, int) except +
         np.PyArrayObject* executeNext() except +
         void stop() except +
 
@@ -118,12 +113,6 @@ cdef extern from "PyArray.hpp" namespace "pdal::python":
         Array(np.PyArrayObject*) except +
 
 
-cdef extern from "PyPipeline.hpp" namespace "pdal::python":
-    void addArrayReaders(PipelineExecutor*, vector[shared_ptr[Array]]) except +
-    np.PyArrayObject* viewToNumpyArray(PointViewPtr) except +
-    np.PyArrayObject* meshToNumpyArray(const TriangularMesh*) except +
-
-
 @cython.internal
 cdef class PipelineResultsMixin:
     @property
@@ -179,21 +168,17 @@ cdef class Pipeline(PipelineResultsMixin):
     @property
     def arrays(self):
         cdef PipelineExecutor* executor = self._get_executor()
-        if not executor.executed():
-            raise RuntimeError("Pipeline has not been executed!")
         output = []
-        for view in executor.getManagerConst().views():
+        for view in executor.views():
             output.append(<object>viewToNumpyArray(view))
             Py_DECREF(output[-1])
         return output
 
     @property
     def meshes(self):
         cdef PipelineExecutor* executor = self._get_executor()
-        if not executor.executed():
-            raise RuntimeError("Pipeline has not been executed!")
         output = []
-        for view in executor.getManagerConst().views():
+        for view in executor.views():
             output.append(<object>meshToNumpyArray(deref(view).mesh()))
             Py_DECREF(output[-1])
         return output
@@ -204,15 +189,10 @@ cdef class Pipeline(PipelineResultsMixin):
         return self._get_executor().execute()
 
     def iterator(self, int chunk_size=10000, int prefetch = 0):
-        cdef StreamableExecutor* executor = new StreamableExecutor(
-            self._get_json(), chunk_size, prefetch
-        )
-        self._configure_executor(executor)
-        if not executor.getManagerConst().pipelineStreamable():
-            raise RuntimeError("Pipeline is not streamable")
-
         it = PipelineIterator()
-        it.set_executor(executor)
+        it.set_executor(new StreamableExecutor(
+            self._get_json(), self._inputs, self._loglevel, chunk_size, prefetch
+        ))
         return it
 
     #========= non-public properties & methods ===========================================
@@ -232,16 +212,11 @@ cdef class Pipeline(PipelineResultsMixin):
 
     cdef PipelineExecutor* _get_executor(self) except NULL:
         if not self._executor:
-            executor = new PipelineExecutor(self._get_json())
-            self._configure_executor(executor)
-            self._executor.reset(executor)
+            self._executor.reset(new PipelineExecutor(
+                self._get_json(), self._inputs, self._loglevel)
+            )
         return self._executor.get()
 
-    cdef _configure_executor(self, PipelineExecutor* executor):
-        executor.setLogLevel(self._loglevel)
-        executor.read()
-        addArrayReaders(executor, self._inputs)
-
 
 @cython.internal
 cdef class PipelineIterator(PipelineResultsMixin):
@@ -250,10 +225,6 @@ cdef class PipelineIterator(PipelineResultsMixin):
     cdef set_executor(self, StreamableExecutor* executor):
         self._executor.reset(executor)
 
-    @property
-    def schema(self):
-        return json.loads(self._executor.get().getSchema())
-
     def __iter__(self):
         return self