



'''simple docstring'''






import gzip
import hashlib
import json
import multiprocessing
import os
import re
import shutil
import time
from pathlib import Path

import numpy as np
from arguments import PreprocessingArguments
from datasets import load_dataset
from minhash_deduplication import deduplicate_dataset

from transformers import AutoTokenizer, HfArgumentParser


lowerCAmelCase:      Dict		 =		re.compile(r'\s+')
def    lowerCamelCase__ (		_A       ):
  return {"hash": hashlib.mda(re.sub(__snake_case				,    ''				,    example['content']       ).encode('utf-8'       )       ).hexdigest()}
def    lowerCamelCase__ (		_A       ):
  a		:      Union[str, Any]       =  [len(__snake_case       ) for line in example["content"].splitlines()]
  return {"line_mean": np.mean(__snake_case       ), "line_max": max(__snake_case       )}
def    lowerCamelCase__ (		_A       ):
  a		:      Union[str, Any]       =  np.mean([c.isalnum() for c in example['content']]       )
  return {"alpha_frac": alpha_frac}
def    lowerCamelCase__ (		_A				,    _A       ):


  if example["hash"] in uniques:
    uniques.remove(example['hash']       )
    return True
  else:
    return False
def    lowerCamelCase__ (		_A				,    _A=5       ):
  a		:      Dict       =  ["auto-generated", "autogenerated", "automatically generated"]
  a		:      List[Any]       =  example["content"].splitlines()
  for _, line in zip(range(__snake_case       )				,    __snake_case       ):
    for keyword in keywords:
      if keyword in line.lower():
        return {"autogenerated": True}
  else:
    return {"autogenerated": False}
def    lowerCamelCase__ (		_A				,    _A=5				,    _A=0.05       ):
  a		:      int       =  ["unit tests", "test file", "configuration file"]
  a		:      Any       =  example["content"].splitlines()
  a		:      List[Any]       =  0
  a		:      Union[str, Any]       =  0
  # first test
  for _, line in zip(range(__snake_case       )				,    __snake_case       ):
    for keyword in keywords:
      if keyword in line.lower():
        return {"config_or_test": True}
    # second test
  a		:      Dict       =  example["content"].count('\n'       )
  a		:      Optional[Any]       =  int(coeff * nlines       )
  for line in lines:
    count_config += line.lower().count('config'       )
    count_test += line.lower().count('test'       )
    if count_config > threshold or count_test > threshold:
      return {"config_or_test": True}
  return {"config_or_test": False}
def    lowerCamelCase__ (		_A       ):
  a		:      str       =  ["def ", "class ", "for ", "while "]
  a		:      int       =  example["content"].splitlines()
  for line in lines:
    for keyword in keywords:
      if keyword in line.lower():
        return {"has_no_keywords": False}
  return {"has_no_keywords": True}
def    lowerCamelCase__ (		_A				,    _A=4       ):
  a		:      Tuple       =  example["content"].splitlines()
  a		:      Optional[int]       =  0
  for line in lines:
    counter += line.lower().count('='       )
    if counter > minimum:
      return {"has_few_assignments": False}
  return {"has_few_assignments": True}
def    lowerCamelCase__ (		_A       ):
  a		:      List[str]       =  tokenizer(example['content']				,    truncation=__snake_case       )["input_ids"]
  a		:      int       =  len(example['content']       ) / len(__snake_case       )
  return {"ratio": ratio}
def    lowerCamelCase__ (		_A       ):
  a		:      str       =  {}
  results.update(get_hash(__snake_case       )       )
  results.update(line_stats(__snake_case       )       )
  results.update(alpha_stats(__snake_case       )       )
  results.update(char_token_ratio(__snake_case       )       )
  results.update(is_autogenerated(__snake_case       )       )
  results.update(is_config_or_test(__snake_case       )       )
  results.update(has_no_keywords(__snake_case       )       )
  results.update(has_few_assignments(__snake_case       )       )
  return results
def    lowerCamelCase__ (		_A				,    _A				,    _A       ):


  if not check_uniques(__snake_case				,    __snake_case       ):
    return False
  elif example["autogenerated"]:
    return False
  elif example["line_max"] > args.line_max:
    return False
  elif example["line_mean"] > args.line_mean:
    return False
  elif example["alpha_frac"] < args.alpha_frac:
    return False
  elif example["ratio"] < args.min_token_ratio:
    return False
  elif example["config_or_test"] and np.random.rand() <= args.filter_proba:
    return False
  elif example["has_no_keywords"] and np.random.rand() <= args.filter_proba:
    return False
  elif example["has_few_assignments"]:
    return False
  else:
    return True
def    lowerCamelCase__ (		_A       ):


  with open(__snake_case				,    'rb'       ) as f_in:
    with gzip.open(str(__snake_case       ) + '.gz'				,    'wb'				,    compresslevel=6       ) as f_out:
      shutil.copyfileobj(__snake_case				,    __snake_case       )
  os.unlink(__snake_case       )


# Settings
lowerCAmelCase:      Any		 =		HfArgumentParser(PreprocessingArguments)
lowerCAmelCase:      Optional[int]		 =		parser.parse_args()
if args.num_workers is None:
     lowerCAmelCase:      Any		 =		multiprocessing.cpu_count()
lowerCAmelCase:      List[Any]		 =		AutoTokenizer.from_pretrained(args.tokenizer_dir)

# Load dataset
lowerCAmelCase:      int		 =		time.time()
lowerCAmelCase:      Dict		 =		load_dataset(args.dataset_name, split='train')
print(F"Time to load dataset: {time.time()-t_start:.2f}")

# Run preprocessing
lowerCAmelCase:      Optional[Any]		 =		time.time()
lowerCAmelCase:      Tuple		 =		ds.map(preprocess, num_proc=args.num_workers)
print(F"Time to preprocess dataset: {time.time()-t_start:.2f}")

# Deduplicate hashes
lowerCAmelCase:      Dict		 =		set(ds.unique('hash'))
lowerCAmelCase:      List[Any]		 =		len(uniques) / len(ds)
print(F"Fraction of duplicates: {1-frac:.2%}")

# Deduplicate data and apply heuristics
lowerCAmelCase:      List[str]		 =		time.time()
lowerCAmelCase:      Tuple		 =		ds.filter(filter, fn_kwargs={'uniques': uniques, 'args': args})
print(F"Time to filter dataset: {time.time()-t_start:.2f}")
print(F"Size of filtered dataset: {len(ds_filter)}")

# Deduplicate with minhash and jaccard similarity
if args.near_deduplication:
     lowerCAmelCase:      Tuple		 =		time.time()
     lowerCAmelCase:      Optional[Any]		 =		deduplicate_dataset(ds_filter, args.jaccard_threshold)
     print(F"Time to deduplicate dataset: {time.time()-t_start:.2f}")
     print(F"Size of deduplicate dataset: {len(ds_filter)}")

# Save data in batches of samples_per_file
lowerCAmelCase:      Optional[int]		 =		Path(args.output_dir)
output_dir.mkdir(exist_ok=True)

# save duplicate_clusters in the output_dir as artifacts
# not sure it is the right place the save it
if args.near_deduplication:
     with open(output_dir / 'duplicate_clusters.json', 'w') as f:
          json.dump(duplicate_clusters, f)

lowerCAmelCase:      Any		 =		output_dir / "data"
data_dir.mkdir(exist_ok=True)

lowerCAmelCase:      Optional[Any]		 =		time.time()
for file_number, index in enumerate(range(0, len(ds_filter), args.samples_per_file)):
     lowerCAmelCase:      Dict		 =		str(data_dir / F"file-{file_number+1:012}.json")
     lowerCAmelCase:      List[Any]		 =		min(len(ds_filter), index + args.samples_per_file)
     ds_filter.select(list(range(index, end_index))).to_json(file_path)
     compress_file(file_path)

print(F"Time to save dataset: {time.time()-t_start:.2f}")




'''simple docstring'''






import inspect
import math
import tempfile
import unittest

import numpy as np

from transformers import ViTMAEConfig
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
     import torch
     from torch import nn

     from transformers import ViTMAEForPreTraining, ViTMAEModel
     from transformers.models.vit.modeling_vit import VIT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
     from PIL import Image

     from transformers import ViTImageProcessor

class 							a__:







    def __init__(   self				:						List[str]		,    __snake_case				:						Optional[Any]		,    __snake_case				:						Any=13		,    __snake_case				:						int=30		,    __snake_case				:						Union[str, Any]=2		,    __snake_case				:						Optional[int]=3		,    __snake_case				:						Dict=True		,    __snake_case				:						List[str]=True		,    __snake_case				:						Any=32		,    __snake_case				:						List[Any]=5		,    __snake_case				:						List[Any]=4		,    __snake_case				:						Optional[int]=37		,    __snake_case				:						int="gelu"		,    __snake_case				:						Optional[Any]=0.1		,    __snake_case				:						List[str]=0.1		,    __snake_case				:						List[str]=10		,    __snake_case				:						int=0.02		,    __snake_case				:						Optional[int]=3		,    __snake_case				:						Tuple=0.6		,    __snake_case				:						Union[str, Any]=None		,    ):
      a		:      List[str]       =  parent
      a		:      Tuple       =  batch_size
      a		:      Union[str, Any]       =  image_size
      a		:      List[str]       =  patch_size
      a		:      Optional[Any]       =  num_channels
      a		:      Optional[Any]       =  is_training
      a		:      List[Any]       =  use_labels
      a		:      Union[str, Any]       =  hidden_size
      a		:      Dict       =  num_hidden_layers
      a		:      Optional[Any]       =  num_attention_heads
      a		:      Optional[Any]       =  intermediate_size
      a		:      int       =  hidden_act
      a		:      Dict       =  hidden_dropout_prob
      a		:      Optional[Any]       =  attention_probs_dropout_prob
      a		:      Optional[Any]       =  type_sequence_label_size
      a		:      Optional[Any]       =  initializer_range
      a		:      Union[str, Any]       =  mask_ratio
      a		:      Optional[int]       =  scope

      # in ViTMAE, the expected sequence length = (num_patches + 1) * (1 - config.mask_ratio), rounded above
      # (we add 1 for the [CLS] token)
      a		:      Tuple       =  (image_size // patch_size) ** 2
      a		:      List[Any]       =  int(math.ceil((1 - mask_ratio) * (num_patches + 1)			)			)







    def     lowercase_ (   self				:						List[str]			):
      a		:      str       =  floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]			)

      a		:      Tuple       =  None
      if self.use_labels:
        a		:      List[str]       =  ids_tensor([self.batch_size]		,    self.type_sequence_label_size			)

      a		:      Optional[int]       =  self.get_config()

      return config, pixel_values, labels







    def     lowercase_ (   self				:						str			):
      return ViTMAEConfig(
          image_size=self.image_size		,    patch_size=self.patch_size		,    num_channels=self.num_channels		,    hidden_size=self.hidden_size		,    num_hidden_layers=self.num_hidden_layers		,    num_attention_heads=self.num_attention_heads		,    intermediate_size=self.intermediate_size		,    hidden_act=self.hidden_act		,    hidden_dropout_prob=self.hidden_dropout_prob		,    attention_probs_dropout_prob=self.attention_probs_dropout_prob		,    is_decoder=__snake_case		,    initializer_range=self.initializer_range		,    mask_ratio=self.mask_ratio		,    )







    def     lowercase_ (   self				:						Dict		,    __snake_case				:						Dict		,    __snake_case				:						List[Any]		,    __snake_case				:						Union[str, Any]			):
      a		:      Union[str, Any]       =  ViTMAEModel(config=__snake_case			)
      model.to(__snake_case			)
      model.eval()
      a		:      Optional[Any]       =  model(__snake_case			)
      self.parent.assertEqual(result.last_hidden_state.shape		,    (self.batch_size, self.seq_length, self.hidden_size)			)







    def     lowercase_ (   self				:						Optional[Any]		,    __snake_case				:						List[str]		,    __snake_case				:						Optional[int]		,    __snake_case				:						Dict			):
      a		:      Dict       =  ViTMAEForPreTraining(__snake_case			)
      model.to(__snake_case			)
      model.eval()
      a		:      Dict       =  model(__snake_case			)
      a		:      str       =  (self.image_size // self.patch_size) ** 2
      a		:      int       =  self.patch_size**2 * self.num_channels
      self.parent.assertEqual(result.logits.shape		,    (self.batch_size, num_patches, expected_num_channels)			)

      # test greyscale images
      a		:      Any       =  1
      a		:      Optional[int]       =  ViTMAEForPreTraining(__snake_case			)
      model.to(__snake_case			)
      model.eval()
      a		:      int       =  floats_tensor([self.batch_size, 1, self.image_size, self.image_size]			)
      a		:      Optional[Any]       =  model(__snake_case			)
      a		:      Dict       =  self.patch_size**2
      self.parent.assertEqual(result.logits.shape		,    (self.batch_size, num_patches, expected_num_channels)			)







    def     lowercase_ (   self				:						Dict			):
      a		:      List[str]       =  self.prepare_config_and_inputs()
      a     ,			a     ,			a		:      Any       =  config_and_inputs
      a		:      Optional[Any]       =  {'pixel_values': pixel_values}
      return config, inputs_dict

@require_torch
class 							a__(      lowerCamelCase__	,				lowerCamelCase__	,				unittest.TestCase			):
    lowercase__				=					(ViTMAEModel, ViTMAEForPreTraining) if is_torch_available() else ()
    lowercase__				=					{"""feature-extraction""": ViTMAEModel} if is_torch_available() else {}

    lowercase__				=					False
    lowercase__				=					False
    lowercase__				=					False
    lowercase__				=					False







    def     lowercase_ (   self				:						Tuple			):
      a		:      List[str]       =  ViTMAEModelTester(self			)
      a		:      List[str]       =  ConfigTester(self		,    config_class=__snake_case		,    has_text_modality=__snake_case		,    hidden_size=37			)







    def     lowercase_ (   self				:						List[str]			):
      self.config_tester.run_common_tests()







    @unittest.skip(reason='ViTMAE does not use inputs_embeds'			)
    def     lowercase_ (   self				:						List[str]			):
      pass







    def     lowercase_ (   self				:						Dict			):
      a     ,			a		:      List[Any]       =  self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
        a		:      Optional[Any]       =  model_class(__snake_case			)
        self.assertIsInstance(model.get_input_embeddings()		,    (nn.Module)			)
        a		:      Optional[int]       =  model.get_output_embeddings()
        self.assertTrue(x is None or isinstance(__snake_case		,    nn.Linear			)			)







    def     lowercase_ (   self				:						List[str]			):
      a     ,			a		:      Dict       =  self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
        a		:      Any       =  model_class(__snake_case			)
        a		:      Any       =  inspect.signature(model.forward			)
        # signature.parameters is an OrderedDict => so arg_names order is deterministic
        a		:      int       =  [*signature.parameters.keys()]

        a		:      List[Any]       =  ['pixel_values']
        self.assertListEqual(arg_names[:1]		,    __snake_case			)







    def     lowercase_ (   self				:						Any			):
      a		:      Optional[int]       =  self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_model(*__snake_case			)







    def     lowercase_ (   self				:						List[Any]			):
      a		:      Dict       =  self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_pretraining(*__snake_case			)







    def     lowercase_ (   self				:						List[str]		,    __snake_case				:						Optional[Any]		,    __snake_case				:						Union[str, Any]		,    __snake_case				:						Tuple			):
      # make masks reproducible
      np.random.seed(2			)

      a		:      Union[str, Any]       =  int((pt_model.config.image_size // pt_model.config.patch_size) ** 2			)
      a		:      int       =  np.random.uniform(size=(self.model_tester.batch_size, num_patches)			)
      a		:      List[Any]       =  torch.from_numpy(__snake_case			)

      # Add `noise` argument.
      # PT inputs will be prepared in `super().check_pt_tf_models()` with this added `noise` argument
      a		:      List[str]       =  pt_noise

      super().check_pt_tf_models(__snake_case		,    __snake_case		,    __snake_case			)







    def     lowercase_ (   self				:						Optional[int]			):
      a     ,			a		:      List[str]       =  self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
        a		:      Any       =  model_class(__snake_case			)
        model.to(__snake_case			)
        model.eval()
        # make random mask reproducible
        torch.manual_seed(2			)
        with torch.no_grad():
          a		:      Tuple       =  model(**self._prepare_for_class(__snake_case		,    __snake_case			)			)

        a		:      List[str]       =  outputs[0].cpu().numpy()
        a		:      str       =  0

        with tempfile.TemporaryDirectory() as tmpdirname:
          model.save_pretrained(__snake_case			)
          a		:      Dict       =  model_class.from_pretrained(__snake_case			)
          model.to(__snake_case			)
          # make random mask reproducible
          torch.manual_seed(2			)
          with torch.no_grad():
            a		:      int       =  model(**self._prepare_for_class(__snake_case		,    __snake_case			)			)

          # Make sure we don't have nans
          a		:      List[Any]       =  after_outputs[0].cpu().numpy()
          a		:      Any       =  0
          a		:      Any       =  np.amax(np.abs(out_a - out_a			)			)
          self.assertLessEqual(__snake_case		,    1e-5			)







    @unittest.skip(
        reason='ViTMAE returns a random mask + ids_restore in each forward pass. See test_save_load\n    to get deterministic results.'			)
    def     lowercase_ (   self				:						List[Any]			):
      pass







    @unittest.skip(
        reason='ViTMAE returns a random mask + ids_restore in each forward pass. See test_save_load\n    to get deterministic results.'			)
    def     lowercase_ (   self				:						Any			):
      pass







    @unittest.skip(
        reason='ViTMAE returns a random mask + ids_restore in each forward pass. See test_save_load\n    to get deterministic results.'			)
    def     lowercase_ (   self				:						int			):
      pass







    @unittest.skip(reason='ViTMAE returns a random mask + ids_restore in each forward pass. See test_save_load'			)
    def     lowercase_ (   self				:						int			):
      pass







    @unittest.skip('Will be fixed soon by reducing the size of the model used for common tests.'			)
    def     lowercase_ (   self				:						Any			):
      pass







    @slow
    def     lowercase_ (   self				:						Dict			):
      for model_name in VIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
        a		:      str       =  ViTMAEModel.from_pretrained(__snake_case			)
        self.assertIsNotNone(__snake_case			)
def    lowerCamelCase__ (		):
  a		:      Tuple       =  Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'       )
  return image

@require_torch
@require_vision
class 							a__(      unittest.TestCase			):







    @cached_property
    def     lowercase_ (   self				:						Union[str, Any]			):
      return ViTImageProcessor.from_pretrained('facebook/vit-mae-base'			) if is_vision_available() else None







    @slow
    def     lowercase_ (   self				:						Union[str, Any]			):
      # make random mask reproducible across the PT and TF model
      np.random.seed(2			)

      a		:      Union[str, Any]       =  ViTMAEForPreTraining.from_pretrained('facebook/vit-mae-base'			).to(__snake_case			)

      a		:      str       =  self.default_image_processor
      a		:      Dict       =  prepare_img()
      a		:      Union[str, Any]       =  image_processor(images=__snake_case		,    return_tensors='pt'			).to(__snake_case			)

      # prepare a noise vector that will be also used for testing the TF model
      # (this way we can ensure that the PT and TF models operate on the same inputs)
      a		:      Tuple       =  ViTMAEConfig()
      a		:      Optional[Any]       =  int((vit_mae_config.image_size // vit_mae_config.patch_size) ** 2			)
      a		:      List[str]       =  np.random.uniform(size=(1, num_patches)			)

      # forward pass
      with torch.no_grad():
        a		:      Tuple       =  model(**__snake_case		,    noise=torch.from_numpy(__snake_case			).to(device=__snake_case			)			)

      # verify the logits
      a		:      int       =  torch.Size((1, 1_96, 7_68)			)
      self.assertEqual(outputs.logits.shape		,    __snake_case			)

      a		:      Optional[Any]       =  torch.tensor(
          [[-0.0548, -1.7023, -0.9325], [0.3721, -0.5670, -0.2233], [0.8235, -1.3878, -0.3524]]			)

      self.assertTrue(torch.allclose(outputs.logits[0, :3, :3]		,    expected_slice.to(__snake_case			)		,    atol=1e-4			)			)