casa6

import os

import shutil

import copy

from collections import namedtuple

import numpy

from scipy.optimize import curve_fit

import unittest

​

from casatasks.private.casa_transition import is_CASA6

if is_CASA6:

    from casatasks import sdsidebandsplit

    from casatools import quanta

    from casatools import image

    from casatools import ctsys

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

​

    # default isn't used in casatasks

    def default(atask):

        pass

​

    # stack_frame_find

    def stack_frame_find():

        return {}

else:

    from tasks import sdsidebandsplit

    from taskinit import qatool as quanta

    from taskinit import iatool as image

    from casa_stack_manip import stack_frame_find

​

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

​

​

# Gaussian fit

def gauss_func(x, *p):

    amp, center, width, offset = p

    y = amp * numpy.exp(-(x - center)**2 / (2. * width**2)) + offset

    return y

​

​

def gauss_fit(x, y):

    # initial guess

    o = y.mean()

    a = numpy.abs(y - o).max()

    c = x[numpy.where(numpy.abs(y - o) == a)[0]][0]

    w = numpy.abs(y - o).sum() / a

    # print("initial guess: (%f, %f, %f, %f)" % (a,c,w,o))

    return curve_fit(gauss_func, x, y, p0=(a, c, w, o))

​

​

# a named tuple to store spectral information.

# start, end: start and end of a channel range

# max, min: max and min value of the channel range

# peak, center, width: gaussian fit parameters of the channel range

SpectralInfo = namedtuple('SpectralInfo',

                          ['start', 'end', 'max', 'min', 'peak', 'center', 'width', 'offset'])

​

​

class sdsidebandsplitTestBase(unittest.TestCase):

    standard_param = dict(

        imagename=['onepix_noiseless_shift0.image', 'onepix_noiseless_shift-102.image',

                   'onepix_noiseless_shift8.image', 'onepix_noiseless_shift62.image',

                   'onepix_noiseless_shift88.image', 'onepix_noiseless_shift100.image'],

        outfile='separated.image',

        overwrite=False,

        signalshift=[0.0, -102, +8, +62, +88, +100],

        imageshift=[0.0, 102, -8, -62, -88, -100],

        getbothside=False,

        refchan=0.0,

        refval='805GHz',

        otherside=False,

        threshold=0.2

    )

​

    def update_task_param(self, new_param={}):

        """

        Overwrite standard task parameter and return a new dictionary

        with updated parameters.

        Note this task does not check validity of parameter names in

        the input parameter.

​

        Parameter

            new_param : a dictionary of parameter names (key) and

                        values (value) to overwrite standard task

                        execution parameters in tests.

        """

        if type(new_param) is not dict:

            raise TypeError('The input should be a dictionary')

        updated_param = copy.deepcopy(self.standard_param)

        updated_param.update(new_param)

        return updated_param

​