def node_run(input_file, coords_only, bc_settings, bc_grid_weights):
    """Main function to process visibility data on Spark cluster nodes.

    Args:
        input_file (str):
            RDD element containing filename to process.
        coords_only (boolean):
            If true, read only baseline coordinates to define the weights grid.
        bc_settings (pyspark.broadcast.Broadcast):
            Spark broadcast variable containing pipeline settings dictionary.
        bc_grid_weights (pyspark.broadcast.Broadcast):
            Spark broadcast variable containing weights grid. May be None.

    Returns:
        tuple: Output RDD element.
    """
    # Create a logger.
    log = logging.getLogger('pyspark')
    log.setLevel(logging.INFO)
    if len(log.handlers) == 0:
        log.addHandler(logging.StreamHandler(sys.stdout))

    # Create an imager and configure it.
    precision = bc_settings.value['precision']
    imager = oskar.Imager(precision)
    for key, value in bc_settings.value['imager'].items():
        setattr(imager, key, value)
    grid_size = imager.plane_size
    grid_weights = None

    # Get a handle to the input Measurement Set.
    ms_han = oskar.MeasurementSet.open(input_file)

    # Check if doing a first pass.
    if coords_only:
        # If necessary, generate a local weights grid.
        if imager.weighting == 'Uniform':
            grid_weights = numpy.zeros([grid_size, grid_size], dtype=precision)

        # Do a first pass for uniform weighting or W-projection.
        log.info('Reading coordinates from %s', input_file)
        imager.coords_only = True
        process_input_data(ms_han, imager, None, grid_weights)
        imager.coords_only = False

        # Return weights grid and required number of W-planes as RDD element.
        return grid_weights, imager.num_w_planes

    # Allocate a local visibility grid on the node.
    grid_data = numpy.zeros([grid_size, grid_size],
                            dtype='c8' if precision == 'single' else 'c16')

    # Process data according to mode.
    log.info('Reading visibilities from %s', input_file)
    if bc_settings.value['combine']:
        # Get weights grid from Spark Broadcast variable.
        if imager.weighting == 'Uniform':
            grid_weights = bc_grid_weights.value

        # Populate the local visibility grid.
        grid_norm = process_input_data(ms_han, imager, grid_data, grid_weights)

        # Return grid as RDD element.
        log.info('Returning gridded visibilities to RDD')
        return grid_data, grid_norm
    else:
        # If necessary, generate a local weights grid.
        if imager.weighting == 'Uniform':
            grid_weights = numpy.zeros([grid_size, grid_size], dtype=precision)

        # If necessary, do a first pass for uniform weighting or W-projection.
        if imager.weighting == 'Uniform' or imager.algorithm == 'W-projection':
            imager.coords_only = True
            process_input_data(ms_han, imager, None, grid_weights)
            imager.coords_only = False

        # Populate the local visibility grid.
        grid_norm = process_input_data(ms_han, imager, grid_data, grid_weights)

        # Save image by finalising grid.
        output_file = splitext(input_file)[0] + '.fits'
        save_image(imager, grid_data, grid_norm, output_file)
        log.info('Finished. Output file is %s', output_file)
        return 0
def node_run ( input_file , coords_only , bc_settings , bc_grid_weights ) : log = logging . getLogger ( 'pyspark' ) log . setLevel ( logging . INFO ) if len ( log . handlers ) == 0 : log . addHandler ( logging . StreamHandler ( sys . stdout ) ) precision = bc_settings . value [ 'precision' ] imager = oskar . Imager ( precision ) for key , value in bc_settings . value [ 'imager' ] . items ( ) : setattr ( imager , key , value ) grid_size = imager . plane_size grid_weights = None ms_han = oskar . MeasurementSet . open ( input_file ) if coords_only : if imager . weighting == 'Uniform' : grid_weights = numpy . zeros ( [ grid_size , grid_size ] , dtype = precision ) log . info ( 'Reading coordinates from %s' , input_file ) imager . coords_only = True process_input_data ( ms_han , imager , None , grid_weights ) imager . coords_only = False return grid_weights , imager . num_w_planes grid_data = numpy . zeros ( [ grid_size , grid_size ] , dtype = 'c8' if precision == 'single' else 'c16' ) log . info ( 'Reading visibilities from %s' , input_file ) if bc_settings . value [ 'combine' ] : if imager . weighting == 'Uniform' : grid_weights = bc_grid_weights . value grid_norm = process_input_data ( ms_han , imager , grid_data , grid_weights ) log . info ( 'Returning gridded visibilities to RDD' ) return grid_data , grid_norm else : if imager . weighting == 'Uniform' : grid_weights = numpy . zeros ( [ grid_size , grid_size ] , dtype = precision ) if imager . weighting == 'Uniform' or imager . algorithm == 'W-projection' : imager . coords_only = True process_input_data ( ms_han , imager , None , grid_weights ) imager . coords_only = False grid_norm = process_input_data ( ms_han , imager , grid_data , grid_weights ) output_file = splitext ( input_file ) [ 0 ] + '.fits' save_image ( imager , grid_data , grid_norm , output_file ) log . info ( 'Finished. Output file is %s' , output_file ) return 0
Main function to process visibility data on Spark cluster nodes.

    Args:
        input_file (str):
            RDD element containing filename to process.
        coords_only (boolean):
            If true, read only baseline coordinates to define the weights grid.
        bc_settings (pyspark.broadcast.Broadcast):
            Spark broadcast variable containing pipeline settings dictionary.
        bc_grid_weights (pyspark.broadcast.Broadcast):
            Spark broadcast variable containing weights grid. May be None.

    Returns:
        tuple: Output RDD element.