Module pyl4c.apps.calibration
Essential functions and classes for calibration. See, in particular:
OPT_BOUNDS
, which has lower and upper bounds on calibration parameters
You must create a configuration JSON file before calibrating L4C. There is a template available in the directory:
pyl4c/data/fixtures/files
The optimization routines here are best used with multiple --trials
, i.e.,
multiple, random initial parameter values. This can help avoiding falling into
a local optimum when a better solution set is available. However, if you are
repeatedly seeing the message:
Error in objective function; restarting...
Most likely, the bounds on your parameters are producing initial parameter values that are totally unreasonable.
Note that calibration of L4C (i.e., updating the BPLUT) can be performed using
the command-line interface in pyl4c.apps.calibration.main
, for example:
# Build the scratch dataset needed for calibration
python main.py setup
# You can get a preview of what filtering the data would look like...
python main.py pft <pft_number> filter-preview gpp <window_size>
python main.py pft <pft_number> filter-preview reco <window_size>
# Optionally, filter the tower datasets to remove spurious spikes
python main.py filter-all gpp <window_size>
python main.py filter-all reco <window_size>
# Run the GPP calibration for a given Plant Functional Type (PFT)
python main.py pft <pft_number> tune-gpp
# Run the RECO calibration for a given Plant Functional Type (PFT)
python main.py pft <pft_number> tune-reco
# Finally, to dump the updated BPLUT into a CSV or Python pickle file...
python main.py bplut pickle <output_path> --version-id=<version_id>
# Get help on any command with --help, for example:
python main.py plot-gpp --help
Sub-modules
pyl4c.apps.calibration.mcmc
-
Calibration of L4C using Markov Chain Monte Carlo (MCMC). Example use: …
pyl4c.apps.calibration.optimize
-
To calibrate the GPP or RECO model for a specific PFT: …
Functions
def cbar(rh, k_mult, q_rh=75, q_k=50)
-
Expand source code
def cbar(rh, k_mult, q_rh = 75, q_k = 50):
cutoff = np.apply_along_axis( np.percentile, 0, k_mult, q = q_k).reshape((1, k_mult.shape[1])) return np.nanpercentile( np.where(k_mult >= cutoff, np.divide(rh, np.where(k_mult == 0, np.nan, k_mult)), np.nan), q = q_rh, axis = 0)"""
Calculates "Cbar," the time-constant upper quantile of the RH/Kmult ratio. Where Kmult is >/= `q_k`, return the `q_rh` quantile of RH/Kmult; intended for T x N arrays where T is the number of time steps and N is the number of (flux tower) sites. Parameters ---------- rh : numpy.ndarray (T x N) vector of heterotrophic respiration k_mult : numpy.ndarray (T x N) vector of Kmult q_rh : float Percentile of RH/Kmult to return q_k : float Percentile of Kmult below which RH/Kmult values are masked Returns ------- numpy.float64 '''Calculates "Cbar," the time-constant upper quantile of the RH/Kmult ratio. Where Kmult is >/=
q_k
, return theq_rh
quantile of RH/Kmult; intended for T x N arrays where T is the number of time steps and N is the number of (flux tower) sites.Parameters
rh
:numpy.ndarray
- (T x N) vector of heterotrophic respiration
k_mult
:numpy.ndarray
- (T x N) vector of Kmult
q_rh
:float
- Percentile of RH/Kmult to return
q_k
:float
- Percentile of Kmult below which RH/Kmult values are masked
Returns
numpy.float64
def reco(params, tsoil, smsf, reco_tower, gpp_tower, q_rh=75, q_k=50)
-
Expand source code
def reco(params, tsoil, smsf, reco_tower, gpp_tower, q_rh = 75, q_k = 50):
# Calculate RH as (RECO - RA) or (RECO - (faut * GPP)); # globals "reco_tower", "gpp_tower" ra = ((1 - params['CUE']) * gpp_tower) rh = reco_tower - ra rh = np.where(rh < 0, 0, rh) # Mask out negative RH values # Compute Cbar with globals "q_rh" and "q_k" kmult0 = k_mult(params, tsoil, smsf) cbar0 = cbar(rh, kmult0, q_rh, q_k) return ra + (kmult0 * cbar0)"""
Calculate empirical ecosystem respiration, RECO, based on current model parameters and the inferred soil organic carbon (SOC) storage; i.e., this calculation should be used in model calibration when SOC is not a priori known, see `pyl4c.apps.calibration.cbar()`. The expected model parameter names are "CUE" for the carbon use efficiency of plants. Parameters ---------- params : dict A dict-like data structure with named model parameters tsoil : numpy.ndarray (T x N) vector of soil temperature (deg K), where T is the number of time steps, N the number of sites smsf : numpy.ndarray (T x N) vector of surface soil wetness (%), where T is the number of time steps, N the number of sites reco_tower : numpy.ndarray (T x N) vector of observed RECO from eddy covariance tower sites gpp_tower : numpy.ndarray (T x N) vector of observed GPP from eddy covariance tower sites q_rh : int The percentile of RH/Kmult to use in calculating Cbar q_k : int The percentile of Kmult below which RH/Kmult values are masked Returns ------- numpy.ndarray '''Calculate empirical ecosystem respiration, RECO, based on current model parameters and the inferred soil organic carbon (SOC) storage; i.e., this calculation should be used in model calibration when SOC is not a priori known, see
cbar()
. The expected model parameter names are "CUE" for the carbon use efficiency of plants.Parameters
params
:dict
- A dict-like data structure with named model parameters
tsoil
:numpy.ndarray
- (T x N) vector of soil temperature (deg K), where T is the number of time steps, N the number of sites
smsf
:numpy.ndarray
- (T x N) vector of surface soil wetness (%), where T is the number of time steps, N the number of sites
reco_tower
:numpy.ndarray
- (T x N) vector of observed RECO from eddy covariance tower sites
gpp_tower
:numpy.ndarray
- (T x N) vector of observed GPP from eddy covariance tower sites
q_rh
:int
- The percentile of RH/Kmult to use in calculating Cbar
q_k
:int
- The percentile of Kmult below which RH/Kmult values are masked
Returns
numpy.ndarray
def report_fit_stats(obs, pred, weights=array([1]), verbose=True)
-
Expand source code
def report_fit_stats(obs, pred, weights = np.array([1]), verbose = True):
y = np.apply_along_axis(detrend, 0, obs, fill = True) yhat = np.apply_along_axis(detrend, 0, pred, fill = True) rmse = rmsd(obs, pred, weights = weights) ubrmse = rmsd(y, yhat, weights = weights) bias = np.nanmean(np.subtract(pred, obs)) mask = np.logical_or(np.isnan(obs), np.isnan(pred)) r_squared = 1 - np.divide( sum_of_squares( obs[~mask], pred[~mask], add_intercept = False, which = 'sse'), sum_of_squares( obs[~mask], pred[~mask], add_intercept = False, which = 'sst')) if verbose: print('Fit statistics:') print('-- R^2: %s' % ('%.3f' % r_squared).rjust(6)) print('-- RMSE: %s' % ('%.3f' % rmse).rjust(6)) print('-- ubRMSE: %s' % ('%.3f' % ubrmse).rjust(6)) print('-- Bias: %s' % ('%.3f' % bias).rjust(6)) return (r_squared, rmse, ubrmse, bias)"""
Reports the RMSE, ubRMSE, and Bias for observed and predicted values. Parameters ---------- obs : numpy.ndarray Vector of observed ("true") values pred : numpy.ndarray Vector of predicted values weights : numpy.ndarray (Optional) Vector of weights for each sample Returns ------- tuple (R-squared, RMSE, ubRMSE, Bias) '''Reports the RMSE, ubRMSE, and Bias for observed and predicted values.
Parameters
obs
:numpy.ndarray
- Vector of observed ("true") values
pred
:numpy.ndarray
- Vector of predicted values
weights
:numpy.ndarray
- (Optional) Vector of weights for each sample
Returns
tuple
- (R-squared, RMSE, ubRMSE, Bias)
def solve_least_squares(func, init_params, labels, bounds, **kwargs)
-
Expand source code
def solve_least_squares(func, init_params, labels, bounds, **kwargs):
# Update the optimization settings; this loss function produces # an estimate for the FT multiplier that is closest to prior kwargs.setdefault('loss', 'arctan') kwargs.setdefault('method', 'trf') kwargs.setdefault('max_nfev', 500) kwargs.setdefault('ftol', 1e-8) kwargs.setdefault('xtol', 1e-8) kwargs.setdefault('gtol', 1e-8) try: solution = optimize.least_squares( func, init_params, bounds = bounds, **kwargs) except ValueError: below = [ labels[i] for i in np.argwhere(init_params < bounds[0]).flatten().tolist() ] above = [ labels[i] for i in np.argwhere(init_params > bounds[1]).flatten().tolist() ] if np.isnan(init_params).any(): raise ValueError( 'Error in candidate parameter values; residual function probably returning NaNs') else: raise ValueError( '"Infeasibility" error; check lower bound on %s; upper bound on %s' % ( '(None)' if len(below) == 0 else ', '.join(below), '(None)' if len(above) == 0 else ', '.join(above))) return solution"""
Apply constrained, non-linear least-squares optimization. Mostly a wrapper for `scipy.optimize.least_squares()`. Parameters ---------- func : function Function to calculate the residuals init_params : list or tuple or numpy.ndarray Sequence of starting parameters (or "initial guesses") labels : list or tuple or numpy.ndarray Sequence of parameter names bounds : list or tuple 2-element sequence of (lower, upper) bounds where each element is an array Returns ------- scipy.optimize.OptimizeResult '''Apply constrained, non-linear least-squares optimization. Mostly a wrapper for
scipy.optimize.least_squares()
.Parameters
func
:function
- Function to calculate the residuals
init_params
:list
ortuple
ornumpy.ndarray
- Sequence of starting parameters (or "initial guesses")
labels
:list
ortuple
ornumpy.ndarray
- Sequence of parameter names
bounds
:list
ortuple
- 2-element sequence of (lower, upper) bounds where each element is an array
Returns
scipy.optimize.OptimizeResult
Classes
class BPLUT (params_dict=None, labels=None, hdf5_path=None, hdf5_group='BPLUT')
-
Expand source code
class BPLUT(object):
_labels = [ 'LUE', 'CUE', 'tmin0', 'tmin1', 'vpd0', 'vpd1', 'smrz0', 'smrz1', 'smsf0', 'smsf1', 'ft0', 'ft1', 'tsoil', 'decay_rates0', 'decay_rates1', 'decay_rates2', 'f_metabolic', 'f_structural' ] _npft = 10 # Number of (possible) PFT classes _valid_pft = range(1, 9) # Canonical range of valid PFTs def __init__( self, params_dict = None, labels = None, hdf5_path = None, hdf5_group = 'BPLUT'):"""
Represents a Biome Properties Look-Up Table (BPLUT) with PFT classes along the rows and parameters along the columns. If initialized with a `params_dict`, these are the values of the BPLUT. If initialized with an `hdf5_path` but without a `params_dict`, the parameters are read-in from the HDF5 file. '''self.hdf5_group = hdf5_group self.hdf5_path = hdf5_path if labels is not None: print('WARNING: Parameter names ending with a number are assumed to be bounds on ramp functions!') self._labels = labels # Create an in-memory parameter dictionary empty = self._empty_dict(self.labels) if params_dict is None: init_data = empty # No prior BPLUT, use empty table else: init_data = params_dict.copy() # IMPORTANT: Make sure prior BPLUT has all the necessary params for key in empty.keys(): init_data.setdefault(key, empty[key]) # Optionally, maintain a file BPLUT if hdf5_path is not None: # Restore from the file data, filling in NaNs with initial if os.path.exists(hdf5_path): with h5py.File(hdf5_path, 'r') as hdf: if hdf5_group in hdf.keys(): init_data = self.hdf5_restore(hdf, init_data = init_data) # Then, if file dataset doesn't exist, create a new one and store # the initial data with h5py.File(hdf5_path, 'a') as hdf: self.hdf5_flush(hdf, data = init_data) self.data = init_data @property def data(self): 'The parameters dictionary or `dict` instance' return self._data @data.setter def data(self, data): self._data = data @property def labels(self): 'Names of the free parameters' return self._labels def __getitem__(self, key): return self.data[key] def __setitem__(self, key, value): self.data[key] = value def _empty_dict(self, labels = None, dtype = np.float32): ''' Given sequence of labels, convert to legacy, human-readable dict, e.g.: { 'LUE': array([[ nan, 1.17, ..., nan ]]), ... } ''' labels_dedupe = self._canonical(labels) result = dict() for name in labels_dedupe: size = len(list(filter(lambda x: x.startswith(name), labels))) result[name] = np.ones((size, self._npft), dtype = dtype) * np.nan return result def _canonical(self, labels = None):"""
Parameters ---------- params_dict : dict A BPLUT to initialize the new BPLUT labels : tuple or list Names of the parameters hdf5_path : str Path to an HDF5 file to use as a temporary store hdf5_group : str Field name with which to store data in HDF5 file (Default: "BPLUT") '''labels = self.labels if labels is None else labels # Remove numeric suffixes and de-duplicate the list return list(OrderedDict([ (p.strip('0123456789'), 0) for p in labels ]).keys()) def dump(self, src, dest, clean = False):"""
Returns a short list of labels, without duplicates. Specifically, a list of labels like, e.g., ("tmin0", "tmin1", "decay_rates1") becomes ("tmin", "decay_rates"). Parameters ---------- labels : tuple or list or None Returns ------- list '''bplut = pd.read_csv(src, comment = '#', names = BPLUT_HEADER) bplut['LUEmax'] = self.data['LUE'][0,1:9] bplut['Tmin_min_K'] = self.data['tmin'][0,1:9] bplut['Tmin_max_K'] = self.data['tmin'][1,1:9] bplut['VPD_min_Pa'] = self.data['vpd'][0,1:9] bplut['VPD_max_Pa'] = self.data['vpd'][1,1:9] bplut['SMrz_min'] = self.data['smrz'][0,1:9] bplut['SMrz_max'] = self.data['smrz'][1,1:9] bplut['FT_min'] = self.data['ft'][0,1:9] bplut['FT_max'] = self.data['ft'][1,1:9] # Likely not used (=1.0) bplut['Tsoil_beta0']= self.data['tsoil'][0,1:9] bplut['SMtop_min'] = self.data['smsf'][0,1:9] bplut['SMtop_max'] = self.data['smsf'][1,1:9] bplut['kopt'] = self.data['decay_rates'][0,1:9] # If the soil-moisture parameters are close to zero, make them zero if clean: bplut.loc[np.abs(bplut['SMtop_min']) < 1,'SMtop_min'] = 0 bplut.loc[np.abs(bplut['SMrz_min']) < 1,'SMrz_min'] = 0 # "fraut" (fraction of autotrophic respiration) is the complement # of carbon use efficiency (CUE) bplut['fraut'] = 1 - self.data['CUE'].ravel()[1:9] # Strangely enough, the header in the Ops file begins with '#' header = list(BPLUT_HEADER) header[0] = f'#{header[0]}' bplut.to_csv(dest, index = False, header = header) print(f'Wrote BPLUT to: {dest}') def flat(self, pft, labels = None):"""
Writes the BPLUT to a CSV file, formatted for use in L4C Ops. Requires an existing copy of a BPLUT to use as a template. Parameters ---------- src : str Path to the template (original) BPLUT CSV file dest : str Output BPLUT CSV file path clean : bool True to standardize some output fields, e.g., making sure very small parameter values (from optimization) are actually zero (Default: False) '''labels_dedupe = self._canonical(labels) return np.hstack([ self.data[p][:,pft].ravel() if p in self.data.keys() else np.nan for p in labels_dedupe ]) def hdf5_flush(self, hdf, data = None): ''' Writes the current BPLUT to an HDF5 file. Parameters ---------- hdf : h5py.File HDF5 file open for writing data : dict ''' assert hdf.mode != 'r', 'File not open for writing!' data = self.data if data is None else data if self.hdf5_group not in hdf.keys(): hdf.create_group(self.hdf5_group) for key, value in data.items(): if key.startswith('_'): continue # Skip keys that are not parameter names field = '%s/%s' % (self.hdf5_group, key) if key not in hdf[self.hdf5_group].keys(): hdf.create_dataset(field, value.shape, np.float32, value) else: # Overwrite NaNs in the file data _value = hdf[field][:] hdf[field][:] = np.where(np.isnan(_value), value, _value) hdf.flush() def hdf5_restore(self, hdf, init_data = None, dtype = np.float32):"""
Retrieves a flat list of parameters for a specific PFT. Parameters ---------- pft : int Numeric code of the PFT for which to return parameter values labels : tuple or list (Optional) A sequence of parameter names desired, if not all; defaults to returning all parameters Returns ------- numpy.ndarray '''data = dict() if init_data is None else init_data for key in hdf[self.hdf5_group].keys(): # Update the in-memory BPLUT from_hdf5 = hdf[self.hdf5_group][key][:] data[key] = np.where( ~np.isnan(from_hdf5), from_hdf5, init_data.get(key) ).astype(dtype) return data def pickle(self, output_path, version_id = None):"""
Reads in the BPLUT table stored in the HDF5 file. Parameters ---------- hdf : h5py.File init_data : dict Initital data; will be over-written by HDF5 file contents Returns ------- dict '''with open(output_path, 'wb') as file: output = self.data.copy() if version_id is not None: output['_version'] = version_id pickle.dump(output, file) def show(self, pft, param, precision = 2):"""
Writes the current BPLUT parameters, as a dictionary, to a pickle file. Parameters ---------- output_path : str The output path for the pickle file (*.pickle) version_id : str (Optional) The version identifier for this BPLUT '''assert not (pft is None and param is None),\ 'Either one or both must be specified: --pft or --param' set_of_labels = self._canonical() if param is None else [param] set_of_pfts = self._valid_pft if pft is None else [pft] for each in set_of_labels: assert each in self._canonical(), 'Unrecognized parameter: %s' % each for each in set_of_pfts: assert each in range(0, self._npft), 'PFT code out of range' pad = max(len(l) for l in set_of_labels) + 2 fmt_string = '{:>%d} {:>%d}' % (pad, 5 + precision) for pft in set_of_pfts: print('BPLUT parameters for PFT %d:' % pft) for label in set_of_labels: param_values = self.data[label][:,pft] for i, value in enumerate(param_values): # If there are multiple values for a parameter (group), # append a number to the end of the label if len(param_values) > 1: prefix = '%s%d:' % (label, i) else: prefix = '%s:' % label print( fmt_string.format(prefix, ('%%.%df' % precision) % value)) def update(self, pft, values, labels, flush = True):"""
Prints the current BPLUT parameters for a given PFT, the values of a given parameter for all PFTs, or the value of a specific PFT-parameter combination. Parameters ---------- pft : int or None The PFT class param : str or None The name of the parameter precision : int Decimal precision to use for printing numbers '''assert len(values) == len(labels),\ 'Vectors of values and parameter labels must have the same length' if flush: assert self.hdf5_path is not None,\ 'No HDF5 file storage is attached' hdf = h5py.File(self.hdf5_path, 'a') for i, name in enumerate(labels): abbrv = name.strip('0123456789') # In case parameter has multiple levels, like a ramp function # (e.g., "smsf0" and "smsf1" are two rows) dupes = list(filter(lambda x: x.startswith(abbrv), self.labels)) j = dupes.index(name) # If it has one level (e.g., "LUE"), j = 0 self.data[abbrv][j,pft] = values[i] if flush: path = '%s/%s' % (self.hdf5_group, abbrv) hdf[path][:,pft] = self.data[abbrv][:,pft] if flush: hdf.flush() hdf.close()"""
Updates the BPLUT with the specified parameters for a single PFT. Parameters ---------- pft : int The PFT class values : tuple or list Sequence of parameter values, one for each parameter named in `labels` labels : tuple or list Sequence of parameter names, one for each value in `values` flush : bool True to write the result to disk (attached HDF5 file storage) (Default: True) '''Represents a Biome Properties Look-Up Table (BPLUT) with PFT classes along the rows and parameters along the columns.
If initialized with a
params_dict
, these are the values of the BPLUT. If initialized with anhdf5_path
but without aparams_dict
, the parameters are read-in from the HDF5 file.Parameters
params_dict
:dict
- A BPLUT to initialize the new BPLUT
labels
:tuple
orlist
- Names of the parameters
hdf5_path
:str
- Path to an HDF5 file to use as a temporary store
hdf5_group
:str
- Field name with which to store data in HDF5 file (Default: "BPLUT")
Instance variables
prop data
-
Expand source code
@property def data(self): 'The parameters dictionary or `dict` instance' return self._data
The parameters dictionary or
dict
instance prop labels
-
Expand source code
@property def labels(self): 'Names of the free parameters' return self._labels
Names of the free parameters
Methods
def dump(self, src, dest, clean=False)
-
Expand source code
def dump(self, src, dest, clean = False):
bplut = pd.read_csv(src, comment = '#', names = BPLUT_HEADER) bplut['LUEmax'] = self.data['LUE'][0,1:9] bplut['Tmin_min_K'] = self.data['tmin'][0,1:9] bplut['Tmin_max_K'] = self.data['tmin'][1,1:9] bplut['VPD_min_Pa'] = self.data['vpd'][0,1:9] bplut['VPD_max_Pa'] = self.data['vpd'][1,1:9] bplut['SMrz_min'] = self.data['smrz'][0,1:9] bplut['SMrz_max'] = self.data['smrz'][1,1:9] bplut['FT_min'] = self.data['ft'][0,1:9] bplut['FT_max'] = self.data['ft'][1,1:9] # Likely not used (=1.0) bplut['Tsoil_beta0']= self.data['tsoil'][0,1:9] bplut['SMtop_min'] = self.data['smsf'][0,1:9] bplut['SMtop_max'] = self.data['smsf'][1,1:9] bplut['kopt'] = self.data['decay_rates'][0,1:9] # If the soil-moisture parameters are close to zero, make them zero if clean: bplut.loc[np.abs(bplut['SMtop_min']) < 1,'SMtop_min'] = 0 bplut.loc[np.abs(bplut['SMrz_min']) < 1,'SMrz_min'] = 0 # "fraut" (fraction of autotrophic respiration) is the complement # of carbon use efficiency (CUE) bplut['fraut'] = 1 - self.data['CUE'].ravel()[1:9] # Strangely enough, the header in the Ops file begins with '#' header = list(BPLUT_HEADER) header[0] = f'#{header[0]}' bplut.to_csv(dest, index = False, header = header) print(f'Wrote BPLUT to: {dest}')"""
Writes the BPLUT to a CSV file, formatted for use in L4C Ops. Requires an existing copy of a BPLUT to use as a template. Parameters ---------- src : str Path to the template (original) BPLUT CSV file dest : str Output BPLUT CSV file path clean : bool True to standardize some output fields, e.g., making sure very small parameter values (from optimization) are actually zero (Default: False) '''Writes the BPLUT to a CSV file, formatted for use in L4C Ops. Requires an existing copy of a BPLUT to use as a template.
Parameters
src
:str
- Path to the template (original) BPLUT CSV file
dest
:str
- Output BPLUT CSV file path
clean
:bool
- True to standardize some output fields, e.g., making sure very small parameter values (from optimization) are actually zero (Default: False)
def flat(self, pft, labels=None)
-
Expand source code
def flat(self, pft, labels = None):
labels_dedupe = self._canonical(labels) return np.hstack([ self.data[p][:,pft].ravel() if p in self.data.keys() else np.nan for p in labels_dedupe ])"""
Retrieves a flat list of parameters for a specific PFT. Parameters ---------- pft : int Numeric code of the PFT for which to return parameter values labels : tuple or list (Optional) A sequence of parameter names desired, if not all; defaults to returning all parameters Returns ------- numpy.ndarray '''Retrieves a flat list of parameters for a specific PFT.
Parameters
pft
:int
- Numeric code of the PFT for which to return parameter values
labels
:tuple
orlist
- (Optional) A sequence of parameter names desired, if not all; defaults to returning all parameters
Returns
numpy.ndarray
def hdf5_flush(self, hdf, data=None)
-
Expand source code
def hdf5_flush(self, hdf, data = None): ''' Writes the current BPLUT to an HDF5 file. Parameters ---------- hdf : h5py.File HDF5 file open for writing data : dict ''' assert hdf.mode != 'r', 'File not open for writing!' data = self.data if data is None else data if self.hdf5_group not in hdf.keys(): hdf.create_group(self.hdf5_group) for key, value in data.items(): if key.startswith('_'): continue # Skip keys that are not parameter names field = '%s/%s' % (self.hdf5_group, key) if key not in hdf[self.hdf5_group].keys(): hdf.create_dataset(field, value.shape, np.float32, value) else: # Overwrite NaNs in the file data _value = hdf[field][:] hdf[field][:] = np.where(np.isnan(_value), value, _value) hdf.flush()
Writes the current BPLUT to an HDF5 file.
Parameters
hdf
:h5py.File
- HDF5 file open for writing
data
:dict
def hdf5_restore(self, hdf, init_data=None, dtype=numpy.float32)
-
Expand source code
def hdf5_restore(self, hdf, init_data = None, dtype = np.float32):
data = dict() if init_data is None else init_data for key in hdf[self.hdf5_group].keys(): # Update the in-memory BPLUT from_hdf5 = hdf[self.hdf5_group][key][:] data[key] = np.where( ~np.isnan(from_hdf5), from_hdf5, init_data.get(key) ).astype(dtype) return data"""
Reads in the BPLUT table stored in the HDF5 file. Parameters ---------- hdf : h5py.File init_data : dict Initital data; will be over-written by HDF5 file contents Returns ------- dict '''Reads in the BPLUT table stored in the HDF5 file.
Parameters
hdf
:h5py.File
init_data
:dict
- Initital data; will be over-written by HDF5 file contents
Returns
dict
def pickle(self, output_path, version_id=None)
-
Expand source code
def pickle(self, output_path, version_id = None):
with open(output_path, 'wb') as file: output = self.data.copy() if version_id is not None: output['_version'] = version_id pickle.dump(output, file)"""
Writes the current BPLUT parameters, as a dictionary, to a pickle file. Parameters ---------- output_path : str The output path for the pickle file (*.pickle) version_id : str (Optional) The version identifier for this BPLUT '''Writes the current BPLUT parameters, as a dictionary, to a pickle file.
Parameters
output_path
:str
- The output path for the pickle file (*.pickle)
version_id
:str
- (Optional) The version identifier for this BPLUT
def show(self, pft, param, precision=2)
-
Expand source code
def show(self, pft, param, precision = 2):
assert not (pft is None and param is None),\ 'Either one or both must be specified: --pft or --param' set_of_labels = self._canonical() if param is None else [param] set_of_pfts = self._valid_pft if pft is None else [pft] for each in set_of_labels: assert each in self._canonical(), 'Unrecognized parameter: %s' % each for each in set_of_pfts: assert each in range(0, self._npft), 'PFT code out of range' pad = max(len(l) for l in set_of_labels) + 2 fmt_string = '{:>%d} {:>%d}' % (pad, 5 + precision) for pft in set_of_pfts: print('BPLUT parameters for PFT %d:' % pft) for label in set_of_labels: param_values = self.data[label][:,pft] for i, value in enumerate(param_values): # If there are multiple values for a parameter (group), # append a number to the end of the label if len(param_values) > 1: prefix = '%s%d:' % (label, i) else: prefix = '%s:' % label print( fmt_string.format(prefix, ('%%.%df' % precision) % value))"""
Prints the current BPLUT parameters for a given PFT, the values of a given parameter for all PFTs, or the value of a specific PFT-parameter combination. Parameters ---------- pft : int or None The PFT class param : str or None The name of the parameter precision : int Decimal precision to use for printing numbers '''Prints the current BPLUT parameters for a given PFT, the values of a given parameter for all PFTs, or the value of a specific PFT-parameter combination.
Parameters
pft
:int
orNone
- The PFT class
param
:str
orNone
- The name of the parameter
precision
:int
- Decimal precision to use for printing numbers
def update(self, pft, values, labels, flush=True)
-
Expand source code
def update(self, pft, values, labels, flush = True):
assert len(values) == len(labels),\ 'Vectors of values and parameter labels must have the same length' if flush: assert self.hdf5_path is not None,\ 'No HDF5 file storage is attached' hdf = h5py.File(self.hdf5_path, 'a') for i, name in enumerate(labels): abbrv = name.strip('0123456789') # In case parameter has multiple levels, like a ramp function # (e.g., "smsf0" and "smsf1" are two rows) dupes = list(filter(lambda x: x.startswith(abbrv), self.labels)) j = dupes.index(name) # If it has one level (e.g., "LUE"), j = 0 self.data[abbrv][j,pft] = values[i] if flush: path = '%s/%s' % (self.hdf5_group, abbrv) hdf[path][:,pft] = self.data[abbrv][:,pft] if flush: hdf.flush() hdf.close()"""
Updates the BPLUT with the specified parameters for a single PFT. Parameters ---------- pft : int The PFT class values : tuple or list Sequence of parameter values, one for each parameter named in `labels` labels : tuple or list Sequence of parameter names, one for each value in `values` flush : bool True to write the result to disk (attached HDF5 file storage) (Default: True) '''Updates the BPLUT with the specified parameters for a single PFT.
Parameters
pft
:int
- The PFT class
values
:tuple
orlist
- Sequence of parameter values, one for each parameter named in
labels
labels
:tuple
orlist
- Sequence of parameter names, one for each value in
values
flush
:bool
- True to write the result to disk (attached HDF5 file storage) (Default: True)