Source code for lsst.validate.base.metric

# See COPYRIGHT file at the top of the source tree.
from __future__ import print_function, division

import operator
from collections import OrderedDict
import yaml

from .jsonmixin import JsonSerializationMixin
from .datum import Datum
from .spec import Specification

__all__ = ['Metric', 'load_metrics']

[docs]class Metric(JsonSerializationMixin): """Container for the definition of a metric and its specification levels. Metrics can either be instantiated programatically, or from a :ref:`metric YAML file <validate-base-metric-yaml>` with the `from_yaml` class method. .. seealso:: See the :ref:`validate-base-using-metrics` page for usage details. Parameters ---------- name : `str` Name of the metric (e.g., ``'PA1'``). description : `str` Short description about the metric. operator_str : `str` A string, such as ``'<='``, that defines a success test for a measurement (on the left hand side) against the metric specification level (right hand side). specs : `list`, optional A `list` of `Specification` objects that define various specification levels for this metric. parameters : `dict`, optional A `dict` of named `Datum` values that must be known when measuring a metric. reference_doc : `str`, optional The document handle that originally defined the metric (e.g., ``'LPM-17'``). reference_url : `str`, optional The document's URL. reference_page : `str`, optional Page where metric in defined in the reference document. """ name = None """Name of the metric (`str`).""" description = None """Short description of the metric (`str`).""" reference_doc = None """Name of the document that specifies this metric (`str`).""" reference_url = None """URL of the document that specifies this metric (`str`).""" reference_page = None """Page number in the document that specifies this metric (`int`).""" parameters = dict() """`dict` of named :class:`Datum` values that must be known when measuring a metric. Parameters can also be accessed as attributes of the metric. Attribute names are the same as key names in `parameters`. """ def __init__(self, name, description, operator_str, specs=None, parameters=None, reference_doc=None, reference_url=None, reference_page=None): = name self.description = description self.reference_doc = reference_doc self.reference_url = reference_url self.reference_page = reference_page self.operator_str = operator_str if specs is None: self.specs = [] else: self.specs = specs if parameters is None: self.parameters = {} else: assert isinstance(parameters, dict) for key, value in parameters.items(): assert isinstance(key, basestring) assert isinstance(value, Datum) self.parameters = parameters @classmethod
[docs] def from_yaml(cls, metric_name, yaml_doc=None, yaml_path=None, resolve_dependencies=True): """Create a `Metric` instance from a YAML document that defines metrics. .. seealso:: See :ref:`validate-base-metric-yaml` for details on the metric YAML schema. Parameters ---------- metric_name : `str` Name of the metric (e.g., ``'PA1'``). yaml_doc : `dict`, optional A full metric YAML document loaded as a `dict`. Use this option to increase performance by eliminating redundant reads of a common metric YAML file. Alternatively, set ``yaml_path``. yaml_path : `str`, optional The full file path to a metric YAML file. Alternatively, set ``yaml_doc``. resolve_dependencies : `bool`, optional API users should always set this to `True`. The opposite is used only used internally. Raises ------ RuntimeError Raised when neither ``yaml_doc`` or ``yaml_path`` are set. """ if yaml_doc is None and yaml_path is not None: with open(yaml_path) as f: yaml_doc = yaml.load(f) elif yaml_doc is None and yaml_path is None: raise RuntimeError('Set either yaml_doc or yaml_path argument') metric_doc = yaml_doc[metric_name] metric_params = {} if 'parameters' in metric_doc: for param_name, param_data in metric_doc['parameters'].items(): d = Datum(param_data['value'], param_data.get('unit', ''), label=param_data.get('label', None), description=param_data.get('description', None)) metric_params[param_name] = d m = cls( metric_name, description=metric_doc.get('description', None), operator_str=metric_doc['operator'], reference_doc=metric_doc['reference'].get('doc', None), reference_url=metric_doc['reference'].get('url', None), reference_page=metric_doc['reference'].get('page', None), parameters=metric_params) for spec_doc in metric_doc['specs']: deps = None if 'dependencies' in spec_doc and resolve_dependencies: deps = {} for dep_item in spec_doc['dependencies']: if isinstance(dep_item, basestring): # This is a metric name = dep_item d = Metric.from_yaml(name, yaml_doc=yaml_doc, resolve_dependencies=False) elif isinstance(dep_item, dict): # Likely a Datum # in yaml, wrapper object is dict with single key-val name = dep_item.keys()[0] dep_item = dict(dep_item[name]) v = dep_item['value'] unit = dep_item['unit'] d = Datum(v, unit, label=dep_item.get('label', None), description=dep_item.get('description', None)) else: raise RuntimeError( 'Cannot process dependency %r' % dep_item) deps[name] = d spec = Specification(name=spec_doc['level'], quantity=spec_doc['value'], unit=spec_doc['unit'], filter_names=spec_doc.get('filter_names', None), dependencies=deps) m.specs.append(spec) return m
[docs] def from_json(cls, json_data): """Construct a Metric from a JSON dataset. Parameters ---------- json_data : `dict` Metric JSON object. Returns ------- metric : `Metric` Metric from JSON. """ specs = [Specification.from_json(v) for v in json_data['specifications']] params = {k: Datum.from_json(v) for k, v in json_data['parameters'].items()} m = cls(json_data['name'], json_data['description'], json_data['operator_str'], specs=specs, parameters=params, reference_doc=json_data['reference']['doc'], reference_page=json_data['reference']['page'], reference_url=json_data['reference']['url']) return m
def __getattr__(self, key): if key in self.parameters: return self.parameters[key] else: raise AttributeError("%r object has no attribute %r" % (self.__class__, key)) @property def reference(self): """Documentation reference as human-readable text (`str`, read-only). Uses `reference_doc`, `reference_page`, and `reference_url`, as available. """ ref_str = '' if self.reference_doc and self.reference_page: ref_str = '{doc}, p. {page:d}'.format(doc=self.reference_doc, page=self.reference_page) elif self.reference_doc: ref_str = self.reference_doc if self.reference_url and self.reference_doc: ref_str += ', {url}'.format(url=self.reference_url) elif self.reference_url: ref_str = self.reference_url return ref_str @property def operator_str(self): """String representation of comparison operator. The comparison is oriented with the measurement on the left-hand side and the specification level on the right-hand side. """ return self._operator_str @operator_str.setter def operator_str(self, v): # Cache the operator function as a means of validating the input too self._operator = Metric.convert_operator_str(v) self._operator_str = v @property def operator(self): """Binary comparision operator that tests success of a measurement fulfilling a specification of this metric. Measured value is on left side of comparison and specification level is on right side. """ return self._operator @staticmethod
[docs] def convert_operator_str(op_str): """Convert a string representing a binary comparison operator to the operator function itself. Operators are oriented so that the measurement is on the left-hand side, and specification level on the right hand side. The following operators are permitted: ========== ============= ``op_str`` Function ========== ============= ``>=`` `` ``>`` `` ``<`` `` ``<=`` `operator.le` ``==`` `operator.eq` ``!=`` `` ========== ============= Parameters ---------- op_str : `str` A string representing a binary operator. Returns ------- op_func : obj An operator function from the `operator` standard library module. """ operators = {'>=':, '>':, '<':, '<=': operator.le, '==': operator.eq, '!=':} return operators[op_str]
[docs] def get_spec(self, name, filter_name=None): """Get a specification by name and other qualifications. Parameters ---------- name : `str` Name of a specification level (e.g., ``'design'``, ``'minimum'``, ``'stretch'``). filter_name : `str`, optional The name of the optical filter to qualify a filter-dependent specification level. Returns ------- spec : `Specification` The `Specification` that matches the name and other qualifications. Raises ------ RuntimeError If a specification cannot be found. """ # First collect candidate specifications by name candidates = [s for s in self.specs if == name] if len(candidates) == 1: return candidates[0] # Filter down by optical filter if filter_name is not None: candidates = [s for s in candidates if filter_name in s.filter_names] if len(candidates) == 1: return candidates[0] raise RuntimeError( 'No {2} spec found for name={0} filter_name={1}'.format( name, filter_name,
[docs] def get_spec_dependency(self, spec_name, dep_name, filter_name=None): """Get the `Datum` of a specification's dependency. If the dependency is a metric, this method resolves the value of the dependent metric's specification level ``spec_name``. In other words, ``spec_name`` refers to the specification level of both *this* metric and of the dependency metric. Parameters ---------- spec_name : `str` `Specification` name. dep_name : `str` Name of the dependency. filter_name : `str`, optional Name of the optical filter, if this metric's specifications are optical filter dependent. Returns ------- datum : `Datum` The dependency resolved for the metric's specification. """ spec = self.get_spec(spec_name, filter_name=filter_name) dep = spec.dependencies[dep_name] if isinstance(dep, Metric): dep_spec = dep.get_spec(spec_name, filter_name=filter_name) return dep_spec.datum else: # The dependency should be a straight Datum assert isinstance(dep, Datum) is True return dep
[docs] def get_spec_names(self, filter_name=None): """List names of all specification levels defined for this metric; optionally filtering by attributes such as filter name. Parameters ---------- filter_name : `str`, optional Name of the applicable filter, if needed. Returns ------- spec_names : `list` Specification names as a list of strings, e.g. ``['design', 'minimum', 'stretch']``. """ spec_names = [] for spec in self.specs: if (filter_name is not None) and (spec.filter_names is not None) \ and (filter_name not in spec.filter_names): continue spec_names.append( return list(set(spec_names))
[docs] def check_spec(self, quantity, spec_name, filter_name=None): """Compare a measurement against a named specification level. Parameters ---------- value : `astropy.units.Quantity` The measurement value. spec_name : `str` Name of a `Specification` associated with this metric. filter_name : `str`, optional Name of the applicable filter, if needed. Returns ------- passed : `bool` `True` if the value meets the specification, `False` otherwise. """ spec = self.get_spec(spec_name, filter_name=filter_name) return self.operator(quantity, spec.quantity)
@property def json(self): """`dict` that can be serialized as semantic JSON, compatible with the SQUASH metric service. """ ref_doc = { 'doc': self.reference_doc, 'page': self.reference_page, 'url': self.reference_url} return JsonSerializationMixin.jsonify_dict({ 'name':, 'operator_str': self.operator_str, 'reference': ref_doc, 'description': self.description, 'specifications': self.specs, 'parameters': self.parameters})
[docs]def load_metrics(yaml_path): """Load metric from a YAML document into an ordered dictionary of `Metric`\ s. .. seealso:: :ref:`validate-base-metric-yaml`. Parameters ---------- yaml_path : `str` The full file path to a metric YAML file. Returns ------- metrics : `collections.OrderedDict` A dictionary of `Metric` instances, ordered to matched layout of YAML document at YAML path. Keys are names of metrics (`str`). See also -------- Metric.from_yaml Make a single `Metric` instance from a YAML document. """ with open(yaml_path) as f: metrics_doc = _load_ordered_yaml(f) metrics = [] for key in metrics_doc: metrics.append((key, Metric.from_yaml(key, metrics_doc))) return OrderedDict(metrics)
def _load_ordered_yaml(stream, Loader=yaml.Loader, object_pairs_hook=OrderedDict): """Load a YAML document into an OrderedDict Solution from """ class OrderedLoader(Loader): pass def construct_mapping(loader, node): loader.flatten_mapping(node) return object_pairs_hook(loader.construct_pairs(node)) OrderedLoader.add_constructor( yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, construct_mapping) return yaml.load(stream, OrderedLoader)