casa6

from __future__ import absolute_import

​

import unittest

import os

import math

import sys

​

from casatestutils import testhelper as th

​

from casatasks.private.casa_transition import is_CASA6

if is_CASA6:

    from casatasks import nrobeamaverage

    from casatools import ms

    from casatools import table

​

    # default isn't used in casatasks

    def default(atask):

        pass

​

    from casatasks.private.sdutil import tbmanager

    from casatools import ctsys

    datapath=ctsys.resolve('unittest/nrobeamaverage/')

​

else:

    from tasks import nrobeamaverage

    from taskinit import mstool as ms

    from taskinit import tbtool as table

​

    from __main__ import default

    from sdutil import tbmanager

​

    # Define the root for the data files

    datapath = os.environ.get('CASAPATH').split()[0] + "/casatestdata/unittest/nrobeamaverage/"

​

from casatestutils import testhelper as th

​

def check_eq(val, expval, tol=None):

    """Checks that val matches expval within tol."""

    if type(val) == dict:

        for k in val:

            check_eq(val[k], expval[k], tol)

    else:

        try:

            if tol and hasattr(val, '__rsub__'):

                are_eq = abs(val - expval) < tol #absolute check

                if not are_eq:

                    are_eq = abs(val - expval)/abs(expval) < tol #relative check

            else:

                are_eq = val == expval

            if hasattr(are_eq, 'all'):

                are_eq = are_eq.all()

            if not are_eq:

                raise ValueError('!=')

        except ValueError:

            errmsg = "%r != %r" % (val, expval)

            if (len(errmsg) > 66): # 66 = 78 - len('ValueError: ')

                errmsg = "\n%r\n!=\n%r" % (val, expval)

            raise ValueError(errmsg)

        except Exception as e:

            print("Error comparing {} to {}".format(val, expval))

            raise

​

class test_nrobeamaverage(unittest.TestCase):

    def setUp(self):

        default(nrobeamaverage)

​

        self.i_ms = "onon.ms"

        os.system('cp -RH '+ os.path.join(datapath,self.i_ms) +' '+ self.i_ms)

        self.o_ms = "bave.ms"

        self.args = {'infile': self.i_ms, 'outfile': self.o_ms}

​

        self.antid = self._get_antid()

        self.min_antid = 0

        self.st_onsrc = self._get_onsource_stateid()

        self.interval = 21 # in seconds

        self.tol = 1e-5

​

    def tearDown(self):

        os.system('rm -rf ' + self.i_ms)

        os.system('rm -rf ' + self.o_ms)

​

    def _get_antid(self):

        with tbmanager(self.i_ms + '/ANTENNA') as tb:

            acol = tb.getcol('NAME')

        return range(len(acol))

​

    def _get_onsource_stateid(self):

        with tbmanager(self.i_ms + '/STATE') as tb:

            ocol = tb.getcol('OBS_MODE')

        res = None

        for i in range(len(ocol)):