# -*- mode: python; coding: utf-8 -*- # Copyright (c) 2019 Radio Astronomy Software Group # Licensed under the 2-clause BSD License """Tests for MWACorrFITS object.""" import pytest import os import numpy as np from pyuvdata import UVData from pyuvdata.data import DATA_PATH import pyuvdata.tests as uvtest from astropy.io import fits # set up MWA correlator file list testdir = os.path.join(DATA_PATH, 'mwa_corr_fits_testfiles/') testfiles = ['1131733552.metafits', '1131733552_20151116182537_mini_gpubox01_00.fits', '1131733552_20151116182637_mini_gpubox06_01.fits', '1131733552_mini_01.mwaf', '1131733552_mini_06.mwaf', '1131733552_mod.metafits', '1131733552_mini_cotter.uvfits'] filelist = [testdir + i for i in testfiles] def test_ReadMWAWriteUVFits(): """ MWA correlator fits to uvfits loopback test. Read in MWA correlator files, write out as uvfits, read back in and check for object equality. """ mwa_uv = UVData() uvfits_uv = UVData() messages = ['The `phase_data` keyword is deprecated.', 'telescope_location is not set', 'some coarse channel files were not submitted'] category = [DeprecationWarning] + [UserWarning] * 2 uvtest.checkWarnings(mwa_uv.read_mwa_corr_fits, func_args=[filelist[0:2]], func_kwargs={'correct_cable_len': True, 'phase_data': True}, nwarnings=3, message=messages, category=category) testfile = os.path.join(DATA_PATH, 'test/outtest_MWAcorr.uvfits') mwa_uv.write_uvfits(testfile, spoof_nonessential=True) uvfits_uv.read_uvfits(testfile) assert mwa_uv == uvfits_uv phase_center = (mwa_uv.phase_center_ra + 0.01, mwa_uv.phase_center_dec + 0.01) messages = ['The `phase_center` keyword is deprecated.', 'The `phase_data` keyword is deprecated.', 'telescope_location is not set', 'some coarse channel files were not submitted', 'Phasing this UVData object to phase_center_radec'] category = [DeprecationWarning] * 2 + [UserWarning] * 3 uvtest.checkWarnings(mwa_uv.read_mwa_corr_fits, func_args=[filelist[0:2]], func_kwargs={'correct_cable_len': True, 'phase_data': True, 'phase_center': phase_center}, nwarnings=5, message=messages, category=category) testfile = os.path.join(DATA_PATH, 'test/outtest_MWAcorr.uvfits') mwa_uv.write_uvfits(testfile, spoof_nonessential=True) uvfits_uv.read_uvfits(testfile) assert mwa_uv == uvfits_uv @pytest.mark.filterwarnings("ignore:telescope_location is not set. ") @pytest.mark.filterwarnings("ignore:some coarse channel files were not submitted") def test_select_on_read(): mwa_uv = UVData() mwa_uv2 = UVData() mwa_uv.read_mwa_corr_fits(filelist[0:2], correct_cable_len=True) unique_times = np.unique(mwa_uv.time_array) select_times = unique_times[np.where((unique_times >= np.min(mwa_uv.time_array)) & (unique_times <= np.mean(mwa_uv.time_array)))] mwa_uv.select(times=select_times) uvtest.checkWarnings( mwa_uv2.read, func_args=[filelist[0:2]], func_kwargs={'correct_cable_len': True, 'time_range': [np.min(mwa_uv.time_array), np.mean(mwa_uv.time_array)]}, message=['Warning: select on read keyword set, but file_type is "mwa_corr_fits"', 'telescope_location is not set. Using known values for MWA.', 'some coarse channel files were not submitted'], nwarnings=3) assert mwa_uv == mwa_uv2 @pytest.mark.filterwarnings("ignore:telescope_location is not set. ") @pytest.mark.filterwarnings("ignore:some coarse channel files were not submitted") def test_ReadMWA_ReadCotter(): """ Pyuvdata and cotter equality test. Read in MWA correlator files and the corresponding cotter file and check for data array equality. """ mwa_uv = UVData() cotter_uv = UVData() # cotter data has cable correction and is unphased mwa_uv.read(filelist[0:2], correct_cable_len=True) cotter_uv.read(filelist[6]) # cotter doesn't record the auto xy polarizations # due to a possible bug in cotter, the auto yx polarizations are conjugated # fix these before testing data_array autos = np.isclose(mwa_uv.ant_1_array - mwa_uv.ant_2_array, 0.0) cotter_uv.data_array[autos, :, :, 2] = cotter_uv.data_array[autos, :, :, 3] cotter_uv.data_array[autos, :, :, 3] = np.conj(cotter_uv.data_array[autos, :, :, 3]) assert np.allclose(mwa_uv.data_array[:, :, :, :], cotter_uv.data_array[:, :, :, :], atol=1e-4, rtol=0) def test_ReadMWAWriteUVFits_meta_mod(): """ MWA correlator fits to uvfits loopback test with a modified metafits file. Read in MWA correlator files, write out as uvfits, read back in and check for object equality. """ # The metafits file has been modified to contain some coarse channels < 129, # and to have an uncorrected cable length. mwa_uv = UVData() uvfits_uv = UVData() messages = ['telescope_location is not set', 'some coarse channel files were not submitted'] files = [filelist[1], filelist[5]] uvtest.checkWarnings(mwa_uv.read, func_args=[files], func_kwargs={'correct_cable_len': True, 'phase_to_pointing_center': True}, nwarnings=2, message=messages) testfile = os.path.join(DATA_PATH, 'test/outtest_MWAcorr.uvfits') mwa_uv.write_uvfits(testfile, spoof_nonessential=True) uvfits_uv.read_uvfits(testfile) assert mwa_uv == uvfits_uv @pytest.mark.filterwarnings("ignore:telescope_location is not set. ") @pytest.mark.filterwarnings("ignore:some coarse channel files were not submitted") @pytest.mark.filterwarnings("ignore:Combined frequencies are not contiguous") def test_ReadMWA_multi(): """Test reading in two sets of files.""" set1 = filelist[0:2] set2 = [filelist[0], filelist[2]] mwa_uv = UVData() mwa_uv.read([set1, set2]) mwa_uv2 = UVData() messages = ['Please use the generic `read` method', 'telescope_location is not set', 'some coarse channel files were not submitted', 'telescope_location is not set', 'some coarse channel files were not submitted', 'Combined frequencies are not contiguous'] category = [DeprecationWarning] + [UserWarning] * 5 uvtest.checkWarnings(mwa_uv2.read_mwa_corr_fits, func_args=[[set1, set2]], nwarnings=6, message=messages, category=category) assert(mwa_uv == mwa_uv2) @pytest.mark.filterwarnings("ignore:telescope_location is not set. ") @pytest.mark.filterwarnings("ignore:some coarse channel files were not submitted") def test_ReadMWA_multi_concat(): """Test reading in two sets of files with fast concatenation.""" # modify file so that time arrays are matching mod_mini_6 = os.path.join(DATA_PATH, 'test/mini_gpubox06_01.fits') with fits.open(filelist[2]) as mini6: mini6[1].header['time'] = 1447698337 mini6.writeto(mod_mini_6) set1 = filelist[0:2] set2 = [filelist[0], mod_mini_6] mwa_uv = UVData() mwa_uv.read([set1, set2], axis='freq') mwa_uv2 = UVData() messages = ['Please use the generic `read` method', 'telescope_location is not set', 'some coarse channel files were not submitted', 'telescope_location is not set', 'some coarse channel files were not submitted'] category = [DeprecationWarning] + [UserWarning] * 4 uvtest.checkWarnings(mwa_uv2.read_mwa_corr_fits, func_args=[[set1, set2]], func_kwargs={"axis": "freq"}, nwarnings=5, message=messages, category=category) assert(mwa_uv == mwa_uv2) os.remove(mod_mini_6) @pytest.mark.filterwarnings("ignore:telescope_location is not set. ") @pytest.mark.filterwarnings("ignore:some coarse channel files were not submitted") def test_ReadMWA_flags(): """Test handling of flag files.""" mwa_uv = UVData() subfiles = [filelist[0], filelist[1], filelist[3], filelist[4]] messages = ['mwaf files submitted with use_cotter_flags=False', 'telescope_location is not set', 'some coarse channel files were not submitted'] uvtest.checkWarnings(mwa_uv.read, func_args=[subfiles], nwarnings=3, message=messages) del(mwa_uv) mwa_uv = UVData() with pytest.raises(NotImplementedError) as cm: mwa_uv.read(subfiles, use_cotter_flags=True) assert str(cm.value).startswith('reading in cotter flag files') del(mwa_uv) mwa_uv = UVData() with pytest.raises(ValueError) as cm: mwa_uv.read(subfiles[0:2], use_cotter_flags=True) assert str(cm.value).startswith('no flag files submitted') del(mwa_uv) def test_multiple_coarse(): """ Test two coarse channel files. Read in MWA correlator files with two different orderings of the files and check for object equality. """ order1 = [filelist[0:3]] order2 = [filelist[0], filelist[2], filelist[1]] mwa_uv1 = UVData() mwa_uv2 = UVData() messages = ['telescope_location is not set', 'coarse channels are not contiguous for this observation', 'some coarse channel files were not submitted'] uvtest.checkWarnings(mwa_uv1.read, func_args=[order1], nwarnings=3, message=messages) uvtest.checkWarnings(mwa_uv2.read, func_args=[order2], nwarnings=3, message=messages) assert mwa_uv1 == mwa_uv2 def test_fine_channels(): """ Break read_mwa_corr_fits by submitting files with different fine channels. Test that error is raised if files with different numbers of fine channels are submitted. """ mwa_uv = UVData() bad_fine = os.path.join(DATA_PATH, 'test/bad_gpubox06_01.fits') with fits.open(filelist[2]) as mini6: mini6[1].data = np.concatenate((mini6[1].data, mini6[1].data)) mini6.writeto(bad_fine) with pytest.raises(ValueError) as cm: mwa_uv.read([bad_fine, filelist[1]]) assert str(cm.value).startswith('files submitted have different fine') del(mwa_uv) @pytest.mark.parametrize("files,err_msg", [([filelist[0]], "no data files submitted"), ([filelist[1]], "no metafits file submitted"), ([filelist[0], filelist[1], filelist[5]], "multiple metafits files in filelist")]) def test_break_ReadMWAcorrFITS(files, err_msg): """Break read_mwa_corr_fits by submitting files incorrectly.""" mwa_uv = UVData() with pytest.raises(ValueError) as cm: mwa_uv.read(files) assert str(cm.value).startswith(err_msg) del(mwa_uv) def test_file_extension(): """ Break read_mwa_corr_fits by submitting file with the wrong extension. Test that error is raised if a file with an extension that is not fits, metafits, or mwaf is submitted. """ mwa_uv = UVData() bad_ext = os.path.join(DATA_PATH, 'test/1131733552.meta') with fits.open(filelist[0]) as meta: meta.writeto(bad_ext) with pytest.raises(ValueError) as cm: mwa_uv.read(bad_ext, file_type='mwa_corr_fits') assert str(cm.value).startswith('only fits, metafits, and mwaf files supported') del(mwa_uv) def test_diff_obs(): """ Break read_mwa_corr_fits by submitting files from different observations. Test that error is raised if files from different observations are submitted in the same file list. """ mwa_uv = UVData() bad_obs = os.path.join(DATA_PATH, 'test/bad2_gpubox06_01.fits') with fits.open(filelist[2]) as mini6: mini6[0].header['OBSID'] = '1131733555' mini6.writeto(bad_obs) with pytest.raises(ValueError) as cm: mwa_uv.read([bad_obs, filelist[0], filelist[1]]) assert str(cm.value).startswith('files from different observations') del(mwa_uv)