

from __future__ import annotations

import unittest

import numpy as np

from transformers import LayoutLMConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
  import tensorflow as tf

  from transformers.models.layoutlm.modeling_tf_layoutlm import (
      TF_LAYOUTLM_PRETRAINED_MODEL_ARCHIVE_LIST,
      TFLayoutLMForMaskedLM,
      TFLayoutLMForQuestionAnswering,
      TFLayoutLMForSequenceClassification,
      TFLayoutLMForTokenClassification,
      TFLayoutLMModel,
  )







class        a__       :







  """simple docstring"""




  def __init__(			self    : List[Any]				,				UpperCAmelCase__    : List[str]				,				UpperCAmelCase__    : Tuple=1_3				,				UpperCAmelCase__    : str=7				,				UpperCAmelCase__    : Union[str, Any]=True				,				UpperCAmelCase__    : int=True				,				UpperCAmelCase__    : Optional[Any]=True				,				UpperCAmelCase__    : Any=True				,				UpperCAmelCase__    : Tuple=9_9				,				UpperCAmelCase__    : str=3_2				,				UpperCAmelCase__    : Tuple=2				,				UpperCAmelCase__    : Any=4				,				UpperCAmelCase__    : List[str]=3_7				,				UpperCAmelCase__    : Optional[Any]="gelu"				,				UpperCAmelCase__    : List[Any]=0.1				,				UpperCAmelCase__    : Any=0.1				,				UpperCAmelCase__    : Tuple=5_1_2				,				UpperCAmelCase__    : Union[str, Any]=1_6				,				UpperCAmelCase__    : Optional[int]=2				,				UpperCAmelCase__    : int=0.02				,				UpperCAmelCase__    : List[str]=3				,				UpperCAmelCase__    : List[str]=4				,				UpperCAmelCase__    : List[Any]=None				,				UpperCAmelCase__    : List[str]=1_0_0_0				,				)					->Optional[int]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Tuple       =   parent
        SCREAMING_SNAKE_CASE :      Dict       =   batch_size
        SCREAMING_SNAKE_CASE :      Dict       =   seq_length
        SCREAMING_SNAKE_CASE :      int       =   is_training
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   use_input_mask
        SCREAMING_SNAKE_CASE :      Any       =   use_token_type_ids
        SCREAMING_SNAKE_CASE :      Optional[int]       =   use_labels
        SCREAMING_SNAKE_CASE :      Any       =   vocab_size
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   hidden_size
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   num_hidden_layers
        SCREAMING_SNAKE_CASE :      Optional[int]       =   num_attention_heads
        SCREAMING_SNAKE_CASE :      Any       =   intermediate_size
        SCREAMING_SNAKE_CASE :      List[Any]       =   hidden_act
        SCREAMING_SNAKE_CASE :      List[Any]       =   hidden_dropout_prob
        SCREAMING_SNAKE_CASE :      List[Any]       =   attention_probs_dropout_prob
        SCREAMING_SNAKE_CASE :      str       =   max_position_embeddings
        SCREAMING_SNAKE_CASE :      List[Any]       =   type_vocab_size
        SCREAMING_SNAKE_CASE :      str       =   type_sequence_label_size
        SCREAMING_SNAKE_CASE :      List[Any]       =   initializer_range
        SCREAMING_SNAKE_CASE :      Any       =   num_labels
        SCREAMING_SNAKE_CASE :      List[str]       =   num_choices
        SCREAMING_SNAKE_CASE :      Dict       =   scope
        SCREAMING_SNAKE_CASE :      List[Any]       =   range_bbox


  def  _lowercase      (			self    : List[str]		)					->int:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      List[Any]       =   ids_tensor([self.batch_size, self.seq_length]				,				self.vocab_size		)

        # convert bbox to numpy since TF does not support item assignment
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   ids_tensor([self.batch_size, self.seq_length, 4]				,				self.range_bbox		).numpy()
        # Ensure that bbox is legal
        for i in range(bbox.shape[0]		):
              for j in range(bbox.shape[1]		):
                    if bbox[i, j, 3] < bbox[i, j, 1]:
                          SCREAMING_SNAKE_CASE :      Union[str, Any]       =   bbox[i, j, 3]
                          SCREAMING_SNAKE_CASE :      Union[str, Any]       =   bbox[i, j, 1]
                          SCREAMING_SNAKE_CASE :      int       =   t
                    if bbox[i, j, 2] < bbox[i, j, 0]:
                          SCREAMING_SNAKE_CASE :      int       =   bbox[i, j, 2]
                          SCREAMING_SNAKE_CASE :      List[Any]       =   bbox[i, j, 0]
                          SCREAMING_SNAKE_CASE :      Dict       =   t
        SCREAMING_SNAKE_CASE :      List[str]       =   tf.convert_to_tensor(UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      Optional[Any]       =   None
        if self.use_input_mask:
              SCREAMING_SNAKE_CASE :      str       =   random_attention_mask([self.batch_size, self.seq_length]		)

        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   None
        if self.use_token_type_ids:
              SCREAMING_SNAKE_CASE :      Union[str, Any]       =   ids_tensor([self.batch_size, self.seq_length]				,				self.type_vocab_size		)

        SCREAMING_SNAKE_CASE :      int       =   None
        SCREAMING_SNAKE_CASE :      int       =   None
        SCREAMING_SNAKE_CASE :      Dict       =   None
        if self.use_labels:
              SCREAMING_SNAKE_CASE :      Any       =   ids_tensor([self.batch_size]				,				self.type_sequence_label_size		)
              SCREAMING_SNAKE_CASE :      Optional[int]       =   ids_tensor([self.batch_size, self.seq_length]				,				self.num_labels		)
              SCREAMING_SNAKE_CASE :      Any       =   ids_tensor([self.batch_size]				,				self.num_choices		)

        SCREAMING_SNAKE_CASE :      Optional[int]       =   LayoutLMConfig(
            vocab_size=self.vocab_size				,				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				,				max_position_embeddings=self.max_position_embeddings				,				type_vocab_size=self.type_vocab_size				,				initializer_range=self.initializer_range				,				)

        return config, input_ids, bbox, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels


  def  _lowercase      (			self    : Union[str, Any]				,				UpperCAmelCase__    : str				,				UpperCAmelCase__    : str				,				UpperCAmelCase__    : List[Any]				,				UpperCAmelCase__    : Union[str, Any]				,				UpperCAmelCase__    : List[str]				,				UpperCAmelCase__    : Dict				,				UpperCAmelCase__    : Dict				,				UpperCAmelCase__    : Union[str, Any]		)					->Dict:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      int       =   TFLayoutLMModel(config=UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      Optional[int]       =   model(UpperCAmelCase__				,				UpperCAmelCase__				,				attention_mask=UpperCAmelCase__				,				token_type_ids=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      List[str]       =   model(UpperCAmelCase__				,				UpperCAmelCase__				,				token_type_ids=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Tuple       =   model(UpperCAmelCase__				,				UpperCAmelCase__		)
        self.parent.assertEqual(result.last_hidden_state.shape				,				(self.batch_size, self.seq_length, self.hidden_size)		)
        self.parent.assertEqual(result.pooler_output.shape				,				(self.batch_size, self.hidden_size)		)


  def  _lowercase      (			self    : Any				,				UpperCAmelCase__    : Optional[Any]				,				UpperCAmelCase__    : Optional[Any]				,				UpperCAmelCase__    : Optional[Any]				,				UpperCAmelCase__    : Optional[int]				,				UpperCAmelCase__    : str				,				UpperCAmelCase__    : Dict				,				UpperCAmelCase__    : Union[str, Any]				,				UpperCAmelCase__    : Tuple		)					->int:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   TFLayoutLMForMaskedLM(config=UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      Any       =   model(UpperCAmelCase__				,				UpperCAmelCase__				,				attention_mask=UpperCAmelCase__				,				token_type_ids=UpperCAmelCase__				,				labels=UpperCAmelCase__		)
        self.parent.assertEqual(result.logits.shape				,				(self.batch_size, self.seq_length, self.vocab_size)		)


  def  _lowercase      (			self    : Any				,				UpperCAmelCase__    : str				,				UpperCAmelCase__    : List[Any]				,				UpperCAmelCase__    : Dict				,				UpperCAmelCase__    : Union[str, Any]				,				UpperCAmelCase__    : Optional[Any]				,				UpperCAmelCase__    : str				,				UpperCAmelCase__    : List[Any]				,				UpperCAmelCase__    : Union[str, Any]		)					->Any:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      int       =   self.num_labels
        SCREAMING_SNAKE_CASE :      List[Any]       =   TFLayoutLMForSequenceClassification(config=UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      str       =   model(UpperCAmelCase__				,				UpperCAmelCase__				,				attention_mask=UpperCAmelCase__				,				token_type_ids=UpperCAmelCase__		)
        self.parent.assertEqual(result.logits.shape				,				(self.batch_size, self.num_labels)		)


  def  _lowercase      (			self    : str				,				UpperCAmelCase__    : Dict				,				UpperCAmelCase__    : Union[str, Any]				,				UpperCAmelCase__    : List[Any]				,				UpperCAmelCase__    : int				,				UpperCAmelCase__    : Union[str, Any]				,				UpperCAmelCase__    : List[Any]				,				UpperCAmelCase__    : str				,				UpperCAmelCase__    : Optional[Any]		)					->List[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      List[str]       =   self.num_labels
        SCREAMING_SNAKE_CASE :      str       =   TFLayoutLMForTokenClassification(config=UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      Dict       =   model(UpperCAmelCase__				,				UpperCAmelCase__				,				attention_mask=UpperCAmelCase__				,				token_type_ids=UpperCAmelCase__				,				labels=UpperCAmelCase__		)
        self.parent.assertEqual(result.logits.shape				,				(self.batch_size, self.seq_length, self.num_labels)		)


  def  _lowercase      (			self    : Any				,				UpperCAmelCase__    : List[Any]				,				UpperCAmelCase__    : List[str]				,				UpperCAmelCase__    : Tuple				,				UpperCAmelCase__    : int				,				UpperCAmelCase__    : str				,				UpperCAmelCase__    : Union[str, Any]				,				UpperCAmelCase__    : Union[str, Any]				,				UpperCAmelCase__    : Optional[int]		)					->int:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Optional[Any]       =   TFLayoutLMForQuestionAnswering(config=UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      List[str]       =   model(UpperCAmelCase__				,				UpperCAmelCase__				,				attention_mask=UpperCAmelCase__				,				token_type_ids=UpperCAmelCase__		)
        self.parent.assertEqual(result.start_logits.shape				,				(self.batch_size, self.seq_length)		)
        self.parent.assertEqual(result.end_logits.shape				,				(self.batch_size, self.seq_length)		)




  def  _lowercase      (			self    : Union[str, Any]		)					->str:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      List[str]       =   self.prepare_config_and_inputs()
        (
            SCREAMING_SNAKE_CASE	
        ) :      int       =   config_and_inputs
        SCREAMING_SNAKE_CASE :      Optional[int]       =   {
            """input_ids""": input_ids,
            """bbox""": bbox,
            """token_type_ids""": token_type_ids,
            """attention_mask""": input_mask,
        }
        return config, inputs_dict







@require_tf
class        a__       (   UpperCAmelCase	,			UpperCAmelCase	,			unittest.TestCase       ):







  """simple docstring"""




  UpperCAmelCase__      : Optional[Any]			       =(
      (
          TFLayoutLMModel,
          TFLayoutLMForMaskedLM,
          TFLayoutLMForTokenClassification,
          TFLayoutLMForSequenceClassification,
          TFLayoutLMForQuestionAnswering,
      )
      if is_tf_available()
      else ()
  )
  UpperCAmelCase__      : Optional[Any]			       =(
      {
          """feature-extraction""": TFLayoutLMModel,
          """fill-mask""": TFLayoutLMForMaskedLM,
          """text-classification""": TFLayoutLMForSequenceClassification,
          """token-classification""": TFLayoutLMForTokenClassification,
          """zero-shot""": TFLayoutLMForSequenceClassification,
      }
      if is_tf_available()
      else {}
  )
  UpperCAmelCase__      : str			       =False
  UpperCAmelCase__      : Optional[int]			       =True
  UpperCAmelCase__      : Tuple			       =1_0


  def  _lowercase      (			self    : Any		)					->str:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   TFLayoutLMModelTester(self		)
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   ConfigTester(self				,				config_class=UpperCAmelCase__				,				hidden_size=3_7		)


  def  _lowercase      (			self    : str		)					->Dict:
        """simple docstring"""







        self.config_tester.run_common_tests()


  def  _lowercase      (			self    : Optional[int]		)					->Dict:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      str       =   self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_model(*UpperCAmelCase__		)


  def  _lowercase      (			self    : List[Any]		)					->Dict:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      List[Any]       =   self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_for_masked_lm(*UpperCAmelCase__		)


  def  _lowercase      (			self    : Union[str, Any]		)					->Any:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      int       =   self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_for_sequence_classification(*UpperCAmelCase__		)


  def  _lowercase      (			self    : Any		)					->Tuple:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      List[str]       =   self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_for_token_classification(*UpperCAmelCase__		)


  def  _lowercase      (			self    : int		)					->Optional[int]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Optional[Any]       =   self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_for_question_answering(*UpperCAmelCase__		)


  @slow
  def  _lowercase      (			self    : int		)					->Any:
        """simple docstring"""







        for model_name in TF_LAYOUTLM_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
              SCREAMING_SNAKE_CASE :      int       =   TFLayoutLMModel.from_pretrained(UpperCAmelCase__		)
              self.assertIsNotNone(UpperCAmelCase__		)




  @unittest.skip("""Onnx compliancy broke with TF 2.10"""		)
  def  _lowercase      (			self    : Tuple		)					->Union[str, Any]:
        """simple docstring"""







        pass






def 					__lowercase      (     )     ->						str:
      # Here we prepare a batch of 2 sequences to test a LayoutLM forward pass on:
      # fmt: off
      SCREAMING_SNAKE_CASE :      int       =   tf.convert_to_tensor([[101,1019,1014,1016,1037,12849,4747,1004,14246,2278,5439,4524,5002,2930,2193,2930,4341,3208,1005,1055,2171,2848,11300,3531,102],[101,4070,4034,7020,1024,3058,1015,1013,2861,1013,6070,19274,2772,6205,27814,16147,16147,4343,2047,10283,10969,14389,1012,2338,102]]			)  # noqa: E231
      SCREAMING_SNAKE_CASE :      List[str]       =   tf.convert_to_tensor([[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],]			)  # noqa: E231
      SCREAMING_SNAKE_CASE :      Tuple       =   tf.convert_to_tensor([[[0,0,0,0],[423,237,440,251],[427,272,441,287],[419,115,437,129],[961,885,992,912],[256,38,330,58],[256,38,330,58],[336,42,353,57],[360,39,401,56],[360,39,401,56],[411,39,471,59],[479,41,528,59],[533,39,630,60],[67,113,134,131],[141,115,209,132],[68,149,133,166],[141,149,187,164],[195,148,287,165],[195,148,287,165],[195,148,287,165],[295,148,349,165],[441,149,492,166],[497,149,546,164],[64,201,125,218],[1000,1000,1000,1000]],[[0,0,0,0],[662,150,754,166],[665,199,742,211],[519,213,554,228],[519,213,554,228],[134,433,187,454],[130,467,204,480],[130,467,204,480],[130,467,204,480],[130,467,204,480],[130,467,204,480],[314,469,376,482],[504,684,582,706],[941,825,973,900],[941,825,973,900],[941,825,973,900],[941,825,973,900],[610,749,652,765],[130,659,168,672],[176,657,237,672],[238,657,312,672],[443,653,628,672],[443,653,628,672],[716,301,825,317],[1000,1000,1000,1000]]]			)  # noqa: E231
      SCREAMING_SNAKE_CASE :      Any       =   tf.convert_to_tensor([[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]]			)  # noqa: E231
      # these are sequence labels (i.e. at the token level)
      SCREAMING_SNAKE_CASE :      List[str]       =   tf.convert_to_tensor([[-100,10,10,10,9,1,-100,7,7,-100,7,7,4,2,5,2,8,8,-100,-100,5,0,3,2,-100],[-100,12,12,12,-100,12,10,-100,-100,-100,-100,10,12,9,-100,-100,-100,10,10,10,9,12,-100,10,-100]]			)  # noqa: E231
      # fmt: on

      return input_ids, attention_mask, bbox, token_type_ids, labels







@require_tf
class        a__       (   unittest.TestCase       ):







  """simple docstring"""




  @slow
  def  _lowercase      (			self    : int		)					->List[str]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Any       =   TFLayoutLMModel.from_pretrained("""microsoft/layoutlm-base-uncased"""		)

        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   prepare_layoutlm_batch_inputs()

        # forward pass
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   model(input_ids=UpperCAmelCase__				,				bbox=UpperCAmelCase__				,				attention_mask=UpperCAmelCase__				,				token_type_ids=UpperCAmelCase__		)

        # test the sequence output on [0, :3, :3]
        SCREAMING_SNAKE_CASE :      str       =   tf.convert_to_tensor(
            [[0.17_85, -0.19_47, -0.04_25], [-0.32_54, -0.28_07, 0.25_53], [-0.53_91, -0.33_22, 0.33_64]]				,				)

        self.assertTrue(np.allclose(outputs.last_hidden_state[0, :3, :3]				,				UpperCAmelCase__				,				atol=1e-3		)		)

        # test the pooled output on [1, :3]
        SCREAMING_SNAKE_CASE :      Optional[int]       =   tf.convert_to_tensor([-0.65_80, -0.02_14, 0.85_52]		)

        self.assertTrue(np.allclose(outputs.pooler_output[1, :3]				,				UpperCAmelCase__				,				atol=1e-3		)		)


  @slow
  def  _lowercase      (			self    : str		)					->List[str]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Optional[Any]       =   TFLayoutLMForSequenceClassification.from_pretrained("""microsoft/layoutlm-base-uncased"""				,				num_labels=2		)

        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   prepare_layoutlm_batch_inputs()

        # forward pass
        SCREAMING_SNAKE_CASE :      Tuple       =   model(
            input_ids=UpperCAmelCase__				,				bbox=UpperCAmelCase__				,				attention_mask=UpperCAmelCase__				,				token_type_ids=UpperCAmelCase__				,				labels=tf.convert_to_tensor([1, 1]		)				,				)

        # test whether we get a loss as a scalar
        SCREAMING_SNAKE_CASE :      Any       =   outputs.loss
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   (2,)
        self.assertEqual(loss.shape				,				UpperCAmelCase__		)

        # test the shape of the logits
        SCREAMING_SNAKE_CASE :      Tuple       =   outputs.logits
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   (2, 2)
        self.assertEqual(logits.shape				,				UpperCAmelCase__		)


  @slow
  def  _lowercase      (			self    : List[str]		)					->Optional[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Optional[int]       =   TFLayoutLMForTokenClassification.from_pretrained("""microsoft/layoutlm-base-uncased"""				,				num_labels=1_3		)

        SCREAMING_SNAKE_CASE :      List[Any]       =   prepare_layoutlm_batch_inputs()

        # forward pass
        SCREAMING_SNAKE_CASE :      int       =   model(
            input_ids=UpperCAmelCase__				,				bbox=UpperCAmelCase__				,				attention_mask=UpperCAmelCase__				,				token_type_ids=UpperCAmelCase__				,				labels=UpperCAmelCase__		)

        # test the shape of the logits
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   outputs.logits
        SCREAMING_SNAKE_CASE :      str       =   tf.convert_to_tensor((2, 2_5, 1_3)		)
        self.assertEqual(logits.shape				,				UpperCAmelCase__		)




  @slow
  def  _lowercase      (			self    : List[str]		)					->Tuple:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      str       =   TFLayoutLMForQuestionAnswering.from_pretrained("""microsoft/layoutlm-base-uncased"""		)

        SCREAMING_SNAKE_CASE :      str       =   prepare_layoutlm_batch_inputs()

        # forward pass
        SCREAMING_SNAKE_CASE :      List[str]       =   model(input_ids=UpperCAmelCase__				,				bbox=UpperCAmelCase__				,				attention_mask=UpperCAmelCase__				,				token_type_ids=UpperCAmelCase__		)

        # test the shape of the logits
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   tf.convert_to_tensor((2, 2_5)		)
        self.assertEqual(outputs.start_logits.shape				,				UpperCAmelCase__		)
        self.assertEqual(outputs.end_logits.shape				,				UpperCAmelCase__		)







import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


UpperCAmelCase__ :     Any				    =    logging.get_logger(__name__)  # pylint: disable=invalid-name







class        a__       (   UpperCAmelCase       ):







  """simple docstring"""




  def __init__(			self    : Dict				,				UpperCAmelCase__    : AutoencoderKL				,				UpperCAmelCase__    : CLIPTextModel				,				UpperCAmelCase__    : CLIPTokenizer				,				UpperCAmelCase__    : UNetaDConditionModel				,				UpperCAmelCase__    : Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]				,				UpperCAmelCase__    : StableDiffusionSafetyChecker				,				UpperCAmelCase__    : CLIPImageProcessor				,				)					->Optional[Any]:
        """simple docstring"""







        super().__init__()
        self.register_modules(
            vae=UpperCAmelCase__				,				text_encoder=UpperCAmelCase__				,				tokenizer=UpperCAmelCase__				,				unet=UpperCAmelCase__				,				scheduler=UpperCAmelCase__				,				safety_checker=UpperCAmelCase__				,				feature_extractor=UpperCAmelCase__				,				)


  def  _lowercase      (			self    : int				,				UpperCAmelCase__    : Optional[Union[str, int]] = "auto"		)					->Any:
        """simple docstring"""







        if slice_size == "auto":
              # half the attention head size is usually a good trade-off between
              # speed and memory
              SCREAMING_SNAKE_CASE :      int       =   self.unet.config.attention_head_dim // 2
        self.unet.set_attention_slice(UpperCAmelCase__		)


  def  _lowercase      (			self    : Dict		)					->Tuple:
        """simple docstring"""







        self.enable_attention_slicing(UpperCAmelCase__		)




  @torch.no_grad()
  def __call__(			self    : Optional[Any]				,				UpperCAmelCase__    : Union[str, List[str]]				,				UpperCAmelCase__    : int = 5_1_2				,				UpperCAmelCase__    : int = 5_1_2				,				UpperCAmelCase__    : int = 5_0				,				UpperCAmelCase__    : float = 7.5				,				UpperCAmelCase__    : Optional[Union[str, List[str]]] = None				,				UpperCAmelCase__    : Optional[int] = 1				,				UpperCAmelCase__    : float = 0.0				,				UpperCAmelCase__    : Optional[torch.Generator] = None				,				UpperCAmelCase__    : Optional[torch.FloatTensor] = None				,				UpperCAmelCase__    : Optional[str] = "pil"				,				UpperCAmelCase__    : bool = True				,				UpperCAmelCase__    : Optional[Callable[[int, int, torch.FloatTensor], None]] = None				,				UpperCAmelCase__    : int = 1				,				UpperCAmelCase__    : Optional[torch.FloatTensor] = None				,				**UpperCAmelCase__    : Tuple				,				)					->List[Any]:
        """simple docstring"""







        if isinstance(UpperCAmelCase__				,				UpperCAmelCase__		):
              SCREAMING_SNAKE_CASE :      Any       =   1
        elif isinstance(UpperCAmelCase__				,				UpperCAmelCase__		):
              SCREAMING_SNAKE_CASE :      Any       =   len(UpperCAmelCase__		)
        else:
              raise ValueError(f"`prompt` has to be of type `str` or `list` but is {type(UpperCAmelCase__		)}"		)

        if height % 8 != 0 or width % 8 != 0:
              raise ValueError(f"`height` and `width` have to be divisible by 8 but are {height} and {width}."		)

        if (callback_steps is None) or (
            callback_steps is not None and (not isinstance(UpperCAmelCase__				,				UpperCAmelCase__		) or callback_steps <= 0)
        ):
              raise ValueError(
                  f"`callback_steps` has to be a positive integer but is {callback_steps} of type"
                  f" {type(UpperCAmelCase__		)}."		)

        # get prompt text embeddings
        SCREAMING_SNAKE_CASE :      Optional[int]       =   self.tokenizer(
            UpperCAmelCase__				,				padding="""max_length"""				,				max_length=self.tokenizer.model_max_length				,				return_tensors="""pt"""				,				)
        SCREAMING_SNAKE_CASE :      str       =   text_inputs.input_ids

        if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
              SCREAMING_SNAKE_CASE :      Union[str, Any]       =   self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]		)
              logger.warning(
                  """The following part of your input was truncated because CLIP can only handle sequences up to"""
                  f" {self.tokenizer.model_max_length} tokens: {removed_text}"		)
              SCREAMING_SNAKE_CASE :      Tuple       =   text_input_ids[:, : self.tokenizer.model_max_length]

        if text_embeddings is None:
              SCREAMING_SNAKE_CASE :      Any       =   self.text_encoder(text_input_ids.to(self.device		)		)[0]

        # duplicate text embeddings for each generation per prompt, using mps friendly method
        SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      str       =   text_embeddings.shape
        SCREAMING_SNAKE_CASE :      Tuple       =   text_embeddings.repeat(1				,				UpperCAmelCase__				,				1		)
        SCREAMING_SNAKE_CASE :      int       =   text_embeddings.view(bs_embed * num_images_per_prompt				,				UpperCAmelCase__				,				-1		)

        # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
        # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
        # corresponds to doing no classifier free guidance.
        SCREAMING_SNAKE_CASE :      List[str]       =   guidance_scale > 1.0
        # get unconditional embeddings for classifier free guidance
        if do_classifier_free_guidance:
              SCREAMING_SNAKE_CASE :      List[str]
              if negative_prompt is None:
                    SCREAMING_SNAKE_CASE :      Union[str, Any]       =   [""""""]
              elif type(UpperCAmelCase__		) is not type(UpperCAmelCase__		):
                    raise TypeError(
                        f"`negative_prompt` should be the same type to `prompt`, but got {type(UpperCAmelCase__		)} !="
                        f" {type(UpperCAmelCase__		)}."		)
              elif isinstance(UpperCAmelCase__				,				UpperCAmelCase__		):
                    SCREAMING_SNAKE_CASE :      Union[str, Any]       =   [negative_prompt]
              elif batch_size != len(UpperCAmelCase__		):
                    raise ValueError(
                        f"`negative_prompt`: {negative_prompt} has batch size {len(UpperCAmelCase__		)}, but `prompt`:"
                        f" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"
                        """ the batch size of `prompt`."""		)
              else:
                    SCREAMING_SNAKE_CASE :      Optional[int]       =   negative_prompt

              SCREAMING_SNAKE_CASE :      Dict       =   text_input_ids.shape[-1]
              SCREAMING_SNAKE_CASE :      int       =   self.tokenizer(
                  UpperCAmelCase__				,				padding="""max_length"""				,				max_length=UpperCAmelCase__				,				truncation=UpperCAmelCase__				,				return_tensors="""pt"""				,				)
              SCREAMING_SNAKE_CASE :      Union[str, Any]       =   self.text_encoder(uncond_input.input_ids.to(self.device		)		)[0]

              # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
              SCREAMING_SNAKE_CASE :      Optional[int]       =   uncond_embeddings.shape[1]
              SCREAMING_SNAKE_CASE :      List[str]       =   uncond_embeddings.repeat(UpperCAmelCase__				,				UpperCAmelCase__				,				1		)
              SCREAMING_SNAKE_CASE :      int       =   uncond_embeddings.view(batch_size * num_images_per_prompt				,				UpperCAmelCase__				,				-1		)

              # For classifier free guidance, we need to do two forward passes.
              # Here we concatenate the unconditional and text embeddings into a single batch
              # to avoid doing two forward passes
              SCREAMING_SNAKE_CASE :      int       =   torch.cat([uncond_embeddings, text_embeddings]		)

        # get the initial random noise unless the user supplied it

        # Unlike in other pipelines, latents need to be generated in the target device
        # for 1-to-1 results reproducibility with the CompVis implementation.
        # However this currently doesn't work in `mps`.
        SCREAMING_SNAKE_CASE :      Dict       =   (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
        SCREAMING_SNAKE_CASE :      Dict       =   (batch_size * num_images_per_prompt, self.unet.config.in_channels, 6_4, 6_4)
        SCREAMING_SNAKE_CASE :      List[str]       =   text_embeddings.dtype
        if latents is None:
              if self.device.type == "mps":
                    # randn does not exist on mps
                    SCREAMING_SNAKE_CASE :      Union[str, Any]       =   torch.randn(
                        UpperCAmelCase__				,				generator=UpperCAmelCase__				,				device="""cpu"""				,				dtype=UpperCAmelCase__		).to(self.device		)
                    SCREAMING_SNAKE_CASE :      Dict       =   torch.randn(UpperCAmelCase__				,				generator=UpperCAmelCase__				,				device="""cpu"""				,				dtype=UpperCAmelCase__		).to(
                        self.device		)
              else:
                    SCREAMING_SNAKE_CASE :      List[Any]       =   torch.randn(
                        UpperCAmelCase__				,				generator=UpperCAmelCase__				,				device=self.device				,				dtype=UpperCAmelCase__		)
                    SCREAMING_SNAKE_CASE :      Any       =   torch.randn(UpperCAmelCase__				,				generator=UpperCAmelCase__				,				device=self.device				,				dtype=UpperCAmelCase__		)
        else:
              if latents_reference.shape != latents_shape:
                    raise ValueError(f"Unexpected latents shape, got {latents.shape}, expected {latents_shape}"		)
              SCREAMING_SNAKE_CASE :      List[str]       =   latents_reference.to(self.device		)
              SCREAMING_SNAKE_CASE :      Any       =   latents.to(self.device		)

        # This is the key part of the pipeline where we
        # try to ensure that the generated images w/ the same seed
        # but different sizes actually result in similar images
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   (latents_shape[3] - latents_shape_reference[3]) // 2
        SCREAMING_SNAKE_CASE :      Tuple       =   (latents_shape[2] - latents_shape_reference[2]) // 2
        SCREAMING_SNAKE_CASE :      str       =   latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
        SCREAMING_SNAKE_CASE :      List[Any]       =   latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   0 if dx < 0 else dx
        SCREAMING_SNAKE_CASE :      Any       =   0 if dy < 0 else dy
        SCREAMING_SNAKE_CASE :      int       =   max(-dx				,				0		)
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   max(-dy				,				0		)
        # import pdb
        # pdb.set_trace()
        SCREAMING_SNAKE_CASE :      Tuple       =   latents_reference[:, :, dy : dy + h, dx : dx + w]

        # set timesteps
        self.scheduler.set_timesteps(UpperCAmelCase__		)

        # Some schedulers like PNDM have timesteps as arrays
        # It's more optimized to move all timesteps to correct device beforehand
        SCREAMING_SNAKE_CASE :      str       =   self.scheduler.timesteps.to(self.device		)

        # scale the initial noise by the standard deviation required by the scheduler
        SCREAMING_SNAKE_CASE :      Optional[int]       =   latents * self.scheduler.init_noise_sigma

        # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
        # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
        # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
        # and should be between [0, 1]
        SCREAMING_SNAKE_CASE :      Any       =   """eta""" in set(inspect.signature(self.scheduler.step		).parameters.keys()		)
        SCREAMING_SNAKE_CASE :      str       =   {}
        if accepts_eta:
              SCREAMING_SNAKE_CASE :      Any       =   eta

        for i, t in enumerate(self.progress_bar(UpperCAmelCase__		)		):
              # expand the latents if we are doing classifier free guidance
              SCREAMING_SNAKE_CASE :      List[Any]       =   torch.cat([latents] * 2		) if do_classifier_free_guidance else latents
              SCREAMING_SNAKE_CASE :      Any       =   self.scheduler.scale_model_input(UpperCAmelCase__				,				UpperCAmelCase__		)

              # predict the noise residual
              SCREAMING_SNAKE_CASE :      Optional[int]       =   self.unet(UpperCAmelCase__				,				UpperCAmelCase__				,				encoder_hidden_states=UpperCAmelCase__		).sample

              # perform guidance
              if do_classifier_free_guidance:
                    SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      Union[str, Any]       =   noise_pred.chunk(2		)
                    SCREAMING_SNAKE_CASE :      Optional[Any]       =   noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

              # compute the previous noisy sample x_t -> x_t-1
              SCREAMING_SNAKE_CASE :      Dict       =   self.scheduler.step(UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__				,				**UpperCAmelCase__		).prev_sample

              # call the callback, if provided
              if callback is not None and i % callback_steps == 0:
                    callback(UpperCAmelCase__				,				UpperCAmelCase__				,				UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   1 / 0.1_82_15 * latents
        SCREAMING_SNAKE_CASE :      Tuple       =   self.vae.decode(UpperCAmelCase__		).sample

        SCREAMING_SNAKE_CASE :      Tuple       =   (image / 2 + 0.5).clamp(0				,				1		)

        # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
        SCREAMING_SNAKE_CASE :      Dict       =   image.cpu().permute(0				,				2				,				3				,				1		).float().numpy()

        if self.safety_checker is not None:
              SCREAMING_SNAKE_CASE :      Tuple       =   self.feature_extractor(self.numpy_to_pil(UpperCAmelCase__		)				,				return_tensors="""pt"""		).to(
                  self.device		)
              SCREAMING_SNAKE_CASE     ,	SCREAMING_SNAKE_CASE :      Any       =   self.safety_checker(
                  images=UpperCAmelCase__				,				clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype		)		)
        else:
              SCREAMING_SNAKE_CASE :      Optional[Any]       =   None

        if output_type == "pil":
              SCREAMING_SNAKE_CASE :      Any       =   self.numpy_to_pil(UpperCAmelCase__		)

        if not return_dict:
              return (image, has_nsfw_concept)

        return StableDiffusionPipelineOutput(images=UpperCAmelCase__				,				nsfw_content_detected=UpperCAmelCase__		)




