casa6

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

# test_task_phaseshift.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.

#

# [Add the link to the JIRA ticket here once it exists]

#

# Based on the requirements listed in plone found here:

# https://casadocs.readthedocs.io/en/stable/api/tt/casatasks.manipulation.phaseshift.html

#

#

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

import glob

import numpy as np

import os

import shutil

import unittest

​

from casatools import (

        componentlist, ctsys, image, measures, ms, msmetadata, quanta,

        regionmanager, simulator, table

    )

from casatasks import flagdata, imstat, phaseshift, tclean

from casatasks.private import simutil

cl = componentlist()

ia = image()

md = msmetadata()

me = measures()

ms = ms()

qa = quanta()

rg = regionmanager()

sm = simulator()

tb = table()

datadir = os.path.join('unittest', 'phaseshift')

ctsys_resolve = ctsys.resolve

​

datadir = os.path.join('unittest', 'phaseshift')

datapath = ctsys_resolve(os.path.join(datadir, 'refim_twopoints_twochan.ms'))

datapath_Itziar = ctsys_resolve(os.path.join(datadir, 'Itziar.ms'))

datapath_ngc = ctsys_resolve(os.path.join(datadir, 'ngc7538_ut.ms'))

datapath_nep = ctsys_resolve(os.path.join(datadir, 'nep2-shrunk.ms'))

datapath_mms = ctsys_resolve(

    os.path.join(datadir, 'uid___X02_X3d737_X1_01_small.mms')

)

​

​

def change_perms(path):

    os.chmod(path, 0o777)

    for root, dirs, files in os.walk(path):

        for d in dirs:

            os.chmod(os.path.join(root, d), 0o777)

        for f in files:

            os.chmod(os.path.join(root, f), 0o777)

​

​

datacopy = 'datacopy.ms'

datacopy_Itziar = 'Itziar_copy.ms'

datacopy_ngc = 'ngc_copy.ms'

datacopy_nep = 'nep_copy.ms'

datacopy_mms = 'mms_copy.mms'

output = 'phaseshiftout.ms'

​

​

class phaseshift_base_checks(unittest.TestCase):

​

    def check_nrows(self, vis, expected_nrows):

        """ Simple check to ensure the correct number of rows has been produced in the main

        table of an (output) MS. Checks the vis exists on disk and looks at the number of

        rows of the TIME column. """

        tbt = table()

        try:

            tbt.open(vis)

            nrows = len(tbt.getcol('TIME'))

        finally:

            tbt.close()

​

        self.assertTrue(os.path.exists(vis), f"MeasurementSet not found: {vis}")

        self.assertEqual(nrows, expected_nrows,

                         f"Incorrect number of rows found in MS: {vis}")

​