Source
setjy(vis='ngc5921_virtual_model_col.ms', field='1331+305*',model='',standard='Perley-Taylor 99',scalebychan=False, usescratch=False)
#########################################################################
# test_task_flagdata.py
# Copyright (C) 2018
# Associated Universities, Inc. Washington DC, USA.
#
# This script is free software; you can redistribute it and/or modify it
# under the terms of the GNU Library General Public License as published by
# the Free Software Foundation; either version 2 of the License, or (at your
# option) any later version.
#
# This library is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public
# License for more details.
#
#
# Based on the requirements listed in casadocs found here:
# https://casadocs.readthedocs.io/en/stable/api/tt/casatasks.flagging.flagdata.html
#
##########################################################################
import os
import shutil
import unittest
import filecmp
import pprint
import numpy as np
from numpy import array
import ast
import sys
from casatasks import flagcmd, flagdata, mstransform, setjy, delmod, split
from casatools import ctsys, agentflagger, table, measures, quanta
from casatasks.private.parallel.parallel_task_helper import ParallelTaskHelper
from casatasks.private import flaghelper as fh
from casatestutils import testhelper as th
ctsys_resolve = ctsys.resolve
#
# Test of flagdata modes
#
def func_test_eq(result, total, flagged):
print("%s of %s data was flagged, expected %s of %s" % \
(result['flagged'], result['total'], flagged, total))
assert result['total'] == total, \
"%s data in total; %s expected" % (result['total'], total)
assert result['flagged'] == flagged, \
"%s flags set; %s expected" % (result['flagged'], flagged)
def create_input(str_text, filename):
'''Save the string in a text file'''
inp = filename
cmd = str_text
# remove file first
if os.path.exists(inp):
os.system('rm -f '+ inp)
# save to a file
with open(inp, 'w') as f:
f.write(cmd)
f.close()
return
# Path for data
datapath = ctsys_resolve("unittest/flagdata/")
# Pick up alternative data directory to run tests on MMSs
testmms = False
if 'TEST_DATADIR' in os.environ:
DATADIR = str(os.environ.get('TEST_DATADIR'))+'/flagdata/'
if os.path.isdir(DATADIR):
testmms = True
datapath = DATADIR
print('flagdata tests will use data from '+datapath)
# jagonzal (CAS-4287): Add a cluster-less mode to by-pass parallel processing for MMSs as requested
if 'BYPASS_PARALLEL_PROCESSING' in os.environ:
ParallelTaskHelper.bypassParallelProcessing(1)
# Local copy of the agentflagger tool
aflocal = agentflagger()
class test_dict_consolidation(unittest.TestCase):