gedih3.utils ============ .. py:module:: gedih3.utils Classes ------- .. autoapisummary:: gedih3.utils.AtomicFileWriter Functions --------- .. autoapisummary:: gedih3.utils.is_remote_path gedih3.utils.smart_join gedih3.utils.configure_storage gedih3.utils.get_storage_options gedih3.utils.smart_exists gedih3.utils.smart_isdir gedih3.utils.generate_manifest gedih3.utils.smart_glob gedih3.utils.smart_open gedih3.utils.get_package_version gedih3.utils.now gedih3.utils.get_system_resources gedih3.utils.object_series gedih3.utils.json_write gedih3.utils.json_read gedih3.utils.check_nan_only_columns gedih3.utils.is_parquet gedih3.utils.is_hive_directory gedih3.utils.read_parquet_schema gedih3.utils.read_geopackage_schema gedih3.utils.read_feather_schema gedih3.utils.read_h3_database_schema gedih3.utils.read_schema gedih3.utils.h5_is_valid gedih3.utils.release_arrow_pool gedih3.utils.h5_traverse gedih3.utils.h5_info gedih3.utils.h5_var gedih3.utils.h5_meta gedih3.utils.h5_copy_subset gedih3.utils.read_vector_file gedih3.utils.read_img_bounds gedih3.utils.geo_to_umm gedih3.utils.to_geojson gedih3.utils.from_geojson gedih3.utils.parquet_append_columns gedih3.utils.parquet_schema_add_bbox gedih3.utils.parse_h3_partition_dirname gedih3.utils.h3_partition_bbox gedih3.utils.parquet_backfill_bbox gedih3.utils.parquet_merge_files gedih3.utils.parquet_join_columns gedih3.utils.parse_temporal gedih3.utils.parse_spatial gedih3.utils.merge_spatial gedih3.utils.get_dask_client gedih3.utils.dask_safe_wait gedih3.utils.dask_safe_collect gedih3.utils.atomic_parquet_write Module Contents --------------- .. py:function:: is_remote_path(path) Check if path is a remote URL (S3, HTTP, FTP, etc.). .. !! processed by numpydoc !! .. py:function:: smart_join(*parts) os.path.join() that uses forward slashes for remote URLs. On Windows, os.path.join uses backslashes which corrupts URLs like ``http://host:port/path`` into ``http://host:port\path``, causing port-parsing failures in urllib3. .. !! processed by numpydoc !! .. py:function:: configure_storage(protocol='s3', **kwargs) Set storage credentials for a remote protocol. Credentials are stored at module level and automatically used by every ``smart_*`` function (which all flow through ``_get_filesystem`` / ``smart_open``). :Parameters: **protocol** : str Protocol name: ``'s3'``, ``'http'``, ``'https'``, ``'ftp'``, ``'sftp'``, ``'ssh'``. **\*\*kwargs** Protocol-specific options passed to ``fsspec.filesystem()``. ``S3`` (s3fs): ``key``, ``secret``, ``endpoint_url``, ``anon``. ``endpoint_url`` is automatically wrapped into ``client_kwargs={'endpoint_url': ...}`` for s3fs compatibility. ``HTTP/HTTPS`` (aiohttp): ``username``/``password`` (basic auth via ``client_kwargs``) or ``headers`` dict (bearer tokens, API keys). ``FTP``: ``username``, ``password``, ``host``, ``port``. ``SFTP/SSH``: ``username``, ``password`` or ``key_filename`` (path to SSH private key), ``port``. .. rubric:: Examples >>> configure_storage('s3', endpoint_url='http://localhost:7000', anon=True) >>> configure_storage('http', username='user', password='pass') >>> configure_storage('https', headers={'Authorization': 'Bearer tok'}) >>> configure_storage('sftp', username='user', key_filename='/path/to/id_rsa') .. !! processed by numpydoc !! .. py:function:: get_storage_options(protocol=None) Return the stored options for *protocol*. Returns ``{'anon': True}`` for S3 when nothing has been configured (public-bucket default). Other protocols return ``{}``. :Parameters: **protocol** : str or None Protocol name (e.g. ``'s3'``). ``None`` returns ``{}``. :Returns: dict A **copy** of the stored options (safe to mutate). .. !! processed by numpydoc !! .. py:function:: smart_exists(path) os.path.exists() that works with remote paths. .. !! processed by numpydoc !! .. py:function:: smart_isdir(path) os.path.isdir() that works with remote paths. .. !! processed by numpydoc !! .. py:function:: generate_manifest(root_path, pattern='**/*.parquet', manifest_filename=None, tree_shape='h3db', files=None) Atomically write a manifest file listing all matching files. :Parameters: **root_path** : str Database root directory (must be local). **pattern** : str Glob pattern matched at each leaf — meaning depends on ``tree_shape``. For ``'h3db'`` it is matched recursively under each ``h3_NN=*`` partition; for ``'soc'`` it is matched at each ``year/doy/`` leaf; for ``'flat'`` it is matched at the single ``root_path``. The default ``**/*.parquet`` is the H3 layout for backwards compatibility. **manifest_filename** : str, optional Name of the manifest sentinel file. Defaults to the H3 database manifest (``MANIFEST_FILENAME``). Pass ``SOC_MANIFEST_FILENAME`` for the SOC parallel. **tree_shape** : {'h3db', 'soc', 'flat'} Tree topology to walk. Dispatches to the matching walker in :mod:`gedih3.parallel`. The walker requires a registered dask Client at call time — there is **no serial fallback** (matches the package-wide always-parallel contract). **files** : list[str], optional Pre-computed absolute file list to use instead of walking. When the caller already has the file list in memory (e.g. ``cli/gh3_build.py`` after its existing-h5 listing), passing them here avoids a redundant walk. The list must contain absolute paths under ``root_path``. :Returns: str Path to the written manifest file. The write is atomic (``.tmp`` + ``os.replace``) so an interrupted run never leaves a partial manifest at the final path — important when the manifest is also a resume-correctness signal for the next invocation. .. !! processed by numpydoc !! .. py:function:: smart_glob(pattern, recursive=False) glob.glob() that works with remote paths. Uses a _manifest.txt file at the glob root when available, filtering entries by pattern. Falls back to filesystem globbing when no manifest exists. For remote paths, uses fs.find() to list all files under the root, then filters with a glob-to-regex matcher. Results include the full protocol prefix. .. !! processed by numpydoc !! .. py:function:: smart_open(path, mode='r', storage_options=None) open() that works with remote paths. Use as context manager. :Parameters: **path** : str Local or remote file path. **mode** : str File mode (default ``'r'``). **storage_options** : dict, optional Per-call overrides merged on top of the global config. .. !! processed by numpydoc !! .. py:function:: get_package_version() Get the current package version .. !! processed by numpydoc !! .. py:function:: now() .. py:function:: get_system_resources(disk_path: str = None) .. py:function:: object_series(items) Wrap a list of array-like objects in an object-dtype ``pd.Series``. ``pd.Series([...])`` and ``np.empty(..., dtype=object)[:] = items`` both coerce their input through ``np.asarray``, which xarray Datasets refuse ("cannot directly convert an xarray.Dataset into a numpy array"). Assigning positionally into a pre-sized object Series is the construction that keeps the elements intact — use it whenever a ``map_partitions`` result carries Datasets or other array-likes rather than scalars. :Parameters: **items** : sequence Objects to store as Series elements, kept by reference. :Returns: pd.Series Object-dtype Series of ``items``, with a default RangeIndex. .. !! processed by numpydoc !! .. py:function:: json_write(obj, path, mode='w', rewrite=False) .. py:function:: json_read(path, mode='r') .. py:function:: check_nan_only_columns(df, context='', logger=None) Warn about columns that are entirely NaN. :Parameters: **df** : DataFrame or GeoDataFrame Data to check. **context** : str Optional prefix for the warning message. **logger** : logging.Logger, optional Logger instance. If None, uses module-level warnings. :Returns: list Column names that are entirely NaN. .. !! processed by numpydoc !! .. py:function:: is_parquet(file: str) -> bool .. py:function:: is_hive_directory(dir_path: str, match_str='.+=.+') -> bool .. py:function:: read_parquet_schema(path) path: parquet file path returns a pandas.DataFrame with the parquet column structure .. !! processed by numpydoc !! .. py:function:: read_geopackage_schema(path) path: gpkg file path returns a pandas.DataFrame with the gpkg column structure .. !! processed by numpydoc !! .. py:function:: read_feather_schema(path) Read schema from a feather (Arrow IPC) file. :Parameters: **path** : str Path to feather file :Returns: pandas.DataFrame DataFrame with 'column' and 'dtype' columns .. !! processed by numpydoc !! .. py:function:: read_h3_database_schema(db_path) Read parquet schema from an H3 hive-partitioned database. Finds the first parquet file inside any H3 partition directory and reads its schema via read_parquet_schema(). :Parameters: **db_path** : str Path to H3 database root directory (containing h3_XX=* subdirs) :Returns: pandas.DataFrame DataFrame with 'column' and 'dtype' columns :Raises: FileNotFoundError If no H3 partition directories or parquet files found .. !! processed by numpydoc !! .. py:function:: read_schema(path, root=None) Read schema from a data file or dataset directory, auto-detecting format. Supports parquet, feather, gpkg, and HDF5 files. For directories, detects the dataset format from metadata or file extensions. Also detects H3 databases by the presence of the build log file. :Parameters: **path** : str Path to a file or dataset directory :Returns: pandas.DataFrame DataFrame with 'column' and 'dtype' columns (for vector/tabular formats), or 'column', 'dtype', and 'shape' columns (for HDF5) :Raises: FileNotFoundError If no data files found ValueError If format cannot be determined .. !! processed by numpydoc !! .. py:function:: h5_is_valid(file) .. py:function:: release_arrow_pool() -> None Best-effort drain of pyarrow's allocator. pyarrow's transient read/write buffers do not always return to the OS at GC time, which causes long-running worker RSS to climb across successive parquet operations on shared GPFS. Calling ``pa.default_memory_pool().release_unused()`` after each per-file or per-task scope keeps the plateau flat. The pool API may also raise on unusual installations; we swallow exceptions because the drain is an optimization, not a correctness gate. This helper is the single source of truth for the pattern that used to be inlined in 6+ doctor / build call sites and was easy to forget at a new site. .. !! processed by numpydoc !! .. py:function:: h5_traverse(h5_file, root=None) .. py:function:: h5_info(hdf_file, root=None) .. py:function:: h5_var(file, var, col: int = None) .. py:function:: h5_meta(file, var='METADATA/DatasetIdentification') .. py:function:: h5_copy_subset(source_file, dest_file, variables) Copy selected datasets from source to dest HDF5 file. Uses direct path iteration instead of visit_links() to avoid traversing the entire HDF5 tree — critical for S3 performance where each node visit is a range request (~50-100ms). .. !! processed by numpydoc !! .. py:function:: read_vector_file(filepath: str, crs: Union[str, int] = 4326) .. py:function:: read_img_bounds(filepath: str, crs=4326) .. py:function:: geo_to_umm(obj) Converts a GeoDataFrame, shapely Polygon, or GeoJSON dictionary to a UMM-style list of (lon, lat) coordinate tuples for a single polygon. Multi-polygon geometries are reduced to their convex hull since earthaccess/CMR only supports single-polygon spatial queries. Rings with more than ``_CMR_POLYGON_MAX_VERTICES`` points are auto-simplified via iterative ``shapely.simplify`` (doubling tolerance) until the ring fits. CMR/CloudFront returns HTTP 414 (URI too long) when the polygon query string grows past a few hundred vertices; simplifying client-side avoids that failure. .. !! processed by numpydoc !! .. py:function:: to_geojson(geodf) -> Dict .. py:function:: from_geojson(geojson) .. py:function:: parquet_append_columns(df, f: str, tmp_suffix: str = '.col.tmp') .. py:function:: parquet_schema_add_bbox(schema, bbox) .. py:function:: parse_h3_partition_dirname(h3part) Parse a partition dir name like ``'h3_03=830e4afffffffff'`` into ``(cell_id, parent_res)``. Returns ``(None, None)`` on parse failure. .. !! processed by numpydoc !! .. py:function:: h3_partition_bbox(h3_cell_id, parent_res, edge_fraction=_H3_OVERHANG_FRACTION) Return the EPSG:4326 bbox of an H3 cell, padded to safely contain all descendants at any deeper resolution. The buffer derives from the icosahedral-projection distortion measured empirically at H3 face boundaries: a child cell's vertices can sit up to ``edge_fraction × parent_edge_length`` outside the parent cell's own bbox (in metres on the ground, regardless of how deep the child is once the depth gap is ≥ ~5 levels). The default ``0.18`` is the measured asymptote (~14%) × 1.2 safety margin. The buffer is converted to longitude-degrees at the parent's most poleward vertex (cosine-corrected) so the same scalar in degrees is safe for both lat and lon directions of the bbox. :Parameters: **h3_cell_id** : str H3 cell index (hex string) at ``parent_res``. **parent_res** : int Resolution of ``h3_cell_id``. Used to look up the average edge length for the buffer. **edge_fraction** : float, default 0.18 Buffer as a fraction of the parent's edge length. :Returns: list[float] | None ``[minlon, minlat, maxlon, maxlat]`` in EPSG:4326 degrees, or ``None`` if the cell ID cannot be decoded. .. rubric:: Notes Antimeridian-crossing parents produce a loose bbox spanning ~[-180, 180] in longitude (because the simple min/max over boundary vertices folds incorrectly there). This is conservative — the bbox still contains the cell — but predicate pushdown is ineffective for those partitions. GEDI data above the antimeridian is rare (ISS limit ±51.6° latitude); accept the looseness rather than complicate the formula. .. !! processed by numpydoc !! .. py:function:: parquet_backfill_bbox(path) Rewrite a single parquet file in place so its GeoParquet ``geo`` metadata declares a valid ``columns..bbox``. :Returns: str ``'ok'`` if the file already has a valid bbox (no-op), ``'rewritten'`` if a bbox was computed and the file was rewritten, ``'no_geometry'`` if the file has no ``geometry`` column. :Raises: ValueError If the file lacks a ``geo`` schema metadata key (cannot be backfilled without re-merging from source — caller should flag for a full rebuild). .. rubric:: Notes Atomic: writes to ``.bbox.tmp`` then ``os.replace``. A stale tmp from a prior crash is removed before each rewrite. Memory is bounded by the streaming scanner (``batch_size=100_000``), same profile as ``parquet_merge_files``. .. !! processed by numpydoc !! .. py:function:: parquet_merge_files(ofile, flist, check_shots=False, rm_src=False, rows_per_group=100000, bbox=None) Stream-merge parquet files into a single output with a bounded memory footprint. Architecture (per-file iteration, native column projection): - Schema is taken from the first file's footer (one ``pq.read_schema``). Each input fragment is opened sequentially via ``pq.ParquetFile`` and drained via ``iter_batches(batch_size=rows_per_group, columns=schema.names)``. The ``columns=...`` argument has pyarrow's C++ reader **read columns in the target order** — no Python-side reordering, no per-batch reconciler, and any extras the file might have are dropped at read time (less I/O). When the file goes out of scope, its IO state is released — deterministic per-file lifecycle. - **Invariant assumed by design**: all input fragments share an identical column set and dtypes (true in gh3_build because all fragments come from the same ``dask_geopandas.to_parquet`` call). A fragment with a missing target column will raise from pyarrow — that's the right behavior; it surfaces a serious data invariant violation rather than silently null-filling. - Bbox is **provided by the caller** via the ``bbox`` argument when the input has a ``geometry`` column. ``gh3builder.h3_merge_files`` derives it directly from the H3 partition geometry (no data scan). - Row-group accumulator flushes BEFORE appending a batch that would overflow ``rows_per_group``. - Shot-dedup activates only when ``check_shots=True``. :Parameters: **rows_per_group** : int, default 100_000 Output row-group size and per-file iter batch size. **bbox** : list[float] or None ``[minlon, minlat, maxlon, maxlat]`` in EPSG:4326. When provided and the input has a ``geometry`` column, embedded into the GeoParquet ``columns.geometry.bbox`` metadata. :Returns: dict or None ``None`` if ``flist`` is empty. Otherwise a stats dict accumulated online during the merge stream: ``{'shot_count', 'shot_min', 'shot_max', 'dt_min', 'dt_max', 'root_files'}``. Fields are ``None`` when the source column is absent from the schema. Used by ``h3_write_metadata`` to skip re-reading the merged file. Output is written atomically: ``ofile + '.merge.tmp'`` first, then .. ``os.replace`` to ``ofile``. A stale ``.merge.tmp`` from a prior crash is .. cleaned up before the new write. .. .. !! processed by numpydoc !! .. py:function:: parquet_join_columns(flist: List[str], ofile: str, key_col: str = 'shot_number', tmp_suffix: str = '.join.tmp', join_how='left', rows_per_group: int = 100000) Memory-efficient column-wise join of parquet files. Equivalent to pd.concat(axis=1) but processes in batches to avoid loading entire files into memory. :Parameters: **flist** : List[str] Parquet files to join. First file determines row order, index, and metadata. **ofile** : str Output file path. **key_col** : str, default='shot_number' Column for joining (not the index). **join_how** : str, default='left' Join mode passed to ``DataFrame.join``. **rows_per_group** : int, default 100_000 Output row-group size. Matches the fresh-build ``parquet_merge_files`` default so updated files keep a bounded, deterministic per-group size regardless of the base file's existing row-group shape. **tmp_suffix** : str, default='.join.tmp' Temporary file suffix. .. !! processed by numpydoc !! .. py:function:: parse_temporal(temporal) .. py:function:: parse_spatial(spatial) .. py:function:: merge_spatial(existing, new) .. py:function:: get_dask_client() .. py:function:: dask_safe_wait(persisted, show_progress=False) Wait for a persisted Dask collection to finish and re-raise the first worker exception. Side-effect-only equivalent of ``.compute()`` with no driver-side collect. Why this exists: dask >= 2025.2 always inserts a ``RepartitionToFewer(1)`` optimization step inside ``.compute()`` on a multi-partition collection, which collapses all partitions onto a single worker before delivering to the driver. That fan-in deterministically wedges on tunneled multi-node clusters past ~1500 partitions. Workloads that only need side effects (e.g. per-partition file writes) don't need to collect anything — they just need to wait for completion and surface task exceptions. This helper does exactly that, without going through ``.compute()``. Follows the same persist + futures_of + error-result pattern already used in ``gh3builder._create_h3_dataframe`` (proven in production builds). .. !! processed by numpydoc !! .. py:function:: dask_safe_collect(collection, show_progress=False) Persist a Dask DataFrame/Series on the cluster, then gather each partition independently to the driver and concatenate locally. Returns a single pandas/geopandas DataFrame or Series (or list for a Bag). Same motivation as :func:`dask_safe_wait`: bypasses the optimizer's ``RepartitionToFewer(1)`` collapse step inside ``.compute()`` by walking ``to_delayed()`` and gathering through the well-tested scheduler↔client comm path. Each partition becomes one outbound future, so a single unreachable peer fails only its own fetch instead of wedging the whole job. When no distributed client is registered, falls back to a plain ``.compute()`` (synchronous scheduler — no fan-in, no wedge possible). .. !! processed by numpydoc !! .. py:class:: AtomicFileWriter(target_path: str, suffix: str = '.tmp', backup: bool = False, backup_suffix: str = '.bak') Context manager for atomic file writes with automatic rollback on failure. Writes to a temporary file first, then atomically replaces the target file only on successful completion. If an error occurs, the temporary file is cleaned up and the original file (if any) is preserved. :Parameters: **target_path** : str The final destination file path **suffix** : str Suffix for the temporary file (default: '.tmp') **backup** : bool If True, keep a backup of the original file (default: False) **backup_suffix** : str Suffix for backup files (default: '.bak') .. rubric:: Examples >>> with AtomicFileWriter('/path/to/output.parquet') as tmp_path: ... df.to_parquet(tmp_path) # File is atomically renamed to /path/to/output.parquet on success >>> with AtomicFileWriter('/path/to/output.json', backup=True) as tmp_path: ... with open(tmp_path, 'w') as f: ... json.dump(data, f) # Original file backed up to .bak, new file replaces it .. !! processed by numpydoc !! .. py:attribute:: target_path .. py:attribute:: temp_path .. py:attribute:: backup :value: False .. py:attribute:: backup_path .. py:method:: __enter__() -> str .. py:method:: __exit__(exc_type, exc_val, exc_tb) .. py:function:: atomic_parquet_write(df, opath, *, compression=None, max_attempts=3) Atomic parquet write with post-write page-level integrity check + retry. Each attempt: write to ``.tmp`` inside an ``AtomicFileWriter`` context, then stream-read every data page via ``iter_batches`` while the temp file is still uncommitted. If any page deserialization raises, the exception propagates out of the context, the ``.tmp`` is unlinked, and the loop tries again. After ``max_attempts`` failures the last exception is re-raised — the partition errors out cleanly instead of promoting a torn file to the canonical path. Catches the production-observed GPFS/transient-IO class where pyarrow successfully closes a parquet file with corrupt internal page bytes (footer + metadata valid, data pages corrupt). That failure mode silently rides into the renamed final file under plain ``AtomicFileWriter`` because pyarrow's normal write path doesn't checksum or re-read. Streaming verify (``iter_batches`` with bounded batch_size) keeps peak memory bounded to ~one batch's worth of decoded data — no double-buffer of the whole partition. .. !! processed by numpydoc !!