NEON orchestration¶

Module: spectralbridge.pipelines.pipeline.

cross_sensor_cal.pipelines.pipeline¶

Updated October 2025

This module implements the core single-flightline and multi-flightline processing pipeline for NEON hyperspectral flight lines. The pipeline is now:

1. ENVI export from NEON .h5
2. BRDF/topo correction parameter JSON build
3. BRDF + topographic correction
4. Sensor convolution/resampling
5. Parquet export for every ENVI reflectance product
6. DuckDB merge of original/corrected/resampled Parquet tables

Key guarantees: - The stages ALWAYS run in the order above. - Convolution/resampling ALWAYS uses the BRDF+topo corrected ENVI product (_brdfandtopo_corrected_envi.img/.hdr), never the raw .h5. - Each stage is idempotent: * If valid outputs already exist, that stage logs "✅ ... skipping" and returns. * If outputs are missing or corrupted, that stage recomputes them. - The pipeline is restart-safe. You can rerun go_forth_and_multiply() after an interruption and it will resume where it left off without recomputing successfully completed work. - All file naming and file locations are defined centrally by get_flightline_products(). Stages do not hardcode filenames directly.

Typical usage:

from pathlib import Path
from cross_sensor_cal.pipelines.pipeline import go_forth_and_multiply

go_forth_and_multiply(
    base_folder=Path("output_tester"),
    site_code="NIWO",
    year_month="2023-08",
    flight_lines=[
        "NEON_D13_NIWO_DP1_L019-1_20230815_directional_reflectance",
        "NEON_D13_NIWO_DP1_L020-1_20230815_directional_reflectance",
    ],
)

Logs will include messages like: "🔎 ENVI export target for ... is _envi.img / _envi.hdr" "✅ ENVI export already complete ... (skipping heavy export)" "✅ BRDF+topo correction already complete ... (skipping)" "🎯 Convolving corrected reflectance for ..." "🎉 Finished pipeline for "

These logs confirm both correct ordering and skip behavior.

API glossary · Workflow / scientific context

Signatures and documentation below are collected from the current source. Advanced objects are not newly declared stable by inclusion here.

Audience: advanced.

is_valid_envi_pair ¶

is_valid_envi_pair(img_path: Path, hdr_path: Path) -> bool

Return True when both ENVI files exist and are non-empty.

Audience: advanced.

convolve_resample_product ¶

convolve_resample_product(corrected_hdr_path: Path, sensor_srf: dict[str, ndarray], out_stem_resampled: Path, tile_y: int = 100, tile_x: int = 100, progress_label: str | None = None, *, interactive_mode: bool = True, log_every: int = 25, sensor_name: str | None = None, brightness_system_pair: str | None = None) -> dict[str, dict[int, float]]

Stream a corrected BSQ cube through supplied spectral-response weights into canonical target-sensor products.

Audience: advanced.

resample_to_sensor_bands ¶

resample_to_sensor_bands(*, corrected_img_path: Path, corrected_hdr_path: Path, sensor_name: str, method: str, sensor_entry: dict[str, Any] | None = None, resolved_name: str | None = None, sensor_library: dict[str, dict[str, Any]] | None = None) -> dict[str, np.ndarray]

Construct target-band response weights from source wavelength metadata and a registered sensor specification.

Audience: advanced.

write_resampled_product ¶

write_resampled_product(*, arr: dict[str, ndarray], out_path: Path, sensor_name: str, flight_stem: str, corrected_hdr_path: Path, interactive_mode: bool = True, log_every: int = 25) -> dict[str, dict[int, float]]

Delegate target-response convolution and writing using the canonical sensor output stem.

Audience: advanced.

write_resampled_envi_cube ¶

write_resampled_envi_cube(bandstack_array: dict[str, ndarray], img_path: Path, hdr_path: Path, *, corrected_hdr_path: Path, progress_label: str | None = None, interactive_mode: bool = True, log_every: int = 25, sensor_name: str | None = None) -> dict[str, dict[int, float]]

Persist a resampled sensor bandstack to ENVI .img/.hdr files.

Audience: advanced.

stage_download_h5 ¶

stage_download_h5(base_folder: Path, site_code: str, year_month: str, product_code: str, flight_stem: str) -> Path

Ensure <base_folder>/<flight_stem>.h5 exists and is non-empty.

Audience: advanced.

stage_export_envi_from_h5 ¶

stage_export_envi_from_h5(base_folder: Path, product_code: str, flight_stem: str, brightness_offset: float | None = None, *, parallel_mode: bool = False, recover_missing_raw: bool = True, source_h5: Path | str | None = None, sample_start: int | None = None, sample_stop: int | None = None, line_start: int | None = None, line_stop: int | None = None) -> tuple[Path, Path]

Ensure we have the uncorrected ENVI export (.img/.hdr) for this flightline.

Returns:

Type	Description
`(raw_img_path, raw_hdr_path)`

Behavior

Compute the canonical raw ENVI paths inside the per-flightline work directory.
If those paths already exist and validate, SKIP heavy export immediately. This prevents re-loading huge (>20 GB) hyperspectral cubes on reruns.
Otherwise, run neon_to_envi_no_hytools() ONE TIME to generate ENVI.
After export, if the canonical paths STILL aren't valid, raise RuntimeError that includes a diff of which new files appeared. That diff is then used to correct get_flightline_products(), so on the next run we really will skip.

Audience: advanced.

stage_build_and_write_correction_json ¶

stage_build_and_write_correction_json(*, base_folder: Path, product_code: str, flight_stem: str, raw_img_path: Path, raw_hdr_path: Path, use_ndvi_brdf_bins: bool = False, parallel_mode: bool = False, source_h5: Path | str | None = None, sample_start: int | None = None, sample_stop: int | None = None, line_start: int | None = None, line_stop: int | None = None) -> Path

Compute + persist BRDF/topo correction parameters (illumination geometry, slope/aspect, BRDF coefficients, etc.) before applying correction.

Writes the canonical correction JSON, returns its path. Skips if valid.

Audience: advanced.

stage_apply_brdf_topo_correction ¶

stage_apply_brdf_topo_correction(*, base_folder: Path, product_code: str, flight_stem: str, raw_img_path: Path, raw_hdr_path: Path, correction_json_path: Path, use_ndvi_brdf_bins: bool = False, topo_fit_mode: str = 'scene', parallel_mode: bool = False) -> tuple[Path, Path]

Apply BRDF + topographic correction using the precomputed JSON. Produces the canonical corrected ENVI: *_brdfandtopo_corrected_envi.img/.hdr

Returns (corrected_img_path, corrected_hdr_path). Skips if already valid.

Audience: advanced.

stage_convolve_all_sensors ¶

stage_convolve_all_sensors(*, base_folder: Path, product_code: str, flight_stem: str, corrected_img_path: Path | None = None, corrected_hdr_path: Path | None = None, resample_method: str | None = 'convolution', parallel_mode: bool = False, ray_cpus: int | None = None, merge_memory_limit_gb: float | str | None = 64.0, merge_threads: int | None = 4, merge_row_group_size: int | None = None, merge_temp_directory: Path | None = None, polygon_path: Path | str | None = None, polygon_pixel_size: tuple[int, int] | None = (4, 4), polygon_overwrite: bool = False, polygon_min_overlap: float = 0.0, polygon_search_buffer_m: float = 0.0, extraction_mode: str | None = None)

Convolve the BRDF+topo corrected ENVI cube into sensor bandstacks.

Audience: public. Also exported as spectralbridge.process_one_flightline.

process_one_flightline ¶

process_one_flightline(*, base_folder: Path, product_code: str, flight_stem: str, resample_method: str | None = 'convolution', brightness_offset: float | None = None, use_ndvi_brdf_bins: bool = False, parallel_mode: bool = False, parquet_chunk_size: int = 50000, ray_cpus: int | None = None, merge_memory_limit_gb: float | str | None = 64.0, merge_threads: int | None = 4, merge_row_group_size: int | None = None, merge_temp_directory: Path | None = None, polygon_path: Path | str | None = None, polygon_pixel_size: tuple[int, int] | None = (4, 4), polygon_overwrite: bool = False, polygon_min_overlap: float = 0.0, polygon_search_buffer_m: float = 0.0, extraction_mode: str | None = None, topo_fit_mode: str = 'scene', qa_mode: Literal['off', 'standard', 'deep'] = 'standard', source_h5: Path | str | None = None, sample_start: int | None = None, sample_stop: int | None = None, line_start: int | None = None, line_stop: int | None = None)

Run the structured, skip-aware workflow for a single flightline.

Parameters:

Name	Type	Description	Default
`base_folder`	`Path`	Local NEON download and per-flightline output root; canonical filename helpers determine stage paths.	required
`product_code`	`str`	NEON product identifier; the normal directional-reflectance workflow uses DP1.30006.001.	required
`flight_stem`	`str`	Canonical source flightline identifier used by naming helpers, excluding the .h5 extension.	required
`resample_method`	`str \| None`	Target-sensor method: 'convolution' (also selected by None), 'gaussian', 'straight', or the retained 'legacy'/'resample' adapter.	`'convolution'`
`brightness_offset`	`float \| None`	Optional legacy additive brightness offset in stored reflectance units; None preserves the normal correction behavior.	`None`
`use_ndvi_brdf_bins`	`bool`	Opt into the existing NDVI-stratified BRDF fitting behavior rather than the default fit.	`False`
`parallel_mode`	`bool`	Internal orchestration flag controlling nested execution; direct single-flightline callers normally retain False.	`False`
`parquet_chunk_size`	`int`	Positive pixel/window processing budget for bounded extraction; the default and interpretation follow the selected producer.	`50000`
`ray_cpus`	`int \| None`	Optional CPU allocation for Ray-backed work; None delegates to established resource selection.	`None`
`merge_memory_limit_gb`	`float \| str \| None`	DuckDB merge memory budget in GiB when numeric, or an accepted memory string; None follows the merge default.	`64.0`
`merge_threads`	`int \| None`	Positive DuckDB thread budget; None delegates to the established database default.	`4`
`merge_row_group_size`	`int \| None`	Positive rows per written Parquet row group; None in the merge workflow delegates to its established default.	`None`
`merge_temp_directory`	`Path \| None`	DuckDB spill directory; bulk temporary storage must be outside the read-only source tree.	`None`
`polygon_path`	`Path \| str \| None`	Existing polygon vector dataset for extraction; coordinates are handled by the established polygon workflow.	`None`
`polygon_pixel_size`	`tuple[int, int] \| None`	Requested (x, y) pixel-size pair for polygon indexing; None delegates to raster-derived sizing. Units follow the raster coordinate system.	`(4, 4)`
`polygon_overwrite`	`bool`	Rebuild existing polygon-derived products instead of reusing them.	`False`
`polygon_min_overlap`	`float`	Minimum fractional polygon/pixel overlap admitted by the existing extraction routine.	`0.0`
`polygon_search_buffer_m`	`float`	Search expansion in meters for polygon/raster matching; does not change source polygon geometry.	`0.0`
`extraction_mode`	`str \| None`	'full' writes pixel Parquet, 'polygon' requires polygons; None preserves the entry point's documented default extraction behavior.	`None`
`topo_fit_mode`	`str`	Topographic fit mode passed to correction/QA; the default 'scene' retains the existing scene-wide fit.	`'scene'`
`qa_mode`	`Literal['off', 'standard', 'deep']`	QA coverage: 'off', 'standard', or 'deep'; deep diagnostics do not replace scientific validation of correction assumptions.	`'standard'`
`source_h5`	`Path \| str \| None`	Optional explicit existing HDF5 backing this flightline, used for windowed half-route processing without copying the source.	`None`
`sample_start`	`int \| None`	Optional zero-based half-open bounds on the across-track sample axis; None uses the corresponding full-axis bound.	`None`
`sample_stop`	`int \| None`	Optional zero-based half-open bounds on the across-track sample axis; None uses the corresponding full-axis bound.	`None`
`line_start`	`int \| None`	Optional zero-based half-open bounds on the along-track line axis; None uses the corresponding full-axis bound.	`None`
`line_stop`	`int \| None`	Optional zero-based half-open bounds on the along-track line axis; None uses the corresponding full-axis bound.	`None`

Returns:

Type	Description
`None`	No in-memory result. Canonical ENVI, correction JSON, sensor products, extraction tables and QA artifacts are written or reused on disk.

Raises:

Type	Description
`FileNotFoundError`	If a required source or completed artifact is absent.
`RuntimeError`	If required processing or artifact validation cannot complete.
`ValueError`	If the supplied options, dimensions, support or provenance violate the input contract.

Notes

Enforced stage order: 1) export ENVI from H5 2) build correction JSON 3) apply BRDF + topographic correction 4) convolve / resample to target sensors 5) export Parquet sidecars for all reflectance ENVI outputs 6) merge Parquet sidecars into a DuckDB master table 7) optional polygon extraction 8) stage QA from on-disk artifacts, then the legacy flightline QA panel

Each stage: - uses get_flightline_products() for canonical file naming - checks if its outputs already exist and are valid - skips if possible - re-runs only missing or invalid work

Partially written / corrupt files will fail validation and trigger recomputation so partial runs recover safely.

sort_and_sync_files ¶

sort_and_sync_files(base_folder: str, remote_prefix: str = '', sync_files: bool = True)

Generate file sorting list and optionally sync files to iRODS using gocmd.

Parameters: - base_folder: Base directory containing processed files - remote_prefix: Optional custom path to add after i:/iplant/ for remote paths - sync_files: Whether to actually sync files (True) or just generate the list (False)

Audience: public. Also exported as spectralbridge.go_forth_and_multiply.

go_forth_and_multiply ¶

go_forth_and_multiply(base_folder: Path, site_code: str, year_month: str, flight_lines: list[str], *, product_code: str = 'DP1.30006.001', resample_method: str | None = 'convolution', brightness_offset: float | None = None, use_ndvi_brdf_bins: bool = False, max_workers: int = 8, parquet_chunk_size: int = 50000, engine: Literal['thread', 'process', 'ray'] = 'ray', merge_memory_limit_gb: float | str | None = 64.0, merge_threads: int | None = 4, merge_row_group_size: int | None = None, merge_temp_directory: Path | None = None, polygon_path: Path | str | None = None, polygon_pixel_size: tuple[int, int] | None = (4, 4), polygon_overwrite: bool = False, polygon_min_overlap: float = 0.0, polygon_search_buffer_m: float = 0.0, extraction_mode: str | None = None, topo_fit_mode: str = 'scene', qa_mode: Literal['off', 'standard', 'deep'] = 'standard', split_across_track: bool = False) -> None

High-level orchestrator for processing multiple flight lines.

Parameters:

Name	Type	Description	Default
`base_folder`	`Path`	Local NEON download and per-flightline output root; canonical filename helpers determine stage paths.	required
`site_code`	`str`	NEON site code used for source discovery, for example NIWO.	required
`year_month`	`str`	Acquisition month in YYYY-MM form used by NEON download discovery.	required
`flight_lines`	`list[str]`	Canonical NEON flightline identifiers to stage and process; source downloads are performed serially before parallel processing.	required
`product_code`	`str`	NEON product identifier; the normal directional-reflectance workflow uses DP1.30006.001.	`'DP1.30006.001'`
`resample_method`	`str \| None`	Target-sensor method: 'convolution' (also selected by None), 'gaussian', 'straight', or the retained 'legacy'/'resample' adapter.	`'convolution'`
`brightness_offset`	`float \| None`	Optional legacy additive brightness offset in stored reflectance units; None preserves the normal correction behavior.	`None`
`use_ndvi_brdf_bins`	`bool`	Opt into the existing NDVI-stratified BRDF fitting behavior rather than the default fit.	`False`
`max_workers`	`int`	Positive parallel worker budget; eight by default for the batch NEON orchestrator.	`8`
`parquet_chunk_size`	`int`	Positive pixel/window processing budget for bounded extraction; the default and interpretation follow the selected producer.	`50000`
`engine`	`Literal['thread', 'process', 'ray']`	Parallel backend: 'ray', 'thread', or 'process'; existing Ray state must be compatible with the chosen engine.	`'ray'`
`merge_memory_limit_gb`	`float \| str \| None`	DuckDB merge memory budget in GiB when numeric, or an accepted memory string; None follows the merge default.	`64.0`
`merge_threads`	`int \| None`	Positive DuckDB thread budget; None delegates to the established database default.	`4`
`merge_row_group_size`	`int \| None`	Positive rows per written Parquet row group; None in the merge workflow delegates to its established default.	`None`
`merge_temp_directory`	`Path \| None`	DuckDB spill directory; bulk temporary storage must be outside the read-only source tree.	`None`
`polygon_path`	`Path \| str \| None`	Existing polygon vector dataset for extraction; coordinates are handled by the established polygon workflow.	`None`
`polygon_pixel_size`	`tuple[int, int] \| None`	Requested (x, y) pixel-size pair for polygon indexing; None delegates to raster-derived sizing. Units follow the raster coordinate system.	`(4, 4)`
`polygon_overwrite`	`bool`	Rebuild existing polygon-derived products instead of reusing them.	`False`
`polygon_min_overlap`	`float`	Minimum fractional polygon/pixel overlap admitted by the existing extraction routine.	`0.0`
`polygon_search_buffer_m`	`float`	Search expansion in meters for polygon/raster matching; does not change source polygon geometry.	`0.0`
`extraction_mode`	`str \| None`	'full' writes pixel Parquet, 'polygon' requires polygons; None preserves the entry point's documented default extraction behavior.	`None`
`topo_fit_mode`	`str`	Topographic fit mode passed to correction/QA; the default 'scene' retains the existing scene-wide fit.	`'scene'`
`qa_mode`	`Literal['off', 'standard', 'deep']`	QA coverage: 'off', 'standard', or 'deep'; deep diagnostics do not replace scientific validation of correction assumptions.	`'standard'`
`split_across_track`	`bool`	Produce two canonical half-route windows from each original HDF5, retaining shared source provenance; False processes full scenes.	`False`

Returns:

Type	Description
`None`	No in-memory result. Requested flightline products and associated canonical sidecars are persisted under base_folder.

Raises:

Type	Description
`FileNotFoundError`	If a required source or completed artifact is absent.
`RuntimeError`	If required processing or artifact validation cannot complete.
`ValueError`	If the supplied options, dimensions, support or provenance violate the input contract.

Notes

Steps:

Ensure the NEON .h5 for each flight_stem is present locally via :func:stage_download_h5 (performed serially).
Run the per-flightline pipeline (ENVI export → corrections → resample → Parquet) by delegating to :func:process_one_flightline in parallel.

The function maintains canonical naming, idempotent stage behavior, and optional legacy resample translation. max_workers controls Ray CPU parallelism (default eight cores) and is interpreted as the worker budget for thread/process fallbacks. engine selects the parallel backend: Ray (default), threads, or multiprocessing.

split_across_track=False (default) processes each flight_lines entry as a full scene. When True, each original ID is downloaded once into <base_folder>/<flight_stem>.h5 and then processed as two renamed half-route products that share that H5:

north–south elongated footprints → _left / _right line windows (northern half, then southern half)
east–west elongated footprints → _left / _right sample windows

resample_translation_to_other_sensors ¶

resample_translation_to_other_sensors(base_folder: Path)

Run the retained legacy sensor-translation adapter across discovered corrected ENVI products.

Audience: advanced.

process_base_folder ¶

process_base_folder(base_folder: Path, polygon_layer: str, **kwargs)

Processes subdirectories in a base folder, finding raster files and applying analysis.

Audience: advanced.

process_all_subdirectories ¶

process_all_subdirectories(parent_directory: Path, polygon_path)

Searches and processes all subdirectories.

Audience: advanced.

jefe ¶

jefe(base_folder, site_code, year_month, flight_lines, polygon_layer_path: str, remote_prefix: str = '', sync_files: bool = True, resample_method: str = 'convolution', max_workers: int = 1, skip_download_if_present: bool = True, force_config: bool = False, brightness_offset: float = 0.0, use_ndvi_brdf_bins: bool = False, verbose: bool = False, engine: Literal['thread', 'process', 'ray'] = 'thread')

A control function that orchestrates the processing of spectral data. It first calls go_forth_and_multiply to generate necessary data and structures, then processes all subdirectories within the base_folder, and finally sorts and syncs files to iRODS.

Parameters: - base_folder (str): The base directory for both operations. - site_code (str): Site code for go_forth_and_multiply. - year_month (str): Year and month for go_forth_and_multiply. - flight_lines (list): A list of flight lines for go_forth_and_multiply. - polygon_layer_path (str): Path to polygon shapefile or GeoJSON. - remote_prefix (str): Optional custom path to add after i:/iplant/ for remote paths. - sync_files (bool): Whether to sync files to iRODS or just generate the list.