processing_components/visibility/base.py
def create_blockvisibility(config: Configuration,
                           times: numpy.array,
                           frequency: numpy.array,
                           phasecentre: SkyCoord,
                           weight: float = 1.0,
                           polarisation_frame: PolarisationFrame = None,
                           integration_time=1.0,
                           channel_bandwidth=1e6,
                           zerow=False,
                           elevation_limit=None,
                           source='unknown',
                           meta=None,
                           **kwargs) -> BlockVisibility:
    """ Create a BlockVisibility from Configuration, hour angles, and direction of source

    Note that we keep track of the integration time for BDA purposes

    :param config: Configuration of antennas
    :param times: hour angles in radians
    :param frequency: frequencies (Hz] [nchan]
    :param weight: weight of a single sample
    :param phasecentre: phasecentre of observation
    :param channel_bandwidth: channel bandwidths: (Hz] [nchan]
    :param integration_time: Integration time ('auto' or value in s)
    :param polarisation_frame:
    :return: BlockVisibility
    """
    assert phasecentre is not None, "Must specify phase centre"
    
    if polarisation_frame is None:
        polarisation_frame = correlate_polarisation(config.receptor_frame)
    
    latitude = config.location.geodetic[1].to('rad').value
    nch = len(frequency)
    ants_xyz = config.data['xyz']
    nants = len(config.data['names'])

    ntimes = 0
    n_flagged = 0
    for iha, ha in enumerate(times):
    
        # Calculate the positions of the antennas as seen for this hour angle
        # and declination
        _, elevation = hadec_to_azel(ha, phasecentre.dec.rad, latitude)
        if elevation_limit is None or (elevation > elevation_limit):
            ntimes +=1
        else:
            n_flagged += 1

    assert ntimes > 0, "No unflagged points"
    if elevation_limit is not None:
        log.info('create_visibility: flagged %d/%d times below elevation limit %f (rad)' %
                (n_flagged, ntimes, elevation_limit))
    else:
        log.info('create_visibility: created %d times' % (ntimes))
    
    npol = polarisation_frame.npol
    visshape = [ntimes, nants, nants, nch, npol]
    rvis = numpy.zeros(visshape, dtype='complex')
    rweight = weight * numpy.ones(visshape)
    rimaging_weight = numpy.ones(visshape)
    rtimes = numpy.zeros([ntimes])
    ruvw = numpy.zeros([ntimes, nants, nants, 3])
    
    # Do each hour angle in turn
    itime = 0
    for iha, ha in enumerate(times):
        
        # Calculate the positions of the antennas as seen for this hour angle
        # and declination
        ant_pos = xyz_to_uvw(ants_xyz, ha, phasecentre.dec.rad)
        _, elevation = hadec_to_azel(ha, phasecentre.dec.rad, latitude)
        if elevation_limit is None or (elevation > elevation_limit):
            rtimes[itime] = ha * 43200.0 / numpy.pi
            rweight[itime, ...] = 1.0

            # Loop over all pairs of antennas. Note that a2>a1
            for a1 in range(nants):
                for a2 in range(a1 + 1, nants):
                    ruvw[itime, a2, a1, :] = (ant_pos[a2, :] - ant_pos[a1, :])
                    ruvw[itime, a1, a2, :] = (ant_pos[a1, :] - ant_pos[a2, :])
            itime += 1
    
    rintegration_time = numpy.full_like(rtimes, integration_time)
    rchannel_bandwidth = channel_bandwidth
    if zerow:
        ruvw[..., 2] = 0.0
    vis = BlockVisibility(uvw=ruvw, time=rtimes, frequency=frequency, vis=rvis, weight=rweight,
                          imaging_weight=rimaging_weight,
                          integration_time=rintegration_time, channel_bandwidth=rchannel_bandwidth,
                          polarisation_frame=polarisation_frame, source=source, meta=meta)
    vis.phasecentre = phasecentre
    vis.configuration = config
    log.info("create_blockvisibility: %s" % (vis_summary(vis)))
    assert isinstance(vis, BlockVisibility), "vis is not a BlockVisibility: %r" % vis

    return vis
 Create a BlockVisibility from Configuration, hour angles, and direction of source

    Note that we keep track of the integration time for BDA purposes

    :param config: Configuration of antennas
    :param times: hour angles in radians
    :param frequency: frequencies (Hz] [nchan]
    :param weight: weight of a single sample
    :param phasecentre: phasecentre of observation
    :param channel_bandwidth: channel bandwidths: (Hz] [nchan]
    :param integration_time: Integration time ('auto' or value in s)
    :param polarisation_frame:
    :return: BlockVisibility
    
Create a BlockVisibility from Configuration, hour angles, and direction of source
Note that we keep track of the integration time for BDA purposes
Create a BlockVisibility from Configuration hour angles and direction of source Note that we keep track of the integration time for BDA purposes
def create_blockvisibility(config: Configuration,
                           times: numpy.array,
                           frequency: numpy.array,
                           phasecentre: SkyCoord,
                           weight: float = 1.0,
                           polarisation_frame: PolarisationFrame = None,
                           integration_time=1.0,
                           channel_bandwidth=1e6,
                           zerow=False,
                           elevation_limit=None,
                           source='unknown',
                           meta=None,
                           **kwargs) -> BlockVisibility:
    assert phasecentre is not None, "Must specify phase centre"
    if polarisation_frame is None:
        polarisation_frame = correlate_polarisation(config.receptor_frame)
    latitude = config.location.geodetic[1].to('rad').value
    nch = len(frequency)
    ants_xyz = config.data['xyz']
    nants = len(config.data['names'])
    ntimes = 0
    n_flagged = 0
    for iha, ha in enumerate(times):
        _, elevation = hadec_to_azel(ha, phasecentre.dec.rad, latitude)
        if elevation_limit is None or (elevation > elevation_limit):
            ntimes +=1
        else:
            n_flagged += 1
    assert ntimes > 0, "No unflagged points"
    if elevation_limit is not None:
        log.info('create_visibility: flagged %d/%d times below elevation limit %f (rad)' %
                (n_flagged, ntimes, elevation_limit))
    else:
        log.info('create_visibility: created %d times' % (ntimes))
    npol = polarisation_frame.npol
    visshape = [ntimes, nants, nants, nch, npol]
    rvis = numpy.zeros(visshape, dtype='complex')
    rweight = weight * numpy.ones(visshape)
    rimaging_weight = numpy.ones(visshape)
    rtimes = numpy.zeros([ntimes])
    ruvw = numpy.zeros([ntimes, nants, nants, 3])
    itime = 0
    for iha, ha in enumerate(times):
        ant_pos = xyz_to_uvw(ants_xyz, ha, phasecentre.dec.rad)
        _, elevation = hadec_to_azel(ha, phasecentre.dec.rad, latitude)
        if elevation_limit is None or (elevation > elevation_limit):
            rtimes[itime] = ha * 43200.0 / numpy.pi
            rweight[itime, ...] = 1.0
            for a1 in range(nants):
                for a2 in range(a1 + 1, nants):
                    ruvw[itime, a2, a1, :] = (ant_pos[a2, :] - ant_pos[a1, :])
                    ruvw[itime, a1, a2, :] = (ant_pos[a1, :] - ant_pos[a2, :])
            itime += 1
    rintegration_time = numpy.full_like(rtimes, integration_time)
    rchannel_bandwidth = channel_bandwidth
    if zerow:
        ruvw[..., 2] = 0.0
    vis = BlockVisibility(uvw=ruvw, time=rtimes, frequency=frequency, vis=rvis, weight=rweight,
                          imaging_weight=rimaging_weight,
                          integration_time=rintegration_time, channel_bandwidth=rchannel_bandwidth,
                          polarisation_frame=polarisation_frame, source=source, meta=meta)
    vis.phasecentre = phasecentre
    vis.configuration = config
    log.info("create_blockvisibility: %s" % (vis_summary(vis)))
    assert isinstance(vis, BlockVisibility), "vis is not a BlockVisibility: %r" % vis
    return vis
def create_blockvisibility ( config : Configuration , times : numpy . array , frequency : numpy . array , phasecentre : SkyCoord , weight : float = 1.0 , polarisation_frame : PolarisationFrame = None , integration_time = 1.0 , channel_bandwidth = 1e6 , zerow = False , elevation_limit = None , source = 'unknown' , meta = None , ** kwargs ) -> BlockVisibility : assert phasecentre is not None , "Must specify phase centre" if polarisation_frame is None : polarisation_frame = correlate_polarisation ( config . receptor_frame ) latitude = config . location . geodetic [ 1 ] . to ( 'rad' ) . value nch = len ( frequency ) ants_xyz = config . data [ 'xyz' ] nants = len ( config . data [ 'names' ] ) ntimes = 0 n_flagged = 0 for iha , ha in enumerate ( times ) : _ , elevation = hadec_to_azel ( ha , phasecentre . dec . rad , latitude ) if elevation_limit is None or ( elevation > elevation_limit ) : ntimes += 1 else : n_flagged += 1 assert ntimes > 0 , "No unflagged points" if elevation_limit is not None : log . info ( 'create_visibility: flagged %d/%d times below elevation limit %f (rad)' % ( n_flagged , ntimes , elevation_limit ) ) else : log . info ( 'create_visibility: created %d times' % ( ntimes ) ) npol = polarisation_frame . npol visshape = [ ntimes , nants , nants , nch , npol ] rvis = numpy . zeros ( visshape , dtype = 'complex' ) rweight = weight * numpy . ones ( visshape ) rimaging_weight = numpy . ones ( visshape ) rtimes = numpy . zeros ( [ ntimes ] ) ruvw = numpy . zeros ( [ ntimes , nants , nants , 3 ] ) itime = 0 for iha , ha in enumerate ( times ) : ant_pos = xyz_to_uvw ( ants_xyz , ha , phasecentre . dec . rad ) _ , elevation = hadec_to_azel ( ha , phasecentre . dec . rad , latitude ) if elevation_limit is None or ( elevation > elevation_limit ) : rtimes [ itime ] = ha * 43200.0 / numpy . pi rweight [ itime , ... ] = 1.0 for a1 in range ( nants ) : for a2 in range ( a1 + 1 , nants ) : ruvw [ itime , a2 , a1 , : ] = ( ant_pos [ a2 , : ] - ant_pos [ a1 , : ] ) ruvw [ itime , a1 , a2 , : ] = ( ant_pos [ a1 , : ] - ant_pos [ a2 , : ] ) itime += 1 rintegration_time = numpy . full_like ( rtimes , integration_time ) rchannel_bandwidth = channel_bandwidth if zerow : ruvw [ ... , 2 ] = 0.0 vis = BlockVisibility ( uvw = ruvw , time = rtimes , frequency = frequency , vis = rvis , weight = rweight , imaging_weight = rimaging_weight , integration_time = rintegration_time , channel_bandwidth = rchannel_bandwidth , polarisation_frame = polarisation_frame , source = source , meta = meta ) vis . phasecentre = phasecentre vis . configuration = config log . info ( "create_blockvisibility: %s" % ( vis_summary ( vis ) ) ) assert isinstance ( vis , BlockVisibility ) , "vis is not a BlockVisibility: %r" % vis return vis
Create a BlockVisibility from Configuration, hour angles, and direction of source
Note that we keep track of the integration time for BDA purposes
Create a BlockVisibility from Configuration hour angles and direction of source Note that we keep track of the integration time for BDA purposes
""" Create a BlockVisibility from Configuration, hour angles, and direction of source

    Note that we keep track of the integration time for BDA purposes

    :param config: Configuration of antennas
    :param times: hour angles in radians
    :param frequency: frequencies (Hz] [nchan]
    :param weight: weight of a single sample
    :param phasecentre: phasecentre of observation
    :param channel_bandwidth: channel bandwidths: (Hz] [nchan]
    :param integration_time: Integration time ('auto' or value in s)
    :param polarisation_frame:
    :return: BlockVisibility
    """ # Calculate the positions of the antennas as seen for this hour angle # and declination # Do each hour angle in turn # Calculate the positions of the antennas as seen for this hour angle # and declination # Loop over all pairs of antennas. Note that a2>a1
None
config
Configuration
Configuration of antennas
Configuration of antennas
times
numpy.array
hour angles in radians
hour angles in radians
frequency
numpy.array
frequencies (Hz] [nchan]
frequencies ( Hz ] [ nchan ]
phasecentre
SkyCoord
phasecentre of observation
phasecentre of observation
weight
float
weight of a single sample
weight of a single sample
polarisation_frame
PolarisationFrame
None
integration_time
Integration time ('auto' or value in s)
Integration time ( ' auto ' or value in s )
channel_bandwidth
channel bandwidths: (Hz] [nchan]
channel bandwidths : ( Hz ] [ nchan ]
zerow
elevation_limit
source
meta