Source code for pipeline.h.tasks.common.flaggableviewresults

import collections
import copy

import pipeline.infrastructure as infrastructure

LOG = infrastructure.get_logger(__name__)


[docs]class FlaggableViewResults(object): def __init__(self): """ Construct and return a new FlaggableViewResults. """ # views and associated results self.flagging = [] # following are used instead of standard dictionaries so that # missing keys are created automatically as needed self.view = collections.defaultdict(list)
[docs] def addview(self, description, viewresult): self.view[description].append(viewresult)
[docs] def addflags(self, flags): self.flagging += flags
[docs] def add_flag_reason_plane(self, flag_reason_plane, flag_reason_key): for description in self.descriptions(): self.view[description][-1].flag_reason_plane = \ flag_reason_plane[description] self.view[description][-1].flag_reason_key = \ flag_reason_key
[docs] def descriptions(self): return list(self.view.keys())
[docs] def first(self, description): return copy.deepcopy(self.view[description][0])
[docs] def flagcmds(self): return copy.deepcopy(self.flagging)
[docs] def flagged(self): return len(self.flagging) > 0
[docs] def last(self, description): return copy.deepcopy(self.view[description][-1])
[docs] def importfrom(self, result): if isinstance(result, FlaggableViewResults): # copy over the views for description in result.descriptions(): for view in result.view[description]: self.addview(description, view) # copy over the flagging self.flagging += result.flagging # copy over table if present try: self.table = result.table except AttributeError: pass
[docs] def sort_flagcmds(self): # Try to sort the flagging commands by antenna and spw. try: self.flagging.sort(key=lambda k: k.antenna) except: pass try: self.flagging.sort(key=lambda k: k.spw) except: pass