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

import collections

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

LOG = logging.get_logger(__name__)


[docs]class BpSolintQAHandler(pqa.QAPlugin): result_cls = bpsolint.BpSolintResults child_cls = None generating_task = bpsolint.BpSolint
[docs] def handle(self, context, result): vis = result.inputs['vis'] minphaseupints = result.inputs['minphaseupints'] minbpnchans = result.inputs['minbpnchan'] 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_phaseup_snrs(ms, result.spwids, result.phsolints) score2 = self._missing_bandpass_snrs (ms, result.spwids, result.bpsolints) score3 = self._poor_phaseup_solutions(ms, result.spwids, result.nphsolutions, minphaseupints) score4 = self._poor_bandpass_solutions(ms, result.spwids, result.nbpsolutions, minbpnchans) scores = [score1, score2, score3, score4] result.qa.pool.extend(scores)
def _missing_phaseup_snrs(self, ms, spwids, phsolints): """ Check whether there are missing phaseup solints. """ return qacalc.score_missing_phaseup_snrs(ms, spwids, phsolints) def _missing_bandpass_snrs(self, ms, spwids, bpsolints): """ Check whether there are missing bandpass solints. """ return qacalc.score_missing_bandpass_snrs(ms, spwids, bpsolints) def _poor_phaseup_solutions(self, ms, spwids, nphsolutions, min_nsolutions): """ Check whether there are phaseup solutiosn with fewer than the minimum number of solutions """ return qacalc.score_poor_phaseup_solutions(ms, spwids, nphsolutions, min_nsolutions) def _poor_bandpass_solutions(self, ms, spwids, nbpsolutions, min_nsolutions): """ Check whether there are bandpass solutions with fewer than the minimum number of solutions """ return qacalc.score_poor_bandpass_solutions(ms, spwids, nbpsolutions, min_nsolutions)
[docs]class BpSolintListQAHandler(pqa.QAPlugin): """ QA handler for a list containing BpSolintResults. """ result_cls = collections.Iterable child_cls = bpsolint.BpSolintResults generating_task = bpsolint.BpSolint
[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