




import gc
import unittest

from transformers import MODEL_FOR_MASKED_LM_MAPPING, TF_MODEL_FOR_MASKED_LM_MAPPING, FillMaskPipeline, pipeline
from transformers.pipelines import PipelineException
from transformers.testing_utils import (
    is_pipeline_test,
    is_torch_available,
    nested_simplify,
    require_tf,
    require_torch,
    require_torch_gpu,
    slow,
)

from .test_pipelines_common import ANY

@is_pipeline_test
class        lowercase      (    unittest.TestCase):
				__lowerCAmelCase    :    Union[str, Any]         =   MODEL_FOR_MASKED_LM_MAPPING
				__lowerCAmelCase    :    Union[str, Any]         =   TF_MODEL_FOR_MASKED_LM_MAPPING







				def 					a_  (							self   :					Dict       ):



					"""simple docstring"""

					super().tearDown()
					# clean-up as much as possible GPU memory occupied by PyTorch
					gc.collect()
					if is_torch_available():
						import torch

						torch.cuda.empty_cache()







				@require_tf
				def 					a_  (							self   :					List[Any]       ):



					"""simple docstring"""

					A_  :				str	   =		pipeline(task='''fill-mask'''      ,     model='''sshleifer/tiny-distilroberta-base'''      ,     top_k=2      ,     framework='''tf'''       )
					A_  :				int	   =		unmasker('''My name is <mask>'''       )
					self.assertEqual(
					    nested_simplify(_lowerCamelCase      ,     decimals=6       )      ,     [
					        {'''sequence''': '''My name is grouped''', '''score''': 2.1E-05, '''token''': 3_80_15, '''token_str''': ''' grouped'''},
					        {'''sequence''': '''My name is accuser''', '''score''': 2.1E-05, '''token''': 2_55_06, '''token_str''': ''' accuser'''},
					    ]      ,     )

					A_  :				int	   =		unmasker('''The largest city in France is <mask>'''       )
					self.assertEqual(
					    nested_simplify(_lowerCamelCase      ,     decimals=6       )      ,     [
					        {
					            '''sequence''': '''The largest city in France is grouped''',
					            '''score''': 2.1E-05,
					            '''token''': 3_80_15,
					            '''token_str''': ''' grouped''',
					        },
					        {
					            '''sequence''': '''The largest city in France is accuser''',
					            '''score''': 2.1E-05,
					            '''token''': 2_55_06,
					            '''token_str''': ''' accuser''',
					        },
					    ]      ,     )

					A_  :				Optional[Any]	   =		unmasker('''My name is <mask>'''      ,     targets=[''' Patrick''', ''' Clara''', ''' Teven''']      ,     top_k=3       )
					self.assertEqual(
					    nested_simplify(_lowerCamelCase      ,     decimals=6       )      ,     [
					        {'''sequence''': '''My name is Clara''', '''score''': 2E-05, '''token''': 1_36_06, '''token_str''': ''' Clara'''},
					        {'''sequence''': '''My name is Patrick''', '''score''': 2E-05, '''token''': 34_99, '''token_str''': ''' Patrick'''},
					        {'''sequence''': '''My name is Te''', '''score''': 1.9E-05, '''token''': 29_41, '''token_str''': ''' Te'''},
					    ]      ,     )







				@require_torch
				def 					a_  (							self   :					str       ):



					"""simple docstring"""

					A_  :				Optional[int]	   =		pipeline(task='''fill-mask'''      ,     model='''sshleifer/tiny-distilroberta-base'''      ,     top_k=2      ,     framework='''pt'''       )

					A_  :				Tuple	   =		unmasker('''My name is <mask>'''       )
					self.assertEqual(
					    nested_simplify(_lowerCamelCase      ,     decimals=6       )      ,     [
					        {'''sequence''': '''My name is Maul''', '''score''': 2.2E-05, '''token''': 3_56_76, '''token_str''': ''' Maul'''},
					        {'''sequence''': '''My name isELS''', '''score''': 2.2E-05, '''token''': 1_64_16, '''token_str''': '''ELS'''},
					    ]      ,     )

					A_  :				str	   =		unmasker('''The largest city in France is <mask>'''       )
					self.assertEqual(
					    nested_simplify(_lowerCamelCase      ,     decimals=6       )      ,     [
					        {
					            '''sequence''': '''The largest city in France is Maul''',
					            '''score''': 2.2E-05,
					            '''token''': 3_56_76,
					            '''token_str''': ''' Maul''',
					        },
					        {'''sequence''': '''The largest city in France isELS''', '''score''': 2.2E-05, '''token''': 1_64_16, '''token_str''': '''ELS'''},
					    ]      ,     )

					A_  :				Optional[int]	   =		unmasker('''My name is <mask>'''      ,     targets=[''' Patrick''', ''' Clara''', ''' Teven''']      ,     top_k=3       )
					self.assertEqual(
					    nested_simplify(_lowerCamelCase      ,     decimals=6       )      ,     [
					        {'''sequence''': '''My name is Patrick''', '''score''': 2.1E-05, '''token''': 34_99, '''token_str''': ''' Patrick'''},
					        {'''sequence''': '''My name is Te''', '''score''': 2E-05, '''token''': 29_41, '''token_str''': ''' Te'''},
					        {'''sequence''': '''My name is Clara''', '''score''': 2E-05, '''token''': 1_36_06, '''token_str''': ''' Clara'''},
					    ]      ,     )

					A_  :				Optional[Any]	   =		unmasker('''My name is <mask> <mask>'''      ,     top_k=2       )

					self.assertEqual(
					    nested_simplify(_lowerCamelCase      ,     decimals=6       )      ,     [
					        [
					            {
					                '''score''': 2.2E-05,
					                '''token''': 3_56_76,
					                '''token_str''': ''' Maul''',
					                '''sequence''': '''<s>My name is Maul<mask></s>''',
					            },
					            {'''score''': 2.2E-05, '''token''': 1_64_16, '''token_str''': '''ELS''', '''sequence''': '''<s>My name isELS<mask></s>'''},
					        ],
					        [
					            {
					                '''score''': 2.2E-05,
					                '''token''': 3_56_76,
					                '''token_str''': ''' Maul''',
					                '''sequence''': '''<s>My name is<mask> Maul</s>''',
					            },
					            {'''score''': 2.2E-05, '''token''': 1_64_16, '''token_str''': '''ELS''', '''sequence''': '''<s>My name is<mask>ELS</s>'''},
					        ],
					    ]      ,     )







				@require_torch_gpu
				def 					a_  (							self   :					List[str]       ):



					"""simple docstring"""

					A_  :				Dict	   =		pipeline('''fill-mask'''      ,     model='''hf-internal-testing/tiny-random-distilbert'''      ,     device=0      ,     framework='''pt'''       )

					# convert model to fp16
					pipe.model.half()

					A_  :				Any	   =		pipe('''Paris is the [MASK] of France.'''       )
					# We actually don't care about the result, we just want to make sure
					# it works, meaning the float16 tensor got casted back to float32
					# for postprocessing.
					self.assertIsInstance(_lowerCamelCase      ,     _lowerCamelCase       )







				@slow
				@require_torch
				def 					a_  (							self   :					int       ):



					"""simple docstring"""

					A_  :				Union[str, Any]	   =		pipeline(task='''fill-mask'''      ,     model='''distilroberta-base'''      ,     top_k=2      ,     framework='''pt'''       )
					self.run_large_test(_lowerCamelCase       )







				@slow
				@require_tf
				def 					a_  (							self   :					str       ):



					"""simple docstring"""

					A_  :				str	   =		pipeline(task='''fill-mask'''      ,     model='''distilroberta-base'''      ,     top_k=2      ,     framework='''tf'''       )
					self.run_large_test(_lowerCamelCase       )







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



					"""simple docstring"""

					A_  :				Optional[Any]	   =		unmasker('''My name is <mask>'''       )
					self.assertEqual(
					    nested_simplify(_lowerCamelCase       )      ,     [
					        {'''sequence''': '''My name is John''', '''score''': 0.008, '''token''': 6_10, '''token_str''': ''' John'''},
					        {'''sequence''': '''My name is Chris''', '''score''': 0.007, '''token''': 15_73, '''token_str''': ''' Chris'''},
					    ]      ,     )
					A_  :				Optional[Any]	   =		unmasker('''The largest city in France is <mask>'''       )
					self.assertEqual(
					    nested_simplify(_lowerCamelCase       )      ,     [
					        {
					            '''sequence''': '''The largest city in France is Paris''',
					            '''score''': 0.251,
					            '''token''': 22_01,
					            '''token_str''': ''' Paris''',
					        },
					        {
					            '''sequence''': '''The largest city in France is Lyon''',
					            '''score''': 0.214,
					            '''token''': 1_27_90,
					            '''token_str''': ''' Lyon''',
					        },
					    ]      ,     )

					A_  :				Tuple	   =		unmasker('''My name is <mask>'''      ,     targets=[''' Patrick''', ''' Clara''', ''' Teven''']      ,     top_k=3       )
					self.assertEqual(
					    nested_simplify(_lowerCamelCase       )      ,     [
					        {'''sequence''': '''My name is Patrick''', '''score''': 0.005, '''token''': 34_99, '''token_str''': ''' Patrick'''},
					        {'''sequence''': '''My name is Clara''', '''score''': 0.000, '''token''': 1_36_06, '''token_str''': ''' Clara'''},
					        {'''sequence''': '''My name is Te''', '''score''': 0.000, '''token''': 29_41, '''token_str''': ''' Te'''},
					    ]      ,     )







				@require_torch
				def 					a_  (							self   :					int       ):



					"""simple docstring"""

					A_  :				Any	   =		pipeline(task='''fill-mask'''      ,     model='''sshleifer/tiny-distilroberta-base'''      ,     framework='''pt'''       )
					A_  :				Union[str, Any]	   =		None
					A_  :				Union[str, Any]	   =		None
					self.run_pipeline_test(_lowerCamelCase      ,     []       )







				@require_tf
				def 					a_  (							self   :					Tuple       ):



					"""simple docstring"""

					A_  :				List[str]	   =		pipeline(task='''fill-mask'''      ,     model='''sshleifer/tiny-distilroberta-base'''      ,     framework='''tf'''       )
					A_  :				Optional[Any]	   =		None
					A_  :				str	   =		None
					self.run_pipeline_test(_lowerCamelCase      ,     []       )







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



					"""simple docstring"""

					if tokenizer is None or tokenizer.mask_token_id is None:
						self.skipTest('''The provided tokenizer has no mask token, (probably reformer or wav2vec2)'''       )

					A_  :				int	   =		FillMaskPipeline(model=_lowerCamelCase      ,     tokenizer=_lowerCamelCase       )
					A_  :				Optional[int]	   =		[
					    F"""This is another {tokenizer.mask_token} test""",
					]
					return fill_masker, examples







				def 					a_  (							self   :					Optional[Any]      ,     _lowerCamelCase   :					Union[str, Any]      ,     _lowerCamelCase   :					List[str]       ):



					"""simple docstring"""

					A_  :				List[str]	   =		fill_masker.tokenizer
					A_  :				Tuple	   =		fill_masker.model

					A_  :				Union[str, Any]	   =		fill_masker(
					    F"""This is a {tokenizer.mask_token}"""      ,     )
					self.assertEqual(
					    _lowerCamelCase      ,     [
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					    ]      ,     )

					A_  :				Optional[Any]	   =		fill_masker([F"""This is a {tokenizer.mask_token}"""]       )
					self.assertEqual(
					    _lowerCamelCase      ,     [
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					    ]      ,     )

					A_  :				int	   =		fill_masker([F"""This is a {tokenizer.mask_token}""", F"""Another {tokenizer.mask_token} great test."""]       )
					self.assertEqual(
					    _lowerCamelCase      ,     [
					        [
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        ],
					        [
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        ],
					    ]      ,     )

					with self.assertRaises(_lowerCamelCase       ):
						fill_masker([None]       )
					# No mask_token is not supported
					with self.assertRaises(_lowerCamelCase       ):
						fill_masker('''This is'''       )

					self.run_test_top_k(_lowerCamelCase      ,     _lowerCamelCase       )
					self.run_test_targets(_lowerCamelCase      ,     _lowerCamelCase       )
					self.run_test_top_k_targets(_lowerCamelCase      ,     _lowerCamelCase       )
					self.fill_mask_with_duplicate_targets_and_top_k(_lowerCamelCase      ,     _lowerCamelCase       )
					self.fill_mask_with_multiple_masks(_lowerCamelCase      ,     _lowerCamelCase       )







				def 					a_  (							self   :					Dict      ,     _lowerCamelCase   :					List[str]      ,     _lowerCamelCase   :					Optional[int]       ):



					"""simple docstring"""

					A_  :				List[Any]	   =		tokenizer.get_vocab()
					A_  :				Union[str, Any]	   =		sorted(vocab.keys()       )[:2]
					# Pipeline argument
					A_  :				Dict	   =		FillMaskPipeline(model=_lowerCamelCase      ,     tokenizer=_lowerCamelCase      ,     targets=_lowerCamelCase       )
					A_  :				Optional[Any]	   =		fill_masker(F"""This is a {tokenizer.mask_token}"""       )
					self.assertEqual(
					    _lowerCamelCase      ,     [
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					    ]      ,     )
					A_  :				Dict	   =		{vocab[el] for el in targets}
					self.assertEqual({el['''token'''] for el in outputs}      ,     _lowerCamelCase       )
					A_  :				List[Any]	   =		[tokenizer.decode([x]       ) for x in target_ids]
					self.assertEqual({el['''token_str'''] for el in outputs}      ,     set(_lowerCamelCase       )       )

					# Call argument
					A_  :				Tuple	   =		FillMaskPipeline(model=_lowerCamelCase      ,     tokenizer=_lowerCamelCase       )
					A_  :				str	   =		fill_masker(F"""This is a {tokenizer.mask_token}"""      ,     targets=_lowerCamelCase       )
					self.assertEqual(
					    _lowerCamelCase      ,     [
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					    ]      ,     )
					A_  :				int	   =		{vocab[el] for el in targets}
					self.assertEqual({el['''token'''] for el in outputs}      ,     _lowerCamelCase       )
					A_  :				List[str]	   =		[tokenizer.decode([x]       ) for x in target_ids]
					self.assertEqual({el['''token_str'''] for el in outputs}      ,     set(_lowerCamelCase       )       )

					# Score equivalence
					A_  :				Tuple	   =		fill_masker(F"""This is a {tokenizer.mask_token}"""      ,     targets=_lowerCamelCase       )
					A_  :				List[Any]	   =		[top_mask['''token_str'''] for top_mask in outputs]
					A_  :				List[str]	   =		[top_mask['''score'''] for top_mask in outputs]

					# For some BPE tokenizers, `</w>` is removed during decoding, so `token_str` won't be the same as in `targets`.
					if set(_lowerCamelCase       ) == set(_lowerCamelCase       ):
						A_  :				List[Any]	   =		fill_masker(F"""This is a {tokenizer.mask_token}"""      ,     targets=_lowerCamelCase       )
						A_  :				Union[str, Any]	   =		[top_mask['''score'''] for top_mask in unmasked_targets]
						self.assertEqual(nested_simplify(_lowerCamelCase       )      ,     nested_simplify(_lowerCamelCase       )       )

					# Raises with invalid
					with self.assertRaises(_lowerCamelCase       ):
						A_  :				Any	   =		fill_masker(F"""This is a {tokenizer.mask_token}"""      ,     targets=[]       )
					# For some tokenizers, `""` is actually in the vocabulary and the expected error won't raised
					if "" not in tokenizer.get_vocab():
						with self.assertRaises(_lowerCamelCase       ):
							A_  :				int	   =		fill_masker(F"""This is a {tokenizer.mask_token}"""      ,     targets=['''''']       )
						with self.assertRaises(_lowerCamelCase       ):
							A_  :				int	   =		fill_masker(F"""This is a {tokenizer.mask_token}"""      ,     targets=''''''       )







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



					"""simple docstring"""

					A_  :				List[Any]	   =		FillMaskPipeline(model=_lowerCamelCase      ,     tokenizer=_lowerCamelCase      ,     top_k=2       )
					A_  :				Tuple	   =		fill_masker(F"""This is a {tokenizer.mask_token}"""       )
					self.assertEqual(
					    _lowerCamelCase      ,     [
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					    ]      ,     )

					A_  :				Optional[Any]	   =		FillMaskPipeline(model=_lowerCamelCase      ,     tokenizer=_lowerCamelCase       )
					A_  :				Optional[Any]	   =		fill_masker(F"""This is a {tokenizer.mask_token}"""      ,     top_k=2       )
					self.assertEqual(
					    _lowerCamelCase      ,     [
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					    ]      ,     )
					self.assertEqual(nested_simplify(_lowerCamelCase       )      ,     nested_simplify(_lowerCamelCase       )       )







				def 					a_  (							self   :					Optional[Any]      ,     _lowerCamelCase   :					Union[str, Any]      ,     _lowerCamelCase   :					Optional[int]       ):



					"""simple docstring"""

					A_  :				int	   =		tokenizer.get_vocab()
					A_  :				int	   =		FillMaskPipeline(model=_lowerCamelCase      ,     tokenizer=_lowerCamelCase       )

					# top_k=2, ntargets=3
					A_  :				Dict	   =		sorted(vocab.keys()       )[:3]
					A_  :				Dict	   =		fill_masker(F"""This is a {tokenizer.mask_token}"""      ,     top_k=2      ,     targets=_lowerCamelCase       )

					# If we use the most probably targets, and filter differently, we should still
					# have the same results
					A_  :				List[Any]	   =		[el['''token_str'''] for el in sorted(_lowerCamelCase      ,     key=lambda _lowerCamelCase       : x["score"]      ,     reverse=_lowerCamelCase       )]
					# For some BPE tokenizers, `</w>` is removed during decoding, so `token_str` won't be the same as in `targets`.
					if set(_lowerCamelCase       ).issubset(_lowerCamelCase       ):
						A_  :				Optional[int]	   =		fill_masker(F"""This is a {tokenizer.mask_token}"""      ,     top_k=3      ,     targets=_lowerCamelCase       )
						# They should yield exactly the same result
						self.assertEqual(nested_simplify(_lowerCamelCase       )      ,     nested_simplify(_lowerCamelCase       )       )







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



					"""simple docstring"""

					A_  :				Optional[int]	   =		FillMaskPipeline(model=_lowerCamelCase      ,     tokenizer=_lowerCamelCase       )
					A_  :				str	   =		tokenizer.get_vocab()
					# String duplicates + id duplicates
					A_  :				str	   =		sorted(vocab.keys()       )[:3]
					A_  :				Optional[int]	   =		[targets[0], targets[1], targets[0], targets[2], targets[1]]
					A_  :				List[Any]	   =		fill_masker(F"""My name is {tokenizer.mask_token}"""      ,     targets=_lowerCamelCase      ,     top_k=10       )

					# The target list contains duplicates, so we can't output more
					# than them
					self.assertEqual(len(_lowerCamelCase       )      ,     3       )







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



					"""simple docstring"""

					A_  :				List[Any]	   =		FillMaskPipeline(model=_lowerCamelCase      ,     tokenizer=_lowerCamelCase       )

					A_  :				Any	   =		fill_masker(
					    F"""This is a {tokenizer.mask_token} {tokenizer.mask_token} {tokenizer.mask_token}"""      ,     top_k=2       )
					self.assertEqual(
					    _lowerCamelCase      ,     [
					        [
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        ],
					        [
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        ],
					        [
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					            {'''sequence''': ANY(_lowerCamelCase       ), '''score''': ANY(_lowerCamelCase       ), '''token''': ANY(_lowerCamelCase       ), '''token_str''': ANY(_lowerCamelCase       )},
					        ],
					    ]      ,     )











"""simple docstring"""

import gc
import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()

class        lowercase      (    __UpperCAmelCase     ,	__UpperCAmelCase     ,	unittest.TestCase):
				__lowerCAmelCase    :    int         =   StableDiffusionXLImgaImgPipeline
				__lowerCAmelCase    :    List[Any]         =   TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
				__lowerCAmelCase    :    Optional[int]         =   PipelineTesterMixin.required_optional_params - {"""latents"""}
				__lowerCAmelCase    :    Any         =   TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
				__lowerCAmelCase    :    Optional[Any]         =   IMAGE_TO_IMAGE_IMAGE_PARAMS
				__lowerCAmelCase    :    List[str]         =   IMAGE_TO_IMAGE_IMAGE_PARAMS







				def 					a_  (							self   :					Any       ):



					"""simple docstring"""

					torch.manual_seed(0       )
					A_  :				int	   =		UNetaDConditionModel(
					    block_out_channels=(32, 64)      ,     layers_per_block=2      ,     sample_size=32      ,     in_channels=4      ,     out_channels=4      ,     down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')      ,     up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')      ,     attention_head_dim=(2, 4)      ,     use_linear_projection=_lowerCamelCase      ,     addition_embed_type='''text_time'''      ,     addition_time_embed_dim=8      ,     transformer_layers_per_block=(1, 2)      ,     projection_class_embeddings_input_dim=80      ,     cross_attention_dim=64      ,     )
					A_  :				Tuple	   =		EulerDiscreteScheduler(
					    beta_start=0.00085      ,     beta_end=0.012      ,     steps_offset=1      ,     beta_schedule='''scaled_linear'''      ,     timestep_spacing='''leading'''      ,     )
					torch.manual_seed(0       )
					A_  :				Optional[Any]	   =		AutoencoderKL(
					    block_out_channels=[32, 64]      ,     in_channels=3      ,     out_channels=3      ,     down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']      ,     up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']      ,     latent_channels=4      ,     sample_size=1_28      ,     )
					torch.manual_seed(0       )
					A_  :				Dict	   =		CLIPTextConfig(
					    bos_token_id=0      ,     eos_token_id=2      ,     hidden_size=32      ,     intermediate_size=37      ,     layer_norm_eps=1E-05      ,     num_attention_heads=4      ,     num_hidden_layers=5      ,     pad_token_id=1      ,     vocab_size=10_00      ,     hidden_act='''gelu'''      ,     projection_dim=32      ,     )
					A_  :				Any	   =		CLIPTextModel(_lowerCamelCase       )
					A_  :				List[Any]	   =		CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''      ,     local_files_only=_lowerCamelCase       )

					A_  :				int	   =		CLIPTextModelWithProjection(_lowerCamelCase       )
					A_  :				Any	   =		CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''      ,     local_files_only=_lowerCamelCase       )

					A_  :				Union[str, Any]	   =		{
					    '''unet''': unet,
					    '''scheduler''': scheduler,
					    '''vae''': vae,
					    '''text_encoder''': text_encoder,
					    '''tokenizer''': tokenizer,
					    '''text_encoder_2''': text_encoder_a,
					    '''tokenizer_2''': tokenizer_a,
					    # "safety_checker": None,
					    # "feature_extractor": None,
					}
					return components







				def 					a_  (							self   :					Tuple      ,     _lowerCamelCase   :					int      ,     _lowerCamelCase   :					List[Any]=0       ):



					"""simple docstring"""

					A_  :				Dict	   =		floats_tensor((1, 3, 32, 32)      ,     rng=random.Random(_lowerCamelCase       )       ).to(_lowerCamelCase       )
					A_  :				Optional[Any]	   =		image / 2 + 0.5
					if str(_lowerCamelCase       ).startswith('''mps'''       ):
						A_  :				str	   =		torch.manual_seed(_lowerCamelCase       )
					else:
						A_  :				Any	   =		torch.Generator(device=_lowerCamelCase       ).manual_seed(_lowerCamelCase       )
					A_  :				Any	   =		{
					    '''prompt''': '''A painting of a squirrel eating a burger''',
					    '''image''': image,
					    '''generator''': generator,
					    '''num_inference_steps''': 2,
					    '''guidance_scale''': 5.0,
					    '''output_type''': '''numpy''',
					    '''strength''': 0.75,
					}
					return inputs







				def 					a_  (							self   :					List[Any]       ):



					"""simple docstring"""

					A_  :				List[str]	   =		'''cpu'''  # ensure determinism for the device-dependent torch.Generator
					A_  :				Dict	   =		self.get_dummy_components()
					A_  :				Dict	   =		StableDiffusionXLImgaImgPipeline(**_lowerCamelCase       )
					A_  :				int	   =		sd_pipe.to(_lowerCamelCase       )
					sd_pipe.set_progress_bar_config(disable=_lowerCamelCase       )

					A_  :				Union[str, Any]	   =		self.get_dummy_inputs(_lowerCamelCase       )
					A_  :				Tuple	   =		sd_pipe(**_lowerCamelCase       ).images
					A_  :				Dict	   =		image[0, -3:, -3:, -1]

					assert image.shape == (1, 32, 32, 3)

					A_  :				Union[str, Any]	   =		np.array([0.4656, 0.4840, 0.4439, 0.6698, 0.5574, 0.4524, 0.5799, 0.5943, 0.5165]       )

					assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1E-2







				def 					a_  (							self   :					Optional[int]       ):



					"""simple docstring"""

					super().test_attention_slicing_forward_pass(expected_max_diff=3E-3       )







				def 					a_  (							self   :					Dict       ):



					"""simple docstring"""

					super().test_inference_batch_single_identical(expected_max_diff=3E-3       )







				def 					a_  (							self   :					List[Any]       ):



					"""simple docstring"""

					pass







				def 					a_  (							self   :					List[str]       ):



					"""simple docstring"""

					A_  :				Optional[Any]	   =		self.get_dummy_components()
					A_  :				int	   =		StableDiffusionXLImgaImgPipeline(**_lowerCamelCase       )
					A_  :				Dict	   =		sd_pipe.to(_lowerCamelCase       )
					A_  :				List[Any]	   =		sd_pipe.to(_lowerCamelCase       )
					sd_pipe.set_progress_bar_config(disable=_lowerCamelCase       )

					# forward without prompt embeds
					A_  :				Any	   =		self.get_dummy_inputs(_lowerCamelCase       )
					A_  :				Union[str, Any]	   =		3 * ['''this is a negative prompt''']
					A_  :				Any	   =		negative_prompt
					A_  :				Any	   =		3 * [inputs['''prompt''']]

					A_  :				Union[str, Any]	   =		sd_pipe(**_lowerCamelCase       )
					A_  :				List[str]	   =		output.images[0, -3:, -3:, -1]

					# forward with prompt embeds
					A_  :				Optional[Any]	   =		self.get_dummy_inputs(_lowerCamelCase       )
					A_  :				str	   =		3 * ['''this is a negative prompt''']
					A_  :				List[str]	   =		3 * [inputs.pop('''prompt'''       )]

					(
					    (
					    A_
					)   , (
					    A_
					)   , (
					    A_
					)   , (
					    A_
					)   , 
					)  :				List[str]	   =		sd_pipe.encode_prompt(_lowerCamelCase      ,     negative_prompt=_lowerCamelCase       )

					A_  :				str	   =		sd_pipe(
					    **_lowerCamelCase      ,     prompt_embeds=_lowerCamelCase      ,     negative_prompt_embeds=_lowerCamelCase      ,     pooled_prompt_embeds=_lowerCamelCase      ,     negative_pooled_prompt_embeds=_lowerCamelCase      ,     )
					A_  :				Any	   =		output.images[0, -3:, -3:, -1]

					# make sure that it's equal
					assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()       ).max() < 1E-4

@slow
@require_torch_gpu
class        lowercase      (    unittest.TestCase):







				def 					a_  (							self   :					Dict       ):



					"""simple docstring"""

					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()







				def 					a_  (							self   :					Tuple      ,     _lowerCamelCase   :					List[str]      ,     _lowerCamelCase   :					Union[str, Any]="cpu"      ,     _lowerCamelCase   :					Optional[Any]=torch.floataa      ,     _lowerCamelCase   :					Optional[int]=0       ):



					"""simple docstring"""

					A_  :				int	   =		torch.Generator(device=_lowerCamelCase       ).manual_seed(_lowerCamelCase       )
					A_  :				List[Any]	   =		np.random.RandomState(_lowerCamelCase       ).standard_normal((1, 4, 64, 64)       )
					A_  :				Union[str, Any]	   =		torch.from_numpy(_lowerCamelCase       ).to(device=_lowerCamelCase      ,     dtype=_lowerCamelCase       )
					A_  :				Optional[Any]	   =		{
					    '''prompt''': '''a photograph of an astronaut riding a horse''',
					    '''latents''': latents,
					    '''generator''': generator,
					    '''num_inference_steps''': 3,
					    '''guidance_scale''': 7.5,
					    '''output_type''': '''numpy''',
					}
					return inputs







				def 					a_  (							self   :					Tuple       ):



					"""simple docstring"""

					A_  :				str	   =		DiffusionPipeline.from_pretrained('''stabilityai/stable-diffusion-2-base'''       )
					pipe.to(_lowerCamelCase       )
					pipe.set_progress_bar_config(disable=_lowerCamelCase       )

					A_  :				Optional[Any]	   =		self.get_inputs(_lowerCamelCase       )
					A_  :				Union[str, Any]	   =		pipe(**_lowerCamelCase       ).images
					A_  :				List[Any]	   =		image[0, -3:, -3:, -1].flatten()

					assert image.shape == (1, 5_12, 5_12, 3)
					A_  :				int	   =		np.array([0.49493, 0.47896, 0.40798, 0.54214, 0.53212, 0.48202, 0.47656, 0.46329, 0.48506]       )
					assert np.abs(image_slice - expected_slice       ).max() < 7E-3





