Browse Source

Add test for polarization labelling

master
haavee 8 months ago
parent
commit
e1721bfd33
  1. 401
      N20L2/test_N20L2.py

401
N20L2/test_N20L2.py

@ -0,0 +1,401 @@
from __future__ import print_function
import os, sys, copy, glob, shutil, shlex, argparse, tempfile, operator, subprocess, datetime, unittest
# j2ms2 + tConvert come as a pair.
# We need to be able to specify where to take the binaries from.
# Default: environment "j2ms2" and "tConvert".
# When built using cmake in a build directory (CMake favours out-of-source builds) the binaries are accessible as
# path/to/build_dir/apps/[j2ms2|tConvert]/\\1
# However, when installed, they're accessible as:
# path/to/install_dir/bin/{j2ms2,tConvert}
# from python shutil source code (https://github.com/python/cpython/blob/master/Lib/shutil.py)
def _access_check(fn, mode):
return (os.path.exists(fn) and os.access(fn, mode)
and not os.path.isdir(fn))
def which(cmd):
for p in os.environ.get('PATH', os.defpath).split(':'):
candidate = os.path.join(p, cmd)
if _access_check(candidate, os.F_OK|os.X_OK):
return candidate
return None
# argparse "type" function that checks whether it's a writable path
def is_writable_dir(p):
if os.path.isdir(p) and os.access(p, os.W_OK):
return p
raise argparse.ArgumentTypeError( "[is_writable_dir]: '{0}' is not that".format(p) )
def check_previous_results(p):
if os.path.isdir(p):
raise RuntimeError("Previous test results seem to exists at {0}. Please check and remove those first".format(p))
pass
# This class captures the binaries from $PATH - if any.
# None if not found
class EnvironmentBinaries(object):
_j2ms2_ = which('j2ms2')
_tConvert_ = which('tConvert')
_jplotter_ = which('jplotter')
def get_j2ms2(self):
return self._j2ms2_fn_(self)
def get_tConvert(self):
return self._tConvert_fn_(self)
def get_jplotter(self):
return self._jplotter_fn_(self)
def get_python(self):
return self._python_fn_(self)
j2ms2 = property(get_j2ms2)
tConvert = property(get_tConvert)
jplotter = property(get_jplotter)
python = property(get_python)
# path argument baseclass for Argparse;
# will set "__path" to value of the option on the command line
class BaseAction(argparse.Action):
def __init__(self, option_strings, dest, nargs=None, **kwargs):
if nargs is not None:
raise ValueError("nargs not allowed")
# extract the variable to update in the destination
self.p__ = kwargs.pop('path', None)
if self.p__ is None:
raise RuntimeError("This action requires a path= keyword argument")
super(BaseAction, self).__init__(option_strings, dest, **kwargs)
def __call__(self, parser, namespace, values, option_string=None):
# If the namespace does not already have an attribute called destination
# then we install ourselves as one
if not hasattr(namespace, self.dest):
setattr(namespace, self.dest, self)
# Now update the object in the target namespace
setattr(getattr(namespace, self.dest), self.p__, values)
# shorthand to set attributes in o, with name taken from d.keys and to values set by d.values
def updateObj(o, d):
for (nm, fn) in d.items():
setattr(o, nm, fn)
# Modify base class to extract / return paths for j2ms2/tConvert:
def SetBinaries(cls, **kwargs):
fnsToSet = copy.deepcopy(kwargs)
class NewClass(cls):
def __init__(self, *a, **k):
updateObj(self, fnsToSet)
super(NewClass, self).__init__(*a, **k)
def __call__(self, parser, namespace, values, option_string=None):
# when the action is called we need to update stuff
if not hasattr(namespace, self.dest):
setattr(namespace, self.dest, self)
updateObj(getattr(namespace, self.dest), fnsToSet)#self.fnsToSet)
return super(NewClass, self).__call__(parser, namespace, values, option_string)
return NewClass
def Raise(err):
raise RuntimeError(err)
# Create classes that can be given to Argparse as action and default
# into immediately useful objects
DefaultBinaries = SetBinaries(EnvironmentBinaries,
_j2ms2_fn_=lambda _: EnvironmentBinaries._j2ms2_ or Raise("No j2ms2 found in your $PATH"),
_tConvert_fn_=lambda _: EnvironmentBinaries._tConvert_ or Raise("No tConvert found in your $PATH"),
_jplotter_fn_=lambda _: EnvironmentBinaries._jplotter_ or Raise("No jplotter found in your $PATH"),
_python_fn_=lambda _: sys.executable)
# The InstallDir- and BuildDirAction will set the 'path__' attribute if on the command line
InstallDirAction = SetBinaries(BaseAction, _j2ms2_fn_=lambda s: os.path.join(s.path__, "bin/j2ms2"),
_tConvert_fn_=lambda s: os.path.join(s.path__, "bin/tConvert"))
BuildDirAction = SetBinaries(BaseAction, _j2ms2_fn_=lambda s: os.path.join(s.path__, "apps/j2ms2/j2ms2"),
_tConvert_fn_=lambda s: os.path.join(s.path__, "apps/tConvert/tConvert"))
# The jplotterAction will set the 'jplpath__' attribute if on the command line
jplotterAction = SetBinaries(BaseAction, _jplotter_fn_=lambda s: s.jplpath__)
# Following https://stackoverflow.com/a/17259773
# Strip out options specifically meant for us and replace sys.argv[1:] with what remains
binarys = DefaultBinaries()
parsert = argparse.ArgumentParser(add_help=True)
mutexgr = parsert.add_mutually_exclusive_group()
mutexgr.add_argument('--cmake-install-dir', help="test j2ms2/tConvert binaries from install_dir/bin",
dest='binaries', action=InstallDirAction, path='path__', default=binarys)
mutexgr.add_argument('--cmake-build-dir', help="test j2ms2/tConvert binaries from build_dir/app/{j2ms2,tConvert}/",
dest='binaries', action=BuildDirAction, path='path__', default=binarys)
parsert.add_argument('--jplotter', help="Path to jplotter to use (default: from $PATH)",
dest='binaries', action=jplotterAction, path='jplpath__', default=binarys)
# according to https://stackoverflow.com/a/11114654
# https://stackoverflow.com/a/54547257
parsert.add_argument('--tempdir', type=is_writable_dir, default=None, help=argparse.SUPPRESS)
options, args = parsert.parse_known_args()
sys.argv[1:] = args
## Yes, technically tests should be able to run in any order
## but we're testing a toolchain here where e.g. step N+1
## uses a product generated in step N
unittest.TestLoader.sortTestMethodsUsing = None
## Shorthands
RUN = subprocess.Popen
SHLEX = lambda arg : shlex.split(arg) if isinstance(arg, type("")) else arg
# Some preformatted messages
artefacts_msg = \
"""********************************************************************************
* Automatic cleanup inhibited; artefacts left for post-mortem in *
* {0:^70} *
********************************************************************************
""".format
class TTC(unittest.TestCase):
# unittest calls setUp()/tearDown() around _each testfunction_
# so if we want to persist some stuff between calls we better go through classvariables ...
__curdir__ = None
__rootdir__ = None
__tempdir__ = None
__workdir__ = None
__cleanup__ = True
__lasttst__ = False
def setUp(self):
if TTC.__curdir__:
# Since this is called for every test in this run, we must perform the following
# check each time; TTC.__workdir__ is only set if this code has verified that it
# did not exist first
if TTC.__workdir__ is None:
workdir = os.path.join( TTC.__tempdir__ or options.tempdir, 'N20L2' )
raise RuntimeError("Previous test results exist at {0}. Please check+remove those first".format(workdir))
return
print("Preparing test environment ...", end='')
# we must
# (1) know where we are
TTC.__curdir__ = os.getcwd()
# (2) where this specific file is (to find the raw and gold data files)
TTC.__rootdir__ = os.path.abspath( os.path.dirname(__file__) )
# Before creating any garbage, let's make sure the test data is available
# (possibly download it using wget)
self.download_test_data()
# (3) create a temporary directory where to create new data files to compare
# 10 Jun 2021: only do this if the outer layer did not specify one
if options.tempdir is None:
TTC.__tempdir__ = tempfile.mkdtemp()
# Only set the __workdir__ if that was allowed; check_previous_results throws if
# the workdir already exists
workdir = os.path.join( TTC.__tempdir__ or options.tempdir, 'N20L2' )
check_previous_results( workdir );
#sys.stdout.write( "\n=======> Created tmp dir: {0}\n".format(self.tempDir) )
# j2ms2 expects the directory containing data to have the same name as the vexfile
TTC.__workdir__ = workdir
# (3a) assert the workdir does not exist; if test is rerun with same
# --tempdir X/Y/Z argument must erase results of previous run first
TTC.__cleanup__ = True
# And finally, we change directory to the newly created temp dir
# and populate it with the files as necessary
os.mkdir( TTC.__workdir__ )
# Now we can go on with our business
os.chdir( TTC.__workdir__ )
# symlink SFXC job 24427 into the work dir
os.symlink( os.path.join(TTC.__rootdir__, '31334'), './31334' )
# Copy two VEX files and the lis file across
for f in ['N20L2.vix', 'n20l2.vix' ]:
shutil.copyfile( os.path.join(TTC.__rootdir__, f), os.path.join(TTC.__workdir__, f) )
print( " done" )
# Need to download several testfiles
def download_test_data(self):
# Only attempt to download if not already there!
# Check the first / last entries from the tar file, hoping to catch partial #fail
if os.path.isdir( os.path.join(TTC.__rootdir__, '31334') ) and \
os.path.isfile( os.path.join(TTC.__rootdir__, 'n20l2.vix') ):
return
opath = copy.deepcopy(sys.path)
# pre-compute name of downloadfile
bz2data = os.path.join(TTC.__rootdir__, "n20l2_data.tar.bz2")
try:
print("downloading test data...", end='')
# Import the 'wget' python module from our own repo
sys.path.insert(0, os.path.join(TTC.__rootdir__, "../python"))
import wget, tarfile
# Download URL into workarea
wget.download( "http://archive.jive.nl/regression/n20l2_data.tar.bz2",
out= bz2data )
with tarfile.open( bz2data, "r:bz2" ) as tar:
print("inflate+extract...", end='')
tar.extractall( path = TTC.__rootdir__ )
# hope this triggers auto cleanup at some point
del sys.modules['wget']
del wget
finally:
sys.path = opath
if os.path.isfile( bz2data ):
os.unlink( bz2data )
return
def AtestFoo(self):
self.assertEqual(1+1, 2)
#
# Compare the generated MS with the known-good one
# We use the jplotter helper for this
#
# Select amp/phas vs freq from three different scans and sum them up.
# We expect the sum to be numerically identical between the known-good dataset
# and the newly created one.
# cf. RMC [email Jun 9 2021] on N20L2 / job 31334
# "this has just Ef,Jb,Ar -- but Ar-* has good enough SNR to
# have visible differences in amp levels for all four [polarizations]."
__n20l2_comparison__ = "bl ar*; pt anpfreq; avt sum; solint none;"
# Attempt to run j2ms2 in "manual" mode, i.e. specifiying cor files & output MS file on the command line
# j2ms2 -o <ms> 24427/*.cor
# Note: we can do the globbing on "24427/*.cor" since the setup code has done a
# chdir( __workdir__ )
# Note: if the code was just cloned from git there is no 2+ GB binary data
# which means the glob.glob() here returns an empty list which means
# that the manual j2ms2 goes into infinite wait loop which means either
# the process table of the O/S will fill up (e.g. if this regression is run
# through cron) or it has to be SIGINT/SIGKILL'ed which means the
# cleanup doesn't happen and /tmp/ gets clobbered with a lot of trash.
# Neither a nice prospect. So we fix this by making sure the __glob__()
# returns a non-empty list - if an invalid thing gets passed j2ms2 will
# fail with an error, exactly what we want.
__glob__ = classmethod(lambda _: " ".join( glob.glob("31334/*.cor")) or "no-correlator-data" )
def testA(self):
rv = None
cmd = "{0} -o clk_No0002_31334.ms {1}".format( options.binaries.j2ms2, TTC.__glob__())
print("Start", cmd)
with open( os.path.join(TTC.__workdir__, 'testA.log'), 'w' ) as lf:
lf.write( datetime.datetime.now().isoformat() + '\n' )
lf.write( cmd + '\n' )
lf.flush()
process = RUN(SHLEX(cmd), stdout=lf, stderr=lf)
process.wait()
rv = process.returncode
lf.write( "\nj2ms2 exited with return code {0}".format(rv) )
TTC.__cleanup__ = TTC.__cleanup__ and (rv==0)
# Demand that j2ms2 ran w/o error
self.assertEqual(rv, 0)
def testAA(self):
rv = None
cmd = "{python} {compare} '{selection}' {gold_ms} clk_No0002_31334.ms".format(
python = options.binaries.python,
compare = os.path.join(TTC.__rootdir__, "../python", "compare_MS_numerically.py"),
selection = TTC.__n20l2_comparison__,
gold_ms = os.path.join(TTC.__rootdir__, "clk_No0002_31334.ms") )
# we need to prepare the environment
jplEnv = copy.deepcopy(os.environ)
# Make sure the pythonpath to jplotter is added
jplEnv['PYTHONPATH'] = ":".join(filter(operator.truth,
[os.path.dirname(options.binaries.jplotter),
os.environ.get('PYTHONPATH', None)]))
print("Start", cmd)
with open( os.path.join(TTC.__workdir__, 'testAA.log'), 'w' ) as lf:
lf.write( datetime.datetime.now().isoformat() + '\n' )
lf.write( cmd + '\n' )
lf.flush()
process = RUN(SHLEX(cmd), stdout=lf, stderr=lf, env=jplEnv)
process.wait()
rv = process.returncode
lf.write( "\ncomparison exited with return code {0}".format(rv) )
TTC.__cleanup__ = TTC.__cleanup__ and (rv==0)
# Demand that the comparison ran w/o error
self.assertEqual(rv, 0)
# Run tConvert on the lis-file MS
def testB(self):
rv = None
cmd = "{0} clk_No0002_31334.ms CLK_NO0002_31334.IDI".format( options.binaries.tConvert )
print("Start", cmd)
with open( os.path.join(TTC.__workdir__, 'testB.log'), 'w' ) as lf:
lf.write( datetime.datetime.now().isoformat() + '\n' )
lf.write( cmd + '\n' )
lf.flush()
process = RUN(SHLEX(cmd), stdout=lf, stderr=lf)
process.wait()
rv = process.returncode
lf.write( "\ntConvert exited with return code {0}".format(rv) )
TTC.__cleanup__ = TTC.__cleanup__ and (rv==0)
# Demand that tConvert ran w/o error
self.assertEqual(rv, 0)
# Perform the n-way data content diff between gold measurement set and FITS-IDI file
# and those just produced
def testC(self):
rv = None
cmd = "{python} {verify} --ms {gold_ms} --idi {gold_idi} \
--ms clk_No0002_31334.ms --idi CLK_NO0002_31334.IDI".format(
python = options.binaries.python,
verify = os.path.join(TTC.__rootdir__, "../jive-toolchain-verify/compare-ms-idi.py"),
gold_ms = os.path.join(TTC.__rootdir__, "clk_No0002_31334.ms"),
gold_idi = os.path.join(TTC.__rootdir__, "CLK_NO0002_31334.IDI"),
)
print("Start", cmd)
with open( os.path.join(TTC.__workdir__, 'testC.log'), 'w' ) as lf:
lf.write( datetime.datetime.now().isoformat() + '\n' )
lf.write( cmd + '\n' )
lf.flush()
process = RUN(SHLEX(cmd), stdout=lf, stderr=lf)
process.wait()
rv = process.returncode
lf.write( "\ncompare-ms-idi.py exited with return code {0}".format(rv) )
TTC.__cleanup__ = TTC.__cleanup__ and (rv==0)
self.assertEqual(rv, 0)
# Perform the n-way meta data content diff between gold measurement set and FITS-IDI file
# and those just produced
def testD(self):
rv = None
cmd = "{python} {verify} --ms {gold_ms} --idi {gold_idi} \
--ms clk_No0002_31334.ms --idi CLK_NO0002_31334.IDI".format(
python = options.binaries.python,
verify = os.path.join(TTC.__rootdir__, "../jive-toolchain-verify/compare-ms-idi-meta.py"),
gold_ms = os.path.join(TTC.__rootdir__, "clk_No0002_31334.ms"),
gold_idi = os.path.join(TTC.__rootdir__, "CLK_NO0002_31334.IDI"),
)
print("Start", cmd)
with open( os.path.join(TTC.__workdir__, 'testD.log'), 'w' ) as lf:
lf.write( datetime.datetime.now().isoformat() + '\n' )
lf.write( cmd + '\n' )
lf.flush()
process = RUN(SHLEX(cmd), stdout=lf, stderr=lf)
process.wait()
rv = process.returncode
lf.write( "\ncompare-ms-idi-meta.py exited with return code {0}".format(rv) )
TTC.__cleanup__ = TTC.__cleanup__ and (rv==0)
self.assertEqual(rv, 0)
# Leave this one as last test case to trigger potential cleanup
# in the associated tearDown() call
def testZ(self):
TTC.__lasttst__ = True
self.assertEqual(1+1, 2)
#### tearDown()
def tearDown(self):
if not TTC.__lasttst__:
return
# go back to where we started from (mainly pointless but it might
# affect being able to rm the tree if an active process is still
# using (part of) the tree as current dir
os.chdir( TTC.__curdir__ )
# if cleanup not inhibited ... do clean up
if TTC.__cleanup__:
# depending on whether we created the tempdir remove the whole thing or only our workdir
# (TTC.__tempdir__ is only set if no-one passed in a tempdir command line argument)
shutil.rmtree( TTC.__workdir__ if TTC.__tempdir__ is None else TTC.__tempdir__ )
else:
print( artefacts_msg(TTC.__workdir__) )
if __name__ == '__main__':
unittest.main()
Loading…
Cancel
Save