Source code for pipeline.hifa.tasks.gaincalsnr.qa

import collections

import pipeline.infrastructure.pipelineqa as pqa
import pipeline.infrastructure.utils as utils
import pipeline.qa.scorecalculator as qacalc
from . import gaincalsnr

# LOG = logging.get_logger(__name__)


[docs]class GaincalSnrQAHandler(pqa.QAPlugin): result_cls = gaincalsnr.GaincalSnrResults child_cls = None generating_task = gaincalsnr.GaincalSnr
[docs] def handle(self, context, result): vis = result.inputs['vis'] phasesnr = result.inputs['phasesnr'] ms = context.observing_run.get_ms(vis) # Check for existance of spws combinations for which # SNR estimates are missing. ms argument not really # needed for this but include for the moment. score1 = self._missing_phase_snrs(ms, result.spwids, result.snrs) score2 = self._poor_phase_snrs(ms, result.spwids, phasesnr, result.snrs) scores = [score1, score2] result.qa.pool.extend(scores)
def _missing_phase_snrs(self, ms, spwids, snrs): """ Check whether there are missing phase snrs. """ return qacalc.score_missing_phase_snrs(ms, spwids, snrs) def _poor_phase_snrs(self, ms, spwids, phasesnr, snrs): """ Check whether there are poor snrs values """ return qacalc.score_poor_phase_snrs(ms, spwids, phasesnr, snrs)
[docs]class GaincalSnrListQAHandler(pqa.QAPlugin): """ QA handler for a list containing GaincalSnrResults. """ result_cls = collections.Iterable child_cls = gaincalsnr.GaincalSnrResults generating_task = gaincalsnr.GaincalSnr
[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[:] = collated mses = [r.inputs['vis'] for r in result] longmsg = 'No missing derived fluxes in %s' % utils.commafy(mses, quotes=False, conjunction='or') result.qa.all_unity_longmsg = longmsg