code: def call_variants(
      examples_filename,
      checkpoint_path,
      model,
      output_file,
      execution_hardware="auto",
      batch_size=16,
      max_batches=None,
      use_tpu=False,
      primary="",
  ):
      """Main driver of call_variants."""
      if FLAGS.kmp_blocktime:
          os.environ["KMP_BLOCKTIME"] = FLAGS.kmp_blocktime
          logging.vlog(
              3,
              "Set KMP_BLOCKTIME to {}".format(
                  os.environ["KMP_BLOCKTIME"]
              ),
          )
  
      # Read a single TFExample to make sure we're not loading an older version.
      first_example = dv_utils.get_one_example_from_examples_path(
          examples_filename
      )
      if first_example is None:
          logging.warning(
              "Unable to read any records from %s. Output will contain zero records.",
              examples_filename,
          )
          tfrecord.write_tfrecords([], output_file)
          return
  
      example_info_json = dv_utils.get_example_info_json_filename(
          examples_filename, 0
      )
      (
          example_shape,
          example_channels_enum,
      ) = dv_utils.get_shape_and_channels_from_json(example_info_json)
  
      # Check if the checkpoint_path has the same shape.
      if checkpoint_path is not None and example_shape is not None:
          reader = tf.compat.v1.train.NewCheckpointReader(
              checkpoint_path
          )
          shape_map_for_layers = reader.get_variable_to_shape_map()
          first_layer = "InceptionV3/Conv2d_1a_3x3/weights"
          # For a shape map of [3, 3, 6, 32] for the Conv2d_1a_3x3 layer, the 6
          # is the number of channels.
          num_channels_in_checkpoint_model = shape_map_for_layers[
              first_layer
          ][2]
          if num_channels_in_checkpoint_model != example_shape[2]:
              raise ValueError(
                  "The number of channels in examples and checkpoint "
                  "should match, but the checkpoint has {} channels while "
                  "the examples have {}.".format(
                      num_channels_in_checkpoint_model, example_shape[2]
                  )
              )
          input_info_file = os.path.join(
              os.path.dirname(checkpoint_path),
              "model.ckpt.example_info.json",
          )
          (
              ckpt_shape,
              ckpt_channels_enum,
          ) = dv_utils.get_shape_and_channels_from_json(input_info_file)
  
          if ckpt_shape is not None and ckpt_channels_enum is not None:
              if example_shape != ckpt_shape:
                  raise ValueError(
                      f"Shape mismatch in {example_info_json} and {input_info_file}."
                  )
              if example_channels_enum != ckpt_channels_enum:
                  raise ValueError(
                      f"Channels mismatch in {example_info_json} and {input_info_file}."
                  )
          else:
              # We can consider more strictly enforcing this.
              logging.warning(
                  "Starting from v1.4.0, we recommend having a "
                  "model.ckpt.example_info.json file with your model."
              )
  
      # Check accelerator status.
      if execution_hardware not in _ALLOW_EXECUTION_HARDWARE:
          raise ValueError(
              "Unexpected execution_hardware={} value. Allowed values are {}".format(
                  execution_hardware,
                  ",".join(_ALLOW_EXECUTION_HARDWARE),
              )
          )
      init_op = tf.group(
          tf.compat.v1.global_variables_initializer(),
          tf.compat.v1.local_variables_initializer(),
      )
  
      config = tf.compat.v1.ConfigProto()
      if FLAGS.config_string is not None:
          text_format.Parse(FLAGS.config_string, config)
      if execution_hardware == "cpu":
          # Don't overwrite entire dictionary.
          config.device_count["GPU"] = 0
          config.device_count["TPU"] = 0
  
      # Perform sanity check.
      with tf.compat.v1.Session(config=config) as sess:
          sess.run(init_op)
          if execution_hardware == "accelerator":
              if not any(
                  dev.device_type != "CPU"
                  for dev in sess.list_devices()
              ):
                  raise ExecutionHardwareError(
                      "execution_hardware is set to accelerator, but no accelerator "
                      "was found"
                  )
          # TODO. Sort out auto-detection of TPU. Just calling
          # sess.list_devices here doesn't return the correct answer. That can only
          # work later, after the device (on the other VM) has been initialized,
          # which is generally not yet.
  
      # Prepare input stream and estimator.
      tf_dataset = prepare_inputs(
          source_path=examples_filename, use_tpu=use_tpu
      )
      estimator = model.make_estimator(
          batch_size=batch_size,
          master=primary,
          use_tpu=use_tpu,
          session_config=config,
          include_debug_info=FLAGS.include_debug_info,
      )
  
      # Instantiate the prediction "stream", and select the EMA values from
      # the model.
      if checkpoint_path is None:
          # Unit tests use this branch.
          predict_hooks = []
      else:
          predict_hooks = [
              h(checkpoint_path) for h in model.session_predict_hooks()
          ]
      predictions = iter(
          estimator.predict(
              input_fn=tf_dataset,
              checkpoint_path=checkpoint_path,
              hooks=predict_hooks,
          )
      )
  
      # The following code is introduced to be in sync with call_variants
      # where we use multiple writers to write outpts.
      # If output file is already sharded then don't dynamically shard.
      if sharded_file_utils.is_sharded_filename(output_file):
          logging.info(
              "Output is already sharded, so dynamic sharding is disabled."
          )
      else:
          # For call_variants, we always use one writer process.
          total_writer_process = 1
          # Convert output filename to sharded output filename.
          filename_pattern = output_file.replace(
              ".tfrecord.gz",
              "@" + str(total_writer_process) + ".tfrecord.gz",
          )
          output_file = (
              sharded_file_utils.maybe_generate_sharded_filenames(
                  filename_pattern
              )[0]
          )
      # Consume predictions one at a time and write them to output_file.
      logging.info("Writing calls to %s", output_file)
      writer = tfrecord.Writer(output_file)
      with writer:
          start_time = time.time()
          n_examples, n_batches = 0, 0
          while max_batches is None or n_batches <= max_batches:
              try:
                  prediction = next(predictions)
              except (StopIteration, tf.errors.OutOfRangeError):
                  break
              write_variant_call(writer, prediction, use_tpu)
              n_examples += 1
              n_batches = n_examples // batch_size + 1
              duration = time.time() - start_time
  
              logging.log_every_n(
                  logging.INFO,
                  "Processed %s examples in %s batches [%.3f sec per 100]",
                  _LOG_EVERY_N,
                  n_examples,
                  n_batches,
                  (100 * duration) / n_examples,
              )
          # One last log to capture the extra examples.
          duration = time.time() - start_time
          logging.info(
              "Processed %s examples in %s batches [%.3f sec per 100]",
              n_examples,
              n_batches,
              (100 * duration) / n_examples,
          )
  
          logging.info(
              "Done calling variants from a total of %d examples.",
              n_examples,
          )
  
imports: import os
  import time
  from absl import flags
  from absl import logging
  import numpy
  import tensorflow
  from tensorflow import estimator
  from deepvariant import data_providers
  from deepvariant import dv_utils
  from deepvariant import logging_level
  from deepvariant import modeling
  from deepvariant.protos import deepvariant_pb2
  from google.protobuf import text_format
  from third_party.nucleus.io import sharded_file_utils
  from third_party.nucleus.io import tfrecord
  from third_party.nucleus.protos import variants_pb2
  from third_party.nucleus.util import errors
  from third_party.nucleus.util import proto_utils
  from third_party.nucleus.util import variant_utils
label_desc: Use image node_id 5 for calling a global function with example usage: call_variants(examples_filename, checkpoint_path, model, output_file, execution_hardware, batch_size, max_batches, use_tpu, primary) without return types