docs: 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