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