uvcal.py
import numpy as np
from uvbase import UVBase
import parameter as uvp
import version as uvversion
class UVCal(UVBase):
""" A class defining a calibration """
def __init__(self):
radian_tol = 10 * 2 * np.pi * 1e-3 / (60.0 * 60.0 * 360.0)
self._Nfreqs = uvp.UVParameter('Nfreqs',
description='Number of frequency channels',
expected_type=int)
self._Njones = uvp.UVParameter('Njones',
description='Number of polarizations calibration'
'parameters (Number of jones matrix elements.).',
expected_type=int)
self._Ntimes = uvp.UVParameter('Ntimes',
description='Number of times',
expected_type=int)
self._history = uvp.UVParameter('history',
description='String of history, units English',
form='str', expected_type=str)
self._Nspws = uvp.UVParameter('Nspws', description='Number of spectral windows '
'(ie non-contiguous spectral chunks). '
'More than one spectral window is not '
'currently supported.', expected_type=int)
desc = ('Frequency range that gain solutions are valid for.',
'list: (start_frequency, end_frequency) in Hz.')
self._freq_range = uvp.UVParameter('freq_range',
description=desc,
form=(2,),
expected_type=float)
desc = ('Time range (in JD) that gain solutions are valid for.',
'list: (start_time, end_time) in JD.')
self._time_range = uvp.UVParameter('time_range',
description=desc,
form=(2,),
expected_type=float)
desc = ('Name of telescope. e.g. HERA. String.')
self._telescope_name = uvp.UVParameter('telescope_name',
description=desc,
form='str',
expected_type=str)
desc = ('Number of antennas with data present (i.e. number of unique '
'entries in ant_1_array and ant_2_array). May be smaller ' +
'than the number of antennas in the array')
self._Nants_data = uvp.UVParameter('Nants_data', description=desc,
expected_type=int)
desc = ('Number of antennas in the array. May be larger ' +
'than the number of antennas with data')
self._Nants_telescope = uvp.UVParameter('Nants_telescope',
description=desc,
expected_type=int)
desc = ('List of antenna names, shape (Nants_telescope), '
'with numbers given by antenna_numbers (which can be matched '
'to ant_1_array and ant_2_array). There must be one entry '
'here for each unique entry in ant_1_array and '
'ant_2_array, but there may be extras as well.')
self._antenna_names = uvp.UVParameter('antenna_names',
description=desc,
form=('Nants_telescope',),
expected_type=str)
desc = ('List of integer antenna numbers corresponding to antenna_names,'
'shape (Nants_telescope). There must be one '
'entry here for each unique entry in ant_1_array and '
'ant_2_array, but there may be extras as well.')
self._antenna_numbers = uvp.UVParameter('antenna_numbers',
description=desc,
form=('Nants_telescope',),
expected_type=int)
desc = 'Array of frequencies, shape (Nspws, Nfreqs), units Hz'
self._freq_array = uvp.UVParameter('freq_array', description=desc,
form=('Nspws', 'Nfreqs'),
expected_type=np.float,
tols=1e-3) # mHz
desc = ('Channel width of of a frequency bin. Units Hz.')
self._channel_width = uvp.UVParameter('channel_width',
description=desc,
expected_type=np.float,
tols=1e-3)
desc = ('Array of antenna polarization integers, shape (Njones). '
'linear pols -5:-8 (jxx, jyy, jxy, jyx).'
'circular pols -1:-4 (jrr, jll. jrl, jlr).')
self._jones_array = uvp.UVParameter('jones_array',
description=desc,
expected_type=int,
acceptable_vals=list(np.arange(-8, 0)),
form=('Njones',))
desc = ('Array of times, center of integration, shape (Ntimes), ' +
'units Julian Date')
self._time_array = uvp.UVParameter('time_array', description=desc,
form=('Ntimes',),
expected_type=np.float,
tols=1e-3 / (60.0 * 60.0 * 24.0))
desc = ('Integration time of a time bin (s).')
self._integration_time = uvp.UVParameter('integration_time',
description=desc,
expected_type=np.float,
tols=1e-3) # 1ms
desc = ('The convention for applying he calibration solutions to data.'
'Indicates that to calibrate one should divide or multiply '
'uncalibrated data by gains.')
self._gain_convention = uvp.UVParameter('gain_convention', form='str',
expected_type=str,
description=desc,
acceptable_vals=['divide', 'multiply'])
desc = ('Array of flags to be applied to calibrated data (logical OR \
of input and flag generated by calibration). True is flagged.'
'shape: (Nants_data, Nfreqs, Ntimes, Njones), type = bool.')
self._flag_array = uvp.UVParameter('flag_array', description=desc,
form=('Nants_data', 'Nfreqs',
'Ntimes', 'Njones'),
expected_type=np.bool)
desc = ('Array of qualities of calibration solutions \
shape: (Nants_data, Nfreqs, Ntimes, '
'Njones), type = float.')
self._quality_array = uvp.UVParameter('quality_array', description=desc,
form=('Nants_data', 'Nfreqs',
'Ntimes', 'Njones'),
expected_type=np.float)
desc = ('Orientation of the physical dipole corresponding to what is '
'labelled as the x polarization. Values are east '
'(east/west orientation), north (north/south orientation) or '
'unknown.')
self._x_orientation = uvp.UVParameter('x_orientation', description=desc,
expected_type=str,
acceptable_vals=['east', 'north', 'unknown'])
# --- cal_type parameters ---
desc = ('cal type parameter. Values are delay, gain or unknown.')
self._cal_type = uvp.UVParameter('cal_type', form='str',
expected_type=str, value='unknown',
description=desc,
acceptable_vals=['delay', 'gain', 'unknown'])
desc = ('Array of gains, shape: (Nants_data, Nfreqs, Ntimes, '
'Njones), type = complex float.')
self._gain_array = uvp.UVParameter('gain_array', description=desc,
required=False,
form=('Nants_data', 'Nfreqs',
'Ntimes', 'Njones'),
expected_type=np.complex)
desc = ('Array of delays. shape: (Nants_data, Ntimes, Njones), type = float')
self._delay_array = uvp.UVParameter('delay_array', description=desc,
required=False,
form=('Nants_data', 1, 'Ntimes', 'Njones'),
expected_type=np.float)
# --- truly optional parameters ---
desc = ('Array of input flags, True is flagged. shape: (Nants_data, '
'Nfreqs, Ntimes, Njones), type = bool.')
self._input_flag_array = uvp.UVParameter('input_flag_array',
description=desc,
required=False,
form=('Nants_data', 'Nfreqs',
'Ntimes', 'Njones'),
expected_type=np.bool)
desc = ('Origin (on github for e.g) of calibration software. Url and branch.')
self._git_origin_cal = uvp.UVParameter('git_origin_cal', form='str',
expected_type=str,
description=desc,
required=False)
desc = ('Commit hash of calibration software(from git_origin_cal) used'
'to generate solutions.')
self._git_hash_cal = uvp.UVParameter('git_hash_cal', form='str',
expected_type=str,
description=desc,
required=False)
desc = ('Name of observer who calculated solutions in this file.')
self._observer = uvp.UVParameter('observer', form='str',
description=desc,
expected_type=str,
required=False)
# String to add to history of any files written with this version of pyuvdata
self.pyuvdata_version_str = (' Read/written with pyuvdata version: ' +
uvversion.version + '.')
if uvversion.git_hash is not '':
self.pyuvdata_version_str += (' Git origin: ' + uvversion.git_origin +
'. Git hash: ' + uvversion.git_hash +
'. Git branch: ' + uvversion.git_branch +
'. Git description: ' + uvversion.git_description)
super(UVCal, self).__init__()
def set_gain(self):
"""Set cal_type to 'gain' and adjust required parameters."""
self.cal_type = 'gain'
self._gain_array.required = True
self._delay_array.required = False
self._quality_array.form = self._gain_array.form
def set_delay(self):
"""Set cal_type to 'delay' and adjust required parameters."""
self.cal_type = 'delay'
self._gain_array.required = False
self._delay_array.required = True
self._quality_array.form = self._delay_array.form
def set_unknown_cal_type(self):
"""Set cal_type to 'unknown' and adjust required parameters."""
self.cal_type = 'unknown'
self._gain_array.required = False
self._delay_array.required = False
self._quality_array.form = self.gain_array.form
def _convert_from_filetype(self, other):
for p in other:
param = getattr(other, p)
setattr(self, p, param)
def _convert_to_filetype(self, filetype):
if filetype is 'calfits':
import calfits
other_obj = calfits.CALFITS()
else:
raise ValueError('filetype must be calfits.')
for p in self:
param = getattr(self, p)
setattr(other_obj, p, param)
return other_obj
def read_calfits(self, filename, run_check=True, run_check_acceptability=True):
"""
Read in data from a calfits file.
Args:
filename: The uvfits file to read from.
"""
import calfits
uvfits_obj = calfits.CALFITS()
uvfits_obj.read_calfits(filename, run_check=run_check,
run_check_acceptability=run_check_acceptability)
self._convert_from_filetype(uvfits_obj)
del(uvfits_obj)
def write_calfits(self, filename, spoof_nonessential=False,
run_check=True, run_check_acceptability=True, clobber=False):
"""Write data to a calfits file.
Args:
filename: The calfits filename to write to.
spoof_nonessential: Option to spoof the values of optional
UVParameters that are not set but are required for uvfits files.
Default is False.
run_check: Option to check for the existence and proper shapes of
required parameters before writing the file. Default is True.
run_check_acceptability: Option to check acceptability of the values of
required parameters before writing the file. Default is True.
clobber: Overwrite file.
"""
calfits_obj = self._convert_to_filetype('calfits')
calfits_obj.write_calfits(filename,
spoof_nonessential=spoof_nonessential,
run_check=run_check,
run_check_acceptability=run_check_acceptability,
clobber=clobber)
del(calfits_obj)