Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
polarizations: List of polarization integers or strings to read-in.
Ex: ['xx', 'yy', ...]
time_range: len-2 list containing min and max range of times (Julian Date) to read-in.
Ex: [2458115.20, 2458115.40]
"""
from . import miriad
if isinstance(filepath, (list, tuple)):
self.read_miriad(filepath[0], correct_lat_lon=correct_lat_lon,
run_check=run_check, check_extra=check_extra,
run_check_acceptability=run_check_acceptability,
phase_type=phase_type, antenna_nums=antenna_nums,
ant_str=ant_str, bls=bls,
polarizations=polarizations, time_range=time_range)
if len(filepath) > 1:
for f in filepath[1:]:
uv2 = UVData()
uv2.read_miriad(f, correct_lat_lon=correct_lat_lon,
run_check=run_check, check_extra=check_extra,
run_check_acceptability=run_check_acceptability,
phase_type=phase_type, antenna_nums=antenna_nums,
ant_str=ant_str, bls=bls,
polarizations=polarizations, time_range=time_range)
self += uv2
del(uv2)
else:
miriad_obj = miriad.Miriad()
miriad_obj.read_miriad(filepath, correct_lat_lon=correct_lat_lon,
run_check=run_check, check_extra=check_extra,
run_check_acceptability=run_check_acceptability,
phase_type=phase_type, antenna_nums=antenna_nums,
ant_str=ant_str, bls=bls,
polarizations=polarizations, time_range=time_range)
except(ImportError):
# only import skip if importerror raised
# this way, nose import errors aren't raised
# unless this errors is already encountered.
from nose.plugins.skip import SkipTest
raise SkipTest('casacore not detected. casacore'
' must be installed for measurement set functionality')
from . import ms
if isinstance(filepath, (list, tuple)):
self.read_ms(filepath[0], run_check=run_check, check_extra=check_extra,
run_check_acceptability=run_check_acceptability,
data_column=data_column, pol_order=pol_order)
if len(filepath) > 1:
for f in filepath[1:]:
uv2 = UVData()
uv2.read_ms(f, run_check=run_check, check_extra=check_extra,
run_check_acceptability=run_check_acceptability,
data_column=data_column, pol_order=pol_order)
self += uv2
del(uv2)
else:
ms_obj = ms.MS()
ms_obj.read_ms(filepath, run_check=run_check, check_extra=check_extra,
run_check_acceptability=run_check_acceptability,
data_column=data_column, pol_order=pol_order)
self._convert_from_filetype(ms_obj)
del(ms_obj)
Add some extra checks on top of checks on UVBase class.
Check that all required parameters are set reasonably.
Check that required parameters exist and have appropriate shapes.
Optionally check if the values are acceptable.
Args:
run_check_acceptability: Option to check if values in required parameters
are acceptable. Default is True.
"""
# first run the basic check from UVBase
if np.all(self.ant_1_array == self.ant_2_array):
# Special case of only containing auto correlations
self._uvw_array.acceptable_range = (0.0, 0.0)
super(UVData, self).check(run_check_acceptability=run_check_acceptability)
# Check internal consistency of numbers which don't explicitly correspond
# to the shape of another array.
nants_data_calc = int(len(np.unique(self.ant_1_array.tolist() +
self.ant_2_array.tolist())))
if self.Nants_data != nants_data_calc:
raise ValueError('Nants_data must be equal to the number of unique '
'values in ant_1_array and ant_2_array')
if self.Nbls != len(np.unique(self.baseline_array)):
raise ValueError('Nbls must be equal to the number of unique '
'baselines in the data_array')
if self.Ntimes != len(np.unique(self.time_array)):
raise ValueError('Ntimes must be equal to the number of unique '
'times in the time_array')
"""
from . import uvfits
if isinstance(filename, (list, tuple)):
if not read_data:
raise ValueError('read_data cannot be False for a list of uvfits files')
self.read_uvfits(filename[0], antenna_nums=antenna_nums,
antenna_names=antenna_names, ant_str=ant_str,
bls=bls, frequencies=frequencies,
freq_chans=freq_chans, times=times,
polarizations=polarizations, blt_inds=blt_inds,
run_check=run_check, check_extra=check_extra,
run_check_acceptability=run_check_acceptability)
if len(filename) > 1:
for f in filename[1:]:
uv2 = UVData()
uv2.read_uvfits(f, antenna_nums=antenna_nums,
antenna_names=antenna_names, ant_str=ant_str,
bls=bls, frequencies=frequencies,
freq_chans=freq_chans, times=times,
polarizations=polarizations, blt_inds=blt_inds,
run_check=run_check, check_extra=check_extra,
run_check_acceptability=run_check_acceptability)
self += uv2
del(uv2)
else:
uvfits_obj = uvfits.UVFITS()
uvfits_obj.read_uvfits(filename, antenna_nums=antenna_nums,
antenna_names=antenna_names, ant_str=ant_str,
bls=bls, frequencies=frequencies,
freq_chans=freq_chans, times=times,
polarizations=polarizations, blt_inds=blt_inds,
parameters after reading in the file. Default is True.
check_extra: Option to check optional parameters as well as required
ones. Default is True.
run_check_acceptability: Option to check acceptable range of the values of
parameters after reading in the file. Default is True.
Returns:
None
"""
from . import uvh5
if isinstance(filename, (list, tuple)):
self.read_uvh5(filename[0], run_check=run_check, check_extra=check_extra,
run_check_acceptability=run_check_acceptability)
if len(filename) > 1:
for f in filename[1:]:
uv2 = UVData()
uv2.read_uvh5(f, run_check=run_check, check_extra=check_extra,
run_check_acceptability=run_check_acceptability)
self += uv2
del(uv2)
else:
uvh5_obj = uvh5.UVH5()
uvh5_obj.read_uvh5(filename, run_check=run_check, check_extra=check_extra,
run_check_acceptability=run_check_acceptability)
self._convert_from_filetype(uvh5_obj)
del(uvh5_obj)
__all__ = ["input_output_mapping", "MWACorrFITS"]
def input_output_mapping():
"""Build a mapping dictionary from pfb input to output numbers."""
# the polyphase filter bank maps inputs to outputs, which the MWA
# correlator then records as the antenna indices.
# the following is taken from mwa_build_lfiles/mwac_utils.c
# inputs are mapped to outputs via pfb_mapper as follows
# (from mwa_build_lfiles/antenna_mapping.h):
# floor(index/4) + index%4 * 16 = input
# for the first 64 outputs, pfb_mapper[output] = input
return _corr_fits.input_output_mapping()
class MWACorrFITS(UVData):
"""
UVData subclass for reading MWA correlator fits files.
This class should not be interacted with directly; instead use the
read_mwa_corr_fits method on the UVData class.
"""
def correct_cable_length(self, cable_lens):
"""
Apply a cable length correction to the data array.
Parameters
----------
cable_lens : list of strings
A list of strings containing the cable lengths for each antenna.
"""
description='DUT1 (google it) AIPS 117 '
'calls it UT1UTC',
spoof_val=0.0, expected_type=np.float)
self._timesys = uvp.UVParameter('timesys', required=False,
description='We only support UTC',
spoof_val='UTC', form='str')
desc = ('FHD thing we do not understand, something about the time '
'at which the phase center is normal to the chosen UV plane '
'for phasing')
self._uvplane_reference_time = uvp.UVParameter('uvplane_reference_time',
required=False,
description=desc,
spoof_val=0)
super(UVData, self).__init__()
dirty visibilities. Default is False.
run_check: Option to check for the existence and proper shapes of
parameters after reading in the file. Default is True.
check_extra: Option to check optional parameters as well as required
ones. Default is True.
run_check_acceptability: Option to check acceptable range of the values of
parameters after reading in the file. Default is True.
"""
from . import fhd
if isinstance(filelist[0], (list, tuple)):
self.read_fhd(filelist[0], use_model=use_model, run_check=run_check,
check_extra=check_extra,
run_check_acceptability=run_check_acceptability)
if len(filelist) > 1:
for f in filelist[1:]:
uv2 = UVData()
uv2.read_fhd(f, use_model=use_model, run_check=run_check,
check_extra=check_extra,
run_check_acceptability=run_check_acceptability)
self += uv2
del(uv2)
else:
fhd_obj = fhd.FHD()
fhd_obj.read_fhd(filelist, use_model=use_model, run_check=run_check,
check_extra=check_extra,
run_check_acceptability=run_check_acceptability)
self._convert_from_filetype(fhd_obj)
del(fhd_obj)