#! python
# 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
"""
gedih3 Exception Hierarchy
Custom exceptions for structured error handling throughout the package.
All exceptions inherit from GediError for easy catching of package-specific errors.
Usage:
from gedih3.exceptions import GediDownloadError, GediValidationError
try:
download_granule(granule, odir)
except GediDownloadError as e:
logger.error(f"Download failed: {e}")
"""
[docs]
class GediError(Exception):
"""Base exception for all gedih3 errors."""
pass
# =============================================================================
# Network/Download Errors
# =============================================================================
[docs]
class GediNetworkError(GediError):
"""Base exception for network-related errors."""
pass
[docs]
class GediDownloadError(GediNetworkError):
"""Error during file download from NASA DAAC."""
def __init__(self, message: str, granule_id: str = None, attempts: int = None):
self.granule_id = granule_id
self.attempts = attempts
super().__init__(message)
[docs]
class GediAuthenticationError(GediNetworkError):
"""Error during NASA Earthdata authentication."""
pass
[docs]
class GediS3AccessError(GediNetworkError):
"""Error accessing GEDI data via S3."""
pass
[docs]
class GediIncompleteListingError(GediNetworkError):
"""CMR returned fewer granules than its own CMR-Hits count (silent truncation)."""
def __init__(self, message: str, expected: int = None, received: int = None):
self.expected = expected
self.received = received
super().__init__(message)
# =============================================================================
# Validation Errors
# =============================================================================
[docs]
class GediValidationError(GediError):
"""Base exception for validation errors."""
pass
[docs]
class H3ValidationError(GediValidationError):
"""Error in H3 parameter validation."""
def __init__(self, message: str, param_name: str = None, value = None):
self.param_name = param_name
self.value = value
super().__init__(message)
[docs]
class EGIValidationError(GediValidationError):
"""Error in EGI parameter validation."""
def __init__(self, message: str, param_name: str = None, value = None):
self.param_name = param_name
self.value = value
super().__init__(message)
[docs]
class GediProductError(GediValidationError):
"""Error with GEDI product specification."""
pass
[docs]
class GediVariableError(GediValidationError):
"""Error with GEDI variable specification."""
pass
# =============================================================================
# File/IO Errors
# =============================================================================
[docs]
class GediFileError(GediError):
"""Base exception for file-related errors."""
pass
[docs]
class GediHDF5Error(GediFileError):
"""Error reading/processing HDF5 file."""
def __init__(self, message: str, file_path: str = None):
self.file_path = file_path
super().__init__(message)
[docs]
class GediParquetError(GediFileError):
"""Error reading/writing Parquet file."""
def __init__(self, message: str, file_path: str = None):
self.file_path = file_path
super().__init__(message)
[docs]
class GediCorruptedFileError(GediFileError):
"""File is corrupted or unreadable."""
def __init__(self, message: str, file_path: str = None):
self.file_path = file_path
super().__init__(message)
[docs]
class GediTransactionError(GediFileError):
"""Error during atomic file operation."""
def __init__(self, message: str, source_path: str = None, dest_path: str = None):
self.source_path = source_path
self.dest_path = dest_path
super().__init__(message)
# =============================================================================
# Database Errors
# =============================================================================
[docs]
class GediDatabaseError(GediError):
"""Base exception for H3 database errors."""
pass
[docs]
class GediDatabaseNotFoundError(GediDatabaseError):
"""H3 database directory not found."""
pass
[docs]
class GediDatabaseCorruptedError(GediDatabaseError):
"""H3 database is corrupted or inconsistent."""
pass
[docs]
class GediMergeError(GediDatabaseError):
"""Error merging H3 partitions or databases."""
pass
# =============================================================================
# Spatial/Temporal Errors
# =============================================================================
[docs]
class GediSpatialError(GediError):
"""Error with spatial operations or filters."""
pass
[docs]
class GediTemporalError(GediError):
"""Error with temporal operations or filters."""
pass
# =============================================================================
# Processing Errors
# =============================================================================
[docs]
class GediProcessingError(GediError):
"""Base exception for data processing errors."""
pass
[docs]
class GediAggregationError(GediProcessingError):
"""Error during data aggregation."""
pass
[docs]
class GediRasterizationError(GediProcessingError):
"""Error during rasterization."""
pass
[docs]
class GediImageSamplingError(GediProcessingError):
"""Error during raster image sampling at GEDI shot locations."""
pass
[docs]
class GediSpatialJoinError(GediProcessingError):
"""Error during spatial join of vector data to GEDI shot locations."""
pass
# =============================================================================
# Retry Configuration
# =============================================================================
# Default retry settings for network operations
RETRY_DEFAULTS = {
'max_attempts': 3,
'initial_wait': 1.0, # seconds
'max_wait': 60.0, # seconds
'exponential_base': 2.0,
'jitter': True,
}
[docs]
def is_retryable_error(exception: Exception) -> bool:
"""
Determine if an exception is retryable.
Returns True for transient network errors that may succeed on retry.
"""
import socket
import urllib.error
# `requests.exceptions.Timeout` does NOT inherit from builtin
# ``TimeoutError``; it lives under ``requests.exceptions.RequestException``
# → ``IOError``. List it explicitly so injected per-request timeouts
# (see gedih3.daac._install_request_timeouts) reliably trigger the
# download retry loop instead of falling back to string-pattern matching.
try:
from requests.exceptions import (
Timeout as _RequestsTimeout,
ConnectionError as _RequestsConnectionError,
)
_requests_retryable = (_RequestsTimeout, _RequestsConnectionError)
except ImportError:
_requests_retryable = ()
# Network-level errors
retryable_types = (
ConnectionError,
TimeoutError,
socket.timeout,
socket.gaierror,
urllib.error.URLError,
) + _requests_retryable
if isinstance(exception, retryable_types):
return True
# Check for HTTP status codes in exception message
error_msg = str(exception).lower()
retryable_patterns = [
'500', '502', '503', '504', # Server errors
'timeout', 'timed out',
'connection reset',
'connection refused',
'temporary failure',
'service unavailable',
]
return any(pattern in error_msg for pattern in retryable_patterns)
