casa6

​

##################################       Imports        ##################################

############################################################################################

​

import os

import sys

import time

from functools import wraps

import fnmatch

import logging

import filecmp

import unittest

import pickle

import numbers

import operator

import subprocess

import numpy

import six

​

casa5 = False

casa6 = False

​

from casatasks.private.casa_transition import is_CASA6

if is_CASA6:

​

    # CASA 6

    logging.debug("Importing CASAtools")

    import casatools

    logging.debug("Importing CASAtasks")

    import casatasks

    _cb = casatools.calibrater()

    _tb = casatools.table()

    _tbt = casatools.table()

    _ia  = casatools.image()

    _cb = casatools.calibrater()

    from casatasks import casalog

​

    casampi_imported = False

    import importlib

    _casampi_spec = importlib.util.find_spec('casampi')

    if _casampi_spec:

        # don't catch import error from casampi if it is found in the system modules

        from casampi.MPIEnvironment import MPIEnvironment

        casampi_imported = True

    else:

        casalog.post('casampi not available - not testing MPIEnvironment stuff', 'WARN')

​

    def tclean_param_names():

        from casatasks.tclean import _tclean_t

        return _tclean_t.__code__.co_varnames[:_tclean_t.__code__.co_argcount]

​

    casa6 = True

​

else:

​

    # CASA 5

    logging.debug("Import casa6 errors. Trying CASA5...")

    from __main__ import default

    from taskinit import tbtool, mstool, iatool, cbtool

    from taskinit import *

    from casa_stack_manip import stack_find, find_casa

    from mpi4casa.MPIEnvironment import MPIEnvironment

    casampi_imported = True

​

    _tb = tbtool()

    _tbt = tbtool()

    _ia = iatool()

    _cb = cbtool()

    casa = find_casa()

    if casa.has_key('state') and casa['state'].has_key('init_version') and casa['state']['init_version'] > 0:

        casaglobals=True

        casac = stack_find("casac")

        casalog = stack_find("casalog")

​

    def tclean_param_names():

        # alternatively could use from tasks import tclean; tclean.parameters

        from task_tclean import tclean

        return tclean.__code__.co_varnames[:tclean.__code__.co_argcount]

​

    casa5 = True

​

############################################################################################

##################################       imagerhelpers       ###############################

############################################################################################

class TestHelpers:

​

    # For comparison with keywords added by tclean in its output images

    if casampi_imported:

        num_mpi_procs = 1 + len(MPIEnvironment.mpi_server_rank_list())

    else:

        num_mpi_procs = 1