gedih3.utils#

Classes#

AtomicFileWriter

Context manager for atomic file writes with automatic rollback on failure.

Functions#

`is_remote_path`(path)	Check if path is a remote URL (S3, HTTP, FTP, etc.).
`smart_join`(*parts)	os.path.join() that uses forward slashes for remote URLs.
`configure_storage`([protocol])	Set storage credentials for a remote protocol.
`get_storage_options`([protocol])	Return the stored options for protocol.
`smart_exists`(path)	os.path.exists() that works with remote paths.
`smart_isdir`(path)	os.path.isdir() that works with remote paths.
`generate_manifest`(root_path[, pattern, ...])	Atomically write a manifest file listing all matching files.
`smart_glob`(pattern[, recursive])	glob.glob() that works with remote paths.
`smart_open`(path[, mode, storage_options])	open() that works with remote paths. Use as context manager.
`get_package_version`()	Get the current package version
`now`()
`get_system_resources`([disk_path])
`object_series`(items)	Wrap a list of array-like objects in an object-dtype `pd.Series`.
`json_write`(obj, path[, mode, rewrite])
`json_read`(path[, mode])
`check_nan_only_columns`(df[, context, logger])	Warn about columns that are entirely NaN.
`is_parquet`(→ bool)
`is_hive_directory`(→ bool)
`read_parquet_schema`(path)	path: parquet file path
`read_geopackage_schema`(path)	path: gpkg file path
`read_feather_schema`(path)	Read schema from a feather (Arrow IPC) file.
`read_h3_database_schema`(db_path)	Read parquet schema from an H3 hive-partitioned database.
`read_schema`(path[, root])	Read schema from a data file or dataset directory, auto-detecting format.
`h5_is_valid`(file)
`release_arrow_pool`(→ None)	Best-effort drain of pyarrow's allocator.
`h5_traverse`(h5_file[, root])
`h5_info`(hdf_file[, root])
`h5_var`(file, var[, col])
`h5_meta`(file[, var])
`h5_copy_subset`(source_file, dest_file, variables)	Copy selected datasets from source to dest HDF5 file.
`read_vector_file`(filepath[, crs])
`read_img_bounds`(filepath[, crs])
`geo_to_umm`(obj)	Converts a GeoDataFrame, shapely Polygon, or GeoJSON dictionary to a UMM-style
`to_geojson`(→ Dict)
`from_geojson`(geojson)
`parquet_append_columns`(df, f[, tmp_suffix])
`parquet_schema_add_bbox`(schema, bbox)
`parse_h3_partition_dirname`(h3part)	Parse a partition dir name like `'h3_03=830e4afffffffff'` into
`h3_partition_bbox`(h3_cell_id, parent_res[, edge_fraction])	Return the EPSG:4326 bbox of an H3 cell, padded to safely contain all
`parquet_backfill_bbox`(path)	Rewrite a single parquet file in place so its GeoParquet `geo` metadata
`parquet_merge_files`(ofile, flist[, check_shots, ...])	Stream-merge parquet files into a single output with a bounded memory footprint.
`parquet_join_columns`(flist, ofile[, key_col, ...])	Memory-efficient column-wise join of parquet files. Equivalent to pd.concat(axis=1)
`parse_temporal`(temporal)
`parse_spatial`(spatial)
`merge_spatial`(existing, new)
`get_dask_client`()
`dask_safe_wait`(persisted[, show_progress])	Wait for a persisted Dask collection to finish and re-raise the first
`dask_safe_collect`(collection[, show_progress])	Persist a Dask DataFrame/Series on the cluster, then gather each
`atomic_parquet_write`(df, opath, *[, compression, ...])	Atomic parquet write with post-write page-level integrity check + retry.

Module Contents#

gedih3.utils.is_remote_path(path)[source]#: Check if path is a remote URL (S3, HTTP, FTP, etc.).

gedih3.utils.smart_join(*parts)[source]#

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.

gedih3.utils.configure_storage(protocol='s3', **kwargs)[source]#

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:

protocolstr

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.

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')

gedih3.utils.get_storage_options(protocol=None)[source]#

Return the stored options for protocol.

Returns {'anon': True} for S3 when nothing has been configured (public-bucket default). Other protocols return {}.

Parameters:

protocolstr or None: Protocol name (e.g. 's3'). None returns {}.

Returns:

dict: A copy of the stored options (safe to mutate).

gedih3.utils.smart_exists(path)[source]#: os.path.exists() that works with remote paths.

gedih3.utils.smart_isdir(path)[source]#: os.path.isdir() that works with remote paths.

gedih3.utils.generate_manifest(root_path, pattern='**/*.parquet', manifest_filename=None, tree_shape='h3db', files=None)[source]#

Atomically write a manifest file listing all matching files.

Parameters:

root_pathstr: Database root directory (must be local).
patternstr: 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_filenamestr, 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 gedih3.parallel. The walker requires a registered dask Client at call time — there is no serial fallback (matches the package-wide always-parallel contract).
fileslist[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.

gedih3.utils.smart_glob(pattern, recursive=False)[source]#

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.

gedih3.utils.smart_open(path, mode='r', storage_options=None)[source]#

open() that works with remote paths. Use as context manager.

Parameters:

pathstr: Local or remote file path.
modestr: File mode (default 'r').
storage_optionsdict, optional: Per-call overrides merged on top of the global config.

gedih3.utils.get_package_version()[source]#: Get the current package version

gedih3.utils.now()[source]#

gedih3.utils.get_system_resources(disk_path: str = None)[source]#

gedih3.utils.object_series(items)[source]#

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:

itemssequence: Objects to store as Series elements, kept by reference.

Returns:

pd.Series: Object-dtype Series of items, with a default RangeIndex.

gedih3.utils.json_write(obj, path, mode='w', rewrite=False)[source]#

gedih3.utils.json_read(path, mode='r')[source]#

gedih3.utils.check_nan_only_columns(df, context='', logger=None)[source]#

Warn about columns that are entirely NaN.

Parameters:

dfDataFrame or GeoDataFrame: Data to check.
contextstr: Optional prefix for the warning message.
loggerlogging.Logger, optional: Logger instance. If None, uses module-level warnings.

Returns:

list: Column names that are entirely NaN.

gedih3.utils.is_parquet(file: str) → bool[source]#

gedih3.utils.is_hive_directory(dir_path: str, match_str='.+=.+') → bool[source]#

gedih3.utils.read_parquet_schema(path)[source]#

path: parquet file path

returns a pandas.DataFrame with the parquet column structure

gedih3.utils.read_geopackage_schema(path)[source]#

path: gpkg file path

returns a pandas.DataFrame with the gpkg column structure

gedih3.utils.read_feather_schema(path)[source]#

Read schema from a feather (Arrow IPC) file.

Parameters:

pathstr: Path to feather file

Returns:

pandas.DataFrame: DataFrame with ‘column’ and ‘dtype’ columns

gedih3.utils.read_h3_database_schema(db_path)[source]#

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_pathstr: 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

gedih3.utils.read_schema(path, root=None)[source]#

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:

pathstr: 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

gedih3.utils.h5_is_valid(file)[source]#

gedih3.utils.release_arrow_pool() → None[source]#

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.

gedih3.utils.h5_traverse(h5_file, root=None)[source]#

gedih3.utils.h5_info(hdf_file, root=None)[source]#

gedih3.utils.h5_var(file, var, col: int = None)[source]#

gedih3.utils.h5_meta(file, var='METADATA/DatasetIdentification')[source]#

gedih3.utils.h5_copy_subset(source_file, dest_file, variables)[source]#

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).

gedih3.utils.read_vector_file(filepath: str, crs: str | int = 4326)[source]#

gedih3.utils.read_img_bounds(filepath: str, crs=4326)[source]#

gedih3.utils.geo_to_umm(obj)[source]#

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.

gedih3.utils.to_geojson(geodf) → Dict[source]#

gedih3.utils.from_geojson(geojson)[source]#

gedih3.utils.parquet_append_columns(df, f: str, tmp_suffix: str = '.col.tmp')[source]#

gedih3.utils.parquet_schema_add_bbox(schema, bbox)[source]#

gedih3.utils.parse_h3_partition_dirname(h3part)[source]#: Parse a partition dir name like 'h3_03=830e4afffffffff' into (cell_id, parent_res). Returns (None, None) on parse failure.

gedih3.utils.h3_partition_bbox(h3_cell_id, parent_res, edge_fraction=_H3_OVERHANG_FRACTION)[source]#

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_idstr: H3 cell index (hex string) at parent_res.
parent_resint: Resolution of h3_cell_id. Used to look up the average edge length for the buffer.
edge_fractionfloat, 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.

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.

gedih3.utils.parquet_backfill_bbox(path)[source]#

Rewrite a single parquet file in place so its GeoParquet geo metadata declares a valid columns.<primary>.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).

Notes

Atomic: writes to <path>.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.

gedih3.utils.parquet_merge_files(ofile, flist, check_shots=False, rm_src=False, rows_per_group=100000, bbox=None)[source]#

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_groupint, default 100_000: Output row-group size and per-file iter batch size.
bboxlist[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.

gedih3.utils.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)[source]#

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:

flistList[str]: Parquet files to join. First file determines row order, index, and metadata.
ofilestr: Output file path.
key_colstr, default=’shot_number’: Column for joining (not the index).
join_howstr, default=’left’: Join mode passed to DataFrame.join.
rows_per_groupint, 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_suffixstr, default=’.join.tmp’: Temporary file suffix.

gedih3.utils.parse_temporal(temporal)[source]#

gedih3.utils.parse_spatial(spatial)[source]#

gedih3.utils.merge_spatial(existing, new)[source]#

gedih3.utils.get_dask_client()[source]#

gedih3.utils.dask_safe_wait(persisted, show_progress=False)[source]#

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).

gedih3.utils.dask_safe_collect(collection, show_progress=False)[source]#

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 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).

class gedih3.utils.AtomicFileWriter(target_path: str, suffix: str = '.tmp', backup: bool = False, backup_suffix: str = '.bak')[source]#

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_pathstr: The final destination file path
suffixstr: Suffix for the temporary file (default: ‘.tmp’)
backupbool: If True, keep a backup of the original file (default: False)
backup_suffixstr: Suffix for backup files (default: ‘.bak’)

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

target_path#

temp_path#

backup = False#

backup_path#

__enter__() → str[source]#

__exit__(exc_type, exc_val, exc_tb)[source]#

gedih3.utils.atomic_parquet_write(df, opath, *, compression=None, max_attempts=3)[source]#

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.