Refactor auto-injection of PDAL stage factory methods and include the respective options in their docstring

Author: gsakkis

pdal/__init__.py

@@ -1,9 +1,12 @@
 __version__ = "2.4.2"
-__all__ = ["Pipeline",  "Reader", "Filter", "Writer", "dimensions", "info"]
+__all__ = ["Pipeline", "Stage", "Reader", "Filter", "Writer", "dimensions", "info"]
 
+from .drivers import inject_pdal_drivers
 from .libpdalpython import getDimensions, getInfo
-from .pipeline import Pipeline, Reader, Filter, Writer
+from .pipeline import Filter, Pipeline, Reader, Stage, Writer
 
+inject_pdal_drivers()
 dimensions = getDimensions()
 info = getInfo()
-del getDimensions, getInfo
+
+del inject_pdal_drivers, getDimensions, getInfo

pdal/drivers.py

@@ -0,0 +1,88 @@
+import json
+import subprocess
+from dataclasses import dataclass, field
+from typing import Callable, ClassVar, Mapping, Optional, Sequence, Type
+
+from .pipeline import Filter, Reader, Stage, Writer
+
+
+@dataclass
+class Option:
+    name: str
+    description: str
+    default: Optional[str]
+
+    def __repr__(self) -> str:
+        if self.default is not None:
+            return f"{self.name}={self.default!r}: {self.description}"
+        else:
+            return f"{self.name}: {self.description}"
+
+
+@dataclass
+class Driver:
+    name: str
+    short_name: str = field(init=False)
+    type: Type[Stage] = field(init=False)
+    description: str
+    options: Optional[Sequence[Option]]
+
+    def __post_init__(self) -> None:
+        prefix, _, suffix = self.name.partition(".")
+        self.type = self._prefix_to_type[prefix]
+        self.short_name = suffix
+
+    @property
+    def factory(self) -> Callable[..., Stage]:
+        factory: Callable[..., Stage]
+        if self.type is Reader:
+            factory = lambda filename, **kwargs: Reader(
+                filename, type=self.name, **kwargs
+            )
+        elif self.type is Writer:
+            factory = lambda filename=None, **kwargs: Writer(
+                filename, type=self.name, **kwargs
+            )
+        else:
+            factory = lambda **kwargs: Filter(type=self.name, **kwargs)
+        factory.__name__ = self.short_name
+        factory.__qualname__ = f"{self.type.__name__}.{self.short_name}"
+        factory.__doc__ = self.description
+        if self.options:
+            factory.__doc__ += "\n\n"
+            factory.__doc__ += "\n".join(map(repr, self.options))
+        return factory
+
+    _prefix_to_type: ClassVar[Mapping[str, Type[Stage]]] = {
+        "readers": Reader,
+        "filters": Filter,
+        "writers": Writer,
+    }
+
+
+def inject_pdal_drivers() -> None:
+    drivers = json.loads(
+        subprocess.run(["pdal", "--drivers", "--showjson"], capture_output=True).stdout
+    )
+    options = dict(
+        json.loads(
+            subprocess.run(
+                ["pdal", "--options", "all", "--showjson"], capture_output=True
+            ).stdout
+        )
+    )
+    for d in drivers:
+        name = d["name"]
+        d_options = options.get(name)
+        if d_options is not None:
+            d_options = [
+                Option(o["name"], o["description"], o.get("default")) for o in d_options
+            ]
+            # move filename option first
+            try:
+                i = next(i for i, opt in enumerate(d_options) if opt.name == "filename")
+                d_options.insert(0, d_options.pop(i))
+            except StopIteration:
+                pass
+        driver = Driver(name, d["description"], d_options)
+        setattr(driver.type, driver.short_name, staticmethod(driver.factory))

pdal/pipeline.py

@@ -2,18 +2,13 @@
 
 import glob
 import json
-import subprocess
 from abc import ABC, abstractmethod
 from typing import Any, Container, Dict, Iterator, List, Optional, Sequence, Union, cast
 
 import numpy as np
 
 from . import libpdalpython
 
-_PDAL_DRIVERS = json.loads(
-    subprocess.run(["pdal", "--drivers", "--showjson"], capture_output=True).stdout
-)
-
 
 class Pipeline(libpdalpython.Pipeline):
     def __init__(
@@ -73,13 +68,6 @@ def _json(self) -> str:
 
 
 class Stage:
-    def __init_subclass__(cls, type_prefix: Optional[str] = None) -> None:
-        for driver in _PDAL_DRIVERS:
-            name = driver["name"]
-            prefix, _, suffix = name.partition(".")
-            if prefix == type_prefix:
-                cls._add_constructor(name, suffix, driver["description"])
-
     def __init__(self, **options: Any):
         self._options = options
 
@@ -106,14 +94,6 @@ def pipeline(self, *arrays: np.ndarray) -> Pipeline:
     def __or__(self, other: Union[Stage, Pipeline]) -> Pipeline:
         return Pipeline((self, other))
 
-    @classmethod
-    def _add_constructor(cls, type: str, name: str, description: str) -> None:
-        constructor = lambda cls, *args, **kwargs: cls(*args, **kwargs, type=type)
-        constructor.__name__ = name
-        constructor.__qualname__ = f"{cls.__name__}.{name}"
-        constructor.__doc__ = description
-        setattr(cls, name, classmethod(constructor))
-
 
 class InferableTypeStage(ABC, Stage):
     @staticmethod
@@ -129,19 +109,19 @@ def type(self) -> str:
             return self.infer_type(self._options["filename"])
 
 
-class Reader(InferableTypeStage, type_prefix="readers"):
+class Reader(InferableTypeStage):
     infer_type = staticmethod(libpdalpython.infer_reader_driver)
 
     def __init__(self, filename: str, **options: Any):
         super().__init__(filename=filename, **options)
 
 
-class Filter(Stage, type_prefix="filters"):
+class Filter(Stage):
     def __init__(self, type: str, **options: Any):
         super().__init__(type=type, **options)
 
 
-class Writer(InferableTypeStage, type_prefix="writers"):
+class Writer(InferableTypeStage):
     infer_type = staticmethod(libpdalpython.infer_writer_driver)
 
     def __init__(self, filename: Optional[str] = None, **options: Any):