"""
Functions that Flagging Visibility and Visibility.
The flags of Visibility has axes [chan, pol, z, y, x] where z, y, x
are spatial axes in either sky or Fourier plane. The order in the WCS
is reversed so the grid_WCS describes UU, VV, WW, STOKES, FREQ axes.
"""
__all__ = ["flagging_visibility", "flagging_aoflagger"]
import logging
import os
import pkgutil
import numpy
log = logging.getLogger("rascil-logger")
[docs]
def flagging_visibility(
bvis, baselines=None, antennas=None, channels=None, polarisations=None
):
"""Flagging Visibility
:param bvis: Visibility
:param baselines: The list of baseline numbers to flag
:param antennas: The list of antenna number to flag
:param channels: The list of Channel number to flag
:param polarisations: The list of polarisations to flag
:return: Visibility
"""
if polarisations is None:
polarisations = []
if antennas is None:
antennas = []
if channels is None:
channels = []
if baselines is None:
baselines = []
for baseline in baselines:
bvis["flags"].data[:, baseline, ...] = 1
for channel in channels:
bvis["flags"].data[..., channel, :] = 1
for pol in polarisations:
bvis["flags"].data[..., pol] = 1
for ibaseline, (a1, a2) in enumerate(bvis.baselines.data):
if a1 in antennas or a2 in antennas:
bvis["flags"].data[:, ibaseline, ...] = 1
return bvis
[docs]
def flagging_aoflagger(vis, strategy_name):
"""Flagging Visibility using the AOFlagger package
The flagging strategy can be telescope name (AARTFAAC, ARECIBO, ARECIBO 305M,
BIGHORNS, EVLA, JVLA, LOFAR, MWA, PARKES, PKS, ATPKSMB, or WSRT) or a LUA file
like the ones in
https://gitlab.com/aroffringa/aoflagger/-/tree/master/data/strategies
You can define a new strategy interactively using the AOFlagger rfigui
(https://aoflagger.readthedocs.io/en/latest/using_rfigui.html)
:param vis: Visibility object
:param strategy_name: Strategy to use: can be a LUA file or a telescope name. If
the strategy for the selected telescope is not available, a generic strategy
is used.
:return: Visibility where the flags field has been updated
"""
try:
import aoflagger
except ImportError:
log.warning("AOFlagger is not installed. Not executing flagging.")
return vis
flagger = aoflagger.AOFlagger()
if strategy_name.endswith(".lua"):
# Load specific LUA strategy
strategy = flagger.load_strategy_file(path)
else:
# Use the LUA strategy corresponding to the desired telescope
name_uppercase = strategy_name.upper()
if name_uppercase == "AARTFAAC":
path = flagger.find_strategy_file(aoflagger.TelescopeId.AARTFAAC)
elif name_uppercase == "ARECIBO" or name_uppercase == "ARECIBO 305M":
path = flagger.find_strategy_file(aoflagger.TelescopeId.Arecibo)
elif name_uppercase == "BIGHORNS":
path = flagger.find_strategy_file(aoflagger.TelescopeId.Bighorns)
elif name_uppercase == "EVLA" or name_uppercase == "JVLA":
path = flagger.find_strategy_file(aoflagger.TelescopeId.JVLA)
elif name_uppercase == "LOFAR":
path = flagger.find_strategy_file(aoflagger.TelescopeId.LOFAR)
elif name_uppercase == "MWA":
path = flagger.find_strategy_file(aoflagger.TelescopeId.MWA)
elif (
name_uppercase == "PARKES"
or name_uppercase == "PKS"
or name_uppercase == "ATPKSMB"
):
path = flagger.find_strategy_file(aoflagger.TelescopeId.Parkes)
elif name_uppercase == "WSRT":
path = flagger.find_strategy_file(aoflagger.TelescopeId.WSRT)
else:
path = flagger.find_strategy_file(aoflagger.TelescopeId.Generic)
strategy = flagger.load_strategy_file(path)
ntimes = vis.vis.data.shape[0]
nbaselines = vis.vis.data.shape[1]
nch = vis.vis.data.shape[2]
npol = vis.vis.data.shape[3]
for bl in range(nbaselines):
data = flagger.make_image_set(ntimes, nch, npol * 2)
for imgindex in range(npol):
# AOFlagger uses the axis in a different order than Rascil
values = numpy.swapaxes(vis.vis.data[:, bl, :, imgindex], 0, 1)
data.set_image_buffer(2 * imgindex, numpy.real(values))
data.set_image_buffer(2 * imgindex + 1, numpy.imag(values))
flags = strategy.run(data)
# A flag denotes that the values at that time-frequency position should
# be ignored for all polarizations
for pol in range(npol):
vis.flags.data[:, bl, :, pol] = numpy.swapaxes(
flags.get_buffer().astype(numpy.int32), 0, 1
)
return vis