Source code for lsst.validate.base.job

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

from .jsonmixin import JsonSerializationMixin
from .blob import BlobBase, DeserializedBlob
from .measurement import MeasurementBase, DeserializedMeasurement

__all__ = ['Job']

[docs]class Job(JsonSerializationMixin): """A `Job` wraps all measurements and blob metadata associated with a validation run. Measurements and blobs can be added both during initialization, or anytime afterwards with the `register_measurement` and `register_blob` methods. The `get_measurement` method lets you search for a stored measurement based on the `Metric` name (and filter or specification name, if the measurement is dependent on those). Use the `Job.json` attribute to access a json-serializable `dict` of all measurements and blobs associated with the `Job`. A `Job` can only contain measurements against one dataset at a time. Typically, `Job`\ s are uploaded to SQUASH separately for each tested dataset. Parameters ---------- measurements : `list`, optional List of `MeasurementBase`-derived objects. Additional measurements can be added with the `register_measurement` method. blobs : `list`, optional List of `BlobBase`-derived objects. Additional blobs can be added with the `register_blob` method. """ def __init__(self, measurements=None, blobs=None): self._measurements = [] self._measurement_ids = set() self._blobs = [] self._blob_ids = set() if measurements: for m in measurements: self.register_measurement(m) if blobs: for b in blobs: self.register_blob(b)
[docs] def register_measurement(self, m): """Add a measurement object to the `Job`. Registering a measurement also automatically registers all linked blobs. Parameters ---------- m : `MeasurementBase`-type object A measurement object. """ assert isinstance(m, MeasurementBase) if m.identifier not in self._measurement_ids: self._measurements.append(m) self._measurement_ids.add(m.identifier) for name, b in m.blobs.items(): self.register_blob(b)
@property def measurements(self): """Measurement iterator.""" for m in self._measurements: yield m
[docs] def get_measurement(self, metric_name, spec_name=None, filter_name=None): """Get a measurement corresponding to the given criteria. Parameters ---------- metric_name : `str` Name of the `Metric` for the requested measurement. spec_name : `str`, optional Name of the specification level if the measurement algorithm is dependent on the specification level of a metric. filter_name : `str`, optional Name of the optical filter if the measurement is specific to a filter. Returns ------- measurement : `MeasurementBase`-type object The single measurement instance that fulfills the search criteria. Raises ------ RuntimeError Raised when a measurement cannot be found, either because no such measurement exists or because the request is ambiguous (``spec_name`` or ``filter_name`` need to be set). """ candidates = [m for m in self._measurements if m.label == metric_name] if len(candidates) == 1: candidate = candidates[0] if spec_name is not None and candidate.spec_name is not None: assert candidate.spec_name == spec_name if filter_name is not None and candidate.filter_name is not None: assert candidate.filter_name == filter_name return candidate # Filter by spec_name if spec_name is not None: candidates = [m for m in candidates if m.spec_name == spec_name] if len(candidates) == 1: candidate = candidates[0] if filter_name is not None and candidate.filter_name is not None: assert candidate.filter_name == filter_name return candidate # Filter by filter_name if filter_name is not None: candidates = [m for m in candidates if m.filter_name == filter_name] if len(candidates) == 1: return candidates[0] raise RuntimeError('Measurement not found', metric_name, spec_name)
[docs] def register_blob(self, b): """Add a blob object to the `Job`. Parameters ---------- b : `BlobBase`-type object A blob object. """ assert isinstance(b, BlobBase) if b.identifier not in self._blob_ids: self._blobs.append(b) self._blob_ids.add(b.identifier)
@property def blobs(self): """Blob iterator.""" for b in self._blobs: yield b @classmethod
[docs] def from_json(cls, json_data): """Construct a Job and constituent objects from a JSON dataset. Parameters ---------- json_data : `dict` Job JSON object (as produced by `json`). Returns ------- job : `Job`-type Job from JSON. """ blobs = [DeserializedBlob.from_json(doc) for doc in json_data['blobs']] measurements = [ DeserializedMeasurement.from_json(doc, blobs_json=json_data['blobs']) for doc in json_data['measurements']] job = cls(measurements=measurements, blobs=blobs) return job
@property def json(self): """`Job` data as a JSON-serialiable `dict`.""" doc = JsonSerializationMixin.jsonify_dict({ 'measurements': self._measurements, 'blobs': self._blobs}) return doc @property def metric_names(self): """Names of `Metric`\ s measured in this `Job` (`list`).""" metric_names = [] for m in self._measurements: if m.value is not None: if not in metric_names: metric_names.append( return metric_names @property def spec_levels(self): """`list` of names of specification levels that are available for `Metric`\ s measured in this `Job`. """ spec_names = [] for m in self._measurements: for spec in m.metric.specs: if not in spec_names: spec_names.append( return spec_names