gedih3.utils#

Attributes#

NON_LOCAL_PREFIXES

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 handled by fsspec (S3, HTTP, FTP, ...).
`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.
`split_s3_host_url`(url)	Split `s3://host:port/bucket/...` into `(endpoint_url, s3://bucket/...)`.
`resolve_s3_source`(path)	Normalize an `s3://host:port/bucket/...` source for the Python API.
`smart_exists`(path)	os.path.exists() that works with remote paths.
`smart_database_exists`(path)	`smart_exists()` for a gedih3 database / dataset root.
`smart_isdir`(path)	os.path.isdir() that works with remote paths.
`smart_isfile`(path)	True when path is a single data file rather than a dataset directory.
`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.
`smart_open_columnar`(path[, storage_options])	Open a remote parquet file for pyarrow, with no client-side read-ahead.
`read_parquet_coalesced`(source[, columns, geo])	Read a parquet source with pyarrow range coalescing enabled.
`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])
`json_read_cached`(path)	`json_read()` memoized per path — for read-only metadata sidecars.
`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)
`region_to_geometry`(spatial)	Normalize any accepted region argument to one shapely geometry in EPSG:4326.
`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 handled by fsspec (S3, HTTP, FTP, …).

This is the fsspec notion of remote, used by every smart_* helper. GDAL’s virtual file systems (/vsicurl/ and friends) are deliberately excluded — pyarrow/fsspec cannot read them. See NON_LOCAL_PREFIXES for the broader set that merely must not be treated as a local filesystem path.

gedih3.utils.NON_LOCAL_PREFIXES = ('http://', 'https://', 's3://', 'gs://', 'gcs://', 'ftp://', 'sftp://', 'ssh://', '/vsicurl/',...#

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.

S3 defaults to anon=True in two cases:

nothing at all has been configured for S3 (the long-standing public-bucket default), or
non-credential options (e.g. endpoint_url) were configured, no credential-bearing option is among them, and the botocore chain has no ambient credentials (env vars / shared credentials file) to fall back on — e.g. a read-only rclone serve s3 or MinIO with auth disabled. Without this s3fs raises NoCredentialsError, which its own exists() swallows into a bare False — surfacing as a misleading “not found”.

When ambient AWS credentials exist and the caller configured only non-credential options, the chain is left to do its job — forcing anon=True there would break working env-var / profile setups. 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.split_s3_host_url(url)[source]#

Split s3://host:port/bucket/... into (endpoint_url, s3://bucket/...).

A bucket name cannot contain :, so a netloc with an explicit port is unambiguously a server, not a bucket. Port 443 implies https, anything else http (self-hosted servers are overwhelmingly plain http; configure an explicit endpoint for TLS on a non-standard port). Returns (None, url) for anything that is not the host form — plain-bucket S3 URLs and non-S3 paths pass through untouched.

The single source of truth for this rule: the CLI’s endpoint_from_s3_urls and the Python API’s resolve_s3_source() both delegate here, so the two layers cannot drift.

Raises:

GediValidationError: If the URL names a server but no bucket (s3://host:port).

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

Normalize an s3://host:port/bucket/... source for the Python API.

The CLI layer performs this normalization in setup_storage; the Python entry points (gh3_load, egi_load, gh3_select_partitions, gh3_read_meta, …) call this so the same URL works in both — without it, s3fs treats host:port as a bucket name. Precedence mirrors the CLI: an explicit configure_storage('s3', endpoint_url=...) / --s3-endpoint always wins (a differing URL host logs a warning instead of silently rerouting); an endpoint this function derived earlier is replaced by a later, different host URL — each call names its own server. The path is normalized to plain s3://bucket/... either way; non-S3 and plain-bucket paths are returned unchanged.

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

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

smart_exists() for a gedih3 database / dataset root.

Object stores have no directories: an existence check on a bucket or prefix degrades to a LIST, which some servers answer only after walking the whole tree (rclone serve s3 over a 12k-partition dataset takes minutes; a real S3/MinIO is fast but still pays a round trip). Every gh3 tree carries at least one sidecar at its root, so probe those first — a HeadObject each, and the metadata readers that follow need them anyway. Falls back to the generic directory check so plain parquet directories (no sidecar) still resolve.

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

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

True when path is a single data file rather than a dataset directory.

Remote paths answer from the extension instead of probing: object stores have no directories, so isdir() on a bucket or prefix degrades to a LIST — the exact O(N) round trip the manifest sentinel exists to avoid, and one that rclone serve s3 answers only after walking the whole tree. Every gh3 single-file output carries one of the known extensions, so the answer is knowable for free. Local paths use os.path.isfile. A trailing slash is the one unambiguous “this is a directory” signal a URL can carry, so it always answers False — even when the directory name ends in a data extension (e.g. gh3_extract -o s3://bucket/out.parquet writes a directory).

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.smart_open_columnar(path, storage_options=None)[source]#

Open a remote parquet file for pyarrow, with no client-side read-ahead.

fsspec’s default file cache fetches a whole block (5 MB for HTTP) around every read. Pyarrow already knows exactly which byte ranges it wants and coalesces them itself (pre_buffer=True), so that block is pure over-fetch — and a column-projected read touches dozens of small, scattered ranges. Measured against rclone serve http on a 1.8 GB GEDI partition, projecting one column (18 MB of column chunks + footer): 358 MB pulled off GPFS with the default cache, 18 MB with cache_type='none' plus pyarrow’s own coalescing.

Returns an open binary file object (not a context manager wrapper); callers use it with with. Local paths are opened directly — the kernel page cache already does the right thing.

gedih3.utils.read_parquet_coalesced(source, columns=None, geo=True, **kwargs)[source]#

Read a parquet source with pyarrow range coalescing enabled.

Pair with smart_open_columnar() on remote reads: that turns off fsspec’s block read-ahead, this makes pyarrow fetch the column chunks it needs in as few ranges as possible.

geopandas.read_parquet forwards **kwargs to pyarrow.parquet.read_table, so pre_buffer=True reaches the reader. pandas.read_parquet does not coalesce (measured: 50 scattered server reads vs 26 for read_table on the same query), so the non-geo path calls read_table directly and converts — verified to produce an identical frame, dtypes included, on GEDI partitions. use_pandas_metadata=True matches what pandas.read_parquet always sets: without it a column-projected read drops the index column recorded in the pandas metadata and the frame comes back with a RangeIndex instead of e.g. h3_12.

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.json_read_cached(path)[source]#

json_read() memoized per path — for read-only metadata sidecars.

A single CLI invocation asks the build log a dozen questions (h3_columns, h3_columns_dtypes, h3_partition_level, h3_partition_ids, gedi_version) through two dozen call sites, and each one re-fetched the whole file. That log is 21 MB on a continental database, and one gh3_aggregate was measured opening it 13 times over HTTP — ~270 MB of metadata traffic to answer questions that cannot change mid-query.

Local paths are validated by (mtime_ns, size, inode), so a build that rewrites the log in the same process still sees fresh state — the inode covers the case where an AtomicFileWriter os.replace lands inside the filesystem’s timestamp granularity. Remote paths are held for the process lifetime: gedih3 never writes to a remote database, and a stat over the network would cost the very round trip the cache exists to avoid. Use json_read() for anything read-modify-write.

The returned object IS the shared cache entry — treat it as read-only. Mutating it (e.g. .append() on a returned list) poisons every later read in the process; copy first if you must modify.

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.region_to_geometry(spatial)[source]#

Normalize any accepted region argument to one shapely geometry in EPSG:4326.

Accepts what the CLIs accept — a vector-file path, a "W,S,E,N" string, a [W, S, E, N] list, a GeoDataFrame/GeoSeries, or a shapely geometry — so every spatial-selection path agrees on what a region is. Shared by intersect_h3_geometries() (H3 partition selection) and the EGI partition selection in gh3driver; keeping one normalizer is what stops the two index types from drifting apart on region handling.

Raises:

TypeError: If the input is not one of the accepted forms.

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.