class_docstr: Allow per-line lazy access to multiple text files using numpy memmap.
docstr: Args:
      dataset_paths: list of JSONL file paths.
      newline_int: ASCII code to use to interpret newlines in file.
      header_lines: number of header lines in JSON files.
      workers: number of workers to use for creating index files.
      tokenizer: tokenizer to use to convert text to tokens.
      build_index_fn: a callable build_index_fn(fn, newline_int) -> midx [np.array]
          that returns the index of newlines in a file fn must be pickleable
          (to be used in multiprocessing.Pool.map).
      sort_dataset_paths: whether to sort datasets by paths.
      index_mapping_dir: directory to save the index mapping to.
          If None, will write to the same folder as the dataset.
code: def __init__(
      self,
      dataset_paths: List[str],
      newline_int: Optional[int] = 10,
      header_lines: Optional[int] = 0,
      workers: Optional[int] = None,
      tokenizer: Optional[Type["TokenizerSpec"]] = None,
      build_index_fn: Optional[
          Callable[[str, Optional[int]], bool]
      ] = _build_index_from_memdata,
      sort_dataset_paths: Optional[bool] = True,
      index_mapping_dir: Optional[str] = None,
  ):
      """
      Args:
          dataset_paths: list of JSONL file paths.
          newline_int: ASCII code to use to interpret newlines in file.
          header_lines: number of header lines in JSON files.
          workers: number of workers to use for creating index files.
          tokenizer: tokenizer to use to convert text to tokens.
          build_index_fn: a callable build_index_fn(fn, newline_int) -> midx [np.array]
              that returns the index of newlines in a file fn must be pickleable
              (to be used in multiprocessing.Pool.map).
          sort_dataset_paths: whether to sort datasets by paths.
          index_mapping_dir: directory to save the index mapping to.
              If None, will write to the same folder as the dataset.
      """
      super().__init__()
      self.mdata_midx_list = []
  
      # Make a single string into a list
      if isinstance(dataset_paths, str):
          dataset_paths = [dataset_paths]
  
      if len(dataset_paths) < 1:
          raise ValueError(
              "files_list must contain at leat one file name"
          )
  
      self._newline_int = newline_int
      # skip first N lines
      self._header_lines = header_lines
      self._files_list = dataset_paths
      self._worker = workers
      self.tokenizer = tokenizer
      self._sort_dataset_paths = sort_dataset_paths
  
      if sort_dataset_paths:
          self._files_list = sorted(self._files_list)
  
      logging.info(f"Building data files")
      # load all files into memmap
      is_distributed = (
          torch.distributed.is_available()
          and torch.distributed.is_initialized()
      )
  
      if not is_distributed or (
          is_distributed and torch.distributed.get_rank() == 0
      ):
          # Create index files on global rank 0.
          build_index_files(
              dataset_paths,
              newline_int,
              workers=self._worker,
              build_index_fn=build_index_fn,
              index_mapping_dir=index_mapping_dir,
          )
  
      if is_distributed:
          torch.distributed.barrier()
  
      if is_distributed and AppState().local_rank == 0:
          # If we are in a distributed multi-node set-up and index files are not stored on
          # a shared filesystem, then the index files created on global rank 0 are only
          # accessible to the workers on that node.
          #
          # Two cases may occur here:
          #
          # 1. case of a shared filesystem, or global_rank==0: the index files are present in
          #    the locally available filesystem, calling build_index_files() again is a no-op.
          # 2. case of a non-shared filesystem, and global_rank>0: the index files are not
          #    present in the locally available filesystem, calling build_index_files() again
          #    will create them.
          #
          # Outcome in all cases: all nodes have access to the index files in their filesystem.
          build_index_files(
              dataset_paths,
              newline_int,
              workers=self._worker,
              build_index_fn=build_index_fn,
              index_mapping_dir=index_mapping_dir,
          )
  
      if is_distributed:
          torch.distributed.barrier()
  
      logging.info(f"Loading data files")
      start_time = time.time()
      mdata_midx_list = [
          self.load_file(fn, index_mapping_dir)
          for fn in self._files_list
      ]
      logging.info(
          f"Time loading {len(mdata_midx_list)} mem-mapped files: {datetime.timedelta(seconds=time.time() - start_time)}"
      )
  
      logging.info("Computing global indices")
      midx_bins = np.cumsum(
          [(len(midx) - header_lines) for _, midx in mdata_midx_list]
      )
  
      self.midx_bins = midx_bins
      self.mdata_midx_list = mdata_midx_list
  
      # figure out size of the dataset
      self._size = self.midx_bins[-1]
  
imports: import datetime
  import json
  import multiprocessing
  import os
  import pickle
  import time
  from functools import partial
  from typing import Callable, List, Optional, Type
  import numpy
  import torch
  from nemo.core import Dataset
  from nemo.utils import AppState, logging
label_desc: Use image node_id 1 to create a new TextMemMapDataset object from inherited base classes: Dataset with example: obj = TextMemMapDataset(dataset_paths, newline_int, header_lines, workers, tokenizer, build_index_fn, sort_dataset_paths, index_mapping_dir)