Add streamable property to Stage and Pipeline

Author: gsakkis

pdal/drivers.py

@@ -1,10 +1,12 @@
 import json
 import subprocess
 from dataclasses import dataclass, field
-from typing import Callable, ClassVar, Mapping, Optional, Sequence, Type
+from typing import Callable, ClassVar, FrozenSet, Mapping, Optional, Sequence, Type
 
 from .pipeline import Filter, Reader, Stage, Writer
 
+StreamableTypes: FrozenSet
+
 
 @dataclass
 class Option:
@@ -67,6 +69,7 @@ def inject_pdal_drivers() -> None:
             ).stdout
         )
     )
+    streamable = []
     for d in drivers:
         name = d["name"]
         d_options = [Option(**option_dict) for option_dict in (options.get(name) or ())]
@@ -78,3 +81,7 @@ def inject_pdal_drivers() -> None:
             pass
         driver = Driver(name, d["description"], d_options)
         setattr(driver.type, driver.short_name, staticmethod(driver.factory))
+        if d["streamable"]:
+            streamable.append(driver.name)
+    global StreamableTypes
+    StreamableTypes = frozenset(streamable)

pdal/pipeline.py

@@ -11,7 +11,7 @@
 except ModuleNotFoundError:  # pragma: no cover
     Mesh = None
 
-from . import libpdalpython
+from . import drivers, libpdalpython
 
 LogLevelToPDAL = {
     logging.ERROR: 0,
@@ -42,6 +42,10 @@ def __init__(
     def stages(self) -> List[Stage]:
         return list(self._stages)
 
+    @property
+    def streamable(self) -> bool:
+        return all(stage.streamable for stage in self._stages)
+
     @property
     def schema(self) -> Any:
         return json.loads(super().schema)
@@ -124,6 +128,10 @@ def __init__(self, **options: Any):
     def type(self) -> str:
         return cast(str, self._options["type"])
 
+    @property
+    def streamable(self) -> bool:
+        return self.type in drivers.StreamableTypes
+
     @property
     def tag(self) -> Optional[str]:
         return self._options.get("tag")

test/test_pipeline.py

@@ -223,6 +223,41 @@ def test_infer_stage_type(self):
         assert pdal.Reader().type == ""
         assert pdal.Writer().type == ""
 
+    def test_streamable(self):
+        """Can we distinguish streamable from non-streamable stages and pipeline"""
+        rs = pdal.Reader(type="readers.las", filename="foo")
+        assert rs.streamable is True
+        assert pdal.Reader.las("foo").streamable is True
+        assert pdal.Reader("foo.las").streamable is True
+
+        rn = pdal.Reader(type="readers.pts", filename="foo")
+        assert rn.streamable is False
+        assert pdal.Reader.pts("foo").streamable is False
+        assert pdal.Reader("foo.pts").streamable is False
+
+        fs = pdal.Filter(type="filters.crop")
+        assert fs.streamable is True
+        assert pdal.Filter.crop().streamable is True
+
+        fn = pdal.Filter(type="filters.cluster")
+        assert fn.streamable is False
+        assert pdal.Filter.cluster().streamable is False
+
+        ws = pdal.Writer(type="writers.ogr", filename="foo")
+        assert ws.streamable is True
+        assert pdal.Writer.ogr(filename="foo").streamable is True
+        assert pdal.Writer("foo.shp").streamable is True
+
+        wn = pdal.Writer(type="writers.glb", filename="foo")
+        assert wn.streamable is False
+        assert pdal.Writer.gltf("foo").streamable is False
+        assert pdal.Writer("foo.glb").streamable is False
+
+        assert (rs | fs | ws).streamable is True
+        assert (rn | fs | ws).streamable is False
+        assert (rs | fn | ws).streamable is False
+        assert (rs | fs | wn).streamable is False
+
     # fails against PDAL master; see https://github.com/PDAL/PDAL/issues/3566
     @pytest.mark.xfail
     @pytest.mark.parametrize("filename", ["reproject.json", "reproject.py"])