# Copyright (C) 2026, University of Maryland. All Rights Reserved.
# Authors: Tiago de Conto, Amelia Grace Holcomb
# For commercial licensing inquiries, contact UM Ventures at umdtechtransfer@umd.edu
"""
GEDI data file operations and HDF5 I/O.
This module provides classes and functions for working with GEDI HDF5 files,
including parsing filenames, extracting data, and building Dask DataFrames.
"""
import os
import re
import glob
import itertools
from typing import Dict, List, Optional, Union
from pathlib import Path
import h5py
import yaml
import numpy as np
import pandas as pd
import geopandas as gpd
from numba import njit
from datetime import datetime
import dask
import dask.dataframe
from earthaccess.store import EarthAccessFile
from .config import GEDI_BEAMS, GEDI_MISSION_START, GEDI_START_DATE, GH3_DEFAULT_SOC_DIR, SOC_MANIFEST_FILENAME, get_default_vars_file, _get_versioned, _GEDI_MIN_VARS
from .utils import h5_copy_subset, h5_info, _read_manifest, generate_manifest
from .logging_config import get_logger
from .exceptions import GediValidationError, GediProductError, GediVariableError, GediFileError
logger = get_logger(__name__)
[docs]
def soc_prod_from_file(file_path: Union[str, Path]) -> str:
"""
Extract GEDI product code from a file path.
Parameters
----------
file_path : str or Path
Path to a GEDI HDF5 file
Returns
-------
str
Product code (e.g., 'L2A', 'L4A')
"""
f = GEDIFile(file_path)
prod_str = f"L{f.product[-1]}{f.level}"
return prod_str
[docs]
def soc_from_file(file_path: Union[str, Path]) -> str:
"""
Get the SOC root directory from a GEDI file path.
Parameters
----------
file_path : str or Path
Path to a GEDI HDF5 file
Returns
-------
str
Path to the SOC root directory
"""
return os.path.dirname(os.path.dirname(os.path.dirname(file_path)))
def _soc_manifest_path(soc_dir: str) -> str:
"""Return the absolute path of the SOC manifest sentinel for *soc_dir*."""
return os.path.join(soc_dir.rstrip('/'), SOC_MANIFEST_FILENAME)
def _read_soc_manifest(soc_dir: str) -> Optional[List[str]]:
"""Read ``_soc_manifest.txt`` from a SOC root and return absolute paths.
Returns ``None`` when the manifest does not exist; callers should fall
back to the recursive glob. Thin wrapper around the shared
:func:`gedih3.utils._read_manifest` (cached, manifest-name aware) —
the SOC manifest stores relative paths just like the H3 database
manifest, so the only SOC-specific behavior is filling in the
sentinel filename and joining results back to absolute paths.
"""
rels = _read_manifest(soc_dir, manifest_filename=SOC_MANIFEST_FILENAME)
if rels is None:
return None
root = soc_dir.rstrip('/')
return [os.path.join(root, rel) for rel in rels]
[docs]
def write_soc_manifest(soc_dir: str, files=None) -> int:
"""(Re)generate the SOC manifest sentinel by walking *soc_dir* once.
Producer-driven refresh (R2): every code path that mutates a SOC
tree (`gh3_download`, `gh3_build --download`, `s3_etl_subset`,
`gh3_build -i` exit, `gh3_doctor --fix soc_health`) calls this on
exit. Cheaper than letting every consumer re-walk the tree.
Walks the SOC tree in parallel via
:func:`gedih3.parallel.walk_soc_parallel` (year/doy fan-out on the
registered dask Client). Returns the number of files written to
the manifest (0 when *soc_dir* is missing / empty).
Parameters
----------
soc_dir
SOC root directory.
files : list[str], optional
Pre-computed absolute file list to use instead of walking. When
the caller has already enumerated the tree (e.g.
`cli/gh3_build.py` after its existing-h5 listing), passing the
list here avoids the redundant parallel walk.
"""
if not os.path.isdir(soc_dir):
return 0
if files is None:
from .parallel import walk_soc_parallel
files = walk_soc_parallel(soc_dir, pattern='GEDI*.h5')
if not files:
return 0
generate_manifest(soc_dir, pattern='GEDI*.h5',
manifest_filename=SOC_MANIFEST_FILENAME,
tree_shape='soc', files=files)
return len(files)
# The file-name fields after a granule's orbit/track that rank its releases
# within one GEDI version: PPDS, PGE, generation and an optional suffix.
_RELEASE_FIELDS = re.compile(r'_O\d+_\d{2}_T\d+_(\d{2})_(\d{3})_(\d{2})_V\d{3}(?:_(.+?))?\.h5$')
# The same fields as a release tag (gedi_release_tag).
_RELEASE_TAG = re.compile(r'^(\d{2})_(\d{3})_(\d{2})(?:_(.+))?$')
[docs]
def gedi_release_tag(file_name: str) -> Optional[str]:
"""The release fields of a GEDI file name, as a compact tag.
With the granule key, the product and the GEDI version, the tag
identifies the file: it is what the build log records per granule and
product (``releases``), at a fraction of a basename's size.
Parameters
----------
file_name : str
A GEDI file name, path or URL. Only the basename is read.
Returns
-------
str or None
``'PP_PGE_GG'`` (PPDS, PGE, generation), plus ``'_<suffix>'`` when the
name has one; ``None`` for a name outside the GEDI convention.
:func:`gedi_release_rank` ranks tags as it ranks names.
Examples
--------
>>> gedi_release_tag('GEDI02_A_2019108002012_O01956_03_T03909_02_003_01_V002_SGS.h5')
'02_003_01_SGS'
"""
match = _RELEASE_FIELDS.search(os.path.basename(file_name))
if match is None:
return None
ppds, pge, generation, suffix = match.groups()
return f"{ppds}_{pge}_{generation}" + (f"_{suffix}" if suffix else '')
[docs]
def gedi_release_rank(file_name: str) -> tuple:
"""Rank one file of a granule product against the granule's other files.
NASA re-releases granules within one GEDI version with a new PPDS
(geolocation), PGE (software) or generation. The highest rank is the file
to use: PPDS first, then PGE, then generation, then the suffix, where a
suffixed file (e.g. ``_SGS``) beats the plain file of the same release and
suffixes compare alphabetically. For names that follow the GEDI
convention, this is the order a sort by file name gives within one
directory.
Parameters
----------
file_name : str
A GEDI file name, path or URL (only the basename is read), or a
release tag from :func:`gedi_release_tag`.
Returns
-------
tuple
``(ppds, pge, generation, has_suffix, suffix)``, ordered so that a
larger tuple is a later release. A name that does not follow the GEDI
convention ranks below every one that does.
Examples
--------
>>> old = 'GEDI02_A_2019108002012_O01956_03_T03909_02_003_01_V002.h5'
>>> new = 'GEDI02_A_2019108002012_O01956_03_T03909_02_004_01_V002.h5'
>>> gedi_release_rank(new) > gedi_release_rank(old)
True
"""
match = _RELEASE_FIELDS.search(os.path.basename(file_name))
if match is None and '/' not in file_name and not file_name.endswith('.h5'):
match = _RELEASE_TAG.match(file_name)
if match is None:
return (-1, -1, -1, False, '')
ppds, pge, generation, suffix = match.groups()
return (int(ppds), int(pge), int(generation), suffix is not None, suffix or '')
# Duplicate groups soc_file_tree has already reported in this process, keyed
# by (orb_track, product, file paths): a build lists the SOC tree several
# times, and one warning per group is enough, while a further re-release
# changes the key and is reported again. A stopgap until superseded granules
# are tracked in the build log (#47).
_REPORTED_DUPLICATE_FILES = set()
def _report_duplicate_granule_files(flist: pd.DataFrame) -> None:
"""Warn once per granule product that has more than one file in a listing.
``soc_file_tree`` keys files by orbit/granule/track and keeps the
highest-ranked file of each product (``gedi_release_rank``: PPDS, then
PGE, then generation, then the suffix). A second file is a re-release
within the same GEDI version, or an internal variant such as ``_SGS``.
Nothing re-indexes a granule whose file changes: rows already in a
database stay from the file they were built from, while variables
added and products filled later are read from the file chosen here.
Parameters
----------
flist : pandas.DataFrame
The listing, with ``file_paths``, ``orb_track`` and ``prod`` columns,
sorted so that each product's file to use comes last.
"""
files = flist.drop_duplicates('file_paths') # one path listed twice is not a second file
dup = files[files.duplicated(['orb_track', 'prod'], keep=False)]
if dup.empty:
return
new = []
for (orb_track, prod), paths in dup.groupby(['orb_track', 'prod'], sort=True)['file_paths']:
paths = list(paths) # in rank order: the pivot keeps the last one
key = (orb_track, prod, tuple(paths))
if key not in _REPORTED_DUPLICATE_FILES:
_REPORTED_DUPLICATE_FILES.add(key)
new.append((paths[-1], paths[:-1]))
if not new:
return
for used, others in new:
logger.debug(f"Duplicate granule files: using {used} over {', '.join(others)}")
used, others = new[0]
more = f" (and {len(new) - 1} more, listed at DEBUG level)" if len(new) > 1 else ""
logger.warning(
f"{len(new)} granule product(s) have more than one file in the SOC listing (a re-release within "
f"the same GEDI version, or an internal variant such as _SGS). The latest release (PPDS, then PGE, "
f"generation and suffix) is used, "
f"e.g. {os.path.basename(used)} over {', '.join(os.path.basename(o) for o in others)}{more}. "
f"Rows already in a database are not re-indexed from a newer file, and variables added or "
f"products filled later read the file used here. Remove the files you don't want, or exclude "
f"them (exclude=, gh3_build --exclude, which the build log keeps for later runs and gh3_doctor)."
)
[docs]
def soc_file_tree(
file_struct: Union[str, List[str], List[EarthAccessFile]],
to_list: bool = False,
glob_kwargs: Optional[Dict] = None,
exclude: Optional[List[str]] = None,
require_all: bool = True,
) -> Union[Dict[str, Dict[str, str]], List[Dict[str, str]]]:
"""
Build a structured tree of GEDI SOC files grouped by orbit/track.
Organizes GEDI HDF5 files into a nested structure for easy access to
related files across different products (L2A, L2B, L4A, etc.).
Parameters
----------
file_struct : str, list of str, or list of EarthAccessFile
Either a directory path containing GEDI files, or a list of file
paths or EarthAccessFile objects.
to_list : bool, default False
If True, return a list of dicts instead of a nested dict.
glob_kwargs : dict, optional
Keyword arguments passed to gedi_file_glob() for filtering files
(e.g., version, orbit). Pass ``version`` whenever the tree may hold
more than one GEDI release: granules are keyed by orbit/track only,
so two releases of the same granule would collide under one key.
exclude : list of str, optional
fnmatch-style patterns matched against each file's basename.
Any file whose basename matches any pattern is dropped from the
result. Useful to exclude internal/SGS variants (e.g.
``['*_SGS.h5']``) that share the SOC tree with public release files.
require_all : bool, default True
Keep only granules that have a file for every product found in the
listing. ``False`` keeps every granule with whatever products it has
(its dict simply lacks the others) — for callers that judge
completeness against the products they request, e.g. a build admitting
granules whose later products are not published yet.
Returns
-------
dict or list
If to_list=False: dict keyed by orbit_track identifier, with values
being dicts mapping product codes to file paths. A product with more
than one file maps to its latest release (``gedi_release_rank``),
and the duplicates are logged once per process.
If to_list=True: list of product dicts.
Raises
------
GediValidationError
When the listed files span more than one GEDI release (``_V00N``).
One release per tree is the contract; a mixed listing would silently
pair files of different releases under one granule key.
Examples
--------
>>> tree = soc_file_tree('/path/to/soc')
>>> tree['O12345_01_T00001']
{'L2A': '/path/to/GEDI02_A_...h5', 'L4A': '/path/to/GEDI04_A_...h5'}
>>> # Exclude NASA SGS internal files from discovery
>>> tree = soc_file_tree('/path/to/soc', exclude=['*_SGS.h5'])
"""
direct_access = False
if isinstance(file_struct, str) and os.path.isdir(file_struct):
glob_pattern = gedi_file_glob(**glob_kwargs) if glob_kwargs else 'GEDI*.h5'
# Always walk the SOC tree on the dask cluster. We intentionally do
# not consume ``_soc_manifest.txt`` here: external population paths
# (manual rsync, NASA delivery) bypass the producer-driven refresh
# and would leave the manifest stale relative to disk, silently
# narrowing every downstream scan. ``walk_soc_parallel`` fans the
# year/doy scandir across the registered dask Client, so the cost
# of always walking is bounded.
from .parallel import walk_soc_parallel
file_list = walk_soc_parallel(file_struct, pattern=glob_pattern)
elif (direct_access := isinstance(file_struct[0], EarthAccessFile)):
file_list = [i.path for i in file_struct]
elif isinstance(file_struct[0], str):
file_list = file_struct
else:
raise GediValidationError("file_struct must be a directory or a list of file paths (or s3 links)")
if exclude:
import fnmatch
before = len(file_list)
keep_idx = [
i for i, p in enumerate(file_list)
if not any(fnmatch.fnmatch(os.path.basename(p), pat) for pat in exclude)
]
if len(keep_idx) < before:
file_list = [file_list[i] for i in keep_idx]
if direct_access:
file_struct = [file_struct[i] for i in keep_idx]
# Library is silent about exclusions on purpose: soc_file_tree
# is called multiple times per build and the CLI emits one
# user-facing summary at the existing_h5 filter point.
if not file_list:
return [] if to_list else {}
file_array = np.array(file_struct) if direct_access else np.array(file_list)
flist = pd.DataFrame({'file_paths': file_list, 'file_links': file_array})
fidx = flist.file_paths.str.extract(r'.*GEDI(\d{2})_([A-Z])_.*_(O\d+_\d{2}_T\d+).*\.h5$')
valid = fidx.notna().all(axis=1)
flist = flist.loc[valid].copy()
fidx = fidx.loc[valid]
releases = sorted(flist.file_paths.str.extract(r'_V(\d{3})[^/]*\.h5$')[0].dropna().unique())
if len(releases) > 1:
raise GediValidationError(
f"SOC listing mixes GEDI releases {['V' + r for r in releases]}; granules are keyed by "
f"orbit/track only, so files of different releases would be paired silently. Pin one "
f"release (glob_kwargs={{'version': N}}, or a version-filtered file list)."
)
flist['prod'] = 'L' + fidx[0].astype(int).astype(str) + fidx[1]
flist['orb_track'] = fidx[2]
rank = pd.DataFrame([gedi_release_rank(p) for p in flist.file_paths], index=flist.index,
columns=['_ppds', '_pge', '_gen', '_has_suffix', '_suffix'])
flist = pd.concat([flist, rank], axis=1)
# Each product's latest release sorts last, for the pivot's 'last'. Ties
# (one file copied to two directories) fall back to path order.
flist = flist.sort_values(['orb_track', 'prod', *rank.columns, 'file_paths'])
_report_duplicate_granule_files(flist)
flist = flist.pivot_table(index='orb_track', columns='prod', values='file_links', aggfunc='last')
if require_all:
flist = flist.dropna()
soc_tree = flist.T.to_dict()
else:
soc_tree = {k: {p: f for p, f in v.items() if isinstance(f, str) or not pd.isna(f)}
for k, v in flist.T.to_dict().items()}
if to_list:
soc_tree = list(soc_tree.values())
return soc_tree
[docs]
def gedi_subset(source_file, dest_file, variables, subset_beams=None):
beams = [b for b in GEDI_BEAMS if b in subset_beams] if subset_beams else list(GEDI_BEAMS)
beam_variables = [f"{b}/{v}" for b in beams for v in variables]
h5_copy_subset(source_file, dest_file, beam_variables)
if os.path.exists(dest_file):
return dest_file
return None
[docs]
def gedi_file_glob(orbit=None, orbit_granule=None, track=None, product: int=None, level: str=None, ppds: int=None, pge:int=None, generation:int=None, version:int=None):
str_build = lambda x, digits=2: '*' if x is None else f"{x:0{digits}d}"
lev_str = '*' if level is None else level.upper()
prd_str = str_build(product)
orb_str = str_build(orbit, 5)
ogr_str = str_build(orbit_granule)
trk_str = str_build(track, 5)
pge_str = str_build(pge, 3)
gen_str = str_build(generation)
ver_str = str_build(version, 3)
ppd_str = str_build(ppds)
return f"GEDI{prd_str}_{lev_str}_*_O{orb_str}_{ogr_str}_T{trk_str}_{ppd_str}_{pge_str}_{gen_str}_V{ver_str}*.h5"
_PRESET_KEYWORDS = frozenset({'minimal', 'min', 'default', 'def'})
_SPEC_KEYWORDS = _PRESET_KEYWORDS | {'all', '*'}
[docs]
def gedi_vars_expand(product_vars, version=None):
"""Expand per-product variable specifications to variable lists, in place.
Parameters
----------
product_vars : dict
``{product: spec}``. A spec is ``None`` or ``[]`` (every variable),
``['*']`` / ``['all']`` (every variable), a single path to a file
listing variables, a list naming a preset (``minimal``/``min`` or
``default``/``def``), or a plain list of variable names or wildcards.
Names given alongside a preset are kept and added after the preset's
variables (:func:`preset_extra_names`): ``['default', 'energy_total']``
is the default list plus ``energy_total``. ``minimal`` takes
precedence over ``default`` when both are given, and ``all``/``*``
next to a preset add nothing.
version : int, optional
GEDI release whose preset lists to use.
Returns
-------
dict
``product_vars``, with each spec replaced by its variable list
(``None`` for every variable).
"""
for prod, vars in product_vars.items():
if vars is None:
continue
if isinstance(vars, list) and len(vars) == 0:
# Bare flag (e.g. -l2a with no args) → dump everything
product_vars[prod] = None
elif os.path.isfile(vars[0]) and len(vars) == 1:
with open(vars[0], 'r') as f:
product_vars[prod] = [line.strip() for line in f if line.strip() and not line.startswith('#')]
elif "minimal" in vars or "min" in vars:
# Copy: callers append/purge essentials and quality flags in
# place, which must never rewrite the module-level preset table.
preset = list(_get_versioned(_GEDI_MIN_VARS[prod], version))
product_vars[prod] = list(dict.fromkeys(preset + preset_extra_names(vars)))
elif 'default' in vars or 'def' in vars:
with open(get_default_vars_file(prod, version=version), 'r') as f:
preset = [line.strip() for line in f if line.strip() and not line.startswith('#')]
product_vars[prod] = list(dict.fromkeys(preset + preset_extra_names(vars)))
elif "*" in vars or "all" in vars:
product_vars[prod] = None
elif isinstance(vars, list):
continue
else:
raise GediProductError(f"Unknown variable specification for product {prod}: {vars}")
return product_vars
[docs]
def expand_var_wildcards(var_specs, available_vars):
"""Expand fnmatch-style wildcard patterns in variable specifications.
Supports ``*``, ``?``, ``[seq]``, ``[!seq]``. Non-pattern specs pass
through unchanged. Raises :class:`GediValidationError` if a pattern
matches nothing.
"""
import fnmatch
expanded = []
for spec in var_specs:
if any(c in spec for c in ('*', '?', '[', ']')):
matched = fnmatch.filter(available_vars, spec)
if not matched:
raise GediValidationError(
f"Wildcard pattern '{spec}' matched no available variables"
)
expanded.extend(matched)
else:
expanded.append(spec)
return list(dict.fromkeys(expanded)) # deduplicate, preserve order
[docs]
def gedi_vars_from_h5(gedi_file):
with h5py.File(gedi_file, 'r') as f:
b = [i for i in f.keys() if i.upper().startswith('BEAM')][0]
pl_info = h5_info(gedi_file, root=b)
return pl_info.path.str.replace(b,'').str.lstrip('/').tolist()
_static_vars_cache: Dict = {}
[docs]
def gedi_vars_static(product, version=None):
"""Return the canonical variable list for a GEDI ``(product, version)``
from the static manifest shipped in :mod:`gedih3.data` — no I/O on any
target file.
NASA release files of the same ``(product, version)`` always carry an
identical dataset schema, so this is equivalent to
:func:`gedi_vars_from_h5` for canonical release files but free.
Returns ``None`` when no static manifest exists for the requested
combination; callers should fall back to :func:`gedi_vars_from_h5`.
Note: the result reflects a NASA release file's schema. Files that
were previously subset (e.g. via :func:`gedi_subset` or S3 ETL mode)
contain a *subset* of these variables — for those, use
:func:`gedi_vars_from_h5` to introspect what is actually present.
"""
key = (product, version)
if key in _static_vars_cache:
return _static_vars_cache[key]
try:
with open(get_default_vars_file(product, version=version)) as f:
vars_list = [ln.strip() for ln in f if ln.strip() and not ln.startswith('#')]
except (FileNotFoundError, ValueError):
vars_list = None
_static_vars_cache[key] = vars_list
return vars_list
[docs]
def check_soc_file_vars(soc_file, available_products):
file_products = {}
for prod in available_products.keys():
if prod in soc_file:
try:
available_vars = gedi_vars_from_h5(soc_file[prod])
except Exception as e:
# A single broken granule (truncated download, no BEAM
# group, etc.) must not abort the whole validation bag.
# _filter_granules will drop the file via h5_is_valid.
logger.warning(
f"validate_soc_files: skipping unreadable {prod} file "
f"{soc_file[prod]}: {type(e).__name__}: {e}"
)
continue
file_products[prod] = available_vars
return file_products
@dask.delayed
def _check_soc_file_vars(soc_file, available_products):
return check_soc_file_vars(soc_file, available_products)
[docs]
def validate_soc_files(product_vars: Dict, soc_dir: str = GH3_DEFAULT_SOC_DIR,
version: Optional[int] = None,
exclude: Optional[List[str]] = None):
"""Validate ``product_vars`` against the shipped static manifests.
Only authoritative for **canonical NASA release files** validated against
the per-product manifest in ``src/gedih3/data/GEDI*_DATASETS_*.txt`` —
that is, for the fresh-build ``default`` request (Regime A) or for a
resume where the user explicitly re-requests ``default`` for a product
(Regime C). On all other resume paths the build log is the contract,
and ``gh3_build.py`` bypasses this check entirely. Callers are
responsible for filtering ``product_vars`` to only the products they
actually want to validate against the static manifest — passing the
full union of existing + new vars on a resume will produce false
negatives whenever the shipped manifest has drifted relative to the
list saved in the build log.
"""
glob_kwargs = {'version': version} if version is not None else None
soc_files = soc_file_tree(soc_dir, to_list=True, glob_kwargs=glob_kwargs, exclude=exclude)
if not soc_files:
return False, {"error": "No SOC files found in directory"}
from .config import get_default_vars_file
available_products = {}
for prod in soc_files[0].keys():
try:
with open(get_default_vars_file(prod, version=version)) as f:
available_products[prod] = [
ln.strip() for ln in f
if ln.strip() and not ln.startswith('#')
]
except (FileNotFoundError, ValueError) as e:
logger.warning(
f"validate_soc_files: no static manifest for {prod} v{version}; "
f"skipping variable check for this product ({e})"
)
available_products[prod] = []
validation_report = {
"available_products": available_products,
"missing_products": [],
"missing_variables": {},
"can_skip": True
}
for prod, required_vars in product_vars.items():
if prod not in available_products:
validation_report["missing_products"].append(prod)
validation_report["can_skip"] = False
elif required_vars is not None:
try:
required_vars = expand_var_wildcards(required_vars, available_products[prod])
except GediValidationError:
pass # unmatched patterns will surface as missing vars below
missing_vars = [v for v in required_vars if v not in set(available_products[prod])]
if missing_vars:
validation_report["missing_variables"][prod] = missing_vars
validation_report["can_skip"] = False
if not validation_report["can_skip"]:
error_msg = "Cannot continue due to missing data.\n"
if validation_report.get("missing_products"):
error_msg += f"Missing products: {validation_report['missing_products']}\n"
if validation_report.get("missing_variables"):
for prod, missing_vars in validation_report["missing_variables"].items():
error_msg += f"Missing variables in {prod}: {missing_vars}\n"
validation_report['error_msg'] = error_msg.strip()
return validation_report
[docs]
class GEDIFile:
"""
Parser for GEDI HDF5 filenames.
Extracts metadata from GEDI filename convention including orbit, granule,
track, version, and date information.
Attributes
----------
file_path : str
Full path to the GEDI file
full_name : str
Basename of the file
product : str
Product identifier (e.g., 'GEDI02')
level : str
Processing level (e.g., 'A', 'B')
date : datetime
Acquisition date and time
product_code : str
Combined product code (e.g., 'L2A', 'L4A')
orbit : int
Orbit number
orbit_granule : int
Granule number within orbit
track : int
Ground track number
version : int
Data version number
year : str
Year string (e.g., '2019')
doy : str
Day-of-year string (e.g., '338')
mission_week : int
Mission week number since GEDI launch (2018-12-13)
suffix : str or None
Extra filename parts after version field, if any
Examples
--------
>>> gf = GEDIFile('/path/to/GEDI02_A_2020123120000_O12345_01_T00001_02_003_01_V002.h5')
>>> gf.orbit
12345
>>> gf.version
2
"""
def __init__(self, file_path: Union[str, Path, EarthAccessFile]):
if isinstance(file_path, EarthAccessFile):
file_path = file_path.path
self.parse_file(file_path)
[docs]
def __str__(self) -> str:
return yaml.dump(self)
[docs]
def parse_file(self, f: str) -> None:
# check if f is str or s3 link
self.file_path = f
self.file_size = (os.path.getsize(f) / 1e9) if os.path.exists(f) else None
self.full_name = os.path.basename(f)
f_base = re.sub(r'\.h5$', '', self.full_name)
fl = f_base.split('_')
self.product = fl[0]
self.level = fl[1]
self.date = datetime.strptime(fl[2], '%Y%j%H%M%S')
self.date_str = fl[2]
self.doy_date_str = self.date.strftime('%Y%m%d')
self.julian_date_str = self.date.strftime('%Y%j')
self.time_str = self.date.strftime('%H%M%S')
self.orbit = int(fl[3][1:])
self.orbit_granule = int(fl[4])
self.track = int(fl[5][1:])
self.positioning = int(fl[6])
self.pge = int(fl[7])
self.generation = int(fl[8])
self.version = int(fl[9][1:])
self.product_code = f"L{self.product[-1]}{self.level}"
self.year = str(self.date.year)
self.doy = self.date.strftime('%j')
self.mission_week = (self.date - GEDI_MISSION_START).days // 7 + 1
self.suffix = '_'.join(fl[10:]) if len(fl) > 10 else None
[docs]
def get_glob_pattern(self, product:int=1, level:str='B', ppds:int=2):
return gedi_file_glob(self.orbit, self.orbit_granule, self.track, product, level, ppds)
[docs]
@classmethod
def from_filename(cls, file_path):
return cls(file_path)
[docs]
def search_file(self, soc_dir: str = None, product:int=1, level:str='B', other_files=None): # other_files = e.g. http paths from direct s3
pattern = self.get_glob_pattern(product, level)
if other_files is None:
if soc_dir is None:
soc_dir = soc_from_file(self.file_path)
files = glob.glob(os.path.join(soc_dir, '**', pattern), recursive=True)
else:
files = [f for f in other_files if re.match(pattern.replace('*','.*'), os.path.basename(f))]
if not files:
return None
if len(files) == 1:
return files[0]
parsed_files = [GEDIFile.from_filename(f) for f in files]
version_matches = [f for f in parsed_files if f.version == self.version]
if version_matches:
version_matches.sort(key=lambda x: gedi_release_rank(x.file_path), reverse=True)
return version_matches[0].file_path
parsed_files.sort(key=lambda x: (x.version, *gedi_release_rank(x.file_path)), reverse=True)
return parsed_files[0].file_path
[docs]
class GEDIShot(GEDIFile):
"""
Decoder for GEDI shot numbers.
Extracts the orbit, beam, granule and shot index encoded in GEDI shot
numbers. A shot number has 18 digits, ``OOOOOBBRRGNNNNNNNN``: orbit (5),
beam (2), reserved (2), granule within the orbit (1), shot index (8).
The ground track is not encoded.
Attributes
----------
shot : int or array
The shot number(s)
orbit : int or array
Orbit number
orbit_granule : int or array
Granule number within orbit
track : None
Not encoded in the shot number, so file globs match any track
(orbit and granule identify the file).
beam : int or array
Beam number (0-7)
beam_str : str or list
Beam identifier string (e.g., 'BEAM0101')
power : bool or array
True if power beam (beam > 3)
shot_index : int or array
Shot index within the granule
Examples
--------
>>> shot = GEDIShot(19560500300012345)
>>> shot.orbit, shot.orbit_granule, shot.beam_str, shot.shot_index
(1956, 3, 'BEAM0101', 12345)
"""
def __init__(self, shot: Union[int, np.ndarray], build_glob: bool = False):
self.parse_shot(shot)
if build_glob:
self.get_glob_pattern()
[docs]
def parse_shot(self, shot: Union[int, np.ndarray]) -> None:
self.is_scalar = np.isscalar(shot)
self.shot = shot
self.orbit = shot // 10000000000000
self.orbit_granule = shot % 1000000000 // 100000000
self.track = None # not encoded; ``shot // 1e11`` is the orbit and beam digits
self.beam = shot % 10000000000000 // 100000000000
self.power = self.beam > 3
self.reserved = shot % 1000000000000 // 1000000000 % 100
self.shot_index = shot % 100000000
if self.is_scalar:
self.beam_str = f'BEAM{self.beam:04b}'
else:
self.beam_str = [f'BEAM{b:04b}' for b in self.beam]
[docs]
def get_glob_pattern(self, product: int=1, level: str='B', ppds: int=2, pge:int=None, generation:int=None, version:int=None):
"""Glob pattern(s) of the GEDI file(s) holding these shots (any track).
Returns
-------
str or list of str
One pattern for a scalar shot; for an array, one per distinct
granule, sorted. Also stored as ``file_glob_pattern``.
"""
if self.is_scalar:
self.file_glob_pattern = gedi_file_glob(self.orbit, self.orbit_granule, self.track, product, level, ppds, pge, generation, version)
else:
self.file_glob_pattern = [gedi_file_glob(self.orbit[i], self.orbit_granule[i], self.track, product, level, ppds, pge, generation, version) for i in range(len(self.shot))]
self.file_glob_pattern = sorted(set(self.file_glob_pattern))
return self.file_glob_pattern
[docs]
def search_file(self, soc_dir: str):
"""Files under ``soc_dir`` matching ``file_glob_pattern`` (built if absent).
Returns
-------
list of str
Sorted paths; empty if none match.
"""
# A SOC tree is <soc>/<year>/<doy>/: walk it, as GEDIFile.search_file does
# (_soc_manifest.txt is write-only, so smart_glob does not apply).
patterns = getattr(self, 'file_glob_pattern', None) or self.get_glob_pattern()
if self.is_scalar:
patterns = [patterns]
files = set()
for fp in patterns:
files.update(glob.glob(os.path.join(soc_dir, '**', fp), recursive=True))
return sorted(files)
[docs]
@njit
def wfm_pad(wave, n_bins=1420, cval=0):
bin_diff = len(wave) - n_bins
if bin_diff == 0:
return wave
if bin_diff < 0:
padded = np.full(n_bins, cval, dtype=wave.dtype)
padded[:len(wave)] = wave
return padded
if bin_diff > 0:
return wave[:n_bins]
def _validate_h5_columns(columns: List[str]) -> List[str]:
"""
Validate and normalize the column list for HDF5 extraction.
Ensures shot_number is included and adds waveform dependencies.
Parameters
----------
columns : list of str
Requested column names
Returns
-------
list of str
Normalized column list with dependencies added
"""
columns = list(columns) # Make a copy
if 'shot_number' not in columns:
columns.append('shot_number')
if 'rxwaveform' in columns:
columns += ['noise_mean_corrected', 'rx_sample_start_index', 'rx_sample_count']
# dict.fromkeys preserves insertion order; list(set(...)) is hash-seed
# dependent and produced different orders per worker process, breaking
# the driver-derived canonical schema for L1B builds.
columns = list(dict.fromkeys(columns))
if 'txwaveform' in columns:
columns += ['noise_mean_corrected', 'tx_sample_start_index', 'tx_sample_count']
columns = list(dict.fromkeys(columns))
return columns
def _get_beams_to_load(
h5_file: h5py.File,
shots: Optional[np.ndarray],
which_beams: Optional[List[str]]
) -> tuple:
"""
Determine which beams to load based on shots and beam filters.
Parameters
----------
h5_file : h5py.File
Open HDF5 file handle
shots : array, optional
Specific shot numbers to extract
which_beams : list of str, optional
Specific beams to load
Returns
-------
tuple
(beams_to_load, all_beams, shot_beams_array_or_none)
"""
all_beams = [k for k in h5_file.keys() if k.startswith('BEAM')]
beams = all_beams
shot_beams = None
if shots is not None:
shot_beams = np.uint16(shots % 10000000000000 // 100000000000)
beams = [f'BEAM{b:04b}' for b in np.unique(shot_beams)]
if which_beams is not None:
beams = [b for b in beams if b in which_beams]
return beams, all_beams, shot_beams
def _extract_beam_data(
h5_file: h5py.File,
beam: str,
columns: List[str],
shots: Optional[np.ndarray],
shot_beams: Optional[np.ndarray],
include_source: bool
) -> Optional[pd.DataFrame]:
"""
Extract data from a single beam in an HDF5 file.
Parameters
----------
h5_file : h5py.File
Open HDF5 file handle
beam : str
Beam identifier (e.g., 'BEAM0101')
columns : list of str
Variables to extract
shots : array, optional
Specific shot numbers to extract
shot_beams : array, optional
Beam numbers for each shot (pre-computed)
include_source : bool
Add root_beam column
Returns
-------
pd.DataFrame or None
DataFrame with extracted data, or None if no matching shots
"""
shot_ids = h5_file[f"{beam}/shot_number"][:]
if shots is not None:
kint = int(beam[-4:], base=2)
k_shots = shots[shot_beams == kint]
idx = np.where(np.isin(shot_ids, k_shots))
else:
idx = np.arange(0, len(shot_ids))
if len(idx) == 0:
return None
dfs = {}
for j in columns:
try:
if is_wave := (j == 'rxwaveform' or j == 'txwaveform'):
is_tx = j == 'txwaveform'
d = wfm_extract(h5_file, beam, idx, is_tx)
else:
d = h5_file[f"{beam}/{j}"][:][idx]
except KeyError as e:
# Normalize h5py's two missing-path error shapes ("object 'X'
# doesn't exist" vs "component not found") into the single form
# _MISSING_VAR_RE picks up, so _classify_load_h5_failure tags the
# failure as missing_var with the actual variable name and the
# end-of-build recovery advisory groups it correctly.
raise KeyError(f"object '{j}' doesn't exist (h5py: {e})") from e
if d.ndim == 2:
for col in range(d.shape[-1]):
jj = f"{j}_{col:0{4 if is_wave else 3}d}"
dfs[jj] = d[:, col]
else:
dfs[j] = d
df = pd.DataFrame(dfs)
if include_source:
df = df.assign(root_beam=beam)
return df
def _build_dataframe(
beam_frames: List[pd.DataFrame],
fpath: Union[str, Path, EarthAccessFile],
include_source: bool,
dropna: bool
) -> pd.DataFrame:
"""
Build the final DataFrame from extracted beam data.
Parameters
----------
beam_frames : list of pd.DataFrame
DataFrames from each beam
fpath : str, Path, or EarthAccessFile
Source file path (for root_file column)
include_source : bool
Add root_file column
dropna : bool
Drop rows with NaN values
Returns
-------
pd.DataFrame
Combined DataFrame indexed by shot_number
"""
full_df = pd.concat(beam_frames)
full_df = full_df.set_index('shot_number')
if include_source:
full_df = full_df.assign(
root_file=os.path.basename(fpath.path if isinstance(fpath, EarthAccessFile) else fpath)
)
if dropna:
full_df = full_df.dropna()
return full_df
[docs]
def load_h5(
fpath: Union[str, Path, EarthAccessFile],
columns: List[str],
which_beams: Optional[List[str]] = None,
shots: Optional[np.ndarray] = None,
include_source: bool = True,
dropna: bool = True
) -> pd.DataFrame:
"""
Load data from a GEDI HDF5 file into a pandas DataFrame.
Parameters
----------
fpath : str, Path, or EarthAccessFile
Path to the GEDI HDF5 file
columns : list of str
Variable names to extract from the file
which_beams : list of str, optional
Specific beams to load (e.g., ['BEAM0101', 'BEAM1000']).
If None, loads all beams.
shots : array, optional
Specific shot numbers to extract. If None, loads all shots.
include_source : bool, default True
Add root_file and root_beam columns to output
dropna : bool, default True
Drop rows with NaN values
Returns
-------
pd.DataFrame
DataFrame indexed by shot_number containing requested variables
"""
f = h5py.File(fpath, mode='r', locking=False)
# Validate and normalize columns
columns = _validate_h5_columns(columns)
# Determine which beams to load
beams, all_beams, shot_beams = _get_beams_to_load(f, shots, which_beams)
# Handle case where no beams match
if len(beams) == 0:
f.close()
return load_h5(fpath=fpath, columns=columns, include_source=include_source, which_beams=all_beams[:1]).head(0)
# Extract data from each beam
beam_frames = []
for beam in beams:
beam_df = _extract_beam_data(f, beam, columns, shots, shot_beams, include_source)
if beam_df is not None:
beam_frames.append(beam_df)
f.close()
# Build final DataFrame
return _build_dataframe(beam_frames, fpath, include_source, dropna)
[docs]
def load_h5_merged(
prod_files: Dict[str, str],
product_vars: Dict[str, List[str]],
which_beams: Optional[List[str]] = None,
shots: Optional[np.ndarray] = None,
dropna: bool = True,
suffix_all: bool = False
) -> Optional[pd.DataFrame]:
"""
Load and merge data from multiple GEDI product files.
Combines data from different GEDI products (L2A, L2B, L4A, etc.) into a
single DataFrame, joining on shot_number.
Parameters
----------
prod_files : dict
Mapping of product codes to file paths (e.g., {'L2A': '/path/to/l2a.h5'})
product_vars : dict
Mapping of product codes to lists of variable names to extract
which_beams : list of str, optional
Specific beams to load
shots : array, optional
Specific shot numbers to extract
dropna : bool, default True
Drop rows with NaN values
suffix_all : bool, default False
If True, suffix all columns with product code (e.g., 'rh_l2a').
Required when variables have the same name across products.
Returns
-------
pd.DataFrame or None
Merged DataFrame indexed by shot_number, or None if no data
"""
frames = []
for i, (p, vars) in enumerate(product_vars.items()):
ppath = prod_files.get(p)
if ppath is None:
continue
idf = load_h5(fpath=ppath, columns=vars, include_source = suffix_all or i==0, which_beams=which_beams, shots=shots, dropna=dropna)
suffix = f"_{p.lower()}"
if suffix_all:
idf = idf.rename(columns=lambda x: x if x.endswith(suffix) else f"{x}{suffix}")
frames.append(idf)
if not frames:
return None
if not suffix_all:
all_cols = [col for frame in frames for col in frame.columns]
if len(all_cols) != len(set(all_cols)):
raise GediVariableError("Duplicate columns detected. Remove duplicates or use suffix_all=True to avoid conflicts.")
df = pd.concat(frames, axis=1, join='inner')
return df
[docs]
def dask_h5_merged(
prod_files_list: List[Dict[str, str]],
product_vars: Dict[str, List[str]],
which_beams: Optional[List[str]] = None,
shots: Optional[np.ndarray] = None,
dropna: bool = True,
suffix_all: bool = False,
by_beam: bool = False
) -> dask.dataframe.DataFrame:
"""
Create a Dask DataFrame from multiple GEDI product file sets.
Lazily loads and merges GEDI data for distributed processing.
Parameters
----------
prod_files_list : list of dict
List of product file dictionaries from soc_file_tree()
product_vars : dict
Mapping of product codes to variable lists
which_beams : list of str, optional
Specific beams to load
shots : array, optional
Specific shot numbers to extract
dropna : bool, default True
Drop rows with NaN values
suffix_all : bool, default False
Suffix all columns with product code
by_beam : bool, default False
If True, create separate partitions for each beam (more parallelism)
Returns
-------
dask.dataframe.DataFrame
Lazy Dask DataFrame for distributed processing
"""
# Probe files in order to derive the dask metadata schema. A single broken
# granule at the head of the list (corrupt HDF5, missing variable inside a
# BEAM, etc.) must not abort the whole build — skip it with a warning and
# try the next. Per-partition errors at runtime are already swallowed by
# the `_load_h5_merged` closure below.
_meta = None
_skipped = 0
for i, _pf in enumerate(prod_files_list):
try:
_meta = load_h5_merged(
_pf, product_vars=product_vars,
which_beams=GEDI_BEAMS[:1],
dropna=dropna, suffix_all=suffix_all,
).head(0)
if i > 0:
logger.warning(
f"Skipped {i} broken file(s) while deriving metadata schema; using {_pf} as the schema sample"
)
break
except Exception as e:
_skipped += 1
logger.warning(f"Skipping broken file for metadata sample: {_pf} ({e})")
continue
if _meta is None:
raise GediFileError(
f"Could not derive a metadata schema: all {_skipped} candidate file(s) failed to read"
)
def _load_h5_merged(prod_files, **kwargs):
try:
return load_h5_merged(prod_files, **kwargs)
except Exception as e:
logger.warning(f"Error loading file combination: {prod_files}")
return _meta
if by_beam:
beams = GEDI_BEAMS if which_beams is None else which_beams
def load_by_beam(pfiles_beam_tuple, product_vars, shots, dropna, suffix_all):
pfiles, beam = pfiles_beam_tuple
return _load_h5_merged(pfiles, product_vars=product_vars, which_beams=[beam], shots=shots, dropna=dropna, suffix_all=suffix_all)
file_beam_combinations = list(itertools.product(prod_files_list, beams))
return dask.dataframe.from_map(load_by_beam, file_beam_combinations, product_vars=product_vars, shots=shots, dropna=dropna, suffix_all=suffix_all, meta=_meta)
return dask.dataframe.from_map(_load_h5_merged, prod_files_list, which_beams=which_beams, shots=shots, product_vars=product_vars, dropna=dropna, suffix_all=suffix_all, meta=_meta)
[docs]
def add_special_columns(df, lon_col:str=None, lat_col:str=None, dat_col:str=None):
if dat_col:
df = df.assign(datetime=pd.to_datetime(df[dat_col] + GEDI_START_DATE.timestamp(), unit='s').dt.as_unit('s'))
if lon_col and lat_col:
df = gpd.GeoDataFrame(df, geometry=gpd.points_from_xy(df[lon_col], df[lat_col], crs='EPSG:4326'))
return df.copy()
