code: def _read_files(
      file_instructions: Sequence[shard_utils.FileInstruction],
      read_config: read_config_lib.ReadConfig,
      shuffle_files: bool,
      disable_shuffling: bool,
      file_format: file_adapters.FileFormat,
  ) -> tf.data.Dataset:
      """Returns tf.data.Dataset for given file instructions.
  
      Args:
        file_instructions: the information on the files to read including
          `ds.skip().take()`
        read_config: Additional options to configure the input pipeline (e.g. seed,
          num parallel reads,...).
        shuffle_files: Defaults to False. True to shuffle input files.
        disable_shuffling: Specifies if the dataset being read has shuffling
          disabled.
        file_format: Format of the record files in which the dataset will be
          read/written from.
  
      Returns:
        The dataset object.
      """
  
      def assert_cardinality_and_apply_options(ds):
          # If the number of examples read in the tf-record is known, we forward
          # the information to the tf.data.Dataset object.
          # Check the `tf.data.experimental` for backward compatibility with TF <= 2.1
          if (
              read_config.assert_cardinality
              and not read_config.input_context
              and hasattr(  # TODO(epot): Restore cardinality
                  tf.data.experimental, "assert_cardinality"
              )
          ):
              # TODO(b/154963426): Replace by per-shard cardinality (warning if
              # `experimental_interleave_sort_fn` is set).
              cardinality = sum(f.take for f in file_instructions)
              ds = ds.apply(
                  tf.data.experimental.assert_cardinality(cardinality)
              )
  
          return ds.with_options(
              read_config.options
          )  # Additional users options
  
      def validate_input_context():
          if not read_config.input_context:
              raise ValueError(
                  "Cannot shard the pipeline with undefined `input_context`."
              )
          try:
              num_input_pipelines = int(
                  read_config.input_context.num_input_pipelines
              )
          except TypeError as e:
              logging.warning(
                  "Unable to validate `read_config.input_context`: %s",
                  e,
              )
              return
          if (
              num_input_pipelines > 1
              and len(file_instructions) < num_input_pipelines
          ):
              raise ValueError(
                  "Cannot shard the pipeline with given `input_context`."
                  "`num_shards={}` but `num_input_pipelines={}`. This means that some "
                  "workers won't read any data. To shard the data, you may want to "
                  "use the subsplit API instead: "
                  "https://www.tensorflow.org/datasets/splits".format(
                      len(file_instructions),
                      read_config.input_context.num_input_pipelines,
                  )
              )
  
      # Eventually apply a transformation to the instruction function.
      # This allow the user to have direct control over the interleave order.
      if read_config.experimental_interleave_sort_fn is not None:
          file_instructions = (
              read_config.experimental_interleave_sort_fn(
                  file_instructions
              )
          )
  
      do_skip = any(f.skip > 0 for f in file_instructions)
      do_take = any(not f.takes_all for f in file_instructions)
  
      tfds_id_prefixes = _get_tfds_id_prefixes(file_instructions)
  
      # Transpose the list[dict] into dict[list]
      tensor_inputs = _Instruction(
          filename=[
              os.path.basename(i.filename) for i in file_instructions
          ],
          filepath=[os.fspath(i.filename) for i in file_instructions],
          tfds_id_prefix=[
              tfds_id_prefixes[i.filename] for i in file_instructions
          ],
          # skip/take need to be converted to int64 explicitly
          skip=np.array(
              [i.skip for i in file_instructions], dtype=np.int64
          ),
          take=np.array(
              [i.take for i in file_instructions], dtype=np.int64
          ),
      )
  
      cycle_length = read_config.interleave_cycle_length
      block_length = read_config.interleave_block_length
  
      if cycle_length == read_config_lib.MISSING:
          cycle_length = _get_default_interleave_cycle_length(
              disable_shuffling=disable_shuffling
          )
  
      if disable_shuffling:
          _verify_read_config_for_ordered_dataset(
              read_config,
              interleave_cycle_length=cycle_length,
              shuffle_files=shuffle_files,
          )
  
      unique_folders = set(
          os.path.dirname(path) for path in tensor_inputs.filepath
      )
      logging.info(
          "Creating a tf.data.Dataset reading %d files located in folders: %s.",
          len(tensor_inputs.filepath),
          ", ".join(sorted(unique_folders)),
      )
  
      instruction_ds = tf.data.Dataset.from_tensor_slices(tensor_inputs)
  
      # On distributed environments, we can shard per-file if a
      # `tf.distribute.InputContext` object is provided (e.g. from
      # `experimental_distribute_datasets_from_function`)
      if read_config.input_context:
          validate_input_context()
          instruction_ds = instruction_ds.shard(
              num_shards=read_config.input_context.num_input_pipelines,
              index=read_config.input_context.input_pipeline_id,
          )
  
      # If shuffle is True, we shuffle the instructions/shards
      if shuffle_files:
          instruction_ds = instruction_ds.shuffle(
              len(file_instructions),
              seed=read_config.shuffle_seed,
              reshuffle_each_iteration=read_config.shuffle_reshuffle_each_iteration,
          )
  
      if read_config.repeat_filenames:
          instruction_ds = instruction_ds.repeat()
  
      deterministic = True
      # If shuffling is true and the seed is not set, then the pipeline is allowed
      # to be non-deterministic. Note that setting the deterministic option in
      # tf.data.Options() sets this option globally. This means that also nested
      # fields will appear in undeterministic order.
      if (
          shuffle_files
          and read_config.shuffle_seed is None
          and tf_compat.get_option_deterministic(read_config.options)
          is None
      ):
          deterministic = False
  
      ds = instruction_ds.interleave(
          functools.partial(
              _get_dataset_from_filename,
              do_skip=do_skip,
              do_take=do_take,
              file_format=file_format,
              add_tfds_id=read_config.add_tfds_id,
              override_buffer_size=read_config.override_buffer_size,
          ),
          cycle_length=cycle_length,
          block_length=block_length,
          num_parallel_calls=read_config.num_parallel_calls_for_interleave_files,
          deterministic=deterministic,
      )
  
      return assert_cardinality_and_apply_options(ds)
  
imports: from __future__ import annotations
  import functools
  import os
  import re
  from typing import Any, Callable, List, NamedTuple, Optional, Sequence
  from absl import logging
  import numpy
  from tensorflow_datasets.core import example_parser
  from tensorflow_datasets.core import file_adapters
  from tensorflow_datasets.core import splits
  from tensorflow_datasets.core import tf_compat
  from tensorflow_datasets.core import utils
  from tensorflow_datasets.core.utils import file_utils
  from tensorflow_datasets.core.utils import read_config
  from tensorflow_datasets.core.utils import shard_utils
  from tensorflow_datasets.core.utils import tree_utils
  from tensorflow_datasets.core.utils.lazy_imports_utils import tensorflow
label_desc: Use image node_id 6 for calling a global function with example usage: _read_files(file_instructions, read_config, shuffle_files, disable_shuffling, file_format) and returns: assert_cardinality_and_apply_options, ds