Source code for pipeline.hifv.tasks.importdata.qa

import collections

import pipeline.infrastructure.logging as logging
import pipeline.infrastructure.pipelineqa as pqa
import pipeline.qa.scorecalculator as qacalc
import pipeline.infrastructure.utils as utils
from pipeline.h.tasks.importdata import qa as hqa
from . import importdata

LOG = logging.get_logger(__name__)


[docs]class VLAImportDataQAHandler(hqa.ImportDataQAHandler, pqa.QAPlugin): result_cls = importdata.VLAImportDataResults child_cls = None generating_task = importdata.VLAImportData
[docs] def handle(self, context, result): score1 = self._check_intents(result.mses) score2 = self._check_history_column(result.mses, result.inputs) scores = [score1, score2] result.qa.pool.extend(scores)
@staticmethod def _check_history_column(mses, inputs): """ Check whether any of the measurement sets has entries in the history column, potentially signifying a non-pristine data set. """ qascore = hqa.ImportDataQAHandler._check_history_column(mses, inputs) if qascore.score < 1.0: qascore.longmsg += " If hanning smooth has already been applied in previous execution, invoking hifv_hanning in a following stage means smoothing twice." return qascore def _check_intents(self, mses): """ Check each measurement set in the list for a set of required intents. TODO Should we terminate execution on missing intents? """ return qacalc.score_missing_intents(mses, array_type='VLA')
[docs]class VLAImportDataListQAHandler(pqa.QAPlugin): result_cls = collections.Iterable child_cls = importdata.VLAImportDataResults
[docs] def handle(self, context, result): # collate the QAScores from each child result, pulling them into our # own QAscore list collated = utils.flatten([r.qa.pool[:] for r in result]) result.qa.pool.extend(collated)