gedih3.parallel =============== .. py:module:: gedih3.parallel .. autoapi-nested-parse:: 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 :class:`~dask.distributed.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. .. !! processed by numpydoc !! Attributes ---------- .. autoapisummary:: gedih3.parallel.logger Functions --------- .. autoapisummary:: gedih3.parallel.parallel_map gedih3.parallel.walk_soc_parallel gedih3.parallel.walk_h3db_parallel gedih3.parallel.walk_flat_parallel gedih3.parallel.check_manifest_freshness Module Contents --------------- .. py:data:: logger .. py:function:: 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]] 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 :class:`~dask.distributed.Client` to be registered (via :func:`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 :class:`gedih3.exceptions.GediError` when no client is registered. :Parameters: **items** : list Iterable of work items (e.g. partition directory paths). **fn** : callable Top-level (picklable) function. Signature: ``fn(item, **broadcast)``. **args** : argparse.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, unit** : str Dispatch log prefix and unit name used in the periodic ``N/M done`` counter line. **batch_size** : int, 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``. .. !! processed by numpydoc !! .. py:function:: walk_soc_parallel(soc_dir: str, *, pattern: str = 'GEDI*.h5', exclude: Optional[List[str]] = None, batch_size: int = 32) -> List[str] Parallel year/doy walk over a SOC tree. Driver enumerates year and doy subdirs via :func:`os.scandir` (bounded: ~7 years × ~300 doys = ~2000 single-level syscalls on the driver). Workers run one non-recursive scandir per doy via :func:`_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. .. !! processed by numpydoc !! .. py:function:: walk_h3db_parallel(db_root: str, *, pattern: str = '*.parquet', batch_size: int = 64) -> List[str] Parallel per-partition walk over an H3 database. Driver enumerates ``h3_NN=*`` partition directories via one :func:`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. .. !! processed by numpydoc !! .. py:function:: walk_flat_parallel(dir_path: str, *, pattern: str = '*.parquet') -> List[str] Flat single-directory listing. Provided for API symmetry with the other two walkers; no dask dispatch (the tree has a single leaf). .. !! processed by numpydoc !! .. py:function:: check_manifest_freshness(manifest_path: str, root_dir: str, *, raise_on_stale: bool = False, remedy: str = '') -> bool 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 :class:`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. .. !! processed by numpydoc !!