Source code for src.data_model

"""Classes to describe "abstract" properties of model data: aspects that are
independent of any model, experiment, or hosting protocol.

This is based on the `CF standard terminology
<https://cfconventions.org/Data/cf-conventions/cf-conventions-1.9/cf-conventions.html>`__.
"""
import abc
import dataclasses as dc
import itertools
import typing
from src import util
import src.units  # fully qualify name to reduce confusion with "units" attributes
import src.core
import logging
_log = logging.getLogger(__name__)

[docs] class AbstractDMCoordinate(abc.ABC): """Defines interface (set of attributes) for :class:`DMCoordinate` objects. """ @property @abc.abstractmethod def name(self): """Coordinate name (str).""" pass @property @abc.abstractmethod def standard_name(self): """Coordinate CF standard name (str).""" pass @property @abc.abstractmethod def units(self): """Coordinate units (str, or :class:`~src.units.Units`).""" pass @property @abc.abstractmethod def axis(self): """Coordinate axis identifier (str, 'X', 'Y', 'Z', 'T').""" pass @property @abc.abstractmethod def bounds(self): """Associated bounds variable for the coordinate, if present (else None.) """ pass @property @abc.abstractmethod def value(self): """If a `scalar coordinate <http://cfconventions.org/Data/cf-conventions/cf-conventions-1.8/cf-conventions.html#scalar-coordinate-variables>`__, value of the coordinate in units given by :meth:`units`. """ pass @property @abc.abstractmethod def is_scalar(self): """Whether the coordinate is a `scalar coordinate <http://cfconventions.org/Data/cf-conventions/cf-conventions-1.8/cf-conventions.html#scalar-coordinate-variables>`__ (bool). """ pass @property @abc.abstractmethod def has_bounds(self): """Whether the coordinate has an associated bounds variable (bool). """ pass
[docs] class AbstractDMDependentVariable(abc.ABC): """Defines interface (set of attributes) for "dependent variables" (data defined as a function of one or more dimension coordinates), which inherit from :class:`DMDimensions` in this implementation. """ @property @abc.abstractmethod def name(self): """Variable name (str).""" pass @property @abc.abstractmethod def standard_name(self): """Variable CF standard name (str).""" pass @property @abc.abstractmethod def units(self): """Variable units (str, or :class:`~src.units.Units`).""" pass @property @abc.abstractmethod def dims(self): pass @property @abc.abstractmethod def scalar_coords(self): pass @property @abc.abstractmethod def axes(self): pass @property @abc.abstractmethod def all_axes(self): pass @property @abc.abstractmethod def X(self): """X axis coordinate of variable, if defined.""" pass @property @abc.abstractmethod def Y(self): """Y axis coordinate of variable, if defined.""" pass @property @abc.abstractmethod def Z(self): """Z axis coordinate of variable, if defined.""" pass @property @abc.abstractmethod def T(self): """T axis coordinate of variable, if defined.""" pass @property @abc.abstractmethod def is_static(self): """Whether the variable has time dependence (bool).""" pass
[docs] class AbstractDMCoordinateBounds(AbstractDMDependentVariable): """Defines interface (set of attributes) for :class:`DMCoordinateBounds` objects. """ @property @abc.abstractmethod def coord(self): """DMCoordinate object which this object is the bounds of.""" pass
# ------------------------------------------------------------------------------ _AXIS_NAMES = ('X', 'Y', 'Z', 'T') _ALL_AXIS_NAMES = _AXIS_NAMES + ('BOUNDS', 'OTHER')
[docs] @util.mdtf_dataclass class DMBoundsDimension(object): """Placeholder object to represent the bounds dimension of a :class:`DMCoordinateBounds` object. Not a dimension coordinate, and strictly speaking we should make another set of classes for dimensions. """ name: str = util.MANDATORY standard_name = 'bounds' units = src.units.Units('1') axis = 'BOUNDS' bounds = None value = None @property def has_bounds(self): """Whether the coordinate has an associated bounds variable (bool). Always False for this class. """ return False @property def is_scalar(self): """Whether the coordinate is a `scalar coordinate <http://cfconventions.org/Data/cf-conventions/cf-conventions-1.8/cf-conventions.html#scalar-coordinate-variables>`__ (bool). Always False for this class. """ return False
@util.mdtf_dataclass class _DMCoordinateShared(object): """Fields common to all :class:`AbstractDMCoordinate` child classes which aren't fixed to particular values. ``value`` is our mechanism for implementing CF convention `scalar coordinates <http://cfconventions.org/Data/cf-conventions/cf-conventions-1.8/cf-conventions.html#scalar-coordinate-variables>`__. """ standard_name: str = util.MANDATORY units: src.units.Units = util.MANDATORY axis: str = 'OTHER' bounds_var: AbstractDMCoordinateBounds = None value: typing.Union[int, float] = None @property def bounds(self): """The *bounds_var* attribute is stored as a pointer to the actual object representing the bounds variable for this coordinate, but in order to parallel xarray's syntax define 'bounds' to return the name of this variable, not the variable itself. """ if not hasattr(self.bounds_var, 'name'): return None else: return self.bounds_var.name @property def has_bounds(self): """Whether the coordinate has an associated bounds variable (bool). """ return (self.bounds_var is not None) @property def is_scalar(self): """Whether the coordinate is a `scalar coordinate <http://cfconventions.org/Data/cf-conventions/cf-conventions-1.8/cf-conventions.html#scalar-coordinate-variables>`__ (bool). """ return (self.value is not None) def make_scalar(self, new_value): """Returns a copy of the coordinate, converted to a scalar coordinate at value *new_value* (and coordinate's current ``units``.) """ return dc.replace(self, value=new_value)
[docs] @util.mdtf_dataclass class DMCoordinate(_DMCoordinateShared): """Class to describe a single coordinate variable (dimension coordinate or scalar coordinate, in the sense used by the `CF conventions <http://cfconventions.org/Data/cf-conventions/cf-conventions-1.8/cf-conventions.html#terminology>`__). """ name: str = util.MANDATORY """Coordinate name.""" # bounds_var: AbstractDMCoordinateBounds # [scalar] value: int or float standard_name: str = util.MANDATORY """Coordinate CF standard name.""" units: src.units.Units = util.MANDATORY """Coordinate units (str or :class:`~src.units.Units`).""" axis: str = 'OTHER' """Coordinate axis identifier ('X', 'Y', etc.)"""
[docs] @util.mdtf_dataclass class DMLongitudeCoordinate(_DMCoordinateShared): """Class to describe a longitude dimension coordinate. """ name: str = util.MANDATORY """Coordinate name.""" # bounds_var: AbstractDMCoordinateBounds # [scalar] value: int or float standard_name: str = util.MANDATORY """Coordinate CF standard name.""" units: src.units.Units = 'degrees_east' """Coordinate units (str or :class:`~src.units.Units`).""" axis: str = 'X' """Coordinate axis identifier. Always 'X' for this coordinate."""
[docs] @util.mdtf_dataclass class DMLatitudeCoordinate(_DMCoordinateShared): """Class to describe a latitude dimension coordinate. """ name: str = util.MANDATORY """Coordinate name; defaults to 'lat'.""" # bounds_var: AbstractDMCoordinateBounds # [scalar] value: int or float standard_name: str = util.MANDATORY """Coordinate CF standard name.""" units: src.units.Units = 'degrees_north' """Coordinate units (str or :class:`~src.units.Units`).""" axis: str = 'Y' """Coordinate axis identifier. Always 'Y' for this coordinate."""
[docs] @util.mdtf_dataclass class DMVerticalCoordinate(_DMCoordinateShared): """Class to describe a non-parametric vertical coordinate (height or depth), following the `CF conventions <http://cfconventions.org/Data/cf-conventions/cf-conventions-1.8/cf-conventions.html#vertical-coordinate>`__. """ name: str = util.MANDATORY """Coordinate name.""" # bounds_var: AbstractDMCoordinateBounds # [scalar] value: int or float standard_name: str = util.MANDATORY """Coordinate CF standard name.""" units: src.units.Units = "1" # dimensionless vertical coords OK """Coordinate units (str or :class:`~src.units.Units`).""" axis: str = 'Z' """Coordinate axis identifier. Always 'Z' for this coordinate.""" positive: str = util.MANDATORY
[docs] @util.mdtf_dataclass class DMParametricVerticalCoordinate(DMVerticalCoordinate): """Class to describe `parametric vertical coordinates <http://cfconventions.org/Data/cf-conventions/cf-conventions-1.8/cf-conventions.html#parametric-vertical-coordinate>`__. Note that the variable names appearing in ``formula_terms`` aren't parsed here, in order to keep the class hashable. """ computed_standard_name: str = "" """Coordinate CF standard name.""" long_name: str = "" # Don't include formula_terms in testing for equality, since this could # reference different names for the aux coord variables. # TODO: resolve names in formula_terms to references to objects in the data # model. formula_terms: str = dc.field(default=None, compare=False)
[docs] @util.mdtf_dataclass class DMGenericTimeCoordinate(_DMCoordinateShared): """Applies to collections of variables, which may be at different time frequencies (or other attributes). """ name: str = 'time' """Coordinate name; defaults to 'time'.""" # bounds_var: AbstractDMCoordinateBounds # [scalar] value: int or float standard_name: str = 'time' """Coordinate CF standard name.""" units: src.units.Units = "" """Coordinate units (str or :class:`~src.units.Units`).""" axis: str = 'T' """Coordinate axis identifier. Always 'T' for this coordinate.""" calendar: str = "" """CF standard calendar for time data.""" range: typing.Any = None @property def is_static(self): """Check for time-independent data ('fx' in CMIP6 DRS.) Do the comparison by checking date_range against the placeholder value because that's unique -- we may be using a different DateFrequency depending on the data source. """ return (self.range == util.FXDateRange)
[docs] @classmethod def from_instances(cls, *t_coords): """Create new instance from "union" of attributes of one or more *t_coords*. """ if not t_coords: raise ValueError() t_coords = [util.coerce_to_dataclass(t, cls) for t in t_coords] t0 = t_coords.pop(0) if any(t != t0 for t in t_coords): raise ValueError("mismatch") return t0
[docs] @util.mdtf_dataclass class DMTimeCoordinate(DMGenericTimeCoordinate): name: str = util.MANDATORY """Coordinate name.""" # bounds_var: AbstractDMCoordinateBounds # [scalar] value: int or float standard_name: str = 'time' """Coordinate CF standard name.""" units: src.units.Units = util.MANDATORY """Coordinate units (str or :class:`~src.units.Units`).""" axis: str = 'T' """Coordinate axis identifier. Always 'T' for this coordinate.""" calendar: str = "" """CF standard calendar for time data.""" range: util.AbstractDateRange = None """Date range of coordinate.""" frequency: util.AbstractDateFrequency = None """Sampling frequency of coordinate."""
[docs] @classmethod def from_instances(cls, *t_coords): raise NotImplementedError
# Use the "register" method, instead of inheritance, to associate these classes # with their corresponding abstract interfaces, because Python dataclass fields # aren't recognized as implementing an abc.abstractmethod. AbstractDMCoordinate.register(DMCoordinate) AbstractDMCoordinate.register(DMLongitudeCoordinate) AbstractDMCoordinate.register(DMLatitudeCoordinate) AbstractDMCoordinate.register(DMVerticalCoordinate) AbstractDMCoordinate.register(DMParametricVerticalCoordinate) AbstractDMCoordinate.register(DMGenericTimeCoordinate) AbstractDMCoordinate.register(DMTimeCoordinate) AbstractDMCoordinate.register(DMBoundsDimension)
[docs] def coordinate_from_struct(d, class_dict=None, **kwargs): """Attempt to instantiate the correct :class:`DMCoordinate` class based on information in dict *d* (read from JSON file). *class_dict* is an optional dict mapping axes identifiers to the :class:`DMCoordinate` child classes to instantiate. Default is to use :class:`DMLongitudeCoordinate` for 'X', etc. TODO: implement full cf_xarray/MetPy heuristics. """ if class_dict is None: class_dict = { 'X': DMLongitudeCoordinate, 'Y': DMLatitudeCoordinate, 'Z': DMVerticalCoordinate, 'T': DMGenericTimeCoordinate, 'OTHER': DMCoordinate } standard_names = { 'longitude': 'X', 'latitude': 'Y', 'time': 'T' } try: ax = 'OTHER' if 'axis' in d: ax = d['axis'] else: # try to match an axis value (X,Y,T) to the dimension. The standard name of the dimension specified # in the input json file must contain the word "longitude", "latitude", or "time" for k in standard_names.keys(): if k in d.get('standard_name', ""): ax = standard_names[k] return util.coerce_to_dataclass(d, class_dict[ax], **kwargs) except Exception: raise ValueError(f"Couldn't parse coordinate: {repr(d)}")
class _DMPlaceholderCoordinateBase(object): """Dummy base class for placeholder coordinates. Placeholder coordinates are only used in instantiating :class:`~src.core.FieldlistEntry` objects: they're replaced by the appropriate translated coordinates when that object is used to create a :class:`~src.core.TranslatedVarlistEntry` object. """ pass
[docs] @util.mdtf_dataclass class DMPlaceholderCoordinate(_DMCoordinateShared, _DMPlaceholderCoordinateBase): """Dummy base class for placeholder coordinates. Placeholder coordinates are only used in instantiating :class:`~src.core.FieldlistEntry` objects: they're replaced by the appropriate translated coordinates when that object is used to create a :class:`~src.core.TranslatedVarlistEntry` object. """ name: str = 'PLACEHOLDER_COORD' """Coordinate name; defaults to 'PLACEHOLDER_COORD' since this is a temporary object.""" standard_name: str = NotImplemented """Coordinate CF standard name.""" units: src.units.Units = NotImplemented """Coordinate units (str or :class:`~src.units.Units`).""" axis: str = 'OTHER' """Coordinate axis identifier ('X', 'Y', etc.)"""
[docs] @util.mdtf_dataclass class DMPlaceholderXCoordinate(_DMCoordinateShared, _DMPlaceholderCoordinateBase): """Dummy base class for placeholder X axis coordinates. Placeholder coordinates are only used in instantiating :class:`~src.core.FieldlistEntry` objects: they're replaced by the appropriate translated coordinates when that object is used to create a :class:`~src.core.TranslatedVarlistEntry` object. """ name: str = 'PLACEHOLDER_X_COORD' """Coordinate name; defaults to 'PLACEHOLDER_X_COORD' since this is a temporary object.""" standard_name: str = NotImplemented """Coordinate CF standard name.""" units: src.units.Units = NotImplemented """Coordinate units (str or :class:`~src.units.Units`).""" axis: str = 'X' """Coordinate axis identifier ('X', 'Y', etc.)"""
[docs] @util.mdtf_dataclass class DMPlaceholderYCoordinate(_DMCoordinateShared, _DMPlaceholderCoordinateBase): """Dummy base class for placeholder Y axis coordinates. Placeholder coordinates are only used in instantiating :class:`~src.core.FieldlistEntry` objects: they're replaced by the appropriate translated coordinates when that object is used to create a :class:`~src.core.TranslatedVarlistEntry` object. """ name: str = 'PLACEHOLDER_Y_COORD' """Coordinate name; defaults to 'PLACEHOLDER_Y_COORD' since this is a temporary object.""" standard_name: str = NotImplemented """Coordinate CF standard name.""" units: src.units.Units = NotImplemented """Coordinate units (str or :class:`~src.units.Units`).""" axis: str = 'Y' """Coordinate axis identifier ('X', 'Y', etc.)"""
[docs] @util.mdtf_dataclass class DMPlaceholderZCoordinate(_DMCoordinateShared, _DMPlaceholderCoordinateBase): """Dummy base class for placeholder Z axis coordinates. Placeholder coordinates are only used in instantiating :class:`~src.core.FieldlistEntry` objects: they're replaced by the appropriate translated coordinates when that object is used to create a :class:`~src.core.TranslatedVarlistEntry` object. """ name: str = 'PLACEHOLDER_Z_COORD' """Coordinate name; defaults to 'PLACEHOLDER_Z_COORD' since this is a temporary object.""" standard_name: str = NotImplemented """Coordinate CF standard name.""" units: src.units.Units = NotImplemented """Coordinate units (str or :class:`~src.units.Units`).""" axis: str = 'Z' """Coordinate axis identifier ('X', 'Y', etc.)""" positive: str = NotImplemented
[docs] @util.mdtf_dataclass class DMPlaceholderTCoordinate(_DMCoordinateShared, _DMPlaceholderCoordinateBase): """Dummy base class for placeholder T axis coordinates. Placeholder coordinates are only used in instantiating :class:`~src.core.FieldlistEntry` objects: they're replaced by the appropriate translated coordinates when that object is used to create a :class:`~src.core.TranslatedVarlistEntry` object. """ name: str = 'PLACEHOLDER_T_COORD' """Coordinate name; defaults to 'PLACEHOLDER_T_COORD' since this is a temporary object.""" standard_name: str = NotImplemented """Coordinate CF standard name.""" units: src.units.Units = NotImplemented """Coordinate units (str or :class:`~src.units.Units`).""" axis: str = 'T' """Coordinate axis identifier ('X', 'Y', etc.)""" calendar: str = NotImplemented """CF standard calendar for time data.""" range: typing.Any = None """Date range of coordinate.""" @property def is_static(self): """Check for time-independent data ('fx' in CMIP6 DRS.) Do the comparison by checking date_range against the placeholder value because that's unique -- we may be using a different DateFrequency depending on the data source. """ return (self.range == util.FXDateRange)
# ------------------------------------------------------------------------------ @util.mdtf_dataclass class _DMDimensionsMixin(object): """Lookups for the dimensions, and associated dimension coordinates, associated with an array (eg a variable or auxiliary coordinate.) Needs to be included as a parent class of a dataclass. """ coords: dc.InitVar = None dims: list = dc.field(init=False, default_factory=list) scalar_coords: list = dc.field(init=False, default_factory=list) def __post_init__(self, coords=None): if coords is None: # if we're called to rebuild dicts, rather than after __init__ assert (self.dims or self.scalar_coords) coords = self.dims + self.scalar_coords self.dims = [] self.scalar_coords = [] for c in coords: if c.is_scalar: self.scalar_coords.append(c) else: self.dims.append(c) # raises exceptions if axes are inconsistent _ = self.build_axes(self.dims, verify=True) @property def dim_axes(self): """Retrun dict mapping axes labels ('X', 'Y', etc.) to corresponding dimension coordinate objects. """ return self.build_axes(self.dims, verify=False) @property def X(self): """Return X axis dimension coordinate if defined, else None.""" return self.dim_axes.get('X', None) @property def Y(self): """Return Y axis dimension coordinate if defined, else None.""" return self.dim_axes.get('Y', None) @property def Z(self): """Return Z axis dimension coordinate if defined, else None.""" return self.dim_axes.get('Z', None) @property def T(self): """Return T axis dimension coordinate if defined, else None.""" return self.dim_axes.get('T', None) @property def dim_axes_set(self): """Return frozenset of dimension coordinate axes labels.""" return frozenset(self.dim_axes.keys()) @property def is_static(self): """Whether the variable has time dependence (bool).""" return (self.T is None) or (self.T.is_static) def get_scalar(self, ax_name): """If the axis label *ax_name* is a scalar coordinate, return the corresponding :class:`AbstractDMCoordinate` object, otherwise return None. """ for c in self.scalar_coords: if c.axis == ax_name: return c return None def build_axes(self, *coords, verify=True): """Constructs a dict mapping axes labels to dimension coordinates (of type :class:`AbstractDMCoordinate`.) """ if verify: # validate that we don't have duplicate axes d = util.WormDict() verify_d = util.WormDict() for c in itertools.chain(*coords): if c.axis != 'OTHER' and c.axis in verify_d: err_name = getattr(self, 'name', self.__class__.__name__) raise ValueError((f"Duplicate definition of {c.axis} axis in " f"{err_name}: {c}, {verify_d[c.axis]}")) verify_d[c.axis] = c if c.axis in _AXIS_NAMES: d[c.axis] = c return d else: # assume we've already verified, so use a quicker version of same logic return {c.axis: c for c in itertools.chain(*coords) \ if c.axis in _AXIS_NAMES} def change_coord(self, ax_name, new_class=None, **kwargs): """Replace attributes on a given coordinate, but also optionally cast them to new classes. Args: ax_name: Name of the coordinate to modify. new_class (optional): new class to cast the returned coordinate to. kwargs: Set of attribute names and values to replace on the returned copy. """ # TODO: lookup by non-axis name old_coord = getattr(self, ax_name, None) if not old_coord: if ax_name == 'T' and not self.is_static: raise KeyError(f"{self.name} has no {ax_name} axis") if isinstance(new_class, dict): new_coord_class = new_class.pop('self', None) else: new_coord_class = new_class if new_coord_class is None and not isinstance(new_class, dict): # keep all classes new_coord = dc.replace(old_coord, **kwargs) else: if new_coord_class is None: new_coord_class = old_coord.__class__ new_kwargs = dc.asdict(old_coord) else: new_kwargs = util.filter_dataclass(old_coord, new_coord_class) new_kwargs.update(kwargs) if isinstance(new_class, dict): for k, cls_ in new_class.items(): if k in new_kwargs and not isinstance(new_kwargs[k], cls_): new_kwargs[k] = cls_(new_kwargs[k]) new_coord = new_coord_class(**new_kwargs) self.dims[self.dims.index(old_coord)] = new_coord self.__post_init__(None) # rebuild axes dicts
[docs] @util.mdtf_dataclass class DMDependentVariable(_DMDimensionsMixin): """Base class for any "dependent variable": all non-dimension-coordinate information that depends on one or more dimension coordinates. """ name: str = util.MANDATORY standard_name: str = util.MANDATORY units: src.units.Units = "" # not MANDATORY since may be set later from var translation modifier: str = "" component: str = "" associated_files: str = "" rename_coords: bool = True # dims: from _DMDimensionsMixin # scalar_coords: from _DMDimensionsMixin def __post_init__(self, coords=None): """Call :meth:`build_axes` and deal with CF modifier, if any. """ super(DMDependentVariable, self).__post_init__(coords) # raises exceptions if axes are inconsistent _ = self.build_axes(self.dims, self.scalar_coords, verify=True) # if specified, verify that POD modifier attributes are valid if not self.modifier.lower().strip() in (None, ''): _str = src.core.VariableTranslator() if self.modifier not in _str.modifier: raise ValueError(f"Modifier {self.modifier} is not a recognized value.") @property def full_name(self): """Object's full name, to be used in logging and debugging. Preferred because it eliminates irrelevant information in repr(), which is lengthy. """ return '<' + self.name + '>'# synonym here; child classes override def __str__(self): """Condensed string representation. """ if hasattr(self, 'name_in_model') and self.name_in_model: str_ = f"='{self.name_in_model}'" else: str_ = f"{self.standard_name}" attrs_ = [] if not self.is_static and hasattr(self.T, 'frequency'): attrs_.append(str(self.T.frequency)) if self.get_scalar('Z'): lev = self.get_scalar('Z') attrs_.append(f"{lev.value} {lev.units}") if attrs_: str_ += " @ " str_ += ", ".join(attrs_) return f"{self.full_name} ({str_})" @property def axes(self): """Superset of the :meth:`dim_axes` dict (whose values contain coordinate dimensions only) that includes axes corresponding to scalar coordinates. """ return self.build_axes(self.dims, self.scalar_coords, verify=False) @property def axes_set(self): """Superset of the :meth:`dim_axes_set` frozenset (which contains axes labels corresponding to coordinate dimensions only) that includes axes labels corresponding to scalar coordinates. """ return frozenset(self.axes.keys())
[docs] def add_scalar(self, ax, ax_value, **kwargs): """Metadata operation corresponding to taking a slice of a higher-dimensional variable (extracting its values at axis *ax* = *ax_value*). The coordinate corresponding to *ax* is removed from the list of coordinate dimensions and added to the list of scalar coordinates. """ assert ax in self.dim_axes dim = self.dim_axes[ax] new_dim = dc.replace(dim, value=ax_value) new_dims = self.dims.copy() new_dims.remove(dim) new_scalars = self.scalar_coords.copy() new_scalars.add(new_dim) return dc.replace( self, coords=(new_dims + new_scalars), **kwargs )
[docs] def remove_scalar(self, ax, position=-1, **kwargs): """Metadata operation that's the inverse of :meth:`add_scalar`. Given an axis label *ax* that's currently a scalar coordinate, remove the slice value and add it to the list of dimension coordinates at *position* (default end of the list.) """ dim = self.get_scalar(ax) assert dim is not None new_dim = dc.replace(dim, value=None) new_dims = self.dims.copy() new_dims.insert(position, new_dim) new_scalars = self.scalar_coords.copy() new_scalars.remove(dim) return dc.replace( self, coords=(new_dims + new_scalars), **kwargs )
[docs] @util.mdtf_dataclass class DMAuxiliaryCoordinate(DMDependentVariable): """Class to describe `auxiliary coordinate variables <http://cfconventions.org/Data/cf-conventions/cf-conventions-1.8/cf-conventions.html#terminology>`__, as defined in the CF conventions. An example would be lat or lon for data presented in a tripolar grid projection. """ pass
[docs] @util.mdtf_dataclass class DMCoordinateBounds(DMAuxiliaryCoordinate): """Class describing bounds on a dimension coordinate. """ def __post_init__(self, coords=None): super(DMCoordinateBounds, self).__post_init__(coords) # validate dimensions if self.scalar_coords: raise ValueError(("Attempted to create DMCoordinateBounds " f"{self.name} with scalar coordinates: {self.scalar_coords}.")) if len(self.dims) != 2 or \ 'BOUNDS' not in {c.axis for c in self.dims}: raise ValueError(("Attempted to create DMCoordinateBounds " f"{self.name} with improper dimensions: {self.dims}.")) @property def coord(self): """CF dimension coordinate for which this is the bounds. """ for c in self.dims: if c.axis != 'BOUNDS': return c raise ValueError()
[docs] @classmethod def from_coordinate(cls, coord, bounds_dim): """Create instance from a coordinate object *coord*. """ kwargs = {attr: getattr(coord, attr) for attr \ in ('name', 'standard_name', 'units')} if not isinstance(bounds_dim, DMBoundsDimension): bounds_dim = DMBoundsDimension(name=bounds_dim) kwargs['coords'] = [coord, bounds_dim] coord_bounds = cls(**kwargs) coord.bounds_var = coord_bounds return coord_bounds
[docs] @util.mdtf_dataclass class DMVariable(DMDependentVariable): """Class to describe general properties of data variables. """ # name: str # fields inherited from DMDependentVariable # standard_name: str # modifier: str # units: src.units.Units # dims: list # fields inherited from _DMDimensionsMixin # scalar_coords: list pass
# Use the "register" method, instead of inheritance, to associate these classes # with their corresponding abstract interfaces, because Python dataclass fields # aren't recognized as implementing an abc.abstractmethod. AbstractDMDependentVariable.register(DMDependentVariable) AbstractDMDependentVariable.register(DMAuxiliaryCoordinate) AbstractDMDependentVariable.register(DMVariable) AbstractDMCoordinateBounds.register(DMCoordinateBounds)
[docs] @util.mdtf_dataclass class DMDataSet(_DMDimensionsMixin): """Class to describe a collection of one or more variables sharing a set of common dimensions. """ contents: dc.InitVar = util.MANDATORY """All members of the collection (input).""" vars: list = dc.field(init=False, default_factory=list) """List of dependent variables in the collection.""" coord_bounds: list = dc.field(init=False, default_factory=list) """List of bounds coordinates referenced by variables in the collection.""" aux_coords: list = dc.field(init=False, default_factory=list) """List of `auxiliary coordinates <https://cfconventions.org/cf-conventions/cf-conventions.html#data-model-coordinates>`__ referenced by variables in the collection. """ def __post_init__(self, coords=None, contents=None): """Populate shared ``vars``, ``coord_bounds``, ``aux_coords`` attributes from collection of input variables. """ assert coords is None # shouldn't be called with bare coordinates if contents is None: # if we're called to rebuild dicts, rather than after __init__ assert (self.vars or self.coord_bounds or self.aux_coords) contents = self.vars + self.coord_bounds + self.aux_coords self.vars = [] self.coord_bounds = [] self.aux_coords = [] for v in contents: self._classify(v).append(v) # dims, scalar_coords are a union of those in contents # axes must all be the same, except for time axis, which gets described # by a DMGenericTimeCoordinate t_axes = [] coords = [] for v in contents: v_dims = v.dims.copy() if not v.is_static: t_axes.append(v_dims.pop(v_dims.index(v.T))) for c in itertools.chain(v_dims, v.scalar_coords): if c not in coords: coords.append(c) if t_axes: new_t = DMGenericTimeCoordinate.from_instances(*t_axes) coords.append(new_t) # can't have duplicate dims, but duplicate scalar_coords are OK. super(DMDataSet, self).__post_init__(coords)
[docs] def iter_contents(self): """Generator iterating over the full contents of the DataSet (variables, auxiliary coordinates and coordinate bounds.) """ yield from itertools.chain(self.vars, self.aux_coords, self.coord_bounds)
[docs] def iter_vars(self): """Generator iterating over variables and auxiliary coordinates but excluding coordinate bounds. """ yield from itertools.chain(self.vars, self.aux_coords)
def _classify(self, v): assert isinstance(v, DMDependentVariable) if isinstance(v, DMVariable): return self.vars elif isinstance(v, DMCoordinateBounds): return self.coord_bounds else: return self.aux_coords
[docs] def add_contents(self, *vars_): raise NotImplementedError()
[docs] def change_coord(self, ax_name, new_class=None, **kwargs): """Replace attributes on a given coordinate (*ax_name*), but also optionally cast them to new classes. Args: ax_name: Name of the coordinate to modify. new_class (optional): new class to cast the returned coordinate to. kwargs: Set of attribute names and values to replace on the returned copy. """ for v in self.iter_contents(): try: v.change_coord(ax_name, new_class, **kwargs) except ValueError: if v.is_static: continue else: raise # time coord for self derived from those for contents self.__post_init__(None, None)