




'''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


snake_case__ 							=    logging.get_logger(__name__)

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

snake_case__ 							=    {
    """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"""
        )
    },
}

snake_case__ 							=    {"""Helsinki-NLP/opus-mt-en-de""": 5_12}
snake_case__ 							=    {}






class      UpperCamelCase_  (a__  ):







						"""simple docstring"""


						_lowerCAmelCase         =       VOCAB_FILES_NAMES
						_lowerCAmelCase         =       PRETRAINED_VOCAB_FILES_MAP
						_lowerCAmelCase         =       PRETRAINED_INIT_CONFIGURATION
						_lowerCAmelCase         =       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						_lowerCAmelCase         =       ['input_ids', 'attention_mask']
						_lowerCAmelCase         =       re.compile('>>.+<<'  )  # type: re.Pattern







						def __init__(  self   :							int				,			_lowerCamelCase   :							Union[str, Any]				,			_lowerCamelCase   :							str				,			_lowerCamelCase   :							Any				,			_lowerCamelCase   :							Union[str, Any]=None				,			_lowerCamelCase   :							int=None				,			_lowerCamelCase   :							Dict=None				,			_lowerCamelCase   :							str="<unk>"				,			_lowerCamelCase   :							List[str]="</s>"				,			_lowerCamelCase   :							str="<pad>"				,			_lowerCamelCase   :							Optional[Any]=512				,			_lowerCamelCase   :							Optional[Dict[str, Any]] = None				,			_lowerCamelCase   :							Tuple=False				,			**_lowerCamelCase   :							Optional[Any]				,			):





											"""simple docstring"""







											A_			:  Optional[int]			=				{} 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=_lowerCamelCase				,			target_lang=_lowerCamelCase				,			unk_token=_lowerCamelCase				,			eos_token=_lowerCamelCase				,			pad_token=_lowerCamelCase				,			model_max_length=_lowerCamelCase				,			sp_model_kwargs=self.sp_model_kwargs				,			target_vocab_file=_lowerCamelCase				,			separate_vocabs=_lowerCamelCase				,			**_lowerCamelCase				,			)
											assert Path(_lowerCamelCase     ).exists(), f'cannot find spm source {source_spm}'

											A_			:  Optional[Any]			=				separate_vocabs
											A_			:  List[Any]			=				load_json(_lowerCamelCase     )
											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:
																A_			:  Any			=				load_json(_lowerCamelCase     )
																A_			:  Dict			=				{v: k for k, v in self.target_encoder.items()}
																A_			:  List[str]			=				[]
											else:
																A_			:  int			=				{v: k for k, v in self.encoder.items()}
																A_			:  list			=				[k for k in self.encoder if k.startswith('''>>'''     ) and k.endswith('''<<'''     )]

											A_			:  Optional[int]			=				source_lang
											A_			:  int			=				target_lang
											A_			:  Optional[Any]			=				[source_spm, target_spm]

											# load SentencePiece model for pre-processing
											A_			:  Tuple			=				load_spm(_lowerCamelCase				,			self.sp_model_kwargs     )
											A_			:  List[Any]			=				load_spm(_lowerCamelCase				,			self.sp_model_kwargs     )
											A_			:  int			=				self.spm_source
											A_			:  int			=				self.encoder

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

											self._setup_normalizer()







						def    _a     (  self   :							Any     ):





											"""simple docstring"""







											try:
																from sacremoses import MosesPunctNormalizer

																A_			:  Any			=				MosesPunctNormalizer(self.source_lang     ).normalize
											except (ImportError, FileNotFoundError):
																warnings.warn('''Recommended: pip install sacremoses.'''     )
																A_			:  Optional[Any]			=				lambda _lowerCamelCase     : x







						def    _a     (  self   :							Union[str, Any]				,			_lowerCamelCase   :							str     ):





											"""simple docstring"""







											return self.punc_normalizer(_lowerCamelCase     ) if x else ""







						def    _a     (  self   :							Union[str, Any]				,			_lowerCamelCase   :							Any     ):





											"""simple docstring"""







											return self.current_encoder.get(_lowerCamelCase				,			self.current_encoder[self.unk_token]     )







						def    _a     (  self   :							Optional[int]				,			_lowerCamelCase   :							str     ):





											"""simple docstring"""







											A_			:  List[str]			=				self.language_code_re.match(_lowerCamelCase     )
											A_			:  list			=				[match.group(0     )] if match else []
											return code, self.language_code_re.sub(''''''				,			_lowerCamelCase     )







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





											"""simple docstring"""







											A_			,A_			:  Dict			=				self.remove_language_code(_lowerCamelCase     )
											A_			:  Any			=				self.current_spm.encode(_lowerCamelCase				,			out_type=_lowerCamelCase     )
											return code + pieces







						def    _a     (  self   :							List[Any]				,			_lowerCamelCase   :							int     ):





											"""simple docstring"""







											return self.decoder.get(_lowerCamelCase				,			self.unk_token     )







						def    _a     (  self   :							Optional[int]				,			_lowerCamelCase   :							Any				,			**_lowerCamelCase   :							List[str]     ):





											"""simple docstring"""







											return super().batch_decode(_lowerCamelCase				,			**_lowerCamelCase     )







						def    _a     (  self   :							Optional[int]				,			_lowerCamelCase   :							List[Any]				,			**_lowerCamelCase   :							List[str]     ):





											"""simple docstring"""







											return super().decode(_lowerCamelCase				,			**_lowerCamelCase     )







						def    _a     (  self   :							List[str]				,			_lowerCamelCase   :							List[str]     ):





											"""simple docstring"""







											A_			:  str			=				self.spm_source if self._decode_use_source_tokenizer else self.spm_target
											A_			:  List[Any]			=				[]
											A_			:  Dict			=				''''''
											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(_lowerCamelCase     ) + token + " "
																					A_			:  int			=				[]
																else:
																					current_sub_tokens.append(_lowerCamelCase     )
											out_string += sp_model.decode_pieces(_lowerCamelCase     )
											return out_string.strip()







						def    _a     (  self   :							int				,			_lowerCamelCase   :							List[str]				,			_lowerCamelCase   :							Dict=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   :							Union[str, Any]     ):





											"""simple docstring"""







											A_			:  Union[str, Any]			=				self.spm_source
											A_			:  List[Any]			=				self.encoder







						def    _a     (  self   :							Optional[Any]     ):





											"""simple docstring"""







											A_			:  Union[str, Any]			=				self.spm_target
											if self.separate_vocabs:
																A_			:  List[str]			=				self.target_encoder







						@property
						def    _a     (  self   :							Dict     ):





											"""simple docstring"""







											return len(self.encoder     )







						def    _a     (  self   :							List[str]				,			_lowerCamelCase   :							str				,			_lowerCamelCase   :							Optional[str] = None     ):





											"""simple docstring"""







											if not os.path.isdir(_lowerCamelCase     ):
																logger.error(f'Vocabulary path ({save_directory}) should be a directory'     )
																return
											A_			:  Tuple			=				[]

											if self.separate_vocabs:
																A_			:  List[str]			=				os.path.join(
																    _lowerCamelCase				,			(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab''']				,			)
																A_			:  Optional[int]			=				os.path.join(
																    _lowerCamelCase				,			(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''target_vocab_file''']				,			)
																save_json(self.encoder				,			_lowerCamelCase     )
																save_json(self.target_encoder				,			_lowerCamelCase     )
																saved_files.append(_lowerCamelCase     )
																saved_files.append(_lowerCamelCase     )
											else:
																A_			:  List[str]			=				os.path.join(
																    _lowerCamelCase				,			(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab''']     )
																save_json(self.encoder				,			_lowerCamelCase     )
																saved_files.append(_lowerCamelCase     )

											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]				,			):
																A_			:  List[str]			=				os.path.join(
																    _lowerCamelCase				,			(filename_prefix + '''-''' if filename_prefix else '''''') + spm_save_filename     )
																if os.path.abspath(_lowerCamelCase     ) != os.path.abspath(_lowerCamelCase     ) and os.path.isfile(_lowerCamelCase     ):
																					copyfile(_lowerCamelCase				,			_lowerCamelCase     )
																					saved_files.append(_lowerCamelCase     )
																elif not os.path.isfile(_lowerCamelCase     ):
																					with open(_lowerCamelCase				,			'''wb'''     ) as fi:
																										A_			:  Any			=				spm_model.serialized_model_proto()
																										fi.write(_lowerCamelCase     )
																					saved_files.append(_lowerCamelCase     )

											return tuple(_lowerCamelCase     )







						def    _a     (  self   :							Tuple     ):





											"""simple docstring"""







											return self.get_src_vocab()







						def    _a     (  self   :							Tuple     ):





											"""simple docstring"""







											return dict(self.encoder				,			**self.added_tokens_encoder     )







						def    _a     (  self   :							Optional[int]     ):





											"""simple docstring"""







											return dict(self.target_encoder				,			**self.added_tokens_decoder     )







						def __getstate__(  self   :							Dict     ):





											"""simple docstring"""







											A_			:  Any			=				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   :							int				,			_lowerCamelCase   :							Dict     ):





											"""simple docstring"""







											A_			:  List[Any]			=				d

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

											A_			,A_			:  List[Any]			=				(load_spm(_lowerCamelCase				,			self.sp_model_kwargs     ) for f in self.spm_files)
											A_			:  Optional[int]			=				self.spm_source
											self._setup_normalizer()







						def    _a     (  self   :							Any				,			*_lowerCamelCase   :							List[Any]				,			**_lowerCamelCase   :							Tuple     ):





											"""simple docstring"""







											return 1







						def    _a     (  self   :							List[str]				,			_lowerCamelCase   :							Dict     ):





											"""simple docstring"""







											A_			:  Any			=				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   :							List[str]				,			_lowerCamelCase   :							List				,			_lowerCamelCase   :							Optional[List] = None				,			_lowerCamelCase   :							bool = False     ):





											"""simple docstring"""







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






def     snake_case__     (			lowerCamelCase__					: str      ,		lowerCamelCase__					: Dict[str, Any]      ) -> sentencepiece.SentencePieceProcessor:
					A_			:  Any			=				sentencepiece.SentencePieceProcessor(**lowerCamelCase__      )
					spm.Load(lowerCamelCase__      )
					return spm




def     snake_case__     (			lowerCamelCase__					: Optional[int]      ,		lowerCamelCase__					: str      ) -> None:
					with open(lowerCamelCase__      ,		'''w'''      ) as f:
										json.dump(lowerCamelCase__      ,		lowerCamelCase__      ,		indent=2      )





def     snake_case__     (			lowerCamelCase__					: str      ) -> Union[Dict, List]:
					with open(lowerCamelCase__      ,		'''r'''      ) as f:
										return json.load(lowerCamelCase__      )









'''simple docstring'''




import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin


if is_torch_available():
							import torch

if is_vision_available():
							from PIL import Image

							from transformers import SwinaSRImageProcessor
							from transformers.image_transforms import get_image_size






class      UpperCamelCase_  (unittest.TestCase  ):







						"""simple docstring"""


						def __init__(  self   :							Tuple				,			_lowerCamelCase   :							str				,			_lowerCamelCase   :							str=7				,			_lowerCamelCase   :							Tuple=3				,			_lowerCamelCase   :							Any=18				,			_lowerCamelCase   :							Optional[int]=30				,			_lowerCamelCase   :							List[Any]=400				,			_lowerCamelCase   :							int=True				,			_lowerCamelCase   :							Dict=1 / 255				,			_lowerCamelCase   :							List[str]=True				,			_lowerCamelCase   :							Optional[int]=8				,			):





											"""simple docstring"""







											A_			:  Tuple			=				parent
											A_			:  str			=				batch_size
											A_			:  Optional[Any]			=				num_channels
											A_			:  Union[str, Any]			=				image_size
											A_			:  Optional[int]			=				min_resolution
											A_			:  List[Any]			=				max_resolution
											A_			:  int			=				do_rescale
											A_			:  Optional[Any]			=				rescale_factor
											A_			:  Any			=				do_pad
											A_			:  Tuple			=				pad_size







						def    _a     (  self   :							List[Any]     ):





											"""simple docstring"""







											return {
											    "do_rescale": self.do_rescale,
											    "rescale_factor": self.rescale_factor,
											    "do_pad": self.do_pad,
											    "pad_size": self.pad_size,
											}







						def    _a     (  self   :							Tuple				,			_lowerCamelCase   :							List[str]=False				,			_lowerCamelCase   :							Tuple=False				,			_lowerCamelCase   :							str=False     ):





											"""simple docstring"""







											assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

											if equal_resolution:
																A_			:  List[Any]			=				[]
																for i in range(self.batch_size     ):
																					image_inputs.append(
																					    np.random.randint(
																					        255				,			size=(self.num_channels, self.max_resolution, self.max_resolution)				,			dtype=np.uinta     )     )
											else:
																A_			:  List[Any]			=				[]
																for i in range(self.batch_size     ):
																					A_			,A_			:  List[str]			=				np.random.choice(np.arange(self.min_resolution				,			self.max_resolution     )				,			2     )
																					image_inputs.append(np.random.randint(255				,			size=(self.num_channels, width, height)				,			dtype=np.uinta     )     )

											if not numpify and not torchify:
																# PIL expects the channel dimension as last dimension
																A_			:  Any			=				[Image.fromarray(np.moveaxis(_lowerCamelCase				,			0				,			-1     )     ) for x in image_inputs]

											if torchify:
																A_			:  str			=				[torch.from_numpy(_lowerCamelCase     ) for x in image_inputs]

											return image_inputs






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







						"""simple docstring"""


						_lowerCAmelCase         =       SwinaSRImageProcessor if is_vision_available() else None







						def    _a     (  self   :							List[str]     ):





											"""simple docstring"""







											A_			:  int			=				SwinaSRImageProcessingTester(self     )







						@property
						def    _a     (  self   :							Union[str, Any]     ):





											"""simple docstring"""







											return self.image_processor_tester.prepare_image_processor_dict()







						def    _a     (  self   :							Any     ):





											"""simple docstring"""







											A_			:  Optional[int]			=				self.image_processing_class(**self.image_processor_dict     )
											self.assertTrue(hasattr(_lowerCamelCase				,			'''do_rescale'''     )     )
											self.assertTrue(hasattr(_lowerCamelCase				,			'''rescale_factor'''     )     )
											self.assertTrue(hasattr(_lowerCamelCase				,			'''do_pad'''     )     )
											self.assertTrue(hasattr(_lowerCamelCase				,			'''pad_size'''     )     )







						def    _a     (  self   :							Tuple     ):





											"""simple docstring"""







											pass







						def    _a     (  self   :							Optional[int]				,			_lowerCamelCase   :							Optional[int]     ):





											"""simple docstring"""







											A_			,A_			:  Any			=				get_image_size(_lowerCamelCase     )
											A_			:  List[Any]			=				self.image_processor_tester.pad_size

											A_			:  Dict			=				(old_height // size + 1) * size - old_height
											A_			:  Optional[Any]			=				(old_width // size + 1) * size - old_width
											return old_height + pad_height, old_width + pad_width







						def    _a     (  self   :							int     ):





											"""simple docstring"""







											A_			:  Dict			=				self.image_processing_class(**self.image_processor_dict     )
											# create random PIL images
											A_			:  Optional[int]			=				self.image_processor_tester.prepare_inputs(equal_resolution=_lowerCamelCase     )
											for image in image_inputs:
																self.assertIsInstance(_lowerCamelCase				,			Image.Image     )

											# Test not batched input
											A_			:  Any			=				image_processor(image_inputs[0]				,			return_tensors='''pt'''     ).pixel_values
											A_			,A_			:  List[Any]			=				self.calculate_expected_size(np.array(image_inputs[0]     )     )
											self.assertEqual(
											    encoded_images.shape				,			(
											        1,
											        self.image_processor_tester.num_channels,
											        expected_height,
											        expected_width,
											    )				,			)







						def    _a     (  self   :							Optional[Any]     ):





											"""simple docstring"""







											A_			:  Optional[Any]			=				self.image_processing_class(**self.image_processor_dict     )
											# create random numpy tensors
											A_			:  Dict			=				self.image_processor_tester.prepare_inputs(equal_resolution=_lowerCamelCase				,			numpify=_lowerCamelCase     )
											for image in image_inputs:
																self.assertIsInstance(_lowerCamelCase				,			np.ndarray     )

											# Test not batched input
											A_			:  Union[str, Any]			=				image_processor(image_inputs[0]				,			return_tensors='''pt'''     ).pixel_values
											A_			,A_			:  Union[str, Any]			=				self.calculate_expected_size(image_inputs[0]     )
											self.assertEqual(
											    encoded_images.shape				,			(
											        1,
											        self.image_processor_tester.num_channels,
											        expected_height,
											        expected_width,
											    )				,			)







						def    _a     (  self   :							int     ):





											"""simple docstring"""







											A_			:  Union[str, Any]			=				self.image_processing_class(**self.image_processor_dict     )
											# create random PyTorch tensors
											A_			:  List[Any]			=				self.image_processor_tester.prepare_inputs(equal_resolution=_lowerCamelCase				,			torchify=_lowerCamelCase     )
											for image in image_inputs:
																self.assertIsInstance(_lowerCamelCase				,			torch.Tensor     )

											# Test not batched input
											A_			:  int			=				image_processor(image_inputs[0]				,			return_tensors='''pt'''     ).pixel_values
											A_			,A_			:  str			=				self.calculate_expected_size(image_inputs[0]     )
											self.assertEqual(
											    encoded_images.shape				,			(
											        1,
											        self.image_processor_tester.num_channels,
											        expected_height,
											        expected_width,
											    )				,			)



