code: def call_variants(
      examples_filename: str, checkpoint_path: str, output_file: str
  ):
      """Main driver of call_variants."""
      output_queue = multiprocessing.Queue()
  
      # See if GPU is available
      is_gpu_available = (
          True if tf.config.list_physical_devices("GPU") else False
      )
  
      # This means we are in autodetect mode.
      if FLAGS.writer_threads == 0:
          # If GPU is available then use all CPUs for writing.
          total_writer_process = (
              multiprocessing.cpu_count() if is_gpu_available else 1
          )
      else:
          total_writer_process = FLAGS.writer_threads
  
      # 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."
          )
          total_writer_process = 1
          paths = [output_file]
      else:
          # Use maximum _MAX_WRITER_THREADS threads, this is to lower the overhead of
          # spinning up and shutting down many processes.
          total_writer_process = min(
              total_writer_process, _MAX_WRITER_THREADS
          )
          # Convert output filename to sharded output filename.
          output_file = output_file.replace(
              ".tfrecord.gz",
              "@" + str(total_writer_process) + ".tfrecord.gz",
          )
          paths = sharded_file_utils.maybe_generate_sharded_filenames(
              output_file
          )
  
      all_processes = []
      for process_id in range(0, total_writer_process):
          post_processing_process = (
              multiprocessing.get_context().Process(
                  target=post_processing,
                  args=(
                      paths[process_id],
                      output_queue,
                  ),
              )
          )
          all_processes.append(post_processing_process)
          post_processing_process.start()
  
      logging.info(
          "Total %d writing processes started.", len(all_processes)
      )
  
      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 = dv_utils.get_shape_and_channels_from_json(
          example_info_json
      )[0]
  
      if example_shape is None:
          logging.info(
              (
                  "Unable to read shape information from %s. Directly read from "
                  "examples instead."
              ),
              example_info_json,
          )
          example_shape = dv_utils.example_image_shape(first_example)
  
      logging.info("Shape of input examples: %s", str(example_shape))
      use_saved_model = tf.io.gfile.exists(
          checkpoint_path
      ) and tf.io.gfile.exists(f"{checkpoint_path}/saved_model.pb")
      logging.info("Use saved model: %s", str(use_saved_model))
  
      if checkpoint_path is not None:
          if use_saved_model:
              model = tf.saved_model.load(checkpoint_path)
              model_example_info_json = (
                  f"{checkpoint_path}/example_info.json"
              )
              model_example_shape = (
                  dv_utils.get_shape_and_channels_from_json(
                      model_example_info_json
                  )
              )
              # This usually happens in multi-sample cases where the sample_name
              # is added to the prefix, so we remove it.
              if not tf.io.gfile.exists(example_info_json):
                  # Find the sample name by finding file prefix and then the last bit
                  # would be the sample name.
                  filename_prefix = os.path.basename(
                      example_info_json
                  ).split(".")[0]
                  sample_name = filename_prefix.split("_")[-1]
                  expected_filename = example_info_json.replace(
                      "_" + sample_name, ""
                  )
                  if not tf.io.gfile.exists(expected_filename):
                      raise ValueError(
                          f"File {example_info_json} or {expected_filename} not found."
                          "Please check make_examples output."
                      )
                  example_info_json = expected_filename
  
              input_example_shape = (
                  dv_utils.get_shape_and_channels_from_json(
                      example_info_json
                  )
              )
              # These checks make sure we are using the right model with right input.
              if model_example_shape[0] != input_example_shape[0]:
                  # The following has been changed from ValueError to Warning because of
                  # internal
                  logging.warning(
                      "Input shape %s and model shape %s does not match.",
                      str(input_example_shape[0]),
                      str(model_example_shape[0]),
                  )
              if model_example_shape[1] != input_example_shape[1]:
                  # The following has been changed from ValueError to Warning because of
                  # internal
                  logging.warning(
                      "Input channels %s and model channels %s do not match.",
                      str(input_example_shape[1]),
                      str(model_example_shape[1]),
                  )
          else:
              model = modeling.inceptionv3(
                  example_shape, init_backbone_with_imagenet=False
              )
              model.load_weights(checkpoint_path).expect_partial()
  
          enc_image_variant_alt_allele_ds = get_dataset(
              examples_filename, example_shape
          )
  
          batch_no = 0
          n_examples = 0
          n_batches = 0
          start_time = time.time()
          for batch in enc_image_variant_alt_allele_ds:
              if use_saved_model:
                  predictions = model.signatures["serving_default"](
                      batch[1]
                  )
                  predictions = predictions["classification"].numpy()
              else:
                  if not is_gpu_available:
                      # This is faster on CPU but slower on GPU.
                      predictions = model.predict_on_batch(batch[1])
                  else:
                      # This is faster on GPU but slower on CPU.
                      predictions = model(
                          batch[1], training=False
                      ).numpy()
              batch_no += 1
              duration = time.time() - start_time
              n_examples += len(predictions)
              n_batches += 1
              logging.log_every_n(
                  logging.INFO,
                  "Predicted %s examples in %s batches [%.3f sec per 100].",
                  _LOG_EVERY_N_BATCHES,
                  n_examples,
                  n_batches,
                  (100 * duration) / n_examples,
              )
              output_queue.put(
                  (predictions, batch[0], batch[2], batch[3], batch[4])
              )
  
          # Put none values in the queue so the running processes can terminate.
          for _ in range(0, total_writer_process):
              output_queue.put(None)
          for post_processing_process in all_processes:
              post_processing_process.join()
  
imports: import multiprocessing
  import os
  import time
  from typing import Any
  from absl import flags
  from absl import logging
  import numpy
  import tensorflow
  from deepvariant import dv_utils
  from deepvariant import keras_modeling
  from deepvariant import logging_level
  from deepvariant.protos import deepvariant_pb2
  from absl import app
  import multiprocessing
  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 6 for calling a global function with example usage: call_variants(examples_filename, checkpoint_path, output_file) without return types