Skip to content

Processing pipelines#

pyramids.processing turns existing Dataset / FeatureCollection operations into named, self-describing tools that can be chained into a serializable pipeline and run (batched) over one or many inputs — a QGIS-Processing-style workflow layer built on the ops pyramids already owns.

The pieces:

  • a registry of named tools, each with a parameter schema and an input type (Dataset vs FeatureCollection);
  • a Pipeline — an ordered chain of (tool, parameters) steps, serializable to a portable YAML "model" file;
  • a batch run with an error policy and optional process-pool parallelism;
  • a provenance record per run (tool, parameters, timing) that can re-emit the exact pipeline that produced an output.

v1 tool allowlist#

v1 ships a curated, real-signature allowlist (see ADR 0007 for why it is hand-written rather than introspected). List it with pyramids tools:

tool input → output
slope, aspect, hillshade, focal_mean, focal_std Dataset → Array*
to_crs, resample, fill, sieve Dataset → Dataset
interpolate_to_raster FeatureCollection → Dataset
to_h3, voronoi, quadtree, with_centroid, with_coordinates FeatureCollection → FeatureCollection

* The terrain ops natively return a numpy array; inside a pipeline the runner materializes it back into a single-band, georeferenced Dataset (carrying the source raster's geotransform/CRS), so the result is writable to disk and can be chained into a further Dataset step.

The registry is extensible — register a ToolMetadata to add a tool.

Example (Python)#

A cross-type pipeline: interpolate scattered points onto a raster, then compute its slope. interpolate_to_raster is a FeatureCollection op that returns a Dataset, and slope runs on that Dataset — the runner dispatches each step to the correct input type automatically.

from pyramids.feature import FeatureCollection
from pyramids.processing import Pipeline, run

gauges = FeatureCollection.read_file("samples/elevation_points.geojson")

pipe = Pipeline([
    ("interpolate_to_raster", {"column": "elevation", "cell_size": 1000.0}),
    ("slope", {}),
])
pipe.to_yaml("elevation_slope.yaml")            # the portable "model"

result = run(pipe, gauges)                       # batch over one or many inputs
slope_raster = result.outputs[0]                 # a georeferenced Dataset (slope array materialized)
print(result.provenance[0].total_seconds)        # per-run timing

Example (CLI)#

pyramids tools                                   # list registered tools
pyramids tool interpolate_to_raster              # show a tool's parameters
pyramids run elevation_slope.yaml \
    --inputs "samples/*.geojson" --out out/      # batch over a glob, write to out/

run writes each output into --out as <source-stem>_<index> (the batch index keeps same-basename inputs from colliding); existing files in --out are overwritten, so re-running a batch into the same directory replaces the prior outputs. --on-error skip (default) collects failures and continues, --on-error raise fails fast, and --parallel (with --max-workers) fans the batch across a process pool (file-path inputs only — GDAL handles cannot cross process boundaries).

API#

pyramids.processing.Pipeline #

An ordered, validated chain of geoprocessing steps.

Parameters:

Name Type Description Default
steps Iterable[tuple[str, dict[str, Any]]]

An iterable of (tool, parameters) pairs.

 required

Raises:

Type Description
ValueError

If any step names an unknown tool or supplies parameters that fail the tool's schema (validation happens here, at construction).
Source code in src/pyramids/processing/pipeline.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
class Pipeline:
    """An ordered, validated chain of geoprocessing steps.

    Args:
        steps: An iterable of ``(tool, parameters)`` pairs.

    Raises:
        ValueError: If any step names an unknown tool or supplies parameters that fail
            the tool's schema (validation happens here, at construction).
    """

    def __init__(self, steps: Iterable[tuple[str, dict[str, Any]]]):
        built: list[Step] = []
        for index, item in enumerate(steps):
            try:
                name, parameters = item
            except (TypeError, ValueError) as exc:
                raise ValueError(
                    f"pipeline step {index} must be a (tool, parameters) pair, got {item!r}"
                ) from exc
            tool = resolve(name)
            if parameters is None:
                parameters = {}
            elif not isinstance(parameters, dict):
                raise ValueError(
                    f"pipeline step {index}: parameters must be a mapping, got "
                    f"{type(parameters).__name__}"
                )
            parameters = dict(parameters)
            validate_parameters(tool, parameters)
            built.append(Step(name, parameters))
        self._steps = built

    @property
    def steps(self) -> list[Step]:
        """An independent copy of the pipeline's steps.

        Returns fresh :class:`Step` objects with copied ``parameters`` dicts, so
        mutating the returned steps (or their parameters) never affects the pipeline.
        """
        return [Step(step.tool, dict(step.parameters)) for step in self._steps]

    def __iter__(self) -> Iterator[Step]:
        """Iterate over independent copies of the steps (see :attr:`steps`)."""
        return iter(self.steps)

    def __len__(self) -> int:
        return len(self._steps)

    def __repr__(self) -> str:
        inner = ", ".join(f"({s.tool!r}, {s.parameters!r})" for s in self._steps)
        return f"Pipeline([{inner}])"

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Pipeline):
            result: bool = NotImplemented
        else:
            mine = [(s.tool, s.parameters) for s in self._steps]
            theirs = [(s.tool, s.parameters) for s in other._steps]
            result = mine == theirs
        return result

    def to_dict(self) -> dict[str, Any]:
        """Return the pipeline as a plain, YAML-ready mapping.

        Each step's ``parameters`` is copied, and any ``os.PathLike`` value is coerced to
        a plain string so the mapping is safe to `yaml.safe_dump`.
        """
        return {
            "pipeline": [
                {
                    "tool": step.tool,
                    "parameters": _yaml_safe_parameters(step.parameters),
                }
                for step in self._steps
            ]
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> Pipeline:
        """Build a pipeline from a mapping produced by :meth:`to_dict`.

        Args:
            data: A mapping with a ``"pipeline"`` list of ``{"tool", "parameters"}``.

        Returns:
            A validated :class:`Pipeline`.

        Raises:
            ValueError: If ``data`` is not a mapping with a ``"pipeline"`` list, or
                a step is malformed (re-validated through the constructor).
        """
        if not isinstance(data, dict) or "pipeline" not in data:
            raise ValueError(
                "invalid pipeline data: expected a mapping with a 'pipeline' key"
            )
        raw = data["pipeline"]
        if not isinstance(raw, list):
            raise ValueError(
                "invalid pipeline data: 'pipeline' must be a list of steps"
            )
        steps: list[tuple[str, dict[str, Any]]] = []
        for index, step in enumerate(raw):
            if not isinstance(step, dict) or "tool" not in step:
                raise ValueError(
                    f"invalid pipeline step {index}: expected a mapping with a 'tool' key"
                )
            steps.append((step["tool"], step.get("parameters", {})))
        return cls(steps)

    def to_yaml(self, path: str) -> None:
        """Write the pipeline to a portable, version-controllable YAML file.

        Every step's parameters are re-validated with ``for_serialization=True`` first,
        so a pipeline carrying a non-serializable value (array / mask / callable /
        in-memory object) raises here instead of writing a file that cannot be
        loaded back.

        Args:
            path: Destination ``.yaml`` path.

        Raises:
            ValueError: If any step carries a non-serializable parameter value.
        """
        for step in self._steps:
            validate_parameters(
                resolve(step.tool), step.parameters, for_serialization=True
            )
        with open(path, "w", encoding="utf-8") as handle:
            yaml.safe_dump(self.to_dict(), handle, sort_keys=False)

    @classmethod
    def from_yaml(cls, path: str) -> Pipeline:
        """Read a pipeline from a YAML file written by :meth:`to_yaml`.

        Args:
            path: Path to a pipeline YAML file.

        Returns:
            A validated :class:`Pipeline`.

        Raises:
            ValueError: If the file is not a mapping with a ``"pipeline"`` list, or
                a step references an unknown tool / invalid parameters (re-validated via
                the constructor).
        """
        with open(path, encoding="utf-8") as handle:
            data = yaml.safe_load(handle)
        return cls.from_dict(data)

steps property #

An independent copy of the pipeline's steps.

Returns fresh :class:Step objects with copied parameters dicts, so mutating the returned steps (or their parameters) never affects the pipeline.

__iter__() #

Iterate over independent copies of the steps (see :attr:steps).

Source code in src/pyramids/processing/pipeline.py 
85
86
87
def __iter__(self) -> Iterator[Step]:
    """Iterate over independent copies of the steps (see :attr:`steps`)."""
    return iter(self.steps)

from_dict(data) classmethod #

Build a pipeline from a mapping produced by :meth:to_dict.

Parameters:

Name Type Description Default
data dict[str, Any]

A mapping with a "pipeline" list of {"tool", "parameters"}.

 required

Returns:

Type Description
Pipeline

A validated :class:Pipeline.

Raises:

Type Description
ValueError

If data is not a mapping with a "pipeline" list, or a step is malformed (re-validated through the constructor).
Source code in src/pyramids/processing/pipeline.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@classmethod
def from_dict(cls, data: dict[str, Any]) -> Pipeline:
    """Build a pipeline from a mapping produced by :meth:`to_dict`.

    Args:
        data: A mapping with a ``"pipeline"`` list of ``{"tool", "parameters"}``.

    Returns:
        A validated :class:`Pipeline`.

    Raises:
        ValueError: If ``data`` is not a mapping with a ``"pipeline"`` list, or
            a step is malformed (re-validated through the constructor).
    """
    if not isinstance(data, dict) or "pipeline" not in data:
        raise ValueError(
            "invalid pipeline data: expected a mapping with a 'pipeline' key"
        )
    raw = data["pipeline"]
    if not isinstance(raw, list):
        raise ValueError(
            "invalid pipeline data: 'pipeline' must be a list of steps"
        )
    steps: list[tuple[str, dict[str, Any]]] = []
    for index, step in enumerate(raw):
        if not isinstance(step, dict) or "tool" not in step:
            raise ValueError(
                f"invalid pipeline step {index}: expected a mapping with a 'tool' key"
            )
        steps.append((step["tool"], step.get("parameters", {})))
    return cls(steps)

from_yaml(path) classmethod #

Read a pipeline from a YAML file written by :meth:to_yaml.

Parameters:

Name Type Description Default
path str

Path to a pipeline YAML file.

 required

Returns:

Type Description
Pipeline

A validated :class:Pipeline.

Raises:

Type Description
ValueError

If the file is not a mapping with a "pipeline" list, or a step references an unknown tool / invalid parameters (re-validated via the constructor).
Source code in src/pyramids/processing/pipeline.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
@classmethod
def from_yaml(cls, path: str) -> Pipeline:
    """Read a pipeline from a YAML file written by :meth:`to_yaml`.

    Args:
        path: Path to a pipeline YAML file.

    Returns:
        A validated :class:`Pipeline`.

    Raises:
        ValueError: If the file is not a mapping with a ``"pipeline"`` list, or
            a step references an unknown tool / invalid parameters (re-validated via
            the constructor).
    """
    with open(path, encoding="utf-8") as handle:
        data = yaml.safe_load(handle)
    return cls.from_dict(data)

to_dict() #

Return the pipeline as a plain, YAML-ready mapping.

Each step's parameters is copied, and any os.PathLike value is coerced to a plain string so the mapping is safe to yaml.safe_dump.

Source code in src/pyramids/processing/pipeline.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def to_dict(self) -> dict[str, Any]:
    """Return the pipeline as a plain, YAML-ready mapping.

    Each step's ``parameters`` is copied, and any ``os.PathLike`` value is coerced to
    a plain string so the mapping is safe to `yaml.safe_dump`.
    """
    return {
        "pipeline": [
            {
                "tool": step.tool,
                "parameters": _yaml_safe_parameters(step.parameters),
            }
            for step in self._steps
        ]
    }

to_yaml(path) #

Write the pipeline to a portable, version-controllable YAML file.

Every step's parameters are re-validated with for_serialization=True first, so a pipeline carrying a non-serializable value (array / mask / callable / in-memory object) raises here instead of writing a file that cannot be loaded back.

Parameters:

Name Type Description Default
path str

Destination .yaml path.

 required

Raises:

Type Description
ValueError

If any step carries a non-serializable parameter value.
Source code in src/pyramids/processing/pipeline.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def to_yaml(self, path: str) -> None:
    """Write the pipeline to a portable, version-controllable YAML file.

    Every step's parameters are re-validated with ``for_serialization=True`` first,
    so a pipeline carrying a non-serializable value (array / mask / callable /
    in-memory object) raises here instead of writing a file that cannot be
    loaded back.

    Args:
        path: Destination ``.yaml`` path.

    Raises:
        ValueError: If any step carries a non-serializable parameter value.
    """
    for step in self._steps:
        validate_parameters(
            resolve(step.tool), step.parameters, for_serialization=True
        )
    with open(path, "w", encoding="utf-8") as handle:
        yaml.safe_dump(self.to_dict(), handle, sort_keys=False)

pyramids.processing.run(pipeline, inputs, *, on_error='skip', out=None, parallel=False, max_workers=None) #

Run pipeline over inputs and collect the results.

Parameters:

Name Type Description Default
pipeline Pipeline

The :class:Pipeline to apply to each input.

 required
inputs Any

A single path/object, a glob string, a list/tuple of those, or a DatasetCollection. A string containing glob metacharacters (*/?/[) is expanded as a glob — to open a literal path that contains those characters, pass it inside a list.

 required
on_error str

"skip" collects (source, exception) failures and continues; "raise" fails fast on the first error.

 'skip'
out str | None

Optional output directory; when given, each successful output is written there as <source-stem>_<index> (the batch index keeps same-basename inputs from colliding). Existing files at those paths are overwritten — re-running a batch into the same directory replaces prior outputs. Required when parallel=True.

 None
parallel bool

When True, run the batch across a process pool. Because GDAL handles cannot cross process boundaries, this requires file-path inputs and an out directory (outputs are written worker-side and RunResult.outputs holds the written paths, in input order — same as serial mode — not in-memory objects). Only registry tools registered at import (the allowlist) are available in workers. Note that under on_error="raise" the surfaced error is the first worker to complete, which need not be the first input's (serial raises the first input's error).

 False
max_workers int | None

Worker-process count for parallel=True — ignored in serial mode (default: the pool's default, ~CPU count).

 None

Returns:

Name Type Description
A RunResult

class:RunResult with the successful outputs (objects in serial mode,
RunResult

written paths in parallel mode), any failures, and per-input
RunResult

provenance.

Raises:

Type Description
ValueError

If on_error is not "skip"/"raise", or parallel is set without an out directory / with non-path inputs.
Exception

The first per-item error when on_error="raise".
Source code in src/pyramids/processing/runner.py 
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def run(
    pipeline: Pipeline,
    inputs: Any,
    *,
    on_error: str = "skip",
    out: str | None = None,
    parallel: bool = False,
    max_workers: int | None = None,
) -> RunResult:
    """Run ``pipeline`` over ``inputs`` and collect the results.

    Args:
        pipeline: The :class:`Pipeline` to apply to each input.
        inputs: A single path/object, a glob string, a list/tuple of those, or a
            ``DatasetCollection``. A string containing glob metacharacters
            (``*``/``?``/``[``) is expanded as a glob — to open a literal path that
            contains those characters, pass it inside a list.
        on_error: ``"skip"`` collects ``(source, exception)`` failures and
            continues; ``"raise"`` fails fast on the first error.
        out: Optional output directory; when given, each successful output is
            written there as ``<source-stem>_<index>`` (the batch index keeps
            same-basename inputs from colliding). Existing files at those paths are
            overwritten — re-running a batch into the same directory replaces prior
            outputs. Required when ``parallel=True``.
        parallel: When ``True``, run the batch across a process pool. Because GDAL
            handles cannot cross process boundaries, this requires **file-path**
            inputs and an ``out`` directory (outputs are written worker-side and
            ``RunResult.outputs`` holds the written paths, in input order — same as
            serial mode — not in-memory objects). Only registry tools registered at
            import (the allowlist) are available in workers. Note that under
            ``on_error="raise"`` the surfaced error is the first worker to *complete*,
            which need not be the first input's (serial raises the first input's error).
        max_workers: Worker-process count for ``parallel=True`` — ignored in serial
            mode (default: the pool's default, ~CPU count).

    Returns:
        A :class:`RunResult` with the successful ``outputs`` (objects in serial mode,
        written paths in ``parallel`` mode), any ``failures``, and per-input
        ``provenance``.

    Raises:
        ValueError: If ``on_error`` is not ``"skip"``/``"raise"``, or ``parallel``
            is set without an ``out`` directory / with non-path inputs.
        Exception: The first per-item error when ``on_error="raise"``.
    """
    if on_error not in {"skip", "raise"}:
        raise ValueError(f"on_error must be 'skip' or 'raise', got {on_error!r}")
    if parallel:
        if out is None:
            raise ValueError(
                "parallel=True requires an 'out' directory — outputs are written "
                "worker-side, not returned as objects"
            )
        if max_workers is not None and max_workers < 1:
            raise ValueError(f"max_workers must be >= 1, got {max_workers}")
    items = _resolve_inputs(inputs)
    if not items:
        raise ValueError("no inputs to process (the resolved input set is empty)")
    if parallel:
        assert out is not None  # guarded above; narrows str | None -> str for typing
        result = _execute_parallel(pipeline, items, on_error, out, max_workers)
    else:
        result = _execute_serial(pipeline, items, on_error, out)
    return result

pyramids.processing.RunResult dataclass #

The outcome of a batch :func:run.

Attributes:

Name Type Description
outputs list[Any]

The final object produced for each input that succeeded, in input order. In serial mode these are in-memory Dataset/FeatureCollection objects; in parallel mode they are the written output path strings (GDAL handles cannot cross a process boundary).
failures list[tuple[Any, Exception]]

(source, exception) pairs for inputs that failed under the "skip" policy.
provenance list[Provenance]

One :class:~pyramids.processing.provenance.Provenance record per successful input (tool, parameters, and timing per step).
Source code in src/pyramids/processing/runner.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@dataclass
class RunResult:
    """The outcome of a batch :func:`run`.

    Attributes:
        outputs: The final object produced for each input that succeeded, in input
            order. In serial mode these are in-memory `Dataset`/`FeatureCollection`
            objects; in `parallel` mode they are the written output **path strings**
            (GDAL handles cannot cross a process boundary).
        failures: ``(source, exception)`` pairs for inputs that failed under the
            ``"skip"`` policy.
        provenance: One :class:`~pyramids.processing.provenance.Provenance` record
            per successful input (tool, parameters, and timing per step).
    """

    outputs: list[Any] = field(default_factory=list)
    failures: list[tuple[Any, Exception]] = field(default_factory=list)
    provenance: list[Provenance] = field(default_factory=list)

    def __len__(self) -> int:
        """The number of successful outputs (failures are counted separately)."""
        return len(self.outputs)

    @property
    def ok(self) -> bool:
        """Whether every input succeeded (no collected failures)."""
        return not self.failures

ok property #

Whether every input succeeded (no collected failures).

__len__() #

The number of successful outputs (failures are counted separately).

Source code in src/pyramids/processing/runner.py 
54
55
56
def __len__(self) -> int:
    """The number of successful outputs (failures are counted separately)."""
    return len(self.outputs)

pyramids.processing.Provenance dataclass #

The recorded recipe + timing for one processed input.

Attributes:

Name Type Description
source str

A label for the input (a path, or a <Type> marker for an in-memory object).
steps list[StepRecord]

The executed :class:StepRecord entries, in order.
Source code in src/pyramids/processing/provenance.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@dataclass
class Provenance:
    """The recorded recipe + timing for one processed input.

    Attributes:
        source: A label for the input (a path, or a ``<Type>`` marker for an
            in-memory object).
        steps: The executed :class:`StepRecord` entries, in order.
    """

    source: str
    steps: list[StepRecord] = field(default_factory=list)

    @property
    def total_seconds(self) -> float:
        """Total wall-clock time across all steps."""
        return sum(record.seconds for record in self.steps)

    def to_pipeline(self) -> Pipeline:
        """Re-emit the exact :class:`Pipeline` that produced the output.

        Returns:
            A :class:`Pipeline` equal to the one that was run (so it round-trips
            through ``to_yaml``/``from_yaml`` and reproduces the result).
        """
        return Pipeline([(record.tool, record.parameters) for record in self.steps])

total_seconds property #

Total wall-clock time across all steps.

to_pipeline() #

Re-emit the exact :class:Pipeline that produced the output.

Returns:

Name Type Description
A Pipeline

class:Pipeline equal to the one that was run (so it round-trips
Pipeline

through to_yaml/from_yaml and reproduces the result).
Source code in src/pyramids/processing/provenance.py 
50
51
52
53
54
55
56
57
def to_pipeline(self) -> Pipeline:
    """Re-emit the exact :class:`Pipeline` that produced the output.

    Returns:
        A :class:`Pipeline` equal to the one that was run (so it round-trips
        through ``to_yaml``/``from_yaml`` and reproduces the result).
    """
    return Pipeline([(record.tool, record.parameters) for record in self.steps])

pyramids.processing.ToolMetadata dataclass #

Describe one named, addressable pyramids op.

Parameters:

Name Type Description Default
name str

The tool name used in a pipeline and on the CLI.

 required
input_type str

The object the tool runs on — "Dataset" or "FeatureCollection".

 required
output_type str

The object type the tool produces.

 required
parameters tuple[Parameter, ...]

The tool's parameters.

 ()
description str

Human-readable summary.

 ''
method str | None

The method name on the input object; defaults to name.

 None

Raises:

Type Description
ValueError

If input_type/output_type are not valid object types or a parameter name is duplicated.
Source code in src/pyramids/processing/schema.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
@dataclass(frozen=True)
class ToolMetadata:
    """Describe one named, addressable pyramids op.

    Args:
        name: The tool name used in a pipeline and on the CLI.
        input_type: The object the tool runs on — ``"Dataset"`` or
            ``"FeatureCollection"``.
        output_type: The object type the tool produces.
        parameters: The tool's parameters.
        description: Human-readable summary.
        method: The method name on the input object; defaults to ``name``.

    Raises:
        ValueError: If ``input_type``/``output_type`` are not valid object types or a
            parameter name is duplicated.
    """

    name: str
    input_type: str
    output_type: str
    parameters: tuple[Parameter, ...] = ()
    description: str = ""
    method: str | None = None

    def __post_init__(self) -> None:
        if self.input_type not in INPUT_TYPES:
            raise ValueError(
                f"tool {self.name!r}: input_type must be one of "
                f"{sorted(INPUT_TYPES)}, got {self.input_type!r}"
            )
        if self.output_type not in OUTPUT_TYPES:
            raise ValueError(
                f"tool {self.name!r}: output_type must be one of "
                f"{sorted(OUTPUT_TYPES)}, got {self.output_type!r}"
            )
        seen = [p.name for p in self.parameters]
        if len(seen) != len(set(seen)):
            raise ValueError(f"tool {self.name!r}: duplicate parameter names in {seen}")

    @property
    def method_name(self) -> str:
        """The method this tool invokes on its input object."""
        return self.method or self.name

    def param(self, name: str) -> Parameter | None:
        """Return the :class:`Parameter` named ``name`` (or ``None``)."""
        found = None
        for param in self.parameters:
            if param.name == name:
                found = param
                break
        return found

    def help(self) -> str:
        """Render a multi-line help block describing the tool and its parameters."""
        lines = [f"{self.name} ({self.input_type} -> {self.output_type})"]
        if self.description:
            lines.append(f"  {self.description}")
        if self.parameters:
            lines.append("  parameters:")
            lines.extend(f"    {p.help()}" for p in self.parameters)
        else:
            lines.append("  parameters: (none)")
        return "\n".join(lines)

method_name property #

The method this tool invokes on its input object.

help() #

Render a multi-line help block describing the tool and its parameters.

Source code in src/pyramids/processing/schema.py 
237
238
239
240
241
242
243
244
245
246
247
def help(self) -> str:
    """Render a multi-line help block describing the tool and its parameters."""
    lines = [f"{self.name} ({self.input_type} -> {self.output_type})"]
    if self.description:
        lines.append(f"  {self.description}")
    if self.parameters:
        lines.append("  parameters:")
        lines.extend(f"    {p.help()}" for p in self.parameters)
    else:
        lines.append("  parameters: (none)")
    return "\n".join(lines)

param(name) #

Return the :class:Parameter named name (or None).

Source code in src/pyramids/processing/schema.py 
228
229
230
231
232
233
234
235
def param(self, name: str) -> Parameter | None:
    """Return the :class:`Parameter` named ``name`` (or ``None``)."""
    found = None
    for param in self.parameters:
        if param.name == name:
            found = param
            break
    return found

pyramids.processing.Parameter dataclass #

Describe a single tool parameter.

Parameters:

Name Type Description Default
name str

The keyword-argument name passed to the underlying op.

 required
parameter_type str

One of :data:PARAMETER_TYPES.

 required
default Any

Display-only default shown in help. The runner passes only the parameters a step supplies, so this value is never itself applied — the runtime default is always whatever the underlying method uses. Set it to mirror that method's real default so help advertises the true value (e.g. "nearest neighbor" for a resampling method); leave it None only when the method's default is dynamic or unknown.

 None
optional bool

Whether the parameter may be omitted.

 True
description str

Human-readable help text.

 ''
choices tuple[str, ...] | None

Allowed values for an "OptionList" parameter.

 None
serializable bool | None

Override for whether the value can be serialized; when None it is derived from parameter_type.

 None

Raises:

Type Description
ValueError

If parameter_type is unknown or choices is given for a non-OptionList parameter.
Source code in src/pyramids/processing/schema.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
@dataclass(frozen=True)
class Parameter:
    """Describe a single tool parameter.

    Args:
        name: The keyword-argument name passed to the underlying op.
        parameter_type: One of :data:`PARAMETER_TYPES`.
        default: Display-only default shown in ``help``. The runner passes only the
            parameters a step supplies, so this value is never itself applied — the
            *runtime* default is always whatever the underlying method uses. Set it
            to **mirror** that method's real default so ``help`` advertises the true
            value (e.g. ``"nearest neighbor"`` for a resampling ``method``); leave it
            ``None`` only when the method's default is dynamic or unknown.
        optional: Whether the parameter may be omitted.
        description: Human-readable help text.
        choices: Allowed values for an ``"OptionList"`` parameter.
        serializable: Override for whether the value can be serialized; when
            ``None`` it is derived from ``parameter_type``.

    Raises:
        ValueError: If ``parameter_type`` is unknown or ``choices`` is given for a
            non-``OptionList`` parameter.
    """

    name: str
    parameter_type: str
    default: Any = None
    optional: bool = True
    description: str = ""
    choices: tuple[str, ...] | None = None
    serializable: bool | None = None

    def __post_init__(self) -> None:
        if self.parameter_type not in PARAMETER_TYPES:
            raise ValueError(
                f"unknown parameter_type {self.parameter_type!r} for parameter "
                f"{self.name!r}; valid: {sorted(PARAMETER_TYPES)}"
            )
        if self.choices is not None and self.parameter_type != "OptionList":
            raise ValueError(
                f"parameter {self.name!r}: choices are only valid for an "
                f"'OptionList' parameter_type, not {self.parameter_type!r}"
            )

    @property
    def is_serializable(self) -> bool:
        """Whether a value for this parameter can be written to a pipeline file."""
        if self.serializable is not None:
            result = self.serializable
        else:
            result = self.parameter_type in _SERIALIZABLE_TYPES
        return result

    def validate(self, value: Any) -> None:
        """Validate ``value`` against this parameter's type.

        Args:
            value: The value supplied for the parameter.

        Raises:
            ValueError: If ``value`` does not match ``parameter_type`` (this is what
                rejects a numpy array / mask / callable handed to a scalar param).
        """
        pt = self.parameter_type
        ok = True
        if pt == "Float":
            ok = isinstance(value, (int, float)) and not isinstance(value, bool)
        elif pt == "Integer":
            ok = isinstance(value, int) and not isinstance(value, bool)
        elif pt == "Boolean":
            ok = isinstance(value, bool)
        elif pt in {"String", "Field"}:
            ok = isinstance(value, str)
        elif pt == "OptionList":
            ok = isinstance(value, str) and (
                self.choices is None or value in self.choices
            )
        elif pt == "NewFile":
            ok = isinstance(value, (str, os.PathLike))
        # Raster/Vector accept an object or a path; left permissive, flagged non-serializable.
        if not ok:
            expected = pt if self.choices is None else f"one of {list(self.choices)}"
            raise ValueError(
                f"parameter {self.name!r} expects {expected}, got "
                f"{type(value).__name__} ({value!r})"
            )

    def coerce(self, raw: str) -> Any:
        """Coerce a raw CLI string into this parameter's type.

        Public API kept for a future CLI ``--set key=value`` path; not yet wired
        into a shipped command (the ``run`` subcommand reads typed parameters from YAML).

        Args:
            raw: The string value from the command line.

        Returns:
            The value converted to the parameter's Python type.

        Raises:
            ValueError: If ``raw`` cannot be converted (e.g. a non-numeric string
                for a ``"Float"`` parameter, or a value outside ``choices``).
        """
        pt = self.parameter_type
        if pt == "Float":
            result: Any = float(raw)
        elif pt == "Integer":
            result = int(raw)
        elif pt == "Boolean":
            low = raw.strip().lower()
            if low in {"1", "true", "yes", "on"}:
                result = True
            elif low in {"0", "false", "no", "off"}:
                result = False
            else:
                raise ValueError(f"parameter {self.name!r}: {raw!r} is not a boolean")
        else:
            result = raw
            if (
                pt == "OptionList"
                and self.choices is not None
                and raw not in self.choices
            ):
                raise ValueError(
                    f"parameter {self.name!r}: {raw!r} not in {list(self.choices)}"
                )
        return result

    def help(self) -> str:
        """Render a one-line CLI/help description of this parameter."""
        flag = "optional" if self.optional else "required"
        default = "" if self.default is None else f", default={self.default!r}"
        choices = "" if self.choices is None else f" {list(self.choices)}"
        desc = f" — {self.description}" if self.description else ""
        return f"{self.name} ({self.parameter_type}{choices}, {flag}{default}){desc}"

is_serializable property #

Whether a value for this parameter can be written to a pipeline file.

coerce(raw) #

Coerce a raw CLI string into this parameter's type.

Public API kept for a future CLI --set key=value path; not yet wired into a shipped command (the run subcommand reads typed parameters from YAML).

Parameters:

Name Type Description Default
raw str

The string value from the command line.

 required

Returns:

Type Description
Any

The value converted to the parameter's Python type.

Raises:

Type Description
ValueError

If raw cannot be converted (e.g. a non-numeric string for a "Float" parameter, or a value outside choices).
Source code in src/pyramids/processing/schema.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def coerce(self, raw: str) -> Any:
    """Coerce a raw CLI string into this parameter's type.

    Public API kept for a future CLI ``--set key=value`` path; not yet wired
    into a shipped command (the ``run`` subcommand reads typed parameters from YAML).

    Args:
        raw: The string value from the command line.

    Returns:
        The value converted to the parameter's Python type.

    Raises:
        ValueError: If ``raw`` cannot be converted (e.g. a non-numeric string
            for a ``"Float"`` parameter, or a value outside ``choices``).
    """
    pt = self.parameter_type
    if pt == "Float":
        result: Any = float(raw)
    elif pt == "Integer":
        result = int(raw)
    elif pt == "Boolean":
        low = raw.strip().lower()
        if low in {"1", "true", "yes", "on"}:
            result = True
        elif low in {"0", "false", "no", "off"}:
            result = False
        else:
            raise ValueError(f"parameter {self.name!r}: {raw!r} is not a boolean")
    else:
        result = raw
        if (
            pt == "OptionList"
            and self.choices is not None
            and raw not in self.choices
        ):
            raise ValueError(
                f"parameter {self.name!r}: {raw!r} not in {list(self.choices)}"
            )
    return result

help() #

Render a one-line CLI/help description of this parameter.

Source code in src/pyramids/processing/schema.py 
174
175
176
177
178
179
180
def help(self) -> str:
    """Render a one-line CLI/help description of this parameter."""
    flag = "optional" if self.optional else "required"
    default = "" if self.default is None else f", default={self.default!r}"
    choices = "" if self.choices is None else f" {list(self.choices)}"
    desc = f" — {self.description}" if self.description else ""
    return f"{self.name} ({self.parameter_type}{choices}, {flag}{default}){desc}"

validate(value) #

Validate value against this parameter's type.

Parameters:

Name Type Description Default
value Any

The value supplied for the parameter.

 required

Raises:

Type Description
ValueError

If value does not match parameter_type (this is what rejects a numpy array / mask / callable handed to a scalar param).
Source code in src/pyramids/processing/schema.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def validate(self, value: Any) -> None:
    """Validate ``value`` against this parameter's type.

    Args:
        value: The value supplied for the parameter.

    Raises:
        ValueError: If ``value`` does not match ``parameter_type`` (this is what
            rejects a numpy array / mask / callable handed to a scalar param).
    """
    pt = self.parameter_type
    ok = True
    if pt == "Float":
        ok = isinstance(value, (int, float)) and not isinstance(value, bool)
    elif pt == "Integer":
        ok = isinstance(value, int) and not isinstance(value, bool)
    elif pt == "Boolean":
        ok = isinstance(value, bool)
    elif pt in {"String", "Field"}:
        ok = isinstance(value, str)
    elif pt == "OptionList":
        ok = isinstance(value, str) and (
            self.choices is None or value in self.choices
        )
    elif pt == "NewFile":
        ok = isinstance(value, (str, os.PathLike))
    # Raster/Vector accept an object or a path; left permissive, flagged non-serializable.
    if not ok:
        expected = pt if self.choices is None else f"one of {list(self.choices)}"
        raise ValueError(
            f"parameter {self.name!r} expects {expected}, got "
            f"{type(value).__name__} ({value!r})"
        )

pyramids.processing.resolve(name) #

Return the :class:ToolMetadata registered under name.

Parameters:

Name Type Description Default
name str

The tool name referenced by a pipeline or the CLI.

 required

Returns:

Type Description
ToolMetadata

The registered :class:ToolMetadata.

Raises:

Type Description
ValueError

If no tool is registered under name (the message lists the available tool names).
Source code in src/pyramids/processing/registry.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def resolve(name: str) -> ToolMetadata:
    """Return the :class:`ToolMetadata` registered under ``name``.

    Args:
        name: The tool name referenced by a pipeline or the CLI.

    Returns:
        The registered :class:`ToolMetadata`.

    Raises:
        ValueError: If no tool is registered under ``name`` (the message lists the
            available tool names).
    """
    try:
        tool = _REGISTRY[name]
    except KeyError as exc:
        raise ValueError(
            f"unknown tool {name!r}; registered tools: {tool_names()}"
        ) from exc
    return tool

pyramids.processing.tool_names() #

Return the registered tool names, sorted.

Source code in src/pyramids/processing/registry.py 
59
60
61
def tool_names() -> list[str]:
    """Return the registered tool names, sorted."""
    return sorted(_REGISTRY)

pyramids.processing.catalog() #

Return a read-only {name: ToolMetadata} view of the registered tools.

Named catalog (not registry) so it does not shadow the pyramids.processing.registry submodule when re-exported at the package root.

Source code in src/pyramids/processing/registry.py 
64
65
66
67
68
69
70
def catalog() -> Mapping[str, ToolMetadata]:
    """Return a read-only ``{name: ToolMetadata}`` view of the registered tools.

    Named ``catalog`` (not ``registry``) so it does not shadow the
    ``pyramids.processing.registry`` submodule when re-exported at the package root.
    """
    return MappingProxyType(_REGISTRY)

pyramids.processing.register(tool) #

Add tool to the registry (overwriting any tool of the same name).

Source code in src/pyramids/processing/registry.py 
31
32
33
34
def register(tool: ToolMetadata) -> ToolMetadata:
    """Add ``tool`` to the registry (overwriting any tool of the same name)."""
    _REGISTRY[tool.name] = tool
    return tool