class_docstr: Object class to read the index (.idx) file
  
  Args:
      idx_path (str): The path to the index file
  
      multimodal (bool): Whether the dataset is multimodal
class_docstr_tok: ["Object","class","to","read","the","index","(",".idx",")","file","Args",":","idx_path","(","str",")",":","The","path","to","the","index","file","multimodal","(","bool",")",":","Whether","the","dataset","is","multimodal"]
code: def __init__(self, idx_path: str, multimodal: bool) -> None:
      log_single_rank(
          logger,
          logging.INFO,
          f"Load the {type(self).__name__} from {idx_path}",
      )
  
      with open(idx_path, "rb") as stream:
          header = stream.read(9)
          assert (
              header == _INDEX_HEADER
          ), f"bad header, cannot read: {idx_path}"
  
          version = struct.unpack("<Q", stream.read(8))[0]
          assert version == 1, f"bad version, cannot read: {idx_path}"
  
          code = struct.unpack("<B", stream.read(1))[0]
          self.dtype = DType.dtype_from_code(code)
          self.dtype_size = DType.size(self.dtype)
  
          self.sequence_count = struct.unpack("<Q", stream.read(8))[0]
          self.document_count = struct.unpack("<Q", stream.read(8))[0]
  
          offset = stream.tell()
  
      self.bin_buffer_mmap = numpy.memmap(idx_path, mode="r", order="C")
      self.bin_buffer = memoryview(self.bin_buffer_mmap)
  
      log_single_rank(
          logger, logging.INFO, f"\tExtract the sequence lengths"
      )
      t_beg = time.time()
      self.sequence_lengths = numpy.frombuffer(
          self.bin_buffer,
          dtype=numpy.int32,
          count=self.sequence_count,
          offset=offset,
      )
      t_end = time.time()
      log_single_rank(
          logger,
          logging.DEBUG,
          f"\t> time elapsed: {t_end - t_beg:4f} seconds",
      )
  
      log_single_rank(
          logger, logging.INFO, f"\tExtract the sequence pointers"
      )
      t_beg = time.time()
      self.sequence_pointers = numpy.frombuffer(
          self.bin_buffer,
          dtype=numpy.int64,
          count=self.sequence_count,
          offset=offset + self.sequence_lengths.nbytes,
      )
      t_end = time.time()
      log_single_rank(
          logger,
          logging.DEBUG,
          f"\t> time elapsed: {t_end - t_beg:4f} seconds",
      )
  
      log_single_rank(
          logger, logging.INFO, f"\tExtract the document indices"
      )
      t_beg = time.time()
      self.document_indices = numpy.frombuffer(
          self.bin_buffer,
          dtype=numpy.int64,
          count=self.document_count,
          offset=offset
          + self.sequence_lengths.nbytes
          + self.sequence_pointers.nbytes,
      )
      t_end = time.time()
      log_single_rank(
          logger,
          logging.DEBUG,
          f"\t> time elapsed: {t_end - t_beg:4f} seconds",
      )
  
      self.sequence_modes = None
      if multimodal:
          log_single_rank(
              logger, logging.INFO, f"\tExtract the sequence modes"
          )
          t_beg = time.time()
          self.sequence_modes = numpy.frombuffer(
              self.bin_buffer,
              dtype=numpy.int8,
              count=self.sequence_count,
              offset=offset
              + self.sequence_lengths.nbytes
              + self.sequence_pointers.nbytes
              + self.document_indices.nbytes,
          )
          t_end = time.time()
          log_single_rank(
              logger,
              logging.DEBUG,
              f"\t> time elapsed: {t_end - t_beg:4f} seconds",
          )
  
      assert self.sequence_lengths.shape[0] == len(self)
      assert self.sequence_lengths.shape[0] == self.sequence_count
      assert self.sequence_lengths.shape[0] == self.document_indices[-1]
  
      log_single_rank(
          logger,
          logging.INFO,
          f"> total number of sequences: {len(self)}",
      )
      log_single_rank(
          logger,
          logging.INFO,
          f"> total number of documents: {self.document_indices.shape[0] - 1}",
      )
  
imports: import logging
  import os
  import shutil
  import struct
  import time
  from enum import Enum
  from functools import lru_cache
  from itertools import accumulate
  from types import TracebackType
  from typing import List, Optional, Tuple, Type, Union
  import numpy
  import torch
  from megatron.core.datasets.utils import log_single_rank
label_desc: Use image node_id 1 to create a new _IndexReader object from inherited base classes: object with example: obj = _IndexReader(idx_path, multimodal)