Inherit StreamableExecutor from PipelineExecutor

Author: gsakkis

pdal/PyPipeline.cpp

@@ -50,13 +50,6 @@ namespace python
 {
 
 
-void readPipeline(PipelineExecutor* executor, std::string json)
-{
-    std::stringstream strm(json);
-    executor->getManager().readPipeline(strm);
-}
-
-
 void addArrayReaders(PipelineExecutor* executor, std::vector<std::shared_ptr<Array>> arrays)
 {
     // Make the symbols in pdal_base global so that they're accessible

pdal/PyPipeline.hpp

@@ -45,7 +45,6 @@ namespace python
 
 class Array;
 
-void readPipeline(PipelineExecutor* executor, std::string json);
 void addArrayReaders(PipelineExecutor* executor, std::vector<std::shared_ptr<Array>> arrays);
 PyArrayObject* viewToNumpyArray(PointViewPtr view);
 PyArrayObject* meshToNumpyArray(const TriangularMesh* mesh);

pdal/StreamableExecutor.cpp

@@ -226,12 +226,9 @@ char *PythonPointTable::getPoint(PointId idx)
 
 // StreamableExecutor
 
-StreamableExecutor::StreamableExecutor(const char *json, int chunkSize, int prefetch) :
-    m_json(json), m_table(chunkSize, prefetch), m_manager(chunkSize),
-    m_log(Log::makeLog("pdal_python", &m_logStream))
-{
-    m_manager.setLog(m_log);
-}
+StreamableExecutor::StreamableExecutor(std::string const& json, point_count_t chunkSize, int prefetch) :
+    PipelineExecutor(json), m_table(chunkSize, prefetch)
+{}
 
 StreamableExecutor::~StreamableExecutor()
 {
@@ -258,51 +255,10 @@ PyArrayObject *StreamableExecutor::executeNext()
         m_thread->join();
         Py_END_ALLOW_THREADS
         m_thread.reset();
+        m_executed = true;
     }
     return arr;
 }
 
-std::string StreamableExecutor::getMetadata() const
-{
-    return pdal::Utils::toJSON(m_manager.getMetadata().clone("metadata"));
-}
-
-std::string StreamableExecutor::getPipeline() const
-{
-    std::stringstream strm;
-    pdal::PipelineWriter::writePipeline(m_manager.getStage(), strm);
-    return strm.str();
-}
-
-int StreamableExecutor::getLogLevel() const
-{
-    return static_cast<int>(m_log->getLevel());
-}
-
-std::string StreamableExecutor::getLog() const
-{
-    return m_logStream.str();
-}
-
-// Returns the active log level.
-int StreamableExecutor::setLogLevel(int level)
-{
-    if (level < 0 || level > 8)
-        return getLogLevel();
-
-    m_log->setLevel(static_cast<pdal::LogLevel>(level));
-    return level;
-}
-
-bool StreamableExecutor::validate()
-{
-    std::stringstream strm;
-    strm << m_json;
-    m_manager.readPipeline(strm);
-    m_manager.prepare();
-
-    return true;
-}
-
 } // namespace python
 } // namespace pdal

pdal/StreamableExecutor.hpp

@@ -43,6 +43,7 @@
 
 #include <numpy/arrayobject.h>
 
+#include <pdal/PipelineExecutor.hpp>
 #include <pdal/PipelineManager.hpp>
 #include <pdal/PointTable.hpp>
 
@@ -83,28 +84,17 @@ class PythonPointTable : public StreamPointTable
     std::queue<PyArrayObject *> m_arrays;
 };
 
-class StreamableExecutor
+class StreamableExecutor : public PipelineExecutor
 {
 public:
-    StreamableExecutor(const char *json, int chunkSize, int prefetch);
+    StreamableExecutor(std::string const& json, point_count_t chunkSize, int prefetch);
     ~StreamableExecutor();
 
-    PyArrayObject *executeNext();
-    std::string getMetadata() const;
-    std::string getPipeline() const;
-    std::string getSchema() const;
-    std::string getLog() const;
-    int getLogLevel() const;
-    int setLogLevel(int level);
-    bool validate();
+    PyArrayObject* executeNext();
 
 private:
-    std::string m_json;
     int m_prefetch;
     PythonPointTable m_table;
-    PipelineManager m_manager;
-    std::stringstream m_logStream;
-    LogPtr m_log;
     std::unique_ptr<std::thread> m_thread;
 };
 

pdal/__init__.py

@@ -1,18 +1,8 @@
 __version__ = "2.4.2"
-__all__ = [
-    "Pipeline",
-    "PipelineIterator",
-    "Stage",
-    "Reader",
-    "Filter",
-    "Writer",
-    "dimensions",
-    "info",
-]
+__all__ = ["Pipeline", "Stage", "Reader", "Filter", "Writer", "dimensions", "info"]
 
 from . import libpdalpython
 from .drivers import inject_pdal_drivers
-from .libpdalpython import PipelineIterator
 from .pipeline import Filter, Pipeline, Reader, Stage, Writer
 
 inject_pdal_drivers()

pdal/libpdalpython.pyx

@@ -94,6 +94,7 @@ cdef extern from "pdal/PipelineExecutor.hpp" namespace "pdal":
         PipelineExecutor(string) except +
         const PipelineManager& getManagerConst() except +
         bool executed() except +
+        void read() except +
         int execute() except +
         string getPipeline() except +
         string getMetadata() except +
@@ -102,63 +103,29 @@ cdef extern from "pdal/PipelineExecutor.hpp" namespace "pdal":
         void setLogLevel(int) except +
 
 
+cdef extern from "StreamableExecutor.hpp" namespace "pdal::python":
+    cdef cppclass StreamableExecutor(PipelineExecutor):
+        StreamableExecutor(string, int, int) except +
+        np.PyArrayObject* executeNext() except +
+
+
 cdef extern from "PyArray.hpp" namespace "pdal::python":
     cdef cppclass Array:
         Array(np.PyArrayObject*) except +
 
 
 cdef extern from "PyPipeline.hpp" namespace "pdal::python":
-    void readPipeline(PipelineExecutor*, string) except +
     void addArrayReaders(PipelineExecutor*, vector[shared_ptr[Array]]) except +
     np.PyArrayObject* viewToNumpyArray(PointViewPtr) except +
     np.PyArrayObject* meshToNumpyArray(const TriangularMesh*) except +
 
-cdef extern from "StreamableExecutor.hpp" namespace "pdal::python":
-    cdef cppclass StreamableExecutor:
-        StreamableExecutor(const char *, int, int) except +
-        np.PyArrayObject *executeNext() except +
-        bool validate() except +
-        string getMetadata() except +
-        string getPipeline() except +
-        string getSchema() except +
-        string getLog() except +
-        int getLogLevel() except +
-        int setLogLevel(int level) except +
-
-
-cdef class PipelineIterator:
-    cdef int _chunk_size
-    cdef StreamableExecutor *_executor;
-
-    def __cinit__(self, unicode json, list arrays=None, int chunk_size=10000, int prefetch=0):
-        self._chunk_size
-        self._executor = new StreamableExecutor(json.encode("UTF-8"), chunk_size, prefetch)
-#        self._add_arrays(arrays)
-
-    property chunk_size:
-        def __get__(self):
-            return self._chunk_size
-
-    def validate(self):
-        return self._executor.validate()
-
-    def __iter__(self):
-        while True:
-            arrPtr = self._executor.executeNext()
-            if arrPtr is NULL:
-                break
-            arr = <object>arrPtr
-            Py_DECREF(arr)
-            yield arr
-
 
 cdef class Pipeline:
-    cdef unique_ptr[PipelineExecutor] _executor
+    cdef unique_ptr[StreamableExecutor] _executor
     cdef vector[shared_ptr[Array]] _inputs
     cdef int _loglevel
-
-    def execute(self):
-        return self._get_executor().execute()
+    cdef int _chunk_size
+    cdef int _prefetch
 
     #========= writeable properties to be set before execution ===========================
 
@@ -182,6 +149,26 @@ cdef class Pipeline:
         self._loglevel = value
         self._del_executor()
 
+    @property
+    def chunk_size(self):
+        return self._chunk_size
+
+    @chunk_size.setter
+    def chunk_size(self, value):
+        assert value > 0
+        self._chunk_size = value
+        self._del_executor()
+
+    @property
+    def prefetch(self):
+        return self._prefetch
+
+    @prefetch.setter
+    def prefetch(self, value):
+        assert value >= 0
+        self._prefetch = value
+        self._del_executor()
+
     #========= readable properties to be read after execution ============================
 
     @property
@@ -222,6 +209,21 @@ cdef class Pipeline:
             Py_DECREF(output[-1])
         return output
 
+    #========= execution methods =========================================================
+
+    def execute(self):
+        return self._get_executor().execute()
+
+    def __iter__(self):
+        cdef StreamableExecutor* executor = self._get_executor()
+        while True:
+            arr_ptr = executor.executeNext()
+            if arr_ptr is NULL:
+                break
+            arr = <object>arr_ptr
+            Py_DECREF(arr)
+            yield arr
+
     #========= non-public properties & methods ===========================================
 
     def _get_json(self):
@@ -237,12 +239,12 @@ cdef class Pipeline:
     def _del_executor(self):
         self._executor.reset()
 
-    cdef PipelineExecutor* _get_executor(self) except NULL:
+    cdef StreamableExecutor* _get_executor(self) except NULL:
         if not self._executor:
-            json = self._get_json()
-            executor = new PipelineExecutor(json)
+            executor = new StreamableExecutor(self._get_json(),
+                                              self._chunk_size, self._prefetch)
             executor.setLogLevel(self._loglevel)
-            readPipeline(executor, json)
+            executor.read()
             addArrayReaders(executor, self._inputs)
             self._executor.reset(executor)
         return self._executor.get()

pdal/pipeline.py

@@ -28,6 +28,8 @@ def __init__(
         spec: Union[None, str, Sequence[Stage]] = None,
         arrays: Sequence[np.ndarray] = (),
         loglevel: int = logging.ERROR,
+        chunk_size: int = 10000,
+        prefetch: int = 0,
     ):
         super().__init__()
         self._stages: List[Stage] = []
@@ -37,6 +39,8 @@ def __init__(
                 self |= stage
         self.inputs = arrays
         self.loglevel = loglevel
+        self.chunk_size = chunk_size
+        self.prefetch = prefetch
 
     @property
     def stages(self) -> List[Stage]: