# 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
import os, glob
from typing import Dict
from datetime import datetime, timezone
from .config import GEDI_PRODUCTS, GH3_DEFAULT_DOWNLOAD_DIR, GH3_DEFAULT_SOC_DIR, GH3_DEFAULT_H3_DIR, BUILD_LOG_FILENAME, PARTITION_META_FILENAME
from .exceptions import GediValidationError
from .utils import now, json_read, json_write, read_vector_file, to_geojson, from_geojson, parse_spatial, merge_spatial, parse_temporal, get_package_version
from .h3utils import intersect_h3_geometries
from .gedidriver import (GEDIFile, gedi_vars_expand, soc_file_tree, check_soc_file_vars, validate_soc_files,
write_soc_manifest, gedi_release_rank, gedi_release_tag)
from .gh3driver import gh3_list_files
_VALID_STATUSES = (
'DOWNLOADING',
'PROCESSING',
'PARTITIONING',
'MERGING',
'COMPLETED',
'FAILED',
'INTERRUPTED',
'UNKNOWN'
)
# Per-granule per-product status vocabulary (used by gh3_doctor and lazy upgrade
# of legacy log entries). Top-level granule 'status' remains the worst per-product
# value for backwards compatibility with readers that ignore the products map.
PRODUCT_STATUS_INDEXED = 'INDEXED'
PRODUCT_STATUS_PARTIAL_NAN = 'PARTIAL_NAN'
PRODUCT_STATUS_MISSING_COLUMN = 'MISSING_COLUMN'
PRODUCT_STATUS_MISSING_SOURCE = 'MISSING_SOURCE'
PRODUCT_STATUS_FAILED = 'FAILED'
PRODUCT_STATUS_PENDING = 'PENDING'
# Top-level granule status (never a per-product status): the granule was read in
# full by Stage 1 under the recorded scope and yielded zero rows (e.g. entirely
# outside a land-mask ROI). Skipped like INDEXED while the scope is unchanged.
# Set only on positive evidence by ``H3BuildLogger.mark_no_data``; a granule
# found on disk becomes INDEXED again. Valid only for the scope it was recorded
# under: any spatial/temporal widening reopens it (the logger demotes it to
# PENDING when it detects one), as does merging a log that never examined it.
GRANULE_STATUS_NO_DATA = 'NO_DATA'
_VALID_PRODUCT_STATUSES = (
PRODUCT_STATUS_INDEXED,
PRODUCT_STATUS_PARTIAL_NAN,
PRODUCT_STATUS_MISSING_COLUMN,
PRODUCT_STATUS_MISSING_SOURCE,
PRODUCT_STATUS_FAILED,
PRODUCT_STATUS_PENDING,
)
def _products_from_columns(columns, active_products):
"""Return the subset of ``active_products`` whose suffix appears in ``columns``.
A product P is considered present if any column ends with ``_<p>`` (the
suffix convention used by `_add_variables_to_year_file` and `load_h5_merged`).
"""
if not columns or not active_products:
return set()
cols_lower = [str(c).lower() for c in columns]
return {p for p in active_products if any(c.endswith(f"_{p.lower()}") for c in cols_lower)}
def _per_product_status_from_observed(active_products, observed_products):
"""Map active products to INDEXED if observed, else MISSING_COLUMN.
PARTIAL_NAN is not derived here because it requires a row-level scan;
it is populated by gh3_doctor after `_load_filters_from_log`.
"""
return {
p: PRODUCT_STATUS_INDEXED if p in observed_products else PRODUCT_STATUS_MISSING_COLUMN
for p in active_products
}
def _scan_partition_meta_post_build_info(partition_dir, *, meta_filename, active_products, verify=False):
"""Worker: read PARTITION_META JSONs under one h3_* partition and return
the compact aggregate the driver needs to fold into set_post_build_info.
Module-level so it pickles for dask. Per-file errors are non-fatal — the
affected file contributes nothing, matching the legacy serial behavior
(any json_read raise would have aborted the whole scan; we instead skip).
With ``verify=True`` the result also carries ``'unverified'``: True when
this partition's metadata may not list every granule it holds (an
unreadable sidecar, no sidecar at all, or fewer per-year sidecars than
parquet files). ``set_post_build_info`` uses it to refuse the "granule
absent from every partition" inference behind ``NO_DATA``.
"""
out = {
'granules': [],
'date_min': None,
'date_max': None,
'h3_partitions': [],
'columns': [],
'l2a_version': None,
'column_dtypes': {},
'partition_products': set(),
'unverified': False,
}
cols_union: set = set()
metas = glob.glob(os.path.join(partition_dir, f'*{meta_filename}'))
if verify:
n_year_meta = len(glob.glob(os.path.join(partition_dir, '*', f'*{meta_filename}')))
n_year_data = len(glob.glob(os.path.join(partition_dir, '*', '*.parquet')))
if not metas or n_year_meta < n_year_data:
out['unverified'] = True
for mf in metas:
try:
fmeta = json_read(mf) or {}
except Exception:
out['unverified'] = True
continue
gran = fmeta.get('granules', []) or []
drange = fmeta.get('date_range')
cols = fmeta.get('columns', []) or []
if drange:
if out['date_min'] is None or drange[0] < out['date_min']:
out['date_min'] = drange[0]
if out['date_max'] is None or drange[1] > out['date_max']:
out['date_max'] = drange[1]
h3p = fmeta.get('h3_partition')
if h3p is not None:
out['h3_partitions'].append(h3p)
out['granules'].extend(gran)
out['partition_products'].update(_products_from_columns(cols, active_products))
if out['l2a_version'] is None and fmeta.get('l2a_version') is not None:
out['l2a_version'] = fmeta.get('l2a_version')
cols_union.update(cols)
cd = fmeta.get('column_dtypes') or {}
for col_name, dtype in cd.items():
out['column_dtypes'].setdefault(col_name, dtype)
out['columns'] = list(cols_union)
return out
[docs]
def load_log_data(file_path):
if os.path.exists(file_path):
return json_read(file_path)
return {}
[docs]
def merge_temporal(existing, new):
if existing is None:
return None, None
if new is None:
return existing, None
new = parse_temporal(new)
start1, end1 = existing
start2, end2 = new
new_start = None if None in (start1, start2) else min(start1, start2)
new_end = None if None in (end1, end2) else max(end1, end2)
t_full = (new_start, new_end)
if t_full == (None, None):
return t_full, t_full
t_diff = list(t_full)
if new_start is not None and new_start >= start1 and end1 is not None:
t_diff[0] = end1
if new_end is not None and new_end <= end1 and start1 is not None:
t_diff[1] = start1
t_diff = tuple(t_diff)
if t_full == t_diff or t_diff[0] >= t_diff[1]:
t_diff = None
return t_full, t_diff
[docs]
def merge_product_vars(existing_product_vars, new_product_vars=None):
if not new_product_vars:
return existing_product_vars, None
merged_product_vars = existing_product_vars.copy()
update_prods = set()
for prod, vars_list in new_product_vars.items():
if prod not in merged_product_vars:
merged_product_vars[prod] = vars_list
update_prods.add(prod)
elif merged_product_vars[prod] is None or vars_list is None:
merged_product_vars[prod] = None
if vars_list is None and existing_product_vars[prod] is not None:
update_prods.add(prod)
else:
existing_set = set(merged_product_vars[prod])
new_set = set(vars_list)
merged_set = existing_set | new_set
merged_product_vars[prod] = list(merged_set)
if merged_set != existing_set:
update_prods.add(prod)
new_product_vars = None
if len(update_prods) > 0:
new_product_vars = {k:val for k, val in merged_product_vars.items() if k in update_prods}
return merged_product_vars, new_product_vars
[docs]
def resolve_soc_version(soc_dir):
"""Return the GEDI data version a local SOC tree was populated with, or None.
A-priori lookup — no HDF5 open, no recursive walk:
1. the download log's ``gedi_version`` (``gedih3_download_log.json`` at
the SOC root), written by every gh3 producer;
2. else the ``_V00N`` field of the first ``GEDI*.h5`` found by a bounded
``year/doy`` scandir (externally populated trees: manual rsync, NASA
delivery).
Used by ``gh3_build`` / ``gh3_download`` to pin a *fresh* build or
resumed download to the release already on disk when the user did not
pass ``--gedi-version``; the package default (``GEDI_DEFAULT_VERSION``)
only applies when nothing on disk answers the question.
"""
if not soc_dir or not os.path.isdir(soc_dir):
return None
log_path = os.path.join(soc_dir, SOCDownloadLogger._LOG_FILE_NAME)
if os.path.isfile(log_path):
try:
v = json_read(log_path).get('gedi_version')
if v is not None:
return int(v)
except (OSError, ValueError, TypeError):
pass
try:
with os.scandir(soc_dir) as years:
year_dirs = sorted(
(e.path for e in years if e.is_dir() and e.name.isdigit()),
reverse=True,
)
except OSError:
return None
for ydir in year_dirs:
try:
with os.scandir(ydir) as doys:
doy_dirs = sorted((e.path for e in doys if e.is_dir()), reverse=True)
except OSError:
continue
for ddir in doy_dirs:
try:
with os.scandir(ddir) as files:
for e in files:
if e.name.startswith('GEDI') and e.name.endswith('.h5'):
try:
return GEDIFile(e.path).version
except Exception:
continue
except OSError:
continue
return None
[docs]
class SOCDownloadLogger:
_LOG_FILE_NAME = 'gedih3_download_log.json'
_PARENT_DIR = GH3_DEFAULT_SOC_DIR
def __init__(self, product_vars, spatial=None, temporal=None, version=None, dir=None):
if dir is not None:
self._PARENT_DIR = dir
self.gedi_version = version
self.s3_access = False
self.log_file = os.path.join(self._PARENT_DIR, self._LOG_FILE_NAME)
self.log_data = load_log_data(self.log_file)
self.updating = False
# Resume-mode `default`/`minimal` expansion must target the
# existing tree's version, not the package fallback
# (GEDI_DEFAULT_VERSION). Otherwise `gh3_download -l4c default`
# against a v2 download tree expands against the v3 manifest and
# writes the wrong variable names. Peek the log before expansion
# so an absent `--gedi-version` CLI arg adopts the persisted version.
effective_version = version
if version is None and self.log_data:
effective_version = self.log_data.get('gedi_version', version)
persisted_version = self.log_data.get('gedi_version') if self.log_data else None
if version is not None and persisted_version is not None and int(version) != int(persisted_version):
# A SOC tree *can* hold two releases side by side (filenames carry
# the version and every consumer globs by it), so this is not
# fatal — but the log records one version, and a later
# `gh3_build -i` with no --gedi-version pins to whatever is
# written there. Say so loudly.
from .logging_config import get_logger
get_logger(__name__).warning(
f"GEDI version mismatch: download log at {self._PARENT_DIR} records "
f"version {persisted_version}, but version {version} was requested. "
f"The tree will hold both releases and the log will now record "
f"version {version}; prefer one SOC directory per release."
)
if product_vars:
product_vars = gedi_vars_expand(product_vars, version=effective_version)
if not self.log_data:
self.product_vars = product_vars
self.spatial = parse_spatial(spatial)
self.temporal = parse_temporal(temporal)
self.new_spatial = None
self.new_temporal = None
self.new_product_vars = None
return
self._load_filters_from_log()
self.updating = True
self.new_spatial = None
self.new_temporal = None
self.new_product_vars = None
if temporal is not None:
self.temporal, self.new_temporal = merge_temporal(self.temporal, temporal)
if spatial is not None:
self.spatial, self.new_spatial = merge_spatial(self.spatial, spatial)
if product_vars is not None:
self.product_vars, self.new_product_vars = merge_product_vars(self.product_vars, product_vars)
def _load_filters_from_log(self):
self.product_vars = self.log_data.get('products', {})
self.product_vars = {k:val.get('variables') for k,val in self.product_vars.items()}
self.spatial = parse_spatial(self.log_data.get('spatial_filter'))
self.temporal = parse_temporal(self.log_data.get('temporal_filter'))
self.gedi_version = self.log_data.get('gedi_version', self.gedi_version)
self.s3_access = self.log_data.get('s3_access', False)
if 'granules' in self.log_data:
self.granule_info = self.log_data.get('granules')
[docs]
def register_pending_granules(self, granule_list):
"""Register granules as PENDING. Called before download starts.
Parameters
----------
granule_list : list of dict
Each dict must have 'orbit', 'granule', 'track' keys.
"""
if not hasattr(self, 'granule_info'):
self.granule_info = []
existing = {(g['orbit'], g['granule'], g['track']) for g in self.granule_info}
for g in granule_list:
key = (g['orbit'], g['granule'], g['track'])
if key not in existing:
self.granule_info.append({**g, 'status': 'PENDING'})
existing.add(key)
[docs]
def update_granule_status(self, gran_key, status):
"""Update status of a single granule (does NOT auto-save).
Caller controls save frequency via explicit save_log() calls.
Parameters
----------
gran_key : dict
Dict with 'orbit', 'granule', 'track' keys.
status : str
'DOWNLOADED' or 'FAILED'.
"""
if not hasattr(self, 'granule_info'):
return
key = (gran_key['orbit'], gran_key['granule'], gran_key['track'])
for g in self.granule_info:
if (g['orbit'], g['granule'], g['track']) == key:
g['status'] = status
break
[docs]
def set_post_download_info(self):
"""Scan SOC directory and record downloaded granules.
Refreshes ``_soc_manifest.txt`` first so subsequent
``soc_file_tree`` calls (here and on the next resume) skip the
recursive glob over the SOC tree. This is the same pattern the
H3 database uses via ``MANIFEST_FILENAME``.
"""
try:
n_files = write_soc_manifest(self._PARENT_DIR)
except OSError:
# Manifest is an optimization, not a correctness gate; if the
# SOC root is read-only or transiently unavailable the
# downstream glob fallback still produces correct results.
# Defer the count to the post-glob fallback below.
n_files = None
# One release per listing: the tree may hold another release side
# by side, and this log records only the release it tracks.
version = self.gedi_version if self.gedi_version is not None else resolve_soc_version(self._PARENT_DIR)
glob_kwargs = {'version': version} if version is not None else None
soc_files = soc_file_tree(self._PARENT_DIR, to_list=True, glob_kwargs=glob_kwargs)
if n_files is None:
# Manifest write failed; recover the count from the discovery
# step so downstream summary lines stay correct.
n_files = sum(len(d) for d in soc_files)
self.n_files = n_files
granule_info = []
for soc in soc_files:
first_file = list(soc.values())[0]
gfile = GEDIFile(first_file)
gran = {'orbit': gfile.orbit, 'granule': gfile.orbit_granule, 'track': gfile.track}
if gran not in granule_info:
granule_info.append(gran)
if self.gedi_version is None:
self.gedi_version = gfile.version
# Merge with existing tracked granules (preserve status from real-time tracking)
if hasattr(self, 'granule_info') and self.granule_info:
existing = {(g['orbit'], g['granule'], g['track']): g for g in self.granule_info}
for g in granule_info:
key = (g['orbit'], g['granule'], g['track'])
if key in existing:
existing[key].update({k: v for k, v in g.items() if k != 'status'})
if existing[key].get('status') != 'FAILED':
existing[key]['status'] = 'DOWNLOADED'
else:
existing[key] = {**g, 'status': 'DOWNLOADED'}
self.granule_info = list(existing.values())
else:
self.granule_info = [{**g, 'status': 'DOWNLOADED'} for g in granule_info]
[docs]
def get_finished_granules(self):
"""Return skip list when resuming with same filters.
Only returns successfully completed granules, with the status
field stripped for backward compatibility with _filter_granules()
dict comparison. Legacy logs without status fields are treated
as successful.
"""
if (hasattr(self, 'granule_info')
and getattr(self, 'new_product_vars', None) is None
and getattr(self, 'new_temporal', None) is None):
return [
{k: v for k, v in g.items() if k != 'status'}
for g in self.granule_info
if g.get('status') in ('DOWNLOADED', None)
]
return None
[docs]
def get_temporal(self):
if not self.updating or self.new_temporal is None:
return self.temporal
if self.new_spatial is None and self.new_product_vars is None:
return self.new_temporal
return self.temporal
[docs]
def get_spatial(self):
if not self.updating or self.new_spatial is None:
return self.spatial
if self.new_product_vars is None and self.new_temporal is None:
return self.new_spatial
return self.spatial
[docs]
def get_product_vars(self):
if not self.updating or self.new_product_vars is None:
return self.product_vars
if self.new_spatial is None and self.new_temporal is None:
return self.new_product_vars
return self.product_vars
[docs]
def to_dict(self, status):
if status not in _VALID_STATUSES:
raise ValueError(f"Invalid status '{status}'. Must be one of {_VALID_STATUSES}")
product_logs = self.log_data.get('products', {})
for prod in self.get_product_vars().keys():
product_logs[prod] = product_logs.get(prod, {})
product_logs[prod]['status'] = status
product_logs[prod]['last_modified'] = now()
vars_list = self.product_vars.get(prod)
product_logs[prod]['variables'] = sorted(vars_list) if vars_list else vars_list
log_dict = {
'metadata': {
'package_version': get_package_version()
},
'gedi_version': self.gedi_version,
'status': status,
'last_modified': now(),
'spatial_filter': None if self.spatial is None else to_geojson(self.spatial),
'temporal_filter': self.temporal,
"s3_access": self.s3_access,
'products': product_logs
}
if hasattr(self, 'granule_info'):
log_dict['granules'] = self.granule_info
return log_dict
[docs]
def save_log(self, status):
json_write(self.to_dict(status), self.log_file, mode='w', rewrite=True)
[docs]
def stored_exclude(log_data) -> list:
"""The source exclusions a build log keeps (``gh3_build --exclude``).
A single pattern written by hand as a string is one pattern, never its
characters (``'*'`` alone would exclude every file).
Raises
------
GediValidationError
If ``exclude`` is neither a string nor a list of strings.
"""
value = (log_data or {}).get('exclude') or []
if isinstance(value, str):
value = [value]
if not isinstance(value, list) or not all(isinstance(p, str) for p in value):
raise GediValidationError(
f"The build log's 'exclude' must be a list of glob patterns (gh3_build --exclude), got {value!r}")
return list(value)
[docs]
class H3BuildLogger:
_LOG_FILE_NAME = BUILD_LOG_FILENAME
_PARENT_DIR = GH3_DEFAULT_H3_DIR
def __init__(self, product_vars, spatial=None, temporal=None, res:int=12, part:int=3, version:int=None, dir=None, source_mode=None):
if dir is not None:
self._PARENT_DIR = dir
# Capture which products were requested via the `default` keyword
# before expansion replaces it with the literal manifest list. This
# is the bit the resume-path validator gate needs in `gh3_build.py`:
# only the static manifest in `src/gedih3/data/GEDI0*_DATASETS_*.txt`
# is the contract for a `default` request; any other shape (explicit
# list, file path, `*`/`all`, or unmodified resume) treats the build
# log as authoritative and skips manifest consultation entirely.
# Runtime-only — never persisted to disk.
self.default_products = {
prod for prod, v in (product_vars or {}).items()
if isinstance(v, list) and any(
isinstance(s, str) and s in ('default', 'def') for s in v
)
}
# Superset of `default_products` covering every preset keyword
# (`default`/`minimal`/`all`), as opposed to a literal user-supplied
# name, file path, or wildcard. gh3_build's Stage-2 pre-flight
# sample check must skip these too: their variable names come from
# an internal manifest/table, not something the user typed, so a
# mismatch there means the GEDI-version assumption used to resolve
# the preset was wrong — not a user typo — and downstream variable
# resolution (`_expand_product_vars`) already re-derives the
# correct version-specific names regardless. Runtime-only.
self.preset_products = {
prod for prod, v in (product_vars or {}).items()
if isinstance(v, list) and any(
isinstance(s, str) and s in ('default', 'def', 'minimal', 'min', 'all', '*')
for s in v
)
}
# Names the user typed next to a preset keyword (`-l2a default
# energy_total` → {'L2A': ['energy_total']}). Expansion merges them
# into the preset list, but they are not in the static manifest:
# gh3_build's pre-flight checks them against a sample HDF5 instead.
# Runtime-only.
from .gedidriver import preset_extra_names
self.preset_extra_vars = {
prod: names for prod, v in (product_vars or {}).items()
if (names := preset_extra_names(v))
}
self.log_file = os.path.join(self._PARENT_DIR, self._LOG_FILE_NAME)
self.log_data = load_log_data(self.log_file)
# Resume-mode `default`/`minimal` expansion must target the
# existing DB's gedi_version, not the package fallback
# (GEDI_DEFAULT_VERSION). Otherwise `gh3_build -l4c default`
# against an existing v2 DB expands against the v3 manifest,
# requesting variable names that don't exist in the v2 h5 files
# and aborting at validation. Peek the log before expansion so an
# absent `--gedi-version` CLI arg adopts the persisted version.
# A database is single-version by contract: every product in it
# comes from the same GEDI release (enforced by the mismatch check
# below and by the per-version SOC glob in the build).
effective_version = version
if version is None and self.log_data:
effective_version = self.log_data.get('gedi_version', version)
if product_vars:
product_vars = gedi_vars_expand(product_vars, version=effective_version)
self.updating = False
# Set by gh3_build once every stage ran under this run's update: its
# final save (MERGING when merges failed) must not keep it pending.
self.update_ran = False
self.source_mode = source_mode
self.build_start_time = datetime.now(timezone.utc)
self.previous_status = self.log_data.get('status')
# Phased updates (gh3_build --allow-missing-products): once chosen for
# a database, persisted in its log so later runs keep admitting L2A
# before the later products are published.
self.allow_missing_products = bool(self.log_data.get('allow_missing_products'))
# Source exclusions (gh3_build --exclude): persisted, so every later run
# and every listing that takes no --exclude (variable adds, gh3_doctor)
# leaves out the same files the database was built without.
self.exclude = stored_exclude(self.log_data)
if not self.log_data:
self.product_vars = product_vars
self.spatial = parse_spatial(spatial)
self.temporal = parse_temporal(temporal)
# Fresh-build fall-back: argparse defaults are now None (so the
# resume-mismatch check below can be strict). Apply the package
# defaults here when the user didn't specify them.
self.res = res if res is not None else 12
self.part = part if part is not None else 3
self.gedi_version = version
self.new_spatial = None
self.new_temporal = None
self.new_product_vars = None
self.requested_expansion = False
return
self._load_filters_from_log()
if version is not None and self.gedi_version is not None and version != self.gedi_version:
raise GediValidationError(
f"GEDI version mismatch: existing database is version {self.gedi_version}, "
f"but version {version} was requested. Different versions require separate databases."
)
# H3 levels are baked into the database layout (partition
# directory names + per-shot index column). Changing them
# silently on resume produces an inconsistent DB; refuse to
# start, matching the gedi_version pattern above. Only fires
# when the user explicitly passed -h3r/-h3p (argparse defaults
# are None so a naked resume on a non-default DB is safe).
if res is not None and self.res is not None and res != self.res:
raise GediValidationError(
f"H3 resolution mismatch: existing database has h3_resolution_level={self.res}, "
f"but -h3r {res} was requested. The H3 index level is baked into the partition "
f"layout — different levels require separate databases. Drop -h3r to resume against "
f"the existing level, or build a fresh database in a different output directory."
)
if part is not None and self.part is not None and part != self.part:
raise GediValidationError(
f"H3 partition mismatch: existing database has h3_partition_level={self.part}, "
f"but -h3p {part} was requested. The H3 partition level is baked into the directory "
f"layout — different levels require separate databases. Drop -h3p to resume against "
f"the existing level, or build a fresh database in a different output directory."
)
self.updating = True
self.new_spatial = None
self.new_temporal = None
self.new_product_vars = None
prev_temporal, prev_spatial = self.temporal, self.spatial
if temporal is not None:
self.temporal, self.new_temporal = merge_temporal(self.temporal, temporal)
if spatial is not None:
self.spatial, self.new_spatial = merge_spatial(self.spatial, spatial)
if product_vars is not None:
self.product_vars, self.new_product_vars = merge_product_vars(self.product_vars, product_vars)
# The merges report a symmetric difference, so a subset or identical
# request also yields a ``new_*``; only a union beyond what was stored
# widens the scope.
widened = False
if self.new_spatial is not None and prev_spatial is not None:
try:
added = self.spatial.union_all().difference(prev_spatial.union_all())
widened = added.area > 0
except Exception:
added, widened = None, True # cannot tell: reopen, keep the symmetric difference (a superset)
if not widened:
# A narrower or identical region adds nothing (a database's
# scope never shrinks): this is a plain update over the stored
# region. Left set, the symmetric difference made it a
# "spatial-only expansion" over stored-minus-requested, and a
# granule new to the database lost its rows outside that (#42).
self.new_spatial = None
elif added is not None:
# Only the area the request adds: the symmetric difference also
# holds the stored region outside the request (all of it, for a
# disjoint request), which Stage 1 then re-read for nothing.
import geopandas as gpd
self.new_spatial = gpd.GeoDataFrame(geometry=[added], crs=self.spatial.crs)
elif self.new_spatial is not None:
widened = True
if self.new_temporal is not None and not widened:
widened = (tuple(self.temporal or ()) != tuple(prev_temporal or ()))
# NO_DATA holds for the scope it was recorded under. Widening space or
# time (also in a mixed update) makes every such granule unproven; do it
# here so the first save_log of this run persists it, whatever happens
# to Stage 1. A variable-only update runs no Stage 1: zero rows stay so.
if widened:
n_reopen = 0
for g in getattr(self, 'granule_info', None) or []:
if g.get('status') == GRANULE_STATUS_NO_DATA:
g['status'] = 'PENDING'
n_reopen += 1
if n_reopen:
from .logging_config import get_logger
get_logger(__name__).info(
f"{n_reopen} NO_DATA granule(s) reopened: the requested region/dates widen the "
f"scope they were recorded under")
# What this run itself asks to add, before an interrupted run's pending
# update is merged in: gh3_build must not take its merge-only shortcut
# over a widening nothing has read yet.
self.requested_expansion = self.new_spatial is not None or self.new_temporal is not None
self._restore_pending_update()
def _restore_pending_update(self):
"""Resume an update that a previous run started but never completed (#42, #52).
The first ``save_log`` of an expansion already stores the widened
region/dates, so a plain resume would see no ``new_spatial`` /
``new_temporal``: Stage 1 would read the whole scope while skipping
every INDEXED granule, and their rows in the added area would never be
extracted. ``to_dict`` persists the added area as ``_pending_update``
until Stage 1 has run under it; this puts it back into ``new_spatial`` /
``new_temporal``, so every getter answers as it did in the interrupted
run (and Stage 1's scope fingerprint matches its sentinels). An
expansion requested by this run is combined with it: reading a superset
of the added area costs a re-read, never rows.
The same first save stores the new products/variables in the log's
``products``, so a resume would also see no ``new_product_vars``: a
mixed update killed in Phase 1 (or a variable update killed before
Stage 2 started) never ran Stage 2, and the granules already indexed
never got the new columns (#52). They are restored the same way, so the
resume is again a mixed (or variable-only) update.
"""
pending = self.log_data.get('_pending_update')
if not pending:
return
product_vars = pending.get('product_vars')
if product_vars:
self.product_vars = merge_product_vars(self.product_vars or {}, product_vars)[0]
self.new_product_vars = (product_vars if self.new_product_vars is None
else merge_product_vars(self.new_product_vars, product_vars)[0])
spatial = parse_spatial(pending.get('spatial'))
if spatial is not None:
self.new_spatial = (spatial if self.new_spatial is None
else merge_spatial(self.new_spatial, spatial)[0])
temporal = parse_temporal(pending.get('temporal'))
if temporal is not None:
self.new_temporal = (temporal if self.new_temporal is None
else merge_temporal(self.new_temporal, temporal)[0])
from .logging_config import get_logger
get_logger(__name__).debug(
"Build log holds an interrupted update: its added area and variables are processed again")
def _load_filters_from_log(self):
self.product_vars = self.log_data.get('products', {})
self.product_vars = {k:val.get('variables') for k,val in self.product_vars.items()}
self.spatial = parse_spatial(self.log_data.get('spatial_filter'))
self.temporal = parse_temporal(self.log_data.get('temporal_filter'))
self.res = self.log_data.get('h3_resolution_level')
self.part = self.log_data.get('h3_partition_level')
self.gedi_version = self.log_data.get('gedi_version')
if 'granules' in self.log_data:
self.granule_info = self.log_data.get('granules')
# Lazy upgrade: legacy entries lacking the `products` map get one
# derived from their top-level status, matching pre-doctor semantics
# (granule INDEXED ⇒ all then-listed products were INDEXED). The
# on-disk file is only rewritten on the next save_log() call.
active_products = list(self.product_vars.keys())
for g in self.granule_info:
if 'products' not in g and active_products and g.get('status') != GRANULE_STATUS_NO_DATA:
existing_status = g.get('status', PRODUCT_STATUS_INDEXED)
g['products'] = {p: existing_status for p in active_products}
if 'h3_columns' in self.log_data:
self.h3_columns = self.log_data.get('h3_columns')
if 'h3_columns_dtypes' in self.log_data:
self.h3_columns_dtypes = dict(self.log_data.get('h3_columns_dtypes') or {})
if 'h3_partition_ids' in self.log_data:
self.h3_partition_ids = self.log_data.get('h3_partition_ids')
if 'date_range' in self.log_data:
self.date_range = self.log_data.get('date_range')
[docs]
def get_temporal(self):
if not self.updating or self.new_temporal is None:
return self.temporal
if self.new_spatial is None and self.new_product_vars is None:
return self.new_temporal
return self.temporal
[docs]
def get_spatial(self):
if not self.updating or self.new_spatial is None:
return self.spatial
if self.new_product_vars is None and self.new_temporal is None:
return self.new_spatial
return self.spatial
[docs]
def get_product_vars(self):
if not self.updating or self.new_product_vars is None:
return self.product_vars
if self.new_spatial is None and self.new_temporal is None:
return self.new_product_vars
return self.product_vars
def _adding_h3_parts(self):
# No expansion requested → not adding partitions, period. This must
# be checked BEFORE the h3_partition_ids attribute guard: on a resume
# after the first write phase was killed before persisting the
# partition list, `h3_partition_ids` is absent on the loaded log even
# though we are resuming the same spatial plan. Returning True here
# would disable the skip filter in get_finished_granules and cause
# the resume to re-process every already-INDEXED granule.
if self.new_spatial is None:
return False
if not hasattr(self, 'h3_partition_ids'):
# Expansion requested but we have no recorded partition set to
# compare against — treat the entire new set as "added".
return True
new_h3_parts = set(intersect_h3_geometries(self.new_spatial, res=self.part))
existing_parts = set(self.h3_partition_ids)
return not new_h3_parts.issubset(existing_parts)
[docs]
def register_pending_granules(self, granule_list):
"""Register granules as PENDING. Called before build starts.
Parameters
----------
granule_list : list of dict
Each dict must have 'orbit', 'granule', 'track' keys, and may have
``releases``: ``{product: tag}`` (:func:`gedi_release_tag`) of the
source files the build lists for the granule. A newer release
within the GEDI version is detected against the recorded tags
(#47), so a tag is never replaced here: a granule not indexed yet
gains tags only for products it has none for (rows a killed run
left on disk may come from the file recorded first, and an older
tag is reported later rather than a newer one hiding stale rows).
An ``INDEXED`` or ``NO_DATA`` entry, and one registered before
tags were recorded (no ``releases`` key, so rows on disk may come
from any file), are left as they are. No tag means the source is
unknown: an older log, a product added by a variable update, or
an S3-sourced build (which registers no granules).
"""
if not hasattr(self, 'granule_info'):
self.granule_info = []
index = {(g['orbit'], g['granule'], g['track']): g for g in self.granule_info}
for g in granule_list:
key = (g['orbit'], g['granule'], g['track'])
entry = index.get(key)
if entry is None:
index[key] = {**g, 'status': 'PENDING'}
if 'releases' in g:
index[key]['releases'] = dict(g['releases'])
self.granule_info.append(index[key])
elif (g.get('releases') and 'releases' in entry
and entry.get('status') not in ('INDEXED', GRANULE_STATUS_NO_DATA, None)):
for product, tag in g['releases'].items():
entry['releases'].setdefault(product, tag)
[docs]
def update_granule_status(self, gran_key, status):
"""Update status of a single granule (does NOT auto-save).
Caller controls save frequency via explicit save_log() calls.
Parameters
----------
gran_key : dict
Dict with 'orbit', 'granule', 'track' keys.
status : str
'INDEXED' or 'PENDING'.
"""
if not hasattr(self, 'granule_info'):
return
key = (gran_key['orbit'], gran_key['granule'], gran_key['track'])
for g in self.granule_info:
if (g['orbit'], g['granule'], g['track']) == key:
g['status'] = status
break
[docs]
def get_finished_granules(self):
"""Return skip list when resuming with same filters.
Only returns successfully completed granules, with the status and
products fields stripped for backward compatibility with
``_filter_granules()`` dict comparison (which expects bare
``{'orbit', 'granule', 'track'}`` dicts). Legacy logs without status
fields are treated as successful.
``NO_DATA`` granules (read in full, zero rows) are skipped under
exactly the conditions ``INDEXED`` ones are. They never stay skipped
past a scope change that could make them non-empty: ``__init__``
demotes every ``NO_DATA`` granule to ``PENDING`` as soon as the
requested region or dates widen the stored ones, by any area, before
this is consulted.
"""
if (hasattr(self, 'granule_info')
and getattr(self, 'new_product_vars', None) is None
and getattr(self, 'new_temporal', None) is None
and not self._adding_h3_parts()):
return self._finished_granule_list()
return None
def _finished_granule_list(self):
# Keys not part of the original granule identity (``products``: the
# per-product statuses; ``releases``: the source files). Callers get
# bare {'orbit', 'granule', 'track'} dicts, the form legacy logs hold.
_strip = {'status', 'products', 'releases'}
return [
{k: v for k, v in g.items() if k not in _strip}
for g in getattr(self, 'granule_info', None) or []
if g.get('status') in ('INDEXED', GRANULE_STATUS_NO_DATA, None)
]
[docs]
def reads_added_area_only(self):
"""True when Stage 1 reads only the region this update adds (a spatial-only expansion)."""
return (getattr(self, 'updating', False)
and getattr(self, 'new_spatial', None) is not None
and getattr(self, 'new_product_vars', None) is None
and getattr(self, 'new_temporal', None) is None)
[docs]
def stage1_only_granules(self):
"""Granules Stage 1 may read in this run, or None when it may read any.
A spatial-only expansion hands Stage 1 only the added region
(:meth:`get_spatial`). That is exactly what a granule finished under
the stored region still needs. Any other granule (new to the database,
``PENDING``, ``FAILED``, a ``NO_DATA`` reopened by the widening) would
have its rows outside the added region dropped, while its rows inside
made it ``INDEXED`` and kept later builds from ever reading the rest
(#42). Such granules are left as they are for the next build, which
reads them over the whole region.
Returns
-------
list of dict or None
``{'orbit', 'granule', 'track'}`` dicts, as for
:meth:`get_finished_granules`.
"""
if not self.reads_added_area_only():
return None
return self._finished_granule_list()
[docs]
def is_up_to_date(self):
"""Check if database already matches requested parameters.
Returns True when all of these hold:
- Log exists (updating=True)
- No new products, spatial, or temporal changes detected
- No pending variable update from a previous crash
- All tracked granules are INDEXED (or NO_DATA: read, nothing to index)
"""
if not self.updating:
return False
if getattr(self, 'new_product_vars', None) is not None:
return False
if getattr(self, 'new_spatial', None) is not None:
return False
if getattr(self, 'new_temporal', None) is not None:
return False
if self.log_data.get('_pending_variable_update'):
return False
if hasattr(self, 'granule_info') and self.granule_info:
if any(g.get('status') not in ('INDEXED', GRANULE_STATUS_NO_DATA, None)
for g in self.granule_info):
return False
return True
[docs]
def set_post_build_info(self, verify_observed=False):
"""Fold partition metadata into the log: granule status, columns, dates.
Also records ``self._observed_granule_keys``: the ``(orbit, granule,
track)`` of every granule the partition metadata lists, or ``None``
when that listing cannot be trusted to be complete (no partitions, an
unreadable scan). With ``verify_observed=True`` the scan additionally
checks each partition's sidecars against its parquet files and leaves
``None`` on any mismatch; :meth:`mark_no_data` requires a set.
"""
self._observed_granule_keys = None
scan_trusted = True
# Enumerate partition dirs via os.scandir — replaces the prior
# glob.glob('*/*<meta>') which paid 10k+ GPFS metadata round-trips
# in a single driver thread. We then dispatch per-partition JSON
# reads in parallel via parallel_map so the scan plateaus at
# ~seconds instead of tens of minutes on continental-scale DBs.
try:
partition_dirs = sorted(
e.path for e in os.scandir(self._PARENT_DIR)
if e.is_dir(follow_symlinks=False) and e.name.startswith('h3_')
)
except OSError:
return
if not partition_dirs:
# Listed fine and holds no partition: no granule has rows here.
self._observed_granule_keys = set()
return
# Collect indexed granules from partition metadata
indexed_granules = []
h3_parts = []
date_min = None
date_max = None
# Per-granule observed product set: union of products whose suffix is
# present in the columns of any partition that contains the granule.
# Used to derive per-granule per-product status (INDEXED vs MISSING_COLUMN).
active_products = set(self.product_vars.keys())
gran_observed_products = {}
# Schema accumulators: track the union of columns observed and,
# per column, how many partitions store each dtype. Aggregating inside
# the loop (rather than reading the last fmeta) survives mixed
# builds where some partitions predate ``column_dtypes`` — the
# last metadata in glob order may be a legacy partition without
# the field, and reading from it would silently zero the cache
# for the whole DB.
observed_columns: set = set()
observed_dtypes: dict = {} # {column: {dtype: n_partitions}}
# Stream per-partition results. parallel_map yields (item, result)
# pairs as workers finish; we fold them into the aggregates with
# identical semantics to the legacy serial loop.
def _fold(part_result):
nonlocal date_min, date_max
if part_result.get('date_min') is not None:
if date_min is None or part_result['date_min'] < date_min:
date_min = part_result['date_min']
if part_result.get('date_max') is not None:
if date_max is None or part_result['date_max'] > date_max:
date_max = part_result['date_max']
h3_parts.extend(part_result.get('h3_partitions', []))
partition_products = part_result.get('partition_products') or set()
seen_keys = set()
for g in part_result.get('granules', []):
try:
key = (g['orbit'], g['granule'], g['track'])
except (KeyError, TypeError):
continue
if key not in seen_keys:
indexed_granules.append(g)
seen_keys.add(key)
gran_observed_products.setdefault(key, set()).update(partition_products)
if self.gedi_version is None and part_result.get('l2a_version') is not None:
self.gedi_version = part_result['l2a_version']
observed_columns.update(part_result.get('columns', []))
for col_name, dtype in (part_result.get('column_dtypes') or {}).items():
counts = observed_dtypes.setdefault(col_name, {})
counts[dtype] = counts.get(dtype, 0) + 1
from .utils import get_dask_client
client = None
try:
client = get_dask_client()
except Exception:
pass
if client is not None and len(partition_dirs) > 100:
from .parallel import parallel_map
for _, res in parallel_map(
partition_dirs,
_scan_partition_meta_post_build_info,
desc='Scanning partition metadata',
unit='part',
meta_filename=PARTITION_META_FILENAME,
active_products=active_products,
verify=verify_observed,
):
if isinstance(res, Exception):
scan_trusted = False
continue
if res.get('unverified'):
scan_trusted = False
_fold(res)
else:
for pd in partition_dirs:
try:
_res = _scan_partition_meta_post_build_info(
pd,
meta_filename=PARTITION_META_FILENAME,
active_products=active_products,
verify=verify_observed,
)
except Exception:
scan_trusted = False
continue
if _res.get('unverified'):
scan_trusted = False
_fold(_res)
# Deduplicate against any granules that may already have appeared
# in a prior in-memory accumulation — the per-partition seen_keys
# only dedupes within one partition's result.
_seen = set()
_deduped = []
for g in indexed_granules:
try:
k = (g['orbit'], g['granule'], g['track'])
except (KeyError, TypeError):
continue
if k in _seen:
continue
_seen.add(k)
_deduped.append(g)
indexed_granules = _deduped
if scan_trusted:
self._observed_granule_keys = {(g['orbit'], g['granule'], g['track']) for g in indexed_granules}
self.date_range = (date_min, date_max)
self.h3_columns = sorted(observed_columns)
# Empty dict when every partition metadata predates this field.
# A column stored in several dtypes records the one most partitions
# hold (never "whichever partition was scanned first"): updates align
# new data to this, so it must describe the bulk of the database.
self.h3_columns_dtypes = {c: max(n, key=n.get) for c, n in observed_dtypes.items()}
mixed = {c: n for c, n in observed_dtypes.items() if len(n) > 1}
if mixed:
from .logging_config import get_logger
desc = '; '.join(f"{c}: {n}" for c, n in sorted(mixed.items()))
get_logger(__name__).warning(
f"{len(mixed)} column(s) are stored in more than one dtype across partitions "
f"(partitions per dtype): {desc}. Recording the majority dtype. "
f"Run gh3_doctor --check dtype_drift to compare against the source."
)
self.h3_partition_ids = sorted(h3_parts)
# Merge with existing tracked granules (preserve PENDING for unindexed)
indexed_keys = {(g['orbit'], g['granule'], g['track']) for g in indexed_granules}
if hasattr(self, 'granule_info') and self.granule_info:
for g in self.granule_info:
key = (g['orbit'], g['granule'], g['track'])
# MERGE_FAILED stays: the granule is in the metadata of the
# partitions that merged, not of the one that lost its rows.
if key in indexed_keys and g.get('status') != 'MERGE_FAILED':
g['status'] = 'INDEXED'
products = _per_product_status_from_observed(
active_products, gran_observed_products.get(key, set())
)
# MISSING_SOURCE and a backfill's FAILED stay: the partition
# carries the product's columns (null for this granule)
# until a backfill fills them.
for p, s in (g.get('products') or {}).items():
if s in (PRODUCT_STATUS_MISSING_SOURCE, PRODUCT_STATUS_FAILED) and p in products:
products[p] = s
g['products'] = products
# else: keep existing status (PENDING)
# Add any newly discovered granules not previously tracked
existing_keys = {(g['orbit'], g['granule'], g['track']) for g in self.granule_info}
for g in indexed_granules:
key = (g['orbit'], g['granule'], g['track'])
if key not in existing_keys:
self.granule_info.append({
**g,
'status': 'INDEXED',
'products': _per_product_status_from_observed(
active_products, gran_observed_products.get(key, set())
),
})
else:
self.granule_info = [
{
**g,
'status': 'INDEXED',
'products': _per_product_status_from_observed(
active_products,
gran_observed_products.get((g['orbit'], g['granule'], g['track']), set()),
),
}
for g in indexed_granules
]
self.granule_info = sorted(self.granule_info, key=lambda g: (g.get('orbit', 0), g.get('granule', 0), g.get('track', 0)))
[docs]
def stage1_full_scope(self):
"""True when Stage 1 reads under the whole database scope.
``get_spatial`` / ``get_temporal`` hand Stage 1 only the ADDED area (or
dates) on a spatial-only (temporal-only) expansion, so a granule empty
under that diff may hold rows in the old scope. Zero rows then proves
nothing, and :meth:`mark_no_data` refuses. Full scope: a fresh build,
a plain resume or update (the log already holds the union), and a mixed
update's Phase 1. Mirrors the branches of the two getters.
"""
if not getattr(self, 'updating', False):
return True
no_prod = getattr(self, 'new_product_vars', None) is None
temporal_only = (getattr(self, 'new_temporal', None) is not None and no_prod
and getattr(self, 'new_spatial', None) is None)
return not (self.reads_added_area_only() or temporal_only)
[docs]
def mark_no_data(self, listed_keys, incomplete_keys, empty_keys):
"""Record granules Stage 1 read in full that hold no rows as ``NO_DATA``.
Call after the final :meth:`set_post_build_info` of a build whose
Stage 1 drained fully; ``incomplete_keys`` is that run's set of
granules with any task not proven complete, ``empty_keys`` those whose
every beam task is proven complete AND empty. Nothing is marked
without a trustworthy observed set (``set_post_build_info`` could not
verify the partition metadata) or when Stage 1 did not run under the
full database scope (:meth:`stage1_full_scope`). In doubt a granule
stays PENDING: that costs a re-read, a wrong ``NO_DATA`` costs rows.
* ``PENDING`` -> ``NO_DATA`` when listed, in ``empty_keys`` and absent
from every partition's metadata (the last guard matters for tasks
emptied by the skip check, whose rows are already in the database);
* ``NO_DATA`` -> ``PENDING`` when listed and incomplete this run;
* ``MERGE_FAILED`` is never touched.
``NO_DATA`` drops the ``products`` map and ``fill_attempts``: there
is nothing to fill. Does not save. Returns ``{'marked', 'reopened'}``.
Parameters
----------
listed_keys : iterable of tuple
Granules this run listed for Stage 1.
incomplete_keys : iterable of tuple
Granules with any task not proven complete this run.
empty_keys : iterable of tuple
Granules whose every beam task is proven complete and empty.
"""
out = {'marked': 0, 'reopened': 0}
observed = getattr(self, '_observed_granule_keys', None)
if observed is None or listed_keys is None or incomplete_keys is None or empty_keys is None:
return out
listed = set(listed_keys)
incomplete = set(incomplete_keys)
empty = set(empty_keys)
may_mark = self.stage1_full_scope()
for g in getattr(self, 'granule_info', None) or []:
key = (g['orbit'], g['granule'], g['track'])
if key not in listed:
continue
status = g.get('status')
if status == 'MERGE_FAILED':
continue
if key in incomplete:
if status == GRANULE_STATUS_NO_DATA:
g['status'] = 'PENDING'
out['reopened'] += 1
continue
if may_mark and status == 'PENDING' and key in empty and key not in observed:
g['status'] = GRANULE_STATUS_NO_DATA
g.pop('products', None)
g.pop('fill_attempts', None)
out['marked'] += 1
return out
[docs]
def to_dict(self, status):
if status not in _VALID_STATUSES:
raise ValueError(f"Invalid status '{status}'. Must be one of {_VALID_STATUSES}")
product_logs = self.log_data.get('products', {})
for prod in self.get_product_vars().keys():
product_logs[prod] = product_logs.get(prod, {})
product_logs[prod]['status'] = status
product_logs[prod]['last_modified'] = now()
vars_list = self.product_vars.get(prod)
product_logs[prod]['variables'] = sorted(vars_list) if vars_list else vars_list
# Calculate build duration
build_duration = None
if self.build_start_time:
build_duration = (datetime.now(timezone.utc) - self.build_start_time).total_seconds()
log_dict = {
'metadata': {
'package_version': get_package_version()
},
'gedi_version': self.gedi_version,
'h3_resolution_level': self.res,
'h3_partition_level': self.part,
'status': status,
'previous_status': self.previous_status,
'source_mode': self.source_mode,
'last_modified': now(),
'build_duration_seconds': build_duration,
'spatial_filter': None if self.spatial is None else to_geojson(self.spatial),
'temporal_filter': self.temporal,
'products': product_logs
}
if hasattr(self, 'granule_info'):
log_dict['granules'] = self.granule_info
if hasattr(self, 'h3_columns'):
log_dict['h3_columns'] = self.h3_columns
if getattr(self, 'h3_columns_dtypes', None):
log_dict['h3_columns_dtypes'] = self.h3_columns_dtypes
if hasattr(self, 'h3_partition_ids'):
log_dict['h3_partition_ids'] = self.h3_partition_ids
if getattr(self, 'allow_missing_products', False):
log_dict['allow_missing_products'] = True
if getattr(self, 'exclude', None):
log_dict['exclude'] = list(self.exclude)
if hasattr(self, 'date_range'):
log_dict['date_range'] = self.date_range
# Append to update history on terminal statuses
if status in ('COMPLETED', 'FAILED', 'INTERRUPTED'):
action = getattr(self, 'update_action', None) or (
'variable_update' if getattr(self, 'new_product_vars', None) and getattr(self, 'new_spatial', None) is None and getattr(self, 'new_temporal', None) is None else 'build')
update_entry = {
'timestamp': now(),
'action': action,
'status': status,
'duration_seconds': build_duration,
'new_products': sorted(self.new_product_vars.keys()) if getattr(self, 'new_product_vars', None) else None,
}
existing_history = self.log_data.get('update_history', [])
existing_history.append(update_entry)
log_dict['update_history'] = existing_history[-50:] # Keep last 50
# Preserve pending variable update state for crash recovery (P0-B)
if '_pending_variable_update' in self.log_data:
log_dict['_pending_variable_update'] = self.log_data['_pending_variable_update']
# A granule replacement under way (#47): the tmp dir holding its journal,
# so any later run finds it whatever its --tmpdir.
if '_pending_replacement' in self.log_data:
log_dict['_pending_replacement'] = self.log_data['_pending_replacement']
# Pending until the run's stages have run, or until Stage 2 starts: its
# own _pending_variable_update then describes all the remaining work.
# See _restore_pending_update.
new_spatial = getattr(self, 'new_spatial', None)
new_temporal = getattr(self, 'new_temporal', None)
new_product_vars = getattr(self, 'new_product_vars', None)
if (status != 'COMPLETED' and '_pending_variable_update' not in log_dict
and not getattr(self, 'update_ran', False)
and (new_spatial is not None or new_temporal is not None or new_product_vars)):
log_dict['_pending_update'] = {
'spatial': None if new_spatial is None else to_geojson(new_spatial),
'temporal': None if new_temporal is None else list(new_temporal),
'product_vars': new_product_vars or None,
}
return log_dict
[docs]
def save_log(self, status):
json_write(self.to_dict(status), self.log_file, mode='w', rewrite=True)
[docs]
def get_granule_product_status(self, gran_key, product):
"""Return per-product status for a granule, or None if unknown.
Parameters
----------
gran_key : dict or tuple
Either ``{'orbit', 'granule', 'track'}`` or ``(orbit, granule, track)``.
product : str
Product code (e.g. 'L4A').
"""
if not hasattr(self, 'granule_info'):
return None
if isinstance(gran_key, dict):
key = (gran_key['orbit'], gran_key['granule'], gran_key['track'])
else:
key = tuple(gran_key)
for g in self.granule_info:
if (g['orbit'], g['granule'], g['track']) == key:
return (g.get('products') or {}).get(product)
return None
[docs]
def get_product_gaps(self, active_products=None, gap_statuses=None):
"""List granules where any active product is not INDEXED.
Returns a list of ``(gran_key_dict, [missing_products])`` pairs.
Used by gh3_doctor as the fast-path audit (log-only, no parquet scan).
Parameters
----------
active_products : iterable of str, optional
Products to check. Defaults to ``self.product_vars.keys()``.
gap_statuses : iterable of str, optional
Per-product statuses considered as gaps. Defaults to everything except
``INDEXED`` (i.e. PARTIAL_NAN, MISSING_COLUMN, MISSING_SOURCE, FAILED, PENDING).
"""
if not hasattr(self, 'granule_info'):
return []
if active_products is None:
active_products = list(self.product_vars.keys())
active_products = list(active_products)
if gap_statuses is None:
gap_statuses = {
PRODUCT_STATUS_PARTIAL_NAN,
PRODUCT_STATUS_MISSING_COLUMN,
PRODUCT_STATUS_MISSING_SOURCE,
PRODUCT_STATUS_FAILED,
PRODUCT_STATUS_PENDING,
}
else:
gap_statuses = set(gap_statuses)
gaps = []
for g in self.granule_info:
if g.get('status') == GRANULE_STATUS_NO_DATA:
continue # no rows, so no product columns to fill
products_map = g.get('products') or {}
missing = []
for p in active_products:
# Unknown for this entry → treat as PENDING so the doctor
# inspects rather than silently skipping.
status = products_map.get(p, PRODUCT_STATUS_PENDING)
if status in gap_statuses:
missing.append(p)
if missing:
gaps.append(({'orbit': g['orbit'], 'granule': g['granule'], 'track': g['track']}, missing))
return gaps
[docs]
def mark_granule_product(self, gran_key, product, status):
"""Set per-product status for a granule (does not auto-save).
Caller is responsible for invoking ``save_log()`` to persist.
Used by gh3_doctor after a fix or verification step.
"""
if status not in _VALID_PRODUCT_STATUSES:
raise GediValidationError(
f"Invalid per-product status '{status}'. Must be one of {_VALID_PRODUCT_STATUSES}"
)
if not hasattr(self, 'granule_info'):
return False
if isinstance(gran_key, dict):
key = (gran_key['orbit'], gran_key['granule'], gran_key['track'])
else:
key = tuple(gran_key)
for g in self.granule_info:
if (g['orbit'], g['granule'], g['track']) == key:
products_map = g.get('products')
if products_map is None:
products_map = {}
g['products'] = products_map
products_map[product] = status
return True
return False
[docs]
def set_product_statuses(self, updates):
"""Bulk per-product status update (does not auto-save).
``updates`` maps ``(orbit, granule, track)`` to ``{product: status}``.
One pass over ``granule_info`` whatever the number of updates, unlike
repeated :meth:`mark_granule_product` calls. Returns the count applied.
"""
for statuses in updates.values():
bad = set(statuses.values()) - set(_VALID_PRODUCT_STATUSES)
if bad:
raise GediValidationError(f"Invalid per-product status {sorted(bad)}")
n = 0
for g in getattr(self, 'granule_info', None) or []:
statuses = updates.get((g['orbit'], g['granule'], g['track']))
if statuses:
g.setdefault('products', {}).update(statuses)
n += len(statuses)
return n
[docs]
def superseded_releases(self, soc_entries):
"""Indexed granules whose listed source differs from the release read.
Compares each ``INDEXED`` or ``NO_DATA`` granule's recorded
``releases`` with the file a listing would read now. Log-only, no I/O.
A product whose rows name several releases (``mixed_releases``, see
:meth:`adopt_row_releases`) compares its oldest, so a listing of the
newer one reports it until it is re-indexed whole. A product with no
recorded tag is unknown and never reported. A log
with no GEDI version reports nothing: tags carry no version, so a
file of another release would read as a re-release.
Parameters
----------
soc_entries : iterable of dict
``{product: path}`` dicts: per granule, as ``soc_file_tree(...,
to_list=True)`` returns them, or one per file. Where a granule
product is listed more than once, its latest release
(``gedi_release_rank``) is the one compared, the file a build
reads. Only granules with a recorded tag are ranked.
Returns
-------
dict
``{'newer': {key: {product: (recorded, listed)}}, 'older': {...}}``
keyed by ``(orbit, granule, track)``. ``newer``: a re-release
within the GEDI version that the rows do not come from (#47).
``older``: only an older release than the one read is listed;
it is never applied, but variables added and products filled
later would read it.
"""
index = {(g['orbit'], g['granule'], g['track']): {'releases': self.compared_releases(g)}
for g in getattr(self, 'granule_info', None) or []
if (g.get('releases') or g.get('mixed_releases'))
and g.get('status') in ('INDEXED', GRANULE_STATUS_NO_DATA)}
index = {k: v for k, v in index.items() if v['releases']}
out = {'newer': {}, 'older': {}}
if not index or not getattr(self, 'gedi_version', None):
return out
latest = {} # (key, product) -> listed tag
for entry in soc_entries:
for product, path in entry.items():
path = str(getattr(path, 'path', path))
fl = os.path.basename(path).split('_')
try:
key = (int(fl[3][1:]), int(fl[4]), int(fl[5][1:]))
except (IndexError, ValueError):
continue
if product not in (index.get(key) or {}).get('releases', {}):
continue
listed = gedi_release_tag(path)
best = latest.get((key, product))
if listed is not None and (best is None or gedi_release_rank(listed) > gedi_release_rank(best)):
latest[(key, product)] = listed
for (key, product), listed in sorted(latest.items()):
recorded = index[key]['releases'][product]
if listed != recorded:
side = 'newer' if gedi_release_rank(listed) > gedi_release_rank(recorded) else 'older'
out[side].setdefault(key, {})[product] = (recorded, listed)
return out
[docs]
def pending_product_fills(self, statuses=(PRODUCT_STATUS_MISSING_SOURCE,)):
"""Indexed granules still waiting for products published after them.
``{(orbit, granule, track): [product, ...]}`` for every ``INDEXED``
granule with a product in one of ``statuses``. The default,
``MISSING_SOURCE``, is what ``gh3_build`` backfills automatically;
``gh3_doctor`` also passes ``FAILED`` (fills given up on, see
:meth:`record_fill_results`) to retry them on demand. Log-only, no I/O.
"""
statuses = set(statuses)
out = {}
for g in getattr(self, 'granule_info', None) or []:
if g.get('status') != 'INDEXED':
continue
missing = [p for p, s in (g.get('products') or {}).items() if s in statuses]
if missing:
out[(g['orbit'], g['granule'], g['track'])] = missing
return out
[docs]
def record_releases(self, releases):
"""Record the source releases a write is about to read (does not save).
Parameters
----------
releases : dict
``{(orbit, granule, track): {product: tag}}`` (tags from
``gedi_release_tag``), e.g. from ``_build_fill_products``'
``releases_callback`` before it writes. As in
:meth:`register_pending_granules`, a product that has a tag keeps
it (rows already written may come from that release), and an
entry recorded before tags existed (no ``releases`` key) stays
unknown: its other products' rows may come from any file.
Returns
-------
int
Number of tags added.
"""
n = 0
for g in getattr(self, 'granule_info', None) or []:
tags = releases.get((g['orbit'], g['granule'], g['track']))
if tags and 'releases' in g:
known = g['releases']
for product, tag in tags.items():
if product not in known:
known[product] = tag
n += 1
return n
[docs]
@staticmethod
def compared_releases(granule, newest=False):
"""``{product: tag}`` a granule's rows are compared by: its recorded
``releases``, plus the oldest (``newest=True``: the newest) of the
releases a product's rows name when they name several
(``mixed_releases``). Log-only."""
out = {p: (max if newest else min)(tags, key=gedi_release_rank)
for p, tags in (granule.get('mixed_releases') or {}).items() if tags}
out.update(granule.get('releases') or {})
return out
[docs]
def row_release_targets(self):
"""Indexed granules whose releases were never recorded nor read from their rows (#67).
``INDEXED`` granules with no ``releases`` key: a database built before
gedih3 0.20.10 recorded none, and an entry registered by such a version
before it was indexed keeps none (its rows may come from any file).
Their rows name their source files (``root_file_<prod>``), so
:meth:`adopt_row_releases` takes them from there. Empty when the
database has no source column. Log-only.
Returns
-------
set of tuple
``(orbit, granule, track)`` keys.
"""
columns = set(self.log_data.get('h3_columns') or [])
if not any(f"root_file_{p.lower()}" in columns for p in self.product_vars or {}):
return set()
return {(g['orbit'], g['granule'], g['track']) for g in getattr(self, 'granule_info', None) or []
if g.get('status') == PRODUCT_STATUS_INDEXED and 'releases' not in g}
[docs]
def adopt_row_releases(self, releases, keys, mixed=None):
"""Record the releases the database's own rows were read from (does not save).
Parameters
----------
releases : dict
``{(orbit, granule, track): {product: tag}}`` from the rows'
``root_file_<prod>`` columns (``gh3builder.releases_from_rows``),
for products whose rows name a single release. Unlike
:meth:`record_releases` (the files a write is about to read), these
are what the rows hold.
keys : iterable of tuple
The granules whose rows were all read (:meth:`row_release_targets`
less those an unreadable file may hold). Each gets a ``releases``
entry, empty when its rows named none, so it is not read again; a
product not ``INDEXED`` for it stays out.
mixed : dict, optional
``{key: {product: [tags]}}`` for products whose rows name several
releases (a replacement left half done). Kept apart as the
granule's ``mixed_releases``, never as its release: reported by
``gh3_doctor --check superseded_granules``, compared by its oldest
(:meth:`superseded_releases`), and dropped when it is re-indexed.
Returns
-------
int
Number of tags added, mixed ones included.
"""
keys, n, mixed = set(keys), 0, mixed or {}
for g in getattr(self, 'granule_info', None) or []:
key = (g['orbit'], g['granule'], g['track'])
if key not in keys or g.get('status') != PRODUCT_STATUS_INDEXED:
continue
known = g.setdefault('releases', {})
indexed = [p for p in set(releases.get(key) or {}) | set(mixed.get(key) or {})
if p not in known and p not in (g.get('mixed_releases') or {})
and (g.get('products') or {}).get(p, PRODUCT_STATUS_INDEXED) == PRODUCT_STATUS_INDEXED]
for product in sorted(indexed):
if product in (mixed.get(key) or {}):
g.setdefault('mixed_releases', {})[product] = list(mixed[key][product])
else:
known[product] = releases[key][product]
n += 1
return n
[docs]
def mixed_release_granules(self):
"""``{(orbit, granule, track): {product: [tags]}}``: granules whose rows
name several releases of a product (``mixed_releases``). Log-only."""
return {(g['orbit'], g['granule'], g['track']): dict(g['mixed_releases'])
for g in getattr(self, 'granule_info', None) or []
if g.get('mixed_releases') and g.get('status') in ('INDEXED', GRANULE_STATUS_NO_DATA)}
[docs]
def record_fill_results(self, result, max_attempts=3, count_failures=True):
"""Apply a ``_build_fill_products`` result to per-product statuses.
* ``filled`` → ``INDEXED`` (attempt count cleared);
* ``unlocated`` (the product file exists but no database file lists the
granule, so it can never fill) → ``FAILED`` at once;
* ``failed`` (read or merge error, or no row matched) → retried by the
next runs; after ``max_attempts`` consecutive failures → ``FAILED``,
so one bad file does not make every later build rescan. ``FAILED``
is not backfilled automatically; ``gh3_doctor --fix backfill``
retries it. ``count_failures=False`` leaves failures untouched.
Does not save the log. Returns ``{key: [products]}`` newly ``FAILED``.
"""
index = {(g['orbit'], g['granule'], g['track']): g for g in getattr(self, 'granule_info', None) or []}
given_up = {}
for key, prods in (result.get('filled') or {}).items():
g = index.get(tuple(key))
if g is None:
continue
for p in prods:
g.setdefault('products', {})[p] = PRODUCT_STATUS_INDEXED
(g.get('fill_attempts') or {}).pop(p, None)
if 'fill_attempts' in g and not g['fill_attempts']:
del g['fill_attempts']
for key, prods in (result.get('unlocated') or {}).items():
g = index.get(tuple(key))
if g is None:
continue
for p in prods:
g.setdefault('products', {})[p] = PRODUCT_STATUS_FAILED
given_up.setdefault(tuple(key), []).append(p)
if count_failures:
for key, prods in (result.get('failed') or {}).items():
g = index.get(tuple(key))
if g is None:
continue
attempts = g.setdefault('fill_attempts', {})
for p in prods:
attempts[p] = attempts.get(p, 0) + 1
if attempts[p] >= max_attempts:
g.setdefault('products', {})[p] = PRODUCT_STATUS_FAILED
given_up.setdefault(tuple(key), []).append(p)
return given_up
