Source code for pipeline.h.tasks.common.displays.tsys

import collections
import os

import pipeline.infrastructure as infrastructure
import pipeline.infrastructure.utils as utils
import pipeline.infrastructure.renderer.logger as logger
from . import common
from pipeline.infrastructure import casa_tasks

LOG = infrastructure.get_logger(__name__)


[docs]class TsysSummaryChart(object): def __init__(self, context, result, calapp, xaxis='freq', yaxis='tsys'): self._context = context self._result = result self._calapp = calapp self._caltable = calapp.gaintable self._vis = calapp.vis self._xaxis = xaxis self._yaxis = yaxis ms = context.observing_run.get_ms(self._vis) # Set showfdm to True except for "NRO" array. self._showfdm = ms.antenna_array.name != 'NRO' # Get science spws from MS. science_spw_ids = [spw.id for spw in ms.get_spectral_windows(science_windows_only=True)] # Get list of Tsys spws in caltable. wrapper = common.CaltableWrapperFactory.from_caltable(self._caltable) tsys_in_caltable = set(wrapper.spw) # Create a mapping between Tsys and science spws. Sometimes, not all # science spws have a matching Tsys window. self._tsysmap = collections.defaultdict(list) for spw, tsys_spw in enumerate(calapp.spwmap): if spw in science_spw_ids and tsys_spw in tsys_in_caltable: self._tsysmap[tsys_spw].append(spw) # Get base name of figure file(s). self._figfile = self._get_figfile() # Get mapping from Tsys spw to receiver type. self._rxmap = utils.get_receiver_type_for_spws(ms, list(self._tsysmap.keys()))
[docs] def plot(self): plots = [] # Create plot for each Tsys spw. for tsys_spw in self._tsysmap: plots.append(self._get_plot_wrapper(tsys_spw)) for p in plots: if not p: LOG.info("Tsys summary plot wrappers not generated") elif not os.path.exists(p.abspath): LOG.info("Tsys summary plot not generated for {} spw {}" "".format(p.parameters['vis'], p.parameters['tsys_spw'])) return [p for p in plots if p and os.path.exists(p.abspath)]
def _get_figfile(self): return os.path.join(self._context.report_dir, 'stage%s' % self._result.stage_number, 'tsys-%s-summary.png' % os.path.basename(self._vis)) def _create_task(self, spw_arg, showimage=False): task_args = {'vis': self._vis, 'caltable': self._caltable, 'xaxis': self._xaxis, 'yaxis': self._yaxis, 'interactive': False, 'spw': str(spw_arg), 'subplot': 11, 'figfile': self._figfile, 'overlay': 'antenna,time', 'showatm': True, 'showfdm': self._showfdm, 'chanrange': '90%', # CAS-7011 'showimage': showimage, } return casa_tasks.plotbandpass(**task_args) def _get_plot_wrapper(self, tsys_spw): # PIPE-110: Use showimage=True for DSB receiver spws. showimage = self._rxmap.get(tsys_spw, "") == 'DSB' task = self._create_task(tsys_spw, showimage=showimage) # Get prediction of final name of figure files(s), assuming # plotbandpass injects spw ID into every plot filename. root, ext = os.path.splitext(self._figfile) pb_figfile = '%s.spw%0.2d%s' % (root, tsys_spw, ext) if not os.path.exists(pb_figfile): LOG.trace("Creating new plot: {}".format(pb_figfile)) try: task.execute(dry_run=False) except Exception as ex: LOG.error("Could not create plot {}".format(pb_figfile)) LOG.exception(ex) return None parameters = {'vis': os.path.basename(self._vis), 'spw': self._tsysmap[tsys_spw], 'tsys_spw': tsys_spw} wrapper = logger.Plot(pb_figfile, x_axis=self._xaxis, y_axis=self._yaxis, parameters=parameters, command=str(task)) return wrapper
[docs]class TsysPerAntennaChart(common.PlotbandpassDetailBase): def __init__(self, context, result, **kwargs): super(TsysPerAntennaChart, self).__init__(context, result, 'freq', 'tsys', overlay='time', showatm=True, showfdm=True, chanrange='90%', **kwargs) # Get MS and bandpass solution. ms = context.observing_run.get_ms(self._vis) calapp = result.final[0] # Set showfdm to True except for "NRO" array. if ms.antenna_array.name == 'NRO': self._kwargs['showfdm'] = False # Get science spws from MS. science_spw_ids = [spw.id for spw in ms.get_spectral_windows(science_windows_only=True)] # Get list of Tsys spws in caltable. wrapper = common.CaltableWrapperFactory.from_caltable(self._caltable) tsys_in_caltable = set(wrapper.spw) # Create a mapping between Tsys and science spws. Sometimes, not all # science spws have a matching Tsys window. self._tsysmap = collections.defaultdict(list) for spw, tsys_spw in enumerate(calapp.spwmap): if spw in science_spw_ids and tsys_spw in tsys_in_caltable: self._tsysmap[tsys_spw].append(spw) # Get mapping from Tsys spw to receiver type. self._rxmap = utils.get_receiver_type_for_spws(ms, list(self._tsysmap.keys()))
[docs] def plot(self): # PIPE-110: create separate calls to plotbandpass for DSB and non-DSB # receivers. missing_dsb = [(spw_id, ant_id) for spw_id in self._figfile for ant_id in self._antmap if not os.path.exists(self._figfile[spw_id][ant_id]) and self._rxmap.get(spw_id, "") == "DSB"] if missing_dsb: self._create_plotbandpass_task(missing_dsb, showimage=True) missing_nondsb = [(spw_id, ant_id) for spw_id in self._figfile for ant_id in self._antmap if not os.path.exists(self._figfile[spw_id][ant_id]) and self._rxmap.get(spw_id, "") != "DSB"] if missing_nondsb: self._create_plotbandpass_task(missing_nondsb, showimage=False) # Create plot wrappers. wrappers = [] for tsys_spw_id in self._figfile: # PIPE-110: show image sideband for DSB receivers. showimage = self._rxmap.get(tsys_spw_id, "") == "DSB" # some science windows may not have a Tsys window science_spws = self._tsysmap.get(tsys_spw_id, 'N/A') for antenna_id, figfile in self._figfile[tsys_spw_id].items(): ant_name = self._antmap[antenna_id] if os.path.exists(figfile): task = self.create_task(tsys_spw_id, antenna_id, showimage=showimage) wrapper = logger.Plot(figfile, x_axis=self._xaxis, y_axis=self._yaxis, parameters={'vis': self._vis_basename, 'ant': ant_name, 'spw': ','.join([str(i) for i in science_spws]), 'tsys_spw': tsys_spw_id}, command=str(task)) wrappers.append(wrapper) else: LOG.trace('No plotbandpass detail plot found for %s spw ' '%s antenna %s: %s not found', self._vis_basename, tsys_spw_id, ant_name, figfile) return wrappers
def _create_plotbandpass_task(self, missing, showimage=False): LOG.trace('Executing new plotbandpass job for missing figures') spw_ids = ','.join({str(spw_id) for spw_id, _ in missing}) ant_ids = ','.join({str(ant_id) for _, ant_id in missing}) try: task = self.create_task(spw_ids, ant_ids, showimage=showimage) task.execute(dry_run=False) except Exception as ex: LOG.error('Could not create plotbandpass details plots') LOG.exception(ex) return None