



import gc
import random
import unittest

import numpy as np
import torch

from diffusers import (
    DDIMScheduler,
    KandinskyVaaControlnetPipeline,
    KandinskyVaaPriorPipeline,
    UNetaDConditionModel,
    VQModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()


class 	lowerCamelCase__	(       _A							,  unittest.TestCase):



 """simple docstring"""

 a__					: Optional[int] =  KandinskyVaaControlnetPipeline
 a__					: Any =  ["image_embeds", "negative_image_embeds", "hint"]
 a__					: str =  ["image_embeds", "negative_image_embeds", "hint"]
 a__					: List[Any] =  [
     "generator",
     "height",
     "width",
     "latents",
     "guidance_scale",
     "num_inference_steps",
     "return_dict",
     "guidance_scale",
     "num_images_per_prompt",
     "output_type",
     "return_dict",
 ]
 a__					: Union[str, Any] =  False

 @property
 def     snake_case_      (    self					:					str      )   ->	List[Any]:
    return 32

 @property
 def     snake_case_      (    self					:					Optional[int]      )   ->	Optional[int]:
    return 32

 @property
 def     snake_case_      (    self					:					Union[str, Any]      )   ->	Tuple:
    return self.time_input_dim

 @property
 def     snake_case_      (    self					:					str      )   ->	Any:
    return self.time_input_dim * 4

 @property
 def     snake_case_      (    self					:					Optional[int]      )   ->	Tuple:
    return 1_00

 @property
 def     snake_case_      (    self					:					Tuple      )   ->	Union[str, Any]:
    torch.manual_seed(0      )

    _A      =       {
        '''in_channels''': 8,
        # Out channels is double in channels because predicts mean and variance
        '''out_channels''': 8,
        '''addition_embed_type''': '''image_hint''',
        '''down_block_types''': ('''ResnetDownsampleBlock2D''', '''SimpleCrossAttnDownBlock2D'''),
        '''up_block_types''': ('''SimpleCrossAttnUpBlock2D''', '''ResnetUpsampleBlock2D'''),
        '''mid_block_type''': '''UNetMidBlock2DSimpleCrossAttn''',
        '''block_out_channels''': (self.block_out_channels_a, self.block_out_channels_a * 2),
        '''layers_per_block''': 1,
        '''encoder_hid_dim''': self.text_embedder_hidden_size,
        '''encoder_hid_dim_type''': '''image_proj''',
        '''cross_attention_dim''': self.cross_attention_dim,
        '''attention_head_dim''': 4,
        '''resnet_time_scale_shift''': '''scale_shift''',
        '''class_embed_type''': None,
    }

    _A      =       UNetaDConditionModel(**__lowerCAmelCase      )
    return model

 @property
 def     snake_case_      (    self					:					Tuple      )   ->	List[Any]:
    return {
        "block_out_channels": [32, 32, 64, 64],
        "down_block_types": [
            "DownEncoderBlock2D",
            "DownEncoderBlock2D",
            "DownEncoderBlock2D",
            "AttnDownEncoderBlock2D",
        ],
        "in_channels": 3,
        "latent_channels": 4,
        "layers_per_block": 1,
        "norm_num_groups": 8,
        "norm_type": "spatial",
        "num_vq_embeddings": 12,
        "out_channels": 3,
        "up_block_types": ["AttnUpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D", "UpDecoderBlock2D"],
        "vq_embed_dim": 4,
    }

 @property
 def     snake_case_      (    self					:					int      )   ->	List[str]:
    torch.manual_seed(0      )
    _A      =       VQModel(**self.dummy_movq_kwargs      )
    return model

 def     snake_case_      (    self					:					List[str]      )   ->	Dict:
    _A      =       self.dummy_unet
    _A      =       self.dummy_movq

    _A      =       DDIMScheduler(
        num_train_timesteps=10_00			,						beta_schedule='''linear'''			,						beta_start=0.0_0085			,						beta_end=0.012			,						clip_sample=__lowerCAmelCase			,						set_alpha_to_one=__lowerCAmelCase			,						steps_offset=1			,						prediction_type='''epsilon'''			,						thresholding=__lowerCAmelCase			,						)

    _A      =       {
        '''unet''': unet,
        '''scheduler''': scheduler,
        '''movq''': movq,
    }
    return components

 def     snake_case_      (    self					:					Optional[Any]			,						__lowerCAmelCase					:					Union[str, Any]			,						__lowerCAmelCase					:					Tuple=0      )   ->	Optional[Any]:
    _A      =       floats_tensor((1, self.text_embedder_hidden_size)			,						rng=random.Random(__lowerCAmelCase      )      ).to(__lowerCAmelCase      )
    _A      =       floats_tensor((1, self.text_embedder_hidden_size)			,						rng=random.Random(seed + 1      )      ).to(
        __lowerCAmelCase      )

    # create hint
    _A      =       floats_tensor((1, 3, 64, 64)			,						rng=random.Random(__lowerCAmelCase      )      ).to(__lowerCAmelCase      )

    if str(__lowerCAmelCase      ).startswith('''mps'''      ):
       _A      =       torch.manual_seed(__lowerCAmelCase      )
    else:
       _A      =       torch.Generator(device=__lowerCAmelCase      ).manual_seed(__lowerCAmelCase      )
    _A      =       {
        '''image_embeds''': image_embeds,
        '''negative_image_embeds''': negative_image_embeds,
        '''hint''': hint,
        '''generator''': generator,
        '''height''': 64,
        '''width''': 64,
        '''guidance_scale''': 4.0,
        '''num_inference_steps''': 2,
        '''output_type''': '''np''',
    }
    return inputs

 def     snake_case_      (    self					:					Any      )   ->	Union[str, Any]:
    _A      =       '''cpu'''

    _A      =       self.get_dummy_components()

    _A      =       self.pipeline_class(**__lowerCAmelCase      )
    _A      =       pipe.to(__lowerCAmelCase      )

    pipe.set_progress_bar_config(disable=__lowerCAmelCase      )

    _A      =       pipe(**self.get_dummy_inputs(__lowerCAmelCase      )      )
    _A      =       output.images

    _A      =       pipe(
        **self.get_dummy_inputs(__lowerCAmelCase      )			,						return_dict=__lowerCAmelCase			,						)[0]

    _A      =       image[0, -3:, -3:, -1]
    _A      =       image_from_tuple[0, -3:, -3:, -1]

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

    _A      =       np.array(
        [0.695_9826, 0.86_8279, 0.755_8092, 0.6876_9467, 0.8580_5804, 0.6597_7496, 0.4488_5302, 0.595_9111, 0.425_1595]      )

    assert (
        np.abs(image_slice.flatten() - expected_slice      ).max() < 1E-2
    ), f''' expected_slice {expected_slice}, but got {image_slice.flatten()}'''

    assert (
        np.abs(image_from_tuple_slice.flatten() - expected_slice      ).max() < 1E-2
    ), f''' expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}'''


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



 """simple docstring"""

 def     snake_case_      (    self					:					Any      )   ->	List[Any]:
    # clean up the VRAM after each test
    super().tearDown()
    gc.collect()
    torch.cuda.empty_cache()

 def     snake_case_      (    self					:					Tuple      )   ->	List[Any]:
    _A      =       load_numpy(
        '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
        '''/kandinskyv22/kandinskyv22_controlnet_robotcat_fp16.npy'''      )

    _A      =       load_image(
        '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
        '''/kandinskyv22/hint_image_cat.png'''      )
    _A      =       torch.from_numpy(np.array(__lowerCAmelCase      )      ).float() / 255.0
    _A      =       hint.permute(2			,						0			,						1      ).unsqueeze(0      )

    _A      =       KandinskyVaaPriorPipeline.from_pretrained(
        '''kandinsky-community/kandinsky-2-2-prior'''			,						torch_dtype=torch.floataa      )
    pipe_prior.to(__lowerCAmelCase      )

    _A      =       KandinskyVaaControlnetPipeline.from_pretrained(
        '''kandinsky-community/kandinsky-2-2-controlnet-depth'''			,						torch_dtype=torch.floataa      )
    _A      =       pipeline.to(__lowerCAmelCase      )
    pipeline.set_progress_bar_config(disable=__lowerCAmelCase      )

    _A      =       '''A robot, 4k photo'''

    _A      =       torch.Generator(device='''cuda'''      ).manual_seed(0      )
    _A  ,    _A      =       pipe_prior(
        __lowerCAmelCase			,						generator=__lowerCAmelCase			,						num_inference_steps=5			,						negative_prompt=''''''			,						).to_tuple()

    _A      =       torch.Generator(device='''cuda'''      ).manual_seed(0      )
    _A      =       pipeline(
        image_embeds=__lowerCAmelCase			,						negative_image_embeds=__lowerCAmelCase			,						hint=__lowerCAmelCase			,						generator=__lowerCAmelCase			,						num_inference_steps=1_00			,						output_type='''np'''			,						)

    _A      =       output.images[0]

    assert image.shape == (5_12, 5_12, 3)

    assert_mean_pixel_difference(__lowerCAmelCase			,						__lowerCAmelCase      )











import logging

import torch
from torch import nn
from torch.nn import CrossEntropyLoss, MSELoss

from transformers.file_utils import add_start_docstrings, add_start_docstrings_to_model_forward
from transformers.models.bert.modeling_bert import (
    BERT_INPUTS_DOCSTRING,
    BERT_START_DOCSTRING,
    BertEncoder,
    BertModel,
    BertPreTrainedModel,
)


UpperCamelCase    							=      logging.getLogger(__name__)







class 		_A						( UpperCAmelCase_ ):

  def    a					(    self						:	Tuple   ,				lowerCamelCase__						:	Optional[int]   ,				lowerCamelCase__						:	Union[str, Any]   ,				lowerCamelCase__						:	str=None   ,				lowerCamelCase__						:	int=None			):
     """simple docstring"""

     __UpperCamelCase	:			Optional[Any]					=					self.layer[current_layer](lowerCamelCase__   ,				lowerCamelCase__   ,				head_mask[current_layer]			)

     __UpperCamelCase	:			Optional[Any]					=					layer_outputs[0]

     return hidden_states







@add_start_docstrings(
    '''The bare Bert Model transformer with PABEE outputting raw hidden-states without any specific head on top.'''	,				UpperCAmelCase_	,				)
class 		_A						( UpperCAmelCase_ ):

  def __init__(    self						:	Tuple   ,				lowerCamelCase__						:	Tuple			):
     """simple docstring"""

     super().__init__(lowerCamelCase__			)

     __UpperCamelCase	:			List[str]					=					BertEncoderWithPabee(lowerCamelCase__			)

     self.init_weights()
     __UpperCamelCase	:			Optional[int]					=					0
     __UpperCamelCase	:			Tuple					=					0
     __UpperCamelCase	:			Any					=					0

     __UpperCamelCase	:			Optional[Any]					=					0

  def    a					(    self						:	Union[str, Any]   ,				lowerCamelCase__						:	Any			):
     """simple docstring"""

     __UpperCamelCase	:			List[Any]					=					threshold

  def    a					(    self						:	Optional[int]   ,				lowerCamelCase__						:	Union[str, Any]			):
     """simple docstring"""

     __UpperCamelCase	:			int					=					patience

  def    a					(    self						:	Optional[int]			):
     """simple docstring"""

     __UpperCamelCase	:			int					=					0
     __UpperCamelCase	:			Tuple					=					0

  def    a					(    self						:	int			):
     """simple docstring"""

     __UpperCamelCase	:			int					=					self.inference_layers_num / self.inference_instances_num
     __UpperCamelCase	:			str					=					(
         f'*** Patience = {self.patience} Avg. Inference Layers = {avg_inf_layers:.2f} Speed Up ='
         f' {1 - avg_inf_layers / self.config.num_hidden_layers:.2f} ***'
     )
     print(lowerCamelCase__			)

  @add_start_docstrings_to_model_forward(lowerCamelCase__			)
  def    a					(    self						:	Any   ,				lowerCamelCase__						:	List[Any]=None   ,				lowerCamelCase__						:	List[str]=None   ,				lowerCamelCase__						:	List[Any]=None   ,				lowerCamelCase__						:	Optional[Any]=None   ,				lowerCamelCase__						:	List[str]=None   ,				lowerCamelCase__						:	str=None   ,				lowerCamelCase__						:	List[Any]=None   ,				lowerCamelCase__						:	Any=None   ,				lowerCamelCase__						:	List[str]=None   ,				lowerCamelCase__						:	List[str]=None   ,				lowerCamelCase__						:	str=False   ,				):
     """simple docstring"""

     if input_ids is not None and inputs_embeds is not None:
        raise ValueError("""You cannot specify both input_ids and inputs_embeds at the same time"""			)
     elif input_ids is not None:
        __UpperCamelCase	:			Optional[Any]					=					input_ids.size()
     elif inputs_embeds is not None:
        __UpperCamelCase	:			Tuple					=					inputs_embeds.size()[:-1]
     else:
        raise ValueError("""You have to specify either input_ids or inputs_embeds"""			)

     __UpperCamelCase	:			Tuple					=					input_ids.device if input_ids is not None else inputs_embeds.device

     if attention_mask is None:
        __UpperCamelCase	:			Optional[Any]					=					torch.ones(lowerCamelCase__   ,				device=lowerCamelCase__			)
     if token_type_ids is None:
        __UpperCamelCase	:			str					=					torch.zeros(lowerCamelCase__   ,				dtype=torch.long   ,				device=lowerCamelCase__			)

     # We can provide a self-attention mask of dimensions [batch_size, from_seq_length, to_seq_length]
     # ourselves in which case we just need to make it broadcastable to all heads.
     __UpperCamelCase	:			torch.Tensor					=					self.get_extended_attention_mask(lowerCamelCase__   ,				lowerCamelCase__   ,				lowerCamelCase__			)

     # If a 2D ou 3D attention mask is provided for the cross-attention
     # we need to make broadcastable to [batch_size, num_heads, seq_length, seq_length]
     if self.config.is_decoder and encoder_hidden_states is not None:
        __UpperCamelCase	,     __UpperCamelCase	,     __UpperCamelCase	:			Optional[Any]					=					encoder_hidden_states.size()
        __UpperCamelCase	:			Union[str, Any]					=					(encoder_batch_size, encoder_sequence_length)
        if encoder_attention_mask is None:
           __UpperCamelCase	:			Union[str, Any]					=					torch.ones(lowerCamelCase__   ,				device=lowerCamelCase__			)
        __UpperCamelCase	:			Optional[int]					=					self.invert_attention_mask(lowerCamelCase__			)
     else:
        __UpperCamelCase	:			str					=					None

     # Prepare head mask if needed
     # 1.0 in head_mask indicate we keep the head
     # attention_probs has shape bsz x n_heads x N x N
     # input head_mask has shape [num_heads] or [num_hidden_layers x num_heads]
     # and head_mask is converted to shape [num_hidden_layers x batch x num_heads x seq_length x seq_length]
     __UpperCamelCase	:			int					=					self.get_head_mask(lowerCamelCase__   ,				self.config.num_hidden_layers			)

     __UpperCamelCase	:			Any					=					self.embeddings(
         input_ids=lowerCamelCase__   ,				position_ids=lowerCamelCase__   ,				token_type_ids=lowerCamelCase__   ,				inputs_embeds=lowerCamelCase__			)
     __UpperCamelCase	:			Optional[int]					=					embedding_output

     if self.training:
        __UpperCamelCase	:			List[Any]					=					[]
        for i in range(self.config.num_hidden_layers			):
           __UpperCamelCase	:			List[Any]					=					self.encoder.adaptive_forward(
               lowerCamelCase__   ,				current_layer=lowerCamelCase__   ,				attention_mask=lowerCamelCase__   ,				head_mask=lowerCamelCase__			)

           __UpperCamelCase	:			Tuple					=					self.pooler(lowerCamelCase__			)
           __UpperCamelCase	:			str					=					output_layers[i](output_dropout(lowerCamelCase__			)			)
           res.append(lowerCamelCase__			)
     elif self.patience == 0:  # Use all layers for inference
        __UpperCamelCase	:			Union[str, Any]					=					self.encoder(
            lowerCamelCase__   ,				attention_mask=lowerCamelCase__   ,				head_mask=lowerCamelCase__   ,				encoder_hidden_states=lowerCamelCase__   ,				encoder_attention_mask=lowerCamelCase__   ,				)
        __UpperCamelCase	:			Optional[int]					=					self.pooler(encoder_outputs[0]			)
        __UpperCamelCase	:			Union[str, Any]					=					[output_layers[self.config.num_hidden_layers - 1](lowerCamelCase__			)]
     else:
        __UpperCamelCase	:			Union[str, Any]					=					0
        __UpperCamelCase	:			Optional[Any]					=					None
        __UpperCamelCase	:			List[Any]					=					0
        for i in range(self.config.num_hidden_layers			):
           calculated_layer_num += 1
           __UpperCamelCase	:			Union[str, Any]					=					self.encoder.adaptive_forward(
               lowerCamelCase__   ,				current_layer=lowerCamelCase__   ,				attention_mask=lowerCamelCase__   ,				head_mask=lowerCamelCase__			)

           __UpperCamelCase	:			str					=					self.pooler(lowerCamelCase__			)
           __UpperCamelCase	:			Any					=					output_layers[i](lowerCamelCase__			)
           if regression:
              __UpperCamelCase	:			Optional[int]					=					logits.detach()
              if patient_result is not None:
                 __UpperCamelCase	:			Dict					=					patient_result.detach()
              if (patient_result is not None) and torch.abs(patient_result - labels			) < self.regression_threshold:
                 patient_counter += 1
              else:
                 __UpperCamelCase	:			List[str]					=					0
           else:
              __UpperCamelCase	:			Optional[int]					=					logits.detach().argmax(dim=1			)
              if patient_result is not None:
                 __UpperCamelCase	:			Union[str, Any]					=					patient_result.detach().argmax(dim=1			)
              if (patient_result is not None) and torch.all(labels.eq(lowerCamelCase__			)			):
                 patient_counter += 1
              else:
                 __UpperCamelCase	:			str					=					0

           __UpperCamelCase	:			Union[str, Any]					=					logits
           if patient_counter == self.patience:
              break
        __UpperCamelCase	:			List[str]					=					[patient_result]
        self.inference_layers_num += calculated_layer_num
        self.inference_instances_num += 1

     return res







@add_start_docstrings(
    '''Bert Model transformer with PABEE and a sequence classification/regression head on top (a linear layer on top of
    the pooled output) e.g. for GLUE tasks. '''	,				UpperCAmelCase_	,				)
class 		_A						( UpperCAmelCase_ ):

  def __init__(    self						:	Any   ,				lowerCamelCase__						:	Union[str, Any]			):
     """simple docstring"""

     super().__init__(lowerCamelCase__			)
     __UpperCamelCase	:			Dict					=					config.num_labels

     __UpperCamelCase	:			List[Any]					=					BertModelWithPabee(lowerCamelCase__			)
     __UpperCamelCase	:			Optional[int]					=					nn.Dropout(config.hidden_dropout_prob			)
     __UpperCamelCase	:			List[Any]					=					nn.ModuleList(
         [nn.Linear(config.hidden_size   ,				self.config.num_labels			) for _ in range(config.num_hidden_layers			)]			)

     self.init_weights()

  @add_start_docstrings_to_model_forward(lowerCamelCase__			)
  def    a					(    self						:	List[Any]   ,				lowerCamelCase__						:	Union[str, Any]=None   ,				lowerCamelCase__						:	Dict=None   ,				lowerCamelCase__						:	Union[str, Any]=None   ,				lowerCamelCase__						:	List[Any]=None   ,				lowerCamelCase__						:	Optional[Any]=None   ,				lowerCamelCase__						:	Optional[int]=None   ,				lowerCamelCase__						:	Dict=None   ,				):
     """simple docstring"""

     __UpperCamelCase	:			str					=					self.bert(
         input_ids=lowerCamelCase__   ,				attention_mask=lowerCamelCase__   ,				token_type_ids=lowerCamelCase__   ,				position_ids=lowerCamelCase__   ,				head_mask=lowerCamelCase__   ,				inputs_embeds=lowerCamelCase__   ,				output_dropout=self.dropout   ,				output_layers=self.classifiers   ,				regression=self.num_labels == 1   ,				)

     __UpperCamelCase	:			List[str]					=					(logits[-1],)

     if labels is not None:
        __UpperCamelCase	:			Tuple					=					None
        __UpperCamelCase	:			Union[str, Any]					=					0
        for ix, logits_item in enumerate(lowerCamelCase__			):
           if self.num_labels == 1:
              #  We are doing regression
              __UpperCamelCase	:			Union[str, Any]					=					MSELoss()
              __UpperCamelCase	:			Optional[Any]					=					loss_fct(logits_item.view(-1			)   ,				labels.view(-1			)			)
           else:
              __UpperCamelCase	:			Dict					=					CrossEntropyLoss()
              __UpperCamelCase	:			Dict					=					loss_fct(logits_item.view(-1   ,				self.num_labels			)   ,				labels.view(-1			)			)
           if total_loss is None:
              __UpperCamelCase	:			Dict					=					loss
           else:
              total_loss += loss * (ix + 1)
           total_weights += ix + 1
        __UpperCamelCase	:			Tuple					=					(total_loss / total_weights,) + outputs

     return outputs



