<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 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