initial commit

pcmdi_metrics/mjo/mjo_metrics_driver.py

@@ -39,11 +39,8 @@
 import sys
 import time
 from argparse import RawTextHelpFormatter
-from collections import defaultdict
 from shutil import copyfile
 
-from genutil import StringConstructor
-
 import pcmdi_metrics
 from pcmdi_metrics.mean_climate.lib import pmp_parser
 from pcmdi_metrics.mjo.lib import (
@@ -52,6 +49,7 @@
  mjo_metric_ewr_calculation,
  mjo_metrics_to_json,
 )
+from pcmdi_metrics.utils import fill_template, tree
 
 # To avoid below error
 # OpenBLAS blas_thread_init: pthread_create failed for thread XX of 96: Resource temporarily unavailable
@@ -126,18 +124,15 @@
 # case id
 case_id = param.case_id
 
-# Output
+# Output directory
 outdir_template = param.process_templated_argument("results_dir")
-outdir = StringConstructor(
- str(
- outdir_template(output_type="%(output_type)", mip=mip, exp=exp, case_id=case_id)
- )
-)
 
-# Create output directory
+# Create output directories
 for output_type in ["graphics", "diagnostic_results", "metrics_results"]:
- os.makedirs(outdir(output_type=output_type), exist_ok=True)
- print(outdir(output_type=output_type))
+ outdir = fill_template(
+ outdir_template, output_type=output_type, mip=mip, exp=exp, case_id=case_id
+ )
+ os.makedirs(outdir, exist_ok=True)
 
 # Generate CMEC compliant json
 if hasattr(param, "cmec"):
@@ -175,12 +170,6 @@
 # =================================================
 # Declare dictionary for .json record
 # -------------------------------------------------
-
-
-def tree():
- return defaultdict(tree)
-
-
 result_dict = tree()
 
 # Define output json file

pcmdi_metrics/utils/__init__.py

@@ -1,3 +1,5 @@
 from .create_land_sea_mask import apply_landmask, create_land_sea_mask
 from .create_target_grid import create_target_grid
 from .sort_human import sort_human
+from .string_constructor import StringConstructor, fill_template
+from .tree_dict import tree

pcmdi_metrics/utils/string_constructor.py

@@ -0,0 +1,99 @@
+import warnings
+
+
+class StringConstructor:
+ """
+ This class aims at spotting keywords in a string and replacing them.
+ """
+
+ def __init__(self, template=None):
+ """
+ Instantiates a StringConstructor object.
+ """
+ self.template = template
+ # Generate the keys and set them to empty
+ keys = self.keys()
+ for k in keys:
+ setattr(self, k, "")
+
+ def keys(self, template=None):
+ if template is None:
+ template = self.template
+ if template is None:
+ return []
+ # Determine the keywords in the template
+ keys = []
+ template_split = template.split("%(")[1:]
+ if len(template_split) > 0:
+ for k in template_split:
+ sp = k.split(")")
+ if sp[0] not in keys:
+ keys.append(sp[0])
+ return keys
+
+ def construct(self, template=None, **kw):
+ """
+ Accepts a string with an unlimited number of keywords to replace.
+ """
+ if template is None:
+ template = self.template
+ # Replace the keywords with their values
+ for k in self.keys():
+ if k not in kw:
+ warnings.warn(f"Keyword '{k}' not provided for filling the template.")
+ template = template.replace("%(" + k + ")", kw.get(k, getattr(self, k, "")))
+ return template
+
+ def reverse(self, name, debug=False):
+ """
+ The reverse function attempts to take a template and derive its keyword values based on name parameter.
+ """
+ out = {}
+ template = self.template
+ for k in self.keys():
+ sp = template.split("%%(%s)" % k)
+ i1 = name.find(sp[0]) + len(sp[0])
+ j1 = sp[1].find("%(")
+ if j1 == -1:
+ if sp[1] == "":
+ val = name[i1:]
+ else:
+ i2 = name.find(sp[1])
+ val = name[i1:i2]
+ else:
+ i2 = name[i1:].find(sp[1][:j1])
+ val = name[i1 : i1 + i2]
+ template = template.replace("%%(%s)" % k, val)
+ out[k] = val
+ if self.construct(self.template, **out) != name:
+ raise ValueError("Invalid pattern sent")
+ return out
+
+ def __call__(self, *args, **kw):
+ """default call is construct function"""
+ return self.construct(*args, **kw)
+
+
+def fill_template(template: str, **kwargs) -> str:
+ """
+ Fill in a template string with keyword values.
+
+ Parameters
+ ----------
+ - template (str): The template string containing keywords of the form '%(keyword)'.
+ - kwargs (dict): Keyword arguments with values to replace in the template.
+
+ Returns
+ -------
+ - str: The filled-in string with replaced keywords.
+
+ Examples
+ --------
+ >>> from pcmdi_metrics.utils import fill_template
+ >>> template = "This is a %(adjective) %(noun) that %(verb)."
+ >>> filled_string = fill_template(template, adjective="great", noun="example", verb="works")
+ >>> print(filled_string) # It will print "This is a great example that works."
+ """
+ filler = StringConstructor(template)
+ filled_template = filler.construct(**kwargs)
+ return filled_template

pcmdi_metrics/utils/tree_dict.py

@@ -0,0 +1,18 @@
+from collections import defaultdict
+
+
+def tree():
+ """
+ Create a nested defaultdict with itself as the factory.
+
+ Returns:
+ - defaultdict: A nested defaultdict with a default factory of tree itself.
+
+ Examples
+ --------
+ >>> import pcmdi_metrics.utils import tree
+ >>> my_tree = tree()
+ >>> my_tree['level1']['level2']['level3'] = 'value'
+ >>> print(my_tree['level1']['level2']['level3']) # Output: 'value'
+ """
+ return defaultdict(tree)

pcmdi_metrics/utils/utils.py

@@ -0,0 +1,75 @@
+import xarray as xr
+import xcdat as xc
+
+
+def get_axis_list(ds: xr.Dataset) -> list[str]:
+ axes = list(ds.coords.keys())
+ return axes
+
+
+def get_longitude(ds: xr.Dataset) -> xr.DataArray:
+ key_lon = xc.axis.get_dim_keys(ds, axis="X")
+ lon = ds[key_lon]
+ return lon
+
+
+def get_latitude(ds: xr.Dataset) -> xr.DataArray:
+ key_lat = xc.axis.get_dim_keys(ds, axis="Y")
+ lat = ds[key_lat]
+ return lat
+
+
+def select_subset(
+ ds: xr.Dataset, lat: tuple = None, lon: tuple = None, time: tuple = None
+) -> xr.Dataset:
+ """_summary_
+
+ Parameters
+ ----------
+ ds : xr.Dataset
+ _description_
+ lat : tuple, optional
+ _description_, by default None
+ lon : tuple, optional
+ _description_, by default None
+ time : tuple, optional
+ _description_, by default None
+
+ Returns
+ -------
+ xr.Dataset
+ _description_
+
+ Examples
+ ---------
+ Import:
+
+ >>> from pcmdi_metrics.utils import select_subset
+
+ Spatial subsetting:
+
+ >>> (lat1, lat2) = (30, 50)
+ >>> (lon1, lon2) = (110, 130)
+ >>> ds_subset = select_subset(ds, lat=(lat1, lat2), lon=(lon1, lon2))
+
+ Temporal subsetting:
+
+ >>> import cftime
+ >>> time1 = cftime.DatetimeProlepticGregorian(1850, 1, 16, 12, 0, 0, 0)
+ >>> time2 = cftime.DatetimeProlepticGregorian(1851, 1, 16, 12, 0, 0, 0)
+ >>> ds_subset = select_subset(ds, time=(time1, time2))
+ """
+
+ sel_keys = {}
+ if lat is not None:
+ lat_key = xc.axis.get_dim_keys(ds, axis="Y")
+ sel_keys[lat_key] = slice(*lat)
+ if lon is not None:
+ lon_key = xc.axis.get_dim_keys(ds, axis="X")
+ sel_keys[lon_key] = slice(*lon)
+ if time is not None:
+ time_key = xc.axis.get_dim_keys(ds, axis="T")
+ sel_keys[time_key] = slice(*time)
+
+ ds = ds.sel(**sel_keys)
+ return ds

pcmdi_metrics/variability_mode/lib/lib_variability_mode.py

@@ -161,9 +161,7 @@ def variability_metrics_to_json(
  outdir, json_filename, result_dict, model=None, run=None, cmec_flag=False
 ):
  # Open JSON
- JSON = pcmdi_metrics.io.base.Base(
- outdir(output_type="metrics_results"), json_filename
- )
+ JSON = pcmdi_metrics.io.base.Base(outdir, json_filename)
  # Dict for JSON
  json_dict = copy.deepcopy(result_dict)
  if model is not None or run is not None: