Pipeline Module

Dependency graph execution for terrain rendering pipeline.

This module provides a lightweight task dependency system with automatic caching, staleness detection, and execution planning. No external build tools required - all logic stays in Python.

Useful for complex multi-view renders where intermediate results (DEM loading, transforms, mesh creation) can be cached and reused across different camera angles.

TerrainPipeline

class src.terrain.pipeline.TerrainPipeline(dem_dir=None, *, cache_enabled=True, force_rebuild=False, dem_cache_dir=None, mesh_cache_dir=None, verbose=True)[source]

Bases: object

Dependency graph executor for terrain rendering pipeline.

Manages task dependencies, caching, and execution planning without external build system dependencies. All logic stays in Python.

Features: - Automatic staleness detection via hashing - Dry-run execution plans - Reusable cached outputs across multiple views - Per-layer caching support

Tasks in pipeline: 1. load_dem: Load and merge SRTM tiles (cached) 2. apply_transforms: Reproject, flip, downsample (can be cached) 3. detect_water: Identify water bodies from DEM (can be cached) 4. create_mesh: Build Blender geometry (cached, reused across views) 5. render_view: Output to PNG (cached by view/render params)

Dependency graph executor for terrain rendering.

Pipeline tasks:

load_dem: Load and merge SRTM tiles (cached)
apply_transforms: Reproject, flip, downsample (cached)
detect_water: Identify water bodies (cached)
create_mesh: Build Blender geometry (cached, reused across views)
render_view: Output to PNG (cached per view/render params)

Features:

Automatic staleness detection via hashing
Dry-run execution plans (explain())
Reusable cached outputs across multiple views
Per-layer caching support

Example:

from src.terrain.pipeline import TerrainPipeline

# Create pipeline
pipeline = TerrainPipeline(
    dem_dir='data/dem/detroit',
    cache_enabled=True
)

# Show execution plan (dry run)
pipeline.explain('render_view')
# Output:
# Task load_dem: CACHED
# Task apply_transforms: CACHED
# Task detect_water: CACHED
# Task create_mesh: CACHED
# Task render_view: NEEDS COMPUTATION

# Execute pipeline
result = pipeline.execute('render_view')

# Get cache statistics
stats = pipeline.cache_stats()
print(f"Cache hits: {stats['hits']}")
print(f"Cache misses: {stats['misses']}")

# Clear cache for specific task
pipeline.clear_cache(task='render_view')

# Clear all caches
pipeline.clear_cache()

Multi-view rendering:

Cache mesh once, render multiple views:

pipeline = TerrainPipeline(dem_dir='data/dem')

# First view: mesh cached
pipeline.set_params('render_view', {'direction': 'south'})
pipeline.execute('render_view')

# Second view: reuses cached mesh (faster!)
pipeline.set_params('render_view', {'direction': 'north'})
pipeline.execute('render_view')

# Third view: reuses cached mesh again
pipeline.set_params('render_view', {'direction': 'east'})
pipeline.execute('render_view')

Parameters:

dem_dir (Path | str)
cache_enabled (bool)
force_rebuild (bool)
dem_cache_dir (Path | str)
mesh_cache_dir (Path | str)
verbose (bool)

__init__(dem_dir=None, *, cache_enabled=True, force_rebuild=False, dem_cache_dir=None, mesh_cache_dir=None, verbose=True)[source]

Initialize terrain pipeline.

Parameters:

dem_dir (Path | str) – Directory containing SRTM .hgt files
cache_enabled (bool) – Enable caching at all stages
force_rebuild (bool) – Force rebuild even if cache exists (bypasses cache checks)
dem_cache_dir (Path | str) – Custom DEM cache directory
mesh_cache_dir (Path | str) – Custom mesh cache directory
verbose (bool) – Print execution details

load_dem(pattern='*.hgt')[source]

Task: Load and cache raw DEM from SRTM tiles.

Returns:: (dem_array, affine_transform)
Parameters:: pattern (str)
Return type:: Tuple[ndarray, Affine]

apply_transforms(*, target_vertices=1382400, reproject_crs='EPSG:32617', scale_factor=0.0001)[source]

Task: Apply reprojection, flipping, and elevation scaling.

Parameters:

target_vertices (int) – Mesh density target
reproject_crs (str) – Target CRS for reprojection
scale_factor (float) – Elevation scaling factor

Returns:

(transformed_dem, transform)

Return type:

Tuple[ndarray, Affine]

detect_water(*, slope_threshold=0.01, fill_holes=True)[source]

Task: Detect water bodies using slope analysis on unscaled DEM.

Parameters:

slope_threshold (float) – Threshold for slope magnitude (Horn’s method)
fill_holes (bool) – Apply morphological smoothing

Returns:

water_mask (boolean array)

Return type:

ndarray

create_mesh(*, scale_factor=100.0, height_scale=4.0, center_model=True, boundary_extension=True, transform_params=None, water_params=None)[source]

Task: Create Blender mesh from DEM and water mask.

KEY: Mesh is identical for all views and cached at geometry level, not per-view. Different camera angles reuse same mesh.

Cache key includes ALL upstream parameters per dependency graph: - DEM source hash (load_dem) - Transform params (apply_transforms) - Water params (detect_water) - Mesh params (this task)

Parameters:

scale_factor (float) – XY scaling
height_scale (float) – Z scaling for height exaggeration
center_model (bool) – Center mesh at origin
boundary_extension (bool) – Extend boundary for better rendering
transform_params (dict) – Upstream transform parameters (for cache key)
water_params (dict) – Upstream water detection parameters (for cache key)

Returns:

Blender mesh object

render_view(*, view='south', width=960, height=720, distance=0.264, elevation=0.396, focal_length=15, camera_type='PERSP', samples=2048)[source]

Task: Render a view to PNG.

Parameters:

view (str) – Camera direction (north, south, east, west, above)
width (int) – Output width
height (int) – Output height
distance (float) – Camera distance multiplier
elevation (float) – Camera elevation multiplier
focal_length (float) – Focal length
camera_type (str) – PERSP or ORTHO
samples (int) – Render samples

Returns:

Path to rendered PNG

explain(task_name)[source]

Explain what would execute to build a task (show dependency tree).

Shows: - Task dependencies - Execution order - Which tasks would be computed vs cached

Parameters:: task_name (str)
Return type:: None

render_all_views(views=None)[source]

Render all views efficiently.

Builds mesh once, reuses for all views.

Parameters:: views (List[str]) – List of view names
Returns:: Dictionary mapping view names to output paths
Return type:: Dict[str, Path]

cache_stats()[source]

Get cache statistics.

Return type:: Dict

clear_cache()[source]

Clear all caches.

Return type:: int

TaskState

class src.terrain.pipeline.TaskState(name, depends_on=<factory>, params=<factory>, cached=False, computed=False, result=None, cache_key='')[source]

Bases: object

Represents execution state of a task.

Represents execution state of a pipeline task.

Attributes:

name: Task identifier
depends_on: List of task dependencies
params: Task parameters (for cache key)
cached: Whether result is cached
computed: Whether task was executed this run
result: Cached or computed result
cache_key: Hash for cache lookup

Parameters:

name (str)
depends_on (List[str])
params (Dict[str, Any])
cached (bool)
computed (bool)
result (Any)
cache_key (str)

name: str

depends_on: List[str]

params: Dict[str, Any]

cached: bool = False

computed: bool = False

result: Any = None

cache_key: str = ''

__init__(name, depends_on=<factory>, params=<factory>, cached=False, computed=False, result=None, cache_key='')

Parameters:

name (str)
depends_on (List[str])
params (Dict[str, Any])
cached (bool)
computed (bool)
result (Any)
cache_key (str)

Return type:

None

Pipeline Design Patterns

Pattern 1: Multi-view renders

Cache expensive mesh creation, render multiple camera angles:

pipeline = TerrainPipeline('data/dem')

for direction in ['north', 'south', 'east', 'west']:
    pipeline.set_params('render_view', {
        'direction': direction,
        'output': f'render_{direction}.png'
    })
    pipeline.execute('render_view')

Pattern 2: Parameter sweeps

Test different DEM smoothing parameters:

pipeline = TerrainPipeline('data/dem')

for strength in [0.5, 1.0, 1.5, 2.0]:
    pipeline.set_params('apply_transforms', {
        'smooth_strength': strength
    })
    pipeline.execute('render_view')
    # Mesh and render are regenerated, DEM loading is cached

Pattern 3: Development workflow

Fast iteration on render settings:

pipeline = TerrainPipeline('data/dem', cache_enabled=True)

# First run: everything computed
pipeline.execute('render_view')  # ~60s

# Tweak lighting
pipeline.set_params('render_view', {'sun_energy': 1.5})
pipeline.execute('render_view')  # ~5s (reuses mesh)

# Tweak camera
pipeline.set_params('render_view', {'camera_distance': 2.0})
pipeline.execute('render_view')  # ~5s (reuses mesh)

Cache Integration

TerrainPipeline integrates with:

DEMCache - DEM loading cache
MeshCache - Mesh geometry cache

All caches use hash-based validation for automatic staleness detection.

Cache invalidation:

Parameter changes invalidate dependent tasks
Source file changes invalidate all downstream tasks
Manual invalidation via clear_cache()

Performance Notes

Typical speedups (multi-view renders):

First view: ~60s (full pipeline)
Subsequent views: ~5s (cached mesh, only render changes)
10x speedup for camera/lighting iterations

Cache overhead:

Hash computation: ~10-50ms per task
Cache lookup: ~5-20ms per task
Total overhead: ~100-500ms for full pipeline

When to use TerrainPipeline:

Multi-view renders (same DEM, different cameras)
Parameter sweeps (testing smoothing, colors, etc.)
Development iteration (fast render feedback)

When NOT to use:

Single render (overhead > savings)
Rapidly changing DEMs
Memory-constrained systems (caches use disk space)