



"""simple docstring"""




import os
import sys
from pathlib import Path

import pytest

from datasets import Dataset, IterableDataset
from datasets.distributed import split_dataset_by_node

from .utils import execute_subprocess_async, get_torch_dist_unique_port, require_torch


def   lowercase__     (			)     ->			Optional[int]:






      '''simple docstring'''


      lowercase  :      List[str]					=   Dataset.from_dict({'i': range(17				)}				)
      lowercase  :      Optional[Any]					=   len(_UpperCAmelCase				)
      lowercase  :      Dict					=   3
      lowercase  :      Dict					=   [
          split_dataset_by_node(_UpperCAmelCase	,      rank=_UpperCAmelCase	,      world_size=_UpperCAmelCase				) for rank in range(_UpperCAmelCase				)
      ]
      assert sum(len(_UpperCAmelCase				) for ds in datasets_per_rank				) == full_size
      assert len({tuple(x.values()				) for ds in datasets_per_rank for x in ds}				) == full_size


def   lowercase__     (			)     ->			List[Any]:






      '''simple docstring'''


      def gen():
            return ({"i": i} for i in range(17				))

      lowercase  :      List[Any]					=   3
      lowercase  :      Dict					=   IterableDataset.from_generator(_UpperCAmelCase				)
      lowercase  :      Tuple					=   len(list(_UpperCAmelCase				)				)
      lowercase  :      str					=   [
          split_dataset_by_node(_UpperCAmelCase	,      rank=_UpperCAmelCase	,      world_size=_UpperCAmelCase				) for rank in range(_UpperCAmelCase				)
      ]
      assert sum(len(list(_UpperCAmelCase				)				) for ds in datasets_per_rank				) == full_size
      assert len({tuple(x.values()				) for ds in datasets_per_rank for x in ds}				) == full_size


@pytest.mark.parametrize('shards_per_node'	,      [1, 2, 3]				)
def   lowercase__     (			_UpperCAmelCase				)     ->			Optional[int]:






      '''simple docstring'''


      def gen(_UpperCAmelCase				):
            for shard in shards:
                  yield from ({"i": i, "shard": shard} for i in range(17				))

      lowercase  :      Any					=   3
      lowercase  :      str					=   shards_per_node * world_size
      lowercase  :      Optional[int]					=   {'shards': [f'''shard_{shard_idx}.txt''' for shard_idx in range(_UpperCAmelCase				)]}
      lowercase  :      Optional[int]					=   IterableDataset.from_generator(_UpperCAmelCase	,      gen_kwargs=_UpperCAmelCase				)
      lowercase  :      str					=   len(list(_UpperCAmelCase				)				)
      assert full_ds.n_shards == world_size * shards_per_node
      lowercase  :      Any					=   [
          split_dataset_by_node(_UpperCAmelCase	,      rank=_UpperCAmelCase	,      world_size=_UpperCAmelCase				) for rank in range(_UpperCAmelCase				)
      ]
      assert [ds.n_shards for ds in datasets_per_rank] == [shards_per_node] * world_size
      assert sum(len(list(_UpperCAmelCase				)				) for ds in datasets_per_rank				) == full_size
      assert len({tuple(x.values()				) for ds in datasets_per_rank for x in ds}				) == full_size


def   lowercase__     (			)     ->			Tuple:






      '''simple docstring'''


      def gen():
            return ({"i": i} for i in range(17				))

      lowercase  :      List[str]					=   2
      lowercase  :      List[Any]					=   IterableDataset.from_generator(_UpperCAmelCase				)
      lowercase  :      Optional[int]					=   len(list(_UpperCAmelCase				)				)

      lowercase  :      Dict					=   split_dataset_by_node(_UpperCAmelCase	,      rank=0	,      world_size=_UpperCAmelCase				).shuffle(seed=42				)
      assert len(list(_UpperCAmelCase				)				) == 1 + full_size // world_size
      with pytest.raises(_UpperCAmelCase				):
            split_dataset_by_node(_UpperCAmelCase	,      rank=0	,      world_size=_UpperCAmelCase				).shuffle()

      lowercase  :      List[Any]					=   split_dataset_by_node(full_ds.shuffle(seed=42				)	,      rank=0	,      world_size=_UpperCAmelCase				)
      assert len(list(_UpperCAmelCase				)				) == 1 + full_size // world_size
      with pytest.raises(_UpperCAmelCase				):
            split_dataset_by_node(full_ds.shuffle()	,      rank=0	,      world_size=_UpperCAmelCase				)


@pytest.mark.parametrize('streaming'	,      [False, True]				)
@require_torch
@pytest.mark.skipif(os.name == 'nt'	,      reason='execute_subprocess_async doesn\'t support windows'				)
@pytest.mark.integration
def   lowercase__     (			_UpperCAmelCase				)     ->			Dict:






      '''simple docstring'''


      lowercase  :      int					=   2
      lowercase  :      Union[str, Any]					=   get_torch_dist_unique_port()
      lowercase  :      str					=   Path(_UpperCAmelCase				).resolve().parent / 'distributed_scripts' / 'launch_torch_distributed.py'
      lowercase  :      str					=   f'''
        -m torch.distributed.launch
        --nproc_per_node={nproc_per_node}
        --master_port={master_port}
        {test_script}
    '''.split()
      lowercase  :      Any					=   f'''
        --streaming={streaming}
    '''.split()
      lowercase  :      Union[str, Any]					=   [sys.executable] + distributed_args + args
      execute_subprocess_async(_UpperCAmelCase	,      env=os.environ.copy()				)





@pytest.mark.parametrize(
    'nproc_per_node, num_workers'	,      [
        (2, 2),  # each node has 2 shards and each worker has 1 shards
        (3, 2),  # each node uses all the shards but skips examples, and each worker has 2 shards
    ]	,      )
@require_torch
@pytest.mark.skipif(os.name == 'nt'	,      reason='execute_subprocess_async doesn\'t support windows'				)
@pytest.mark.integration
def   lowercase__     (			_UpperCAmelCase	,      _UpperCAmelCase				)     ->			int:






      '''simple docstring'''


      lowercase  :      str					=   True
      lowercase  :      Union[str, Any]					=   get_torch_dist_unique_port()
      lowercase  :      Tuple					=   Path(_UpperCAmelCase				).resolve().parent / 'distributed_scripts' / 'launch_torch_distributed.py'
      lowercase  :      Dict					=   f'''
        -m torch.distributed.launch
        --nproc_per_node={nproc_per_node}
        --master_port={master_port}
        {test_script}
    '''.split()
      lowercase  :      Any					=   f'''
        --streaming={streaming}
        --num_workers={num_workers}
    '''.split()
      lowercase  :      Optional[int]					=   [sys.executable] + distributed_args + args
      execute_subprocess_async(_UpperCAmelCase	,      env=os.environ.copy()				)





"""simple docstring"""




from arguments import TokenizerTrainingArguments
from datasets import load_dataset
from tqdm import tqdm

from transformers import AutoTokenizer, HfArgumentParser
from transformers.models.gpta.tokenization_gpta import bytes_to_unicode


def   lowercase__     (			_UpperCAmelCase=10				)     ->			Any:






      '''simple docstring'''


      for _ in tqdm(range(0	,      args.n_examples	,      _UpperCAmelCase				)				):
            yield [next(_UpperCAmelCase				)[args.text_column] for _ in range(_UpperCAmelCase				)]


# Configuration
_UpperCamelCase:				Union[str, Any]    							=  HfArgumentParser(TokenizerTrainingArguments)
_UpperCamelCase:				Dict    							=  parser.parse_args()

# Base tokenizer
_UpperCamelCase:				List[Any]    							=  AutoTokenizer.from_pretrained(args.base_tokenizer)
_UpperCamelCase:				int    							=  list(bytes_to_unicode().values())

# Load dataset
_UpperCamelCase:				Union[str, Any]    							=  load_dataset(args.dataset_name, split='train', streaming=True)
_UpperCamelCase:				Union[str, Any]    							=  iter(dataset)


# Training and saving
_UpperCamelCase:				Union[str, Any]    							=  tokenizer.train_new_from_iterator(
    batch_iterator(), vocab_size=args.vocab_size, initial_alphabet=base_vocab
)





new_tokenizer.save_pretrained(args.tokenizer_name, push_to_hub=args.push_to_hub)
