gedih3.parallel#

Package-wide parallelism primitives for gedih3.

Houses the always-parallel parallel_map (originally introduced for gh3_doctor diagnoses; promoted here because the build, download, and manifest-writing paths all depend on it now) plus three parallel walker primitives used to regenerate SOC and H3-DB manifests via dask workers instead of driver-side serial recursive globs.

Design rules (carried forward from the v0.8.x build-pipeline lessons and the doctor refactor):

Always parallel. No serial fallback. A registered Client is required; library / notebook callers must wrap their call site in with Client(...) as c: .... Single code path means no quietly-different behavior on small inputs.

Workers receive only what they need. Filter args (exclude patterns, glob patterns, …) flow through parallel_map’s **broadcast kwargs — never via closure capture — so worker fns stay picklable as top-level module functions.

Stream ``as_completed`` instead of persist + compute. Findings accumulate as each future finishes; failures on individual items surface as exception objects in the yielded stream.

Driver-side aggregation is cheap. The walkers flatten + sort the per-leaf results on the driver as they arrive — no large materialized intermediates.

Fail loud. A worker exception in a walker aborts the walk; an incomplete manifest is a worse footgun than a slow re-walk.

Attributes#

logger

Functions#

`parallel_map`(→ Iterator[Tuple[Any, Any]])	Map `fn` across `items` on a dask cluster.
`walk_soc_parallel`(→ List[str])	Parallel year/doy walk over a SOC tree.
`walk_h3db_parallel`(→ List[str])	Parallel per-partition walk over an H3 database.
`walk_flat_parallel`(→ List[str])	Flat single-directory listing. Provided for API symmetry with the
`check_manifest_freshness`(→ bool)	Compare `mtime(manifest_path)` against `mtime(root_dir)`.

Module Contents#

gedih3.parallel.logger#

gedih3.parallel.parallel_map(items: List[Any], fn: Callable[Ellipsis, Any], *, args=None, desc: str = '', unit: str = 'item', batch_size: int = 0, **broadcast) → Iterator[Tuple[Any, Any]][source]#

Map fn across items on a dask cluster.

Yields (item, result) tuples in completion order. When fn raises on a worker, result is the exception instance — callers decide how to surface it (turn into a finding, log, or re-raise).

Requires a dask Client to be registered (via gedih3.utils.get_dask_client()). The gedih3 CLI tools create one at startup; library / notebook callers must wrap their call site in with Client(...) as client: .... Raises gedih3.exceptions.GediError when no client is registered.

Parameters:

itemslist: Iterable of work items (e.g. partition directory paths).
fncallable: Top-level (picklable) function. Signature: fn(item, **broadcast).
argsargparse.Namespace, optional: Accepted for API compatibility with callers that pass it; not consumed because the always-parallel path doesn’t need a tqdm progress bar — the dask dashboard is the live view.
desc, unitstr: Dispatch log prefix and unit name used in the periodic N/M done counter line.
batch_sizeint, optional: When >0, group items into batches of this size and dispatch one dask task per batch (each task internally iterates its slice). Use this when len(items) is in the hundreds-of-thousands — client.map builds a task graph proportional to len(items) and submission/scheduler overhead dominates the actual work beyond ~10k tasks.
**broadcast: Constant keyword arguments forwarded to every call of fn.

gedih3.parallel.walk_soc_parallel(soc_dir: str, *, pattern: str = 'GEDI*.h5', exclude: List[str] | None = None, batch_size: int = 32) → List[str][source]#

Parallel year/doy walk over a SOC tree.

Driver enumerates year and doy subdirs via os.scandir() (bounded: ~7 years × ~300 doys = ~2000 single-level syscalls on the driver). Workers run one non-recursive scandir per doy via _scan_doy_dir(), applying pattern and exclude locally so we never ship unwanted paths back across the network.

Always parallel — requires a registered dask Client.

gedih3.parallel.walk_h3db_parallel(db_root: str, *, pattern: str = '*.parquet', batch_size: int = 64) → List[str][source]#

Parallel per-partition walk over an H3 database.

Driver enumerates h3_NN=* partition directories via one os.scandir() on the DB root (typically ~10k entries at partition level 3). Workers recursively glob each partition (handles both flat and year=NNNN/-nested layouts).

Always parallel — requires a registered dask Client.

gedih3.parallel.walk_flat_parallel(dir_path: str, *, pattern: str = '*.parquet') → List[str][source]#: Flat single-directory listing. Provided for API symmetry with the other two walkers; no dask dispatch (the tree has a single leaf).

gedih3.parallel.check_manifest_freshness(manifest_path: str, root_dir: str, *, raise_on_stale: bool = False, remedy: str = '') → bool[source]#

Compare mtime(manifest_path) against mtime(root_dir).

Returns True when the manifest is at least as new as the root directory’s mtime. Returns False (and logs a loud error / raises) when the root dir was touched after the manifest was written — indicating either a producer crashed before refreshing the manifest or files were dropped in externally (e.g. NASA delivery, manual rsync).

Parameters:

manifest_path: Absolute path to the manifest file.
root_dir: Absolute path to the tree the manifest indexes.
raise_on_stale: When True, raise gedih3.exceptions.GediError on staleness. When False (default), log an ERROR and return False.
remedy: Shell command the user should run to refresh the manifest, included in the error message.