
'''simple docstring'''







import json
import multiprocessing as mp
import re
from collections import defaultdict
from functools import partial
from typing import Dict, List, Optional, Set, Tuple, Type

from datasets import Dataset
from datasketch import MinHash, MinHashLSH
from dpu_utils.utils.iterators import ThreadedIterator
from tqdm import tqdm


__lowerCAmelCase   =       re.compile("""[^A-Za-z_0-9]""")
# parameters used in DuplicationIndex
__lowerCAmelCase   =       1_0
__lowerCAmelCase   =       2_5_6





def    UpperCAmelCase_						(__a :       List[str]					):






   """simple docstring"""

   if len(__a					) < MIN_NUM_TOKENS:
      return None
   _a							:      Optional[int]     =  MinHash(num_perm=__a					)
   for token in set(__a					):
      min_hash.update(token.encode()					)
   return min_hash





def    UpperCAmelCase_						(__a :       str					):






   """simple docstring"""

   return {t for t in NON_ALPHA.split(__a					) if len(t.strip()					) > 0}







class UpperCAmelCase__      :





  """simple docstring"""

  def __init__(						self :	List[str]				,*,
      _a :	float = 0.85				,):


     '''simple docstring'''






     _a							:      int     =  duplication_jaccard_threshold
     _a							:      List[str]     =  NUM_PERM
     _a							:      str     =  MinHashLSH(threshold=self._duplication_jaccard_threshold				,num_perm=self._num_perm			)

     _a							:      Any     =  defaultdict(__lowercase			)


  def     __lowercase						(						self :	List[str]				,_a :	Tuple				,_a :	MinHash			):


     '''simple docstring'''






     _a							:      Union[str, Any]     =  self._index.query(__lowercase			)
     if code_key in self._index.keys:
        print(F"""Duplicate key {code_key}"""			)
        return

     self._index.insert(__lowercase				,__lowercase			)
     if len(__lowercase			) > 0:
        for base_duplicate in close_duplicates:
           if base_duplicate in self._duplicate_clusters:
              self._duplicate_clusters[base_duplicate].add(__lowercase			)
              break
        else:
           self._duplicate_clusters[close_duplicates[0]].add(__lowercase			)


  def     __lowercase						(						self :	List[str]			):


     '''simple docstring'''






     _a							:      Tuple     =  []
     for base, duplicates in self._duplicate_clusters.items():
        _a							:      List[Any]     =  [base] + list(__lowercase			)
        # reformat the cluster to be a list of dict
        _a							:      int     =  [{'''base_index''': el[0], '''repo_name''': el[1], '''path''': el[2]} for el in cluster]
        duplicate_clusters.append(__lowercase			)
     return duplicate_clusters


  def     __lowercase						(						self :	List[str]				,_a :	Optional[Any]			):


     '''simple docstring'''






     _a							:      Dict     =  self.get_duplicate_clusters()
     with open(__lowercase				,'w'			) as f:
        json.dump(__lowercase				,__lowercase			)





def    UpperCAmelCase_						(__a :       str					):






   """simple docstring"""

   _a							:      Optional[Any]     =  element
   _a							:      Tuple     =  get_min_hash([t for t in NON_ALPHA.split(data['content']					) if len(t.strip()					) > 0]					)
   if min_hash is not None:
      return (index, data["repo_name"], data["path"]), min_hash





def    UpperCAmelCase_						(__a :       Type[Dataset]					):






   """simple docstring"""

   with mp.Pool() as pool:
      for data in pool.imap_unordered(
          _compute_min_hash					,      ThreadedIterator(__a					,      max_queue_size=1_0_0_0_0					)					,      chunksize=1_0_0					,      ):
         if data is not None:
            yield data





def    UpperCAmelCase_						(__a :       Type[Dataset]					,      __a :       float					):






   """simple docstring"""

   _a							:      str     =  DuplicationIndex(duplication_jaccard_threshold=__a					)

   for filename, min_hash in tqdm(ThreadedIterator(minhash_iter(enumerate(__a					)					)					,      max_queue_size=1_0_0					)					):
      di.add(__a					,      __a					)

   # Returns a List[Cluster] where Cluster is List[str] with the filenames.
   return di.get_duplicate_clusters()





def    UpperCAmelCase_						(__a :       str					,      __a :       str					):






   """simple docstring"""

   _a							:      Tuple     =  get_tokens(__a					)
   _a							:      List[Any]     =  get_tokens(__a					)
   return len(tokensa & tokensa					) / len(tokensa | tokensa					)


__lowerCAmelCase   =       None





def    UpperCAmelCase_						(__a :       List[str]					,      __a :       str					):






   """simple docstring"""

   _a							:      Any     =  []
   for elementa in cluster:
      _a							:      Tuple     =  _shared_dataset[elementa['''base_index''']]['''content''']
      for elementa in extremes:
         _a							:      int     =  _shared_dataset[elementa['''base_index''']]['''content''']
         if jaccard_similarity(__a					,      __a					) >= jaccard_threshold:
            elementa["copies"] += 1
            break
      else:
         _a							:      Any     =  1
         extremes.append(__a					)
   return extremes





def    UpperCAmelCase_						(__a :       Union[str, Any]					,      __a :       Optional[int]					,      __a :       Union[str, Any]					):






   """simple docstring"""

   global _shared_dataset
   _a							:      Optional[int]     =  dataset
   _a							:      Union[str, Any]     =  []
   _a							:      Optional[int]     =  partial(_find_cluster_extremes_shared					,      jaccard_threshold=__a					)
   with mp.Pool() as pool:
      for extremes in tqdm(
          pool.imap_unordered(
              __a					,      __a					,      )					,      total=len(__a					)					,      ):
         extremes_list.append(__a					)
   return extremes_list





def    UpperCAmelCase_						(__a :       Type[Dataset]					,      __a :       float = 0.85					):






   """simple docstring"""

   _a							:      Optional[int]     =  make_duplicate_clusters(__a					,      __a					)
   _a							:      Union[str, Any]     =  {x['''base_index'''] for cluster in duplicate_clusters for x in cluster}
   _a							:      Optional[int]     =  {}
   _a							:      List[str]     =  find_extremes(__a					,      __a					,      __a					)
   for extremes in extremes_clusters:
      for element in extremes:
         _a							:      Union[str, Any]     =  element
   _a							:      List[str]     =  duplicate_indices - set(extreme_dict.keys()					)
   _a							:      Optional[int]     =  dataset.filter(lambda __a					,      __a					: idx not in remove_indices					,      with_indices=__a					)

   # update duplicate_clusters
   for cluster in duplicate_clusters:
      for element in cluster:
         _a							:      int     =  element['''base_index'''] in extreme_dict
         if element["is_extreme"]:
            _a							:      List[str]     =  extreme_dict[element['''base_index''']]['''copies''']

   print(f"""Original dataset size: {len(__a					)}"""					)
   print(f"""Number of duplicate clusters: {len(__a					)}"""					)
   print(f"""Files in duplicate cluster: {len(__a					)}"""					)
   print(f"""Unique files in duplicate cluster: {len(__a					)}"""					)
   print(f"""Filtered dataset size: {len(__a					)}"""					)

   return ds_filter, duplicate_clusters










"""simple docstring"""





import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment



def 					lowerCAmelCase_      (   SCREAMING_SNAKE_CASE							:				Any     ,    SCREAMING_SNAKE_CASE							:				List[str]=()     ,    SCREAMING_SNAKE_CASE							:				Union[str, Any]=None     ,    SCREAMING_SNAKE_CASE							:				Dict="no"     ,    SCREAMING_SNAKE_CASE							:				int="29500" ):


							'''simple docstring'''







							__lowerCamelCase   :   Optional[int]       =False
							__lowerCamelCase   :   Tuple       =False
							if any(key.startswith('''KAGGLE''' ) for key in os.environ.keys() ):
														__lowerCamelCase   :   Optional[int]       =True
							elif "IPython" in sys.modules:
														__lowerCamelCase   :   int       ='''google.colab''' in str(sys.modules['''IPython'''].get_ipython() )

							try:
														__lowerCamelCase   :   str       =PrecisionType(mixed_precision.lower() )
							except ValueError:
														raise ValueError(
														    F'Unknown mixed_precision mode: {args.mixed_precision.lower()}. Choose between {PrecisionType.list()}.' )

							if (in_colab or in_kaggle) and (os.environ.get('''TPU_NAME'''     ,    SCREAMING_SNAKE_CASE ) is not None):
														# TPU launch
														import torch_xla.distributed.xla_multiprocessing as xmp

														if len(AcceleratorState._shared_state ) > 0:
																					raise ValueError(
																					    '''To train on TPU in Colab or Kaggle Kernel, the `Accelerator` should only be initialized inside '''
																					    '''your training function. Restart your notebook and make sure no cells initializes an '''
																					    '''`Accelerator`.''' )
														if num_processes is None:
																					__lowerCamelCase   :   int       =8

														__lowerCamelCase   :   Optional[Any]       =PrepareForLaunch(SCREAMING_SNAKE_CASE     ,    distributed_type='''TPU''' )
														print(F'Launching a training on {num_processes} TPU cores.' )
														xmp.spawn(SCREAMING_SNAKE_CASE     ,    args=SCREAMING_SNAKE_CASE     ,    nprocs=SCREAMING_SNAKE_CASE     ,    start_method='''fork''' )
							elif in_colab:
														# No need for a distributed launch otherwise as it's either CPU or one GPU.
														if torch.cuda.is_available():
																					print('''Launching training on one GPU.''' )
														else:
																					print('''Launching training on one CPU.''' )
														function(*SCREAMING_SNAKE_CASE )
							else:
														if num_processes is None:
																					raise ValueError(
																					    '''You have to specify the number of GPUs you would like to use, add `num_processes=...` to your call.''' )

														if num_processes > 1:
																					# Multi-GPU launch
																					from torch.multiprocessing import start_processes
																					from torch.multiprocessing.spawn import ProcessRaisedException

																					if len(AcceleratorState._shared_state ) > 0:
																												raise ValueError(
																												    '''To launch a multi-GPU training from your notebook, the `Accelerator` should only be initialized '''
																												    '''inside your training function. Restart your notebook and make sure no cells initializes an '''
																												    '''`Accelerator`.''' )

																					if torch.cuda.is_initialized():
																												raise ValueError(
																												    '''To launch a multi-GPU training from your notebook, you need to avoid running any instruction '''
																												    '''using `torch.cuda` in any cell. Restart your notebook and make sure no cells use any CUDA '''
																												    '''function.''' )

																					# torch.distributed will expect a few environment variable to be here. We set the ones common to each
																					# process here (the other ones will be set be the launcher).
																					with patch_environment(
																					    world_size=SCREAMING_SNAKE_CASE     ,    master_addr='''127.0.01'''     ,    master_port=SCREAMING_SNAKE_CASE     ,    mixed_precision=SCREAMING_SNAKE_CASE ):
																												__lowerCamelCase   :   List[str]       =PrepareForLaunch(SCREAMING_SNAKE_CASE     ,    distributed_type='''MULTI_GPU''' )
																												print(F'Launching training on {num_processes} GPUs.' )
																												try:
																																			start_processes(SCREAMING_SNAKE_CASE     ,    args=SCREAMING_SNAKE_CASE     ,    nprocs=SCREAMING_SNAKE_CASE     ,    start_method='''fork''' )
																												except ProcessRaisedException as e:
																																			if "Cannot re-initialize CUDA in forked subprocess" in e.args[0]:
																																										raise RuntimeError(
																																										    '''CUDA has been initialized before the `notebook_launcher` could create a forked subprocess. '''
																																										    '''This likely stems from an outside import causing issues once the `notebook_launcher()` is called. '''
																																										    '''Please review your imports and test them when running the `notebook_launcher()` to identify '''
																																										    '''which one is problematic.''' ) from e

														else:
																					# No need for a distributed launch otherwise as it's either CPU, GPU or MPS.
																					if is_mps_available():
																												__lowerCamelCase   :   Optional[Any]       ='''1'''
																												print('''Launching training on MPS.''' )
																					elif torch.cuda.is_available():
																												print('''Launching training on one GPU.''' )
																					else:
																												print('''Launching training on CPU.''' )
																					function(*SCREAMING_SNAKE_CASE )



def 					lowerCAmelCase_      (   SCREAMING_SNAKE_CASE							:				str     ,    SCREAMING_SNAKE_CASE							:				Optional[int]=()     ,    SCREAMING_SNAKE_CASE							:				Optional[Any]=2 ):


							'''simple docstring'''







							from torch.multiprocessing import start_processes

							with tempfile.NamedTemporaryFile() as tmp_file:
														# torch.distributed will expect a few environment variable to be here. We set the ones common to each
														# process here (the other ones will be set be the launcher).
														with patch_environment(
														    world_size=SCREAMING_SNAKE_CASE     ,    master_addr='''127.0.01'''     ,    master_port='''29500'''     ,    accelerate_mixed_precision='''no'''     ,    accelerate_debug_rdv_file=tmp_file.name     ,    accelerate_use_cpu='''yes'''     ,    ):
																					__lowerCamelCase   :   Tuple       =PrepareForLaunch(SCREAMING_SNAKE_CASE     ,    debug=SCREAMING_SNAKE_CASE )
																					start_processes(SCREAMING_SNAKE_CASE     ,    args=SCREAMING_SNAKE_CASE     ,    nprocs=SCREAMING_SNAKE_CASE     ,    start_method='''fork''' )




