PDAL Pipeline Validation: Catch Errors Before Processing Point Clouds

import json
import jsonschema

PDAL_PIPELINE_SCHEMA = {
    "type": "array",
    "minItems": 2,
    "items": {
        "type": "object",
        "required": ["type"],
        "properties": {
            "type": {
                "type": "string",
                "pattern": "^(readers|filters|writers)\\."
            },
            "filename": {"type": "string"},
            "inputs": {"type": "array", "items": {"type": "string"}}
        }
    }
}

def validate_json_syntax(pipeline_json: str) -> list[dict]:
    """
    Parse and schema-validate a PDAL pipeline JSON string.
    Returns the parsed pipeline list on success; raises ValueError on failure.
    """
    try:
        pipeline_obj = json.loads(pipeline_json)
    except json.JSONDecodeError as exc:
        raise ValueError(f"JSON parse error at line {exc.lineno}: {exc.msg}") from exc

    try:
        jsonschema.validate(instance=pipeline_obj, schema=PDAL_PIPELINE_SCHEMA)
    except jsonschema.ValidationError as exc:
        raise ValueError(f"Pipeline schema violation: {exc.message}") from exc

    return pipeline_obj

import subprocess

def check_stage_compatibility(pipeline_json: str) -> dict:
    """
    Invoke PDAL's built-in validator to check stage dependencies.
    No data is read; only the pipeline graph is analyzed.
    Raises RuntimeError on incompatibility.
    """
    result = subprocess.run(
        ["pdal", "pipeline", "--validate", "--stdin"],
        input=pipeline_json,
        text=True,
        capture_output=True
    )
    if result.returncode != 0:
        raise RuntimeError(
            f"Stage dependency failure:\n{result.stderr.strip()}"
        )
    return {"status": "compatible", "details": result.stdout.strip()}

def verify_filter_stages(pipeline_obj: list[dict]) -> list[str]:
    """
    Scan filter stages for common configuration issues.
    Returns a list of warning strings; empty list means all checks passed.
    """
    warnings: list[str] = []

    for stage in pipeline_obj:
        stage_type = stage.get("type", "")

        if stage_type == "filters.range":
            limits = stage.get("limits", "")
            if not limits:
                warnings.append(
                    "filters.range: 'limits' is empty — all points will be dropped."
                )
            elif not any(op in limits for op in ["[", "("]):
                warnings.append(
                    f"filters.range: 'limits' value '{limits}' lacks interval brackets."
                )

        if stage_type == "filters.outlier":
            method = stage.get("method", "")
            if method not in ("statistical", "radius"):
                warnings.append(
                    f"filters.outlier: 'method' must be 'statistical' or 'radius', got '{method}'."
                )
            mean_k = stage.get("mean_k", 8)
            multiplier = stage.get("multiplier", 2.0)
            if mean_k < 2:
                warnings.append(
                    f"filters.outlier: mean_k={mean_k} is dangerously low; minimum useful value is 4."
                )
            if multiplier <= 0:
                warnings.append(
                    f"filters.outlier: multiplier={multiplier} must be positive."
                )

        if stage_type == "filters.smrf":
            slope = stage.get("slope", 0.15)
            window = stage.get("window", 18.0)
            if slope <= 0 or slope > 1.0:
                warnings.append(
                    f"filters.smrf: slope={slope} is outside the typical range (0.05–0.4)."
                )
            if window < 1.0:
                warnings.append(
                    f"filters.smrf: window={window} m is unusually small for ground filtering."
                )

    return warnings

import pdal
import psutil
import time
import logging

logger = logging.getLogger(__name__)

def profile_dry_run(
    pipeline_obj: list[dict],
    sample_file: str,
    memory_limit_mb: float = 2048.0
) -> dict:
    """
    Execute the pipeline against a sample file and measure resource usage.
    The first stage's filename is replaced with sample_file.
    Raises RuntimeError if memory consumption exceeds memory_limit_mb.
    """
    test_pipeline = [dict(stage) for stage in pipeline_obj]
    test_pipeline[0]["filename"] = sample_file

    process = psutil.Process()
    mem_before_mb = process.memory_info().rss / (1024 * 1024)

    t0 = time.perf_counter()
    pipeline = pdal.Pipeline(test_pipeline)
    count = pipeline.execute()
    elapsed = time.perf_counter() - t0

    mem_after_mb = process.memory_info().rss / (1024 * 1024)
    delta_mb = mem_after_mb - mem_before_mb

    result = {
        "points_processed": count,
        "duration_sec": round(elapsed, 3),
        "memory_delta_mb": round(delta_mb, 1),
        "memory_ok": delta_mb <= memory_limit_mb,
        "throughput_kpts_sec": round(count / max(elapsed, 0.001) / 1000, 1),
    }

    if not result["memory_ok"]:
        raise RuntimeError(
            f"Dry-run consumed {delta_mb:.0f} MB on {count:,} sample points, "
            f"exceeding limit of {memory_limit_mb:.0f} MB. "
            "Consider tiling input or reducing chunk_size."
        )

    logger.info(
        "Dry-run passed: %d pts in %.3fs, %.1f MB delta",
        count, elapsed, delta_mb
    )
    return result

import numpy as np

REQUIRED_DIMS = {"X", "Y", "Z", "Intensity", "Classification", "ReturnNumber"}

def verify_output_integrity(
    pipeline: pdal.Pipeline,
    expected_count: int,
    tolerance: float = 0.05,
    required_dims: set[str] | None = None
) -> dict:
    """
    Verify pipeline output after pipeline.execute() has been called.
    Args:
        pipeline:       Executed pdal.Pipeline instance.
        expected_count: Baseline point count to compare against.
        tolerance:      Acceptable fractional deviation (default 5%).
        required_dims:  Set of dimension names that must be present in output.
    Returns:
        Dict with per-check results; raises AssertionError on any failure.
    """
    if required_dims is None:
        required_dims = REQUIRED_DIMS

    arrays = pipeline.arrays
    if not arrays or len(arrays[0]) == 0:
        raise AssertionError("Pipeline produced zero output points.")

    actual_count = len(arrays[0])
    count_ratio = abs(actual_count - expected_count) / max(expected_count, 1)
    count_ok = count_ratio <= tolerance

    output_dims = set(arrays[0].dtype.names or [])
    missing_dims = required_dims - output_dims

    metadata_str = str(pipeline.metadata).lower()
    crs_ok = "epsg" in metadata_str or "wkt" in metadata_str

    report = {
        "actual_count": actual_count,
        "expected_count": expected_count,
        "count_deviation": round(count_ratio, 4),
        "count_ok": count_ok,
        "missing_dims": sorted(missing_dims),
        "dims_ok": len(missing_dims) == 0,
        "crs_ok": crs_ok,
    }

    failures = []
    if not count_ok:
        failures.append(
            f"Point count {actual_count:,} deviates {count_ratio:.1%} "
            f"from expected {expected_count:,} (tolerance {tolerance:.0%})."
        )
    if missing_dims:
        failures.append(f"Missing required dimensions: {sorted(missing_dims)}.")
    if not crs_ok:
        failures.append("No CRS reference found in pipeline metadata.")

    if failures:
        raise AssertionError("Output integrity checks failed:\n" + "\n".join(failures))

    return report

import json
import logging

logger = logging.getLogger(__name__)

def run_full_validation(
    pipeline_json: str,
    sample_file: str,
    expected_point_count: int,
    memory_limit_mb: float = 2048.0,
) -> dict:
    """
    Run all five validation phases for a PDAL pipeline definition.

    Args:
        pipeline_json:        Raw JSON string of the PDAL pipeline array.
        sample_file:          Path to a representative LAS/LAZ sample file.
        expected_point_count: Baseline point count from a reference run.
        memory_limit_mb:      Maximum permissible memory delta during dry-run.

    Returns:
        Consolidated validation report dict.

    Raises:
        ValueError on JSON/schema violations.
        RuntimeError on stage incompatibility or resource over-run.
        AssertionError on output integrity failures.
    """
    report: dict = {}

    # Phase 1: JSON syntax and schema
    logger.info("Phase 1: JSON syntax validation")
    pipeline_obj = validate_json_syntax(pipeline_json)
    report["json_valid"] = True

    # Phase 2: Stage dependency resolution via pdal --validate
    logger.info("Phase 2: Stage dependency check")
    compat = check_stage_compatibility(pipeline_json)
    report["stage_compat"] = compat["status"]

    # Phase 3: Filter parameter audit
    logger.info("Phase 3: Filter parameter verification")
    warnings = verify_filter_stages(pipeline_obj)
    report["filter_warnings"] = warnings
    if warnings:
        for w in warnings:
            logger.warning("Filter warning: %s", w)

    # Phase 4: Dry-run profiling
    logger.info("Phase 4: Dry-run profiling against %s", sample_file)
    import pdal
    dry = profile_dry_run(pipeline_obj, sample_file, memory_limit_mb)
    report["dry_run"] = dry

    # Phase 5: Output integrity
    logger.info("Phase 5: Output integrity verification")
    pipeline_instance = pdal.Pipeline(
        [dict(s) for s in pipeline_obj]
    )
    pipeline_instance[0]["filename"] = sample_file  # inject sample
    pipeline_instance.execute()
    integrity = verify_output_integrity(pipeline_instance, expected_point_count)
    report["integrity"] = integrity

    logger.info("All validation phases passed.")
    return report

import pdal
import numpy as np

def assert_pipeline_output(pipeline: pdal.Pipeline) -> None:
    """
    Quick post-execute assertions suitable for embedding in scripts.
    Call immediately after pipeline.execute().
    """
    arrays = pipeline.arrays
    assert arrays and len(arrays) > 0, "No output arrays — pipeline may have filtered all points."

    arr = arrays[0]
    assert len(arr) > 0, "Output point cloud is empty."

    # Verify canonical spatial dimensions are present and finite
    for dim in ("X", "Y", "Z"):
        assert dim in arr.dtype.names, f"Required dimension '{dim}' missing from output."
        vals = arr[dim]
        assert np.isfinite(vals).all(), f"Non-finite values detected in dimension '{dim}'."

    # Verify CRS round-trip (presence of spatial reference in metadata)
    meta_str = str(pipeline.metadata).lower()
    assert "spatialreference" in meta_str or "epsg" in meta_str, \
        "No spatial reference detected in pipeline metadata — CRS may have been dropped."

# .github/workflows/validate-pipelines.yml
name: Validate PDAL Pipelines
on:
  pull_request:
    paths:
      - "pipelines/**/*.json"
      - "tests/test_pipeline_validation.py"

jobs:
  validate:
    runs-on: ubuntu-22.04
    steps:
      - uses: actions/checkout@v4

      - name: Install PDAL and Python dependencies
        run: |
          sudo apt-get update -qq
          sudo apt-get install -y libpdal-dev pdal
          pip install pdal jsonschema psutil pytest numpy

      - name: Run pipeline validation suite
        run: pytest tests/test_pipeline_validation.py -v --tb=short