




'''simple docstring'''







import json
import os
import re
import warnings
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


A					       =logging.get_logger(__name__)

A					       ={
    'source_spm': 'source.spm',
    'target_spm': 'target.spm',
    'vocab': 'vocab.json',
    'target_vocab_file': 'target_vocab.json',
    'tokenizer_config_file': 'tokenizer_config.json',
}

A					       ={
    'source_spm': {
        'Helsinki-NLP/opus-mt-en-de': 'https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/source.spm'
    },
    'target_spm': {
        'Helsinki-NLP/opus-mt-en-de': 'https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/target.spm'
    },
    'vocab': {
        'Helsinki-NLP/opus-mt-en-de': 'https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/vocab.json'
    },
    'tokenizer_config_file': {
        'Helsinki-NLP/opus-mt-en-de': (
            'https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/tokenizer_config.json'
        )
    },
}

A					       ={'Helsinki-NLP/opus-mt-en-de': 5_12}
A					       ={}


class      _a    (      __a					):
       __a							:	Dict  							=   VOCAB_FILES_NAMES
       __a							:	str  							=   PRETRAINED_VOCAB_FILES_MAP
       __a							:	int  							=   PRETRAINED_INIT_CONFIGURATION
       __a							:	int  							=   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
       __a							:	Optional[int]  							=   ["""input_ids""", """attention_mask"""]
       __a							:	int  							=   re.compile(""">>.+<<"""					)  # type: re.Pattern
       def __init__(      self		:      Tuple  ,							lowercase		:      str  ,							lowercase		:      Union[str, Any]  ,							lowercase		:      Optional[int]  ,							lowercase		:      Dict=None  ,							lowercase		:      Optional[int]=None  ,							lowercase		:      List[Any]=None  ,							lowercase		:      int="<unk>"  ,							lowercase		:      Tuple="</s>"  ,							lowercase		:      List[Any]="<pad>"  ,							lowercase		:      Union[str, Any]=512  ,							lowercase		:      Optional[Dict[str, Any]] = None  ,							lowercase		:      List[Any]=False  ,							**lowercase		:      List[Any]  ,							):



           '''simple docstring'''

           UpperCAmelCase     			=						{} if sp_model_kwargs is None else sp_model_kwargs

           super().__init__(
               # bos_token=bos_token,  unused. Start decoding with config.decoder_start_token_id
               source_lang=lowercase  ,							target_lang=lowercase  ,							unk_token=lowercase  ,							eos_token=lowercase  ,							pad_token=lowercase  ,							model_max_length=lowercase  ,							sp_model_kwargs=self.sp_model_kwargs  ,							target_vocab_file=lowercase  ,							separate_vocabs=lowercase  ,							**lowercase  ,							)
           assert Path(lowercase							).exists(), f"cannot find spm source {source_spm}"

           UpperCAmelCase     			=						separate_vocabs
           UpperCAmelCase     			=						load_json(lowercase							)
           if self.unk_token not in self.encoder:
               raise KeyError('''<unk> token must be in vocab'''							)
           assert self.pad_token in self.encoder

           if separate_vocabs:
               UpperCAmelCase     			=						load_json(lowercase							)
               UpperCAmelCase     			=						{v: k for k, v in self.target_encoder.items()}
               UpperCAmelCase     			=						[]
           else:
               UpperCAmelCase     			=						{v: k for k, v in self.encoder.items()}
               UpperCAmelCase     			=						[k for k in self.encoder if k.startswith('''>>'''							) and k.endswith('''<<'''							)]

           UpperCAmelCase     			=						source_lang
           UpperCAmelCase     			=						target_lang
           UpperCAmelCase     			=						[source_spm, target_spm]

           # load SentencePiece model for pre-processing
           UpperCAmelCase     			=						load_spm(lowercase  ,							self.sp_model_kwargs							)
           UpperCAmelCase     			=						load_spm(lowercase  ,							self.sp_model_kwargs							)
           UpperCAmelCase     			=						self.spm_source
           UpperCAmelCase     			=						self.encoder

           # Multilingual target side: default to using first supported language code.

           self._setup_normalizer()
       def        A (      self		:      Dict							):



           '''simple docstring'''

           try:
               from sacremoses import MosesPunctNormalizer

               UpperCAmelCase     			=						MosesPunctNormalizer(self.source_lang							).normalize
           except (ImportError, FileNotFoundError):
               warnings.warn('''Recommended: pip install sacremoses.'''							)
               UpperCAmelCase     			=						lambda lowercase							: x
       def        A (      self		:      Tuple  ,							lowercase		:      str							):



           '''simple docstring'''

           return self.punc_normalizer(lowercase							) if x else ""
       def        A (      self		:      Optional[Any]  ,							lowercase		:      Dict							):



           '''simple docstring'''

           return self.current_encoder.get(lowercase  ,							self.current_encoder[self.unk_token]							)
       def        A (      self		:      Optional[Any]  ,							lowercase		:      str							):



           '''simple docstring'''

           UpperCAmelCase     			=						self.language_code_re.match(lowercase							)
           UpperCAmelCase     			=						[match.group(0							)] if match else []
           return code, self.language_code_re.sub(''''''  ,							lowercase							)
       def        A (      self		:      List[Any]  ,							lowercase		:      str							):



           '''simple docstring'''

           UpperCAmelCase			,	UpperCAmelCase     			=						self.remove_language_code(lowercase							)
           UpperCAmelCase     			=						self.current_spm.encode(lowercase  ,							out_type=lowercase							)
           return code + pieces
       def        A (      self		:      Dict  ,							lowercase		:      int							):



           '''simple docstring'''

           return self.decoder.get(lowercase  ,							self.unk_token							)
       def        A (      self		:      Union[str, Any]  ,							lowercase		:      List[Any]  ,							**lowercase		:      str							):



           '''simple docstring'''

           return super().batch_decode(lowercase  ,							**lowercase							)
       def        A (      self		:      Union[str, Any]  ,							lowercase		:      Optional[Any]  ,							**lowercase		:      Optional[Any]							):



           '''simple docstring'''

           return super().decode(lowercase  ,							**lowercase							)
       def        A (      self		:      Optional[int]  ,							lowercase		:      List[str]							):



           '''simple docstring'''

           UpperCAmelCase     			=						self.spm_source if self._decode_use_source_tokenizer else self.spm_target
           UpperCAmelCase     			=						[]
           UpperCAmelCase     			=						''''''
           for token in tokens:
               # make sure that special tokens are not decoded using sentencepiece model
               if token in self.all_special_tokens:
                   out_string += sp_model.decode_pieces(lowercase							) + token + " "
                   UpperCAmelCase     			=						[]
               else:
                   current_sub_tokens.append(lowercase							)
           out_string += sp_model.decode_pieces(lowercase							)
           return out_string.strip()
       def        A (      self		:      Any  ,							lowercase		:      Any  ,							lowercase		:      Optional[int]=None							):



           '''simple docstring'''

           if token_ids_a is None:
               return token_ids_a + [self.eos_token_id]
           # We don't expect to process pairs, but leave the pair logic for API consistency
           return token_ids_a + token_ids_a + [self.eos_token_id]
       def        A (      self		:      Tuple							):



           '''simple docstring'''

           UpperCAmelCase     			=						self.spm_source
           UpperCAmelCase     			=						self.encoder
       def        A (      self		:      Optional[int]							):



           '''simple docstring'''

           UpperCAmelCase     			=						self.spm_target
           if self.separate_vocabs:
               UpperCAmelCase     			=						self.target_encoder
       @property
       def        A (      self		:      Tuple							):



           '''simple docstring'''

           return len(self.encoder							)
       def        A (      self		:      int  ,							lowercase		:      str  ,							lowercase		:      Optional[str] = None							):



           '''simple docstring'''

           if not os.path.isdir(lowercase							):
               logger.error(f"Vocabulary path ({save_directory}) should be a directory"							)
               return
           UpperCAmelCase     			=						[]

           if self.separate_vocabs:
               UpperCAmelCase     			=						os.path.join(
                   lowercase  ,							(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab''']  ,							)
               UpperCAmelCase     			=						os.path.join(
                   lowercase  ,							(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''target_vocab_file''']  ,							)
               save_json(self.encoder  ,							lowercase							)
               save_json(self.target_encoder  ,							lowercase							)
               saved_files.append(lowercase							)
               saved_files.append(lowercase							)
           else:
               UpperCAmelCase     			=						os.path.join(
                   lowercase  ,							(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab''']							)
               save_json(self.encoder  ,							lowercase							)
               saved_files.append(lowercase							)

           for spm_save_filename, spm_orig_path, spm_model in zip(
               [VOCAB_FILES_NAMES['''source_spm'''], VOCAB_FILES_NAMES['''target_spm''']]  ,							self.spm_files  ,							[self.spm_source, self.spm_target]  ,							):
               UpperCAmelCase     			=						os.path.join(
                   lowercase  ,							(filename_prefix + '''-''' if filename_prefix else '''''') + spm_save_filename							)
               if os.path.abspath(lowercase							) != os.path.abspath(lowercase							) and os.path.isfile(lowercase							):
                   copyfile(lowercase  ,							lowercase							)
                   saved_files.append(lowercase							)
               elif not os.path.isfile(lowercase							):
                   with open(lowercase  ,							'''wb'''							) as fi:
                       UpperCAmelCase     			=						spm_model.serialized_model_proto()
                       fi.write(lowercase							)
                   saved_files.append(lowercase							)

           return tuple(lowercase							)
       def        A (      self		:      int							):



           '''simple docstring'''

           return self.get_src_vocab()
       def        A (      self		:      Union[str, Any]							):



           '''simple docstring'''

           return dict(self.encoder  ,							**self.added_tokens_encoder							)
       def        A (      self		:      Dict							):



           '''simple docstring'''

           return dict(self.target_encoder  ,							**self.added_tokens_decoder							)
       def __getstate__(      self		:      Union[str, Any]							):



           '''simple docstring'''

           UpperCAmelCase     			=						self.__dict__.copy()
           state.update(
               {k: None for k in ['''spm_source''', '''spm_target''', '''current_spm''', '''punc_normalizer''', '''target_vocab_file''']}							)
           return state
       def __setstate__(      self		:      Any  ,							lowercase		:      Dict							):



           '''simple docstring'''

           UpperCAmelCase     			=						d

           # for backward compatibility
           if not hasattr(self  ,							'''sp_model_kwargs'''							):
               UpperCAmelCase     			=						{}

           UpperCAmelCase			,	UpperCAmelCase     			=						(load_spm(lowercase  ,							self.sp_model_kwargs							) for f in self.spm_files)
           UpperCAmelCase     			=						self.spm_source
           self._setup_normalizer()
       def        A (      self		:      int  ,							*lowercase		:      str  ,							**lowercase		:      Tuple							):



           '''simple docstring'''

           return 1
       def        A (      self		:      Optional[int]  ,							lowercase		:      List[Any]							):



           '''simple docstring'''

           UpperCAmelCase     			=						set(self.all_special_ids							)  # call it once instead of inside list comp
           all_special_ids.remove(self.unk_token_id							)  # <unk> is only sometimes special
           return [1 if x in all_special_ids else 0 for x in seq]







       def        A (      self		:      Tuple  ,							lowercase		:      List  ,							lowercase		:      Optional[List] = None  ,							lowercase		:      bool = False							):



           '''simple docstring'''

           if already_has_special_tokens:
               return self._special_token_mask(lowercase							)
           elif token_ids_a is None:
               return self._special_token_mask(lowercase							) + [1]
           else:
               return self._special_token_mask(token_ids_a + token_ids_a							) + [1]






def 		snake_case_				(_a      :				str       ,  _a      :				Dict[str, Any]	):
    UpperCAmelCase     			=						sentencepiece.SentencePieceProcessor(**_a	)
    spm.Load(_a	)
    return spm





def 		snake_case_				(_a      :				Tuple       ,  _a      :				str	):
    with open(_a       ,  '''w'''	) as f:
        json.dump(_a       ,  _a       ,  indent=2	)







def 		snake_case_				(_a      :				str	):
    with open(_a       ,  '''r'''	) as f:
        return json.load(_a	)










'''simple docstring'''







import inspect
import unittest

from transformers import is_torch_available, is_vision_available
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, slow, torch_device

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 transformers import MODEL_MAPPING, PoolFormerConfig, PoolFormerForImageClassification, PoolFormerModel
   from transformers.models.poolformer.modeling_poolformer import POOLFORMER_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
   from PIL import Image

   from transformers import PoolFormerImageProcessor


class      _a    (      __a					):
       def        A (      self		:      Tuple							):



           '''simple docstring'''

           UpperCAmelCase     			=						self.config_class(**self.inputs_dict							)
           self.parent.assertTrue(hasattr(lowercase  ,							'''hidden_sizes'''							)							)
           self.parent.assertTrue(hasattr(lowercase  ,							'''num_encoder_blocks'''							)							)






class      _a    :
       def __init__(      self		:      Any  ,							lowercase		:      Any  ,							lowercase		:      Optional[Any]=13  ,							lowercase		:      List[str]=64  ,							lowercase		:      Optional[Any]=3  ,							lowercase		:      List[str]=4  ,							lowercase		:      Optional[int]=[2, 2, 2, 2]  ,							lowercase		:      Tuple=[8, 4, 2, 1]  ,							lowercase		:      Optional[int]=[16, 32, 64, 128]  ,							lowercase		:      Tuple=[1, 4, 8, 16]  ,							lowercase		:      List[Any]=False  ,							lowercase		:      List[Any]=True  ,							lowercase		:      List[str]="gelu"  ,							lowercase		:      Any=0.1  ,							lowercase		:      Dict=0.02  ,							lowercase		:      str=3  ,							lowercase		:      Optional[Any]=None  ,							):



           '''simple docstring'''

           UpperCAmelCase     			=						parent
           UpperCAmelCase     			=						batch_size
           UpperCAmelCase     			=						image_size
           UpperCAmelCase     			=						num_channels
           UpperCAmelCase     			=						num_encoder_blocks
           UpperCAmelCase     			=						sr_ratios
           UpperCAmelCase     			=						depths
           UpperCAmelCase     			=						hidden_sizes
           UpperCAmelCase     			=						downsampling_rates
           UpperCAmelCase     			=						is_training
           UpperCAmelCase     			=						use_labels
           UpperCAmelCase     			=						hidden_act
           UpperCAmelCase     			=						hidden_dropout_prob
           UpperCAmelCase     			=						initializer_range
           UpperCAmelCase     			=						num_labels
           UpperCAmelCase     			=						scope
       def        A (      self		:      Union[str, Any]							):



           '''simple docstring'''

           UpperCAmelCase     			=						floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]							)

           UpperCAmelCase     			=						None
           if self.use_labels:
               UpperCAmelCase     			=						ids_tensor([self.batch_size, self.image_size, self.image_size]  ,							self.num_labels							)

           UpperCAmelCase     			=						PoolFormerConfig(
               image_size=self.image_size  ,							num_channels=self.num_channels  ,							num_encoder_blocks=self.num_encoder_blocks  ,							depths=self.depths  ,							hidden_sizes=self.hidden_sizes  ,							hidden_act=self.hidden_act  ,							hidden_dropout_prob=self.hidden_dropout_prob  ,							initializer_range=self.initializer_range  ,							)

           return config, pixel_values, labels
       def        A (      self		:      Optional[Any]  ,							lowercase		:      List[Any]  ,							lowercase		:      Dict  ,							lowercase		:      Union[str, Any]							):



           '''simple docstring'''

           UpperCAmelCase     			=						PoolFormerModel(config=lowercase							)
           model.to(lowercase							)
           model.eval()
           UpperCAmelCase     			=						model(lowercase							)
           UpperCAmelCase     			=						UpperCAmelCase     			=						self.image_size // 32.0
           self.parent.assertEqual(
               result.last_hidden_state.shape  ,							(self.batch_size, self.hidden_sizes[-1], expected_height, expected_width)							)







       def        A (      self		:      Union[str, Any]							):



           '''simple docstring'''

           UpperCAmelCase     			=						self.prepare_config_and_inputs()
           UpperCAmelCase			,	UpperCAmelCase			,	UpperCAmelCase     			=						config_and_inputs
           UpperCAmelCase     			=						{'''pixel_values''': pixel_values}
           return config, inputs_dict






@require_torch
class      _a    (      __a				,				__a				,				unittest.TestCase					):
       __a							:	Optional[int]  							=   (PoolFormerModel, PoolFormerForImageClassification) if is_torch_available() else ()
       __a							:	Dict  							=   (
           {"""feature-extraction""": PoolFormerModel, """image-classification""": PoolFormerForImageClassification}
           if is_torch_available()
           else {}
       )

       __a							:	str  							=   False
       __a							:	Optional[int]  							=   False
       __a							:	Tuple  							=   False
       __a							:	Union[str, Any]  							=   False
       __a							:	Dict  							=   False
       def        A (      self		:      Any							):



           '''simple docstring'''

           UpperCAmelCase     			=						PoolFormerModelTester(self							)
           UpperCAmelCase     			=						PoolFormerConfigTester(self  ,							config_class=lowercase							)
       def        A (      self		:      Any							):



           '''simple docstring'''

           self.config_tester.run_common_tests()
       def        A (      self		:      Optional[int]							):



           '''simple docstring'''

           UpperCAmelCase     			=						self.model_tester.prepare_config_and_inputs()
           self.model_tester.create_and_check_model(*lowercase							)
       @unittest.skip('''PoolFormer does not use inputs_embeds'''							)
       def        A (      self		:      Dict							):



           '''simple docstring'''

           pass
       @unittest.skip('''PoolFormer does not have get_input_embeddings method and get_output_embeddings methods'''							)
       def        A (      self		:      Union[str, Any]							):



           '''simple docstring'''

           pass
       def        A (      self		:      Dict							):



           '''simple docstring'''

           def check_hidden_states_output(lowercase		:      int  ,							lowercase		:      Any  ,							lowercase		:      Union[str, Any]							):
               UpperCAmelCase     			=						model_class(lowercase							)
               model.to(lowercase							)
               model.eval()

               with torch.no_grad():
                   UpperCAmelCase     			=						model(**self._prepare_for_class(lowercase  ,							lowercase							)							)

               UpperCAmelCase     			=						outputs.hidden_states

               UpperCAmelCase     			=						self.model_tester.num_encoder_blocks
               self.assertEqual(len(lowercase							)  ,							lowercase							)

               # verify the first hidden states (first block)
               self.assertListEqual(
                   list(hidden_states[0].shape[-3:]							)  ,							[
                       self.model_tester.hidden_sizes[0],
                       self.model_tester.image_size // 4,
                       self.model_tester.image_size // 4,
                   ]  ,							)

           UpperCAmelCase			,	UpperCAmelCase     			=						self.model_tester.prepare_config_and_inputs_for_common()

           for model_class in self.all_model_classes:
               UpperCAmelCase     			=						True
               check_hidden_states_output(lowercase  ,							lowercase  ,							lowercase							)

               # check that output_hidden_states also work using config
               del inputs_dict["output_hidden_states"]
               UpperCAmelCase     			=						True

               check_hidden_states_output(lowercase  ,							lowercase  ,							lowercase							)
       def        A (      self		:      Optional[Any]							):



           '''simple docstring'''

           if not self.model_tester.is_training:
               return

           UpperCAmelCase			,	UpperCAmelCase     			=						self.model_tester.prepare_config_and_inputs_for_common()
           UpperCAmelCase     			=						True

           for model_class in self.all_model_classes:
               if model_class in get_values(lowercase							):
                   continue
               UpperCAmelCase     			=						model_class(lowercase							)
               model.to(lowercase							)
               model.train()
               UpperCAmelCase     			=						self._prepare_for_class(lowercase  ,							lowercase  ,							return_labels=lowercase							)
               UpperCAmelCase     			=						model(**lowercase							).loss
               loss.backward()
       def        A (      self		:      str							):



           '''simple docstring'''

           UpperCAmelCase			,	UpperCAmelCase     			=						self.model_tester.prepare_config_and_inputs_for_common()

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

               UpperCAmelCase     			=						['''pixel_values''']
               self.assertListEqual(arg_names[:1]  ,							lowercase							)







       @slow
       def        A (      self		:      Union[str, Any]							):



           '''simple docstring'''

           for model_name in POOLFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
               UpperCAmelCase     			=						PoolFormerModel.from_pretrained(lowercase							)
               self.assertIsNotNone(lowercase							)






def 		snake_case_				():
    UpperCAmelCase     			=						Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''	)
    return image







@require_torch
class      _a    (      unittest.TestCase					):
       @slow
       def        A (      self		:      str							):



           '''simple docstring'''

           UpperCAmelCase     			=						PoolFormerImageProcessor()
           UpperCAmelCase     			=						PoolFormerForImageClassification.from_pretrained('''sail/poolformer_s12'''							).to(lowercase							)

           UpperCAmelCase     			=						image_processor(images=prepare_img()  ,							return_tensors='''pt'''							).to(lowercase							)

           # forward pass
           with torch.no_grad():
               UpperCAmelCase     			=						model(**lowercase							)

           # verify the logits
           UpperCAmelCase     			=						torch.Size((1, 1_000)							)
           self.assertEqual(outputs.logits.shape  ,							lowercase							)

           UpperCAmelCase     			=						torch.tensor([-0.6113, 0.1685, -0.0492]							).to(lowercase							)
           self.assertTrue(torch.allclose(outputs.logits[0, :3]  ,							lowercase  ,							atol=1E-4							)							)




