

'''simple docstring'''




import html

from ...feature_extraction_utils import BatchFeature, FeatureExtractionMixin
from ...utils import is_bsa_available, logging, requires_backends


if is_bsa_available():
      import bsa
      from bsa import BeautifulSoup


_lowerCAmelCase            =		logging.get_logger(__name__)






class     _SCREAMING_SNAKE_CASE       (       __a      ):



 def __init__(       self						:       List[str]					, **a__						:       str     ):
      requires_backends(self					, ['''bs4''']     )
      super().__init__(**a__     )



 def 					snake_case__    (       self						:       Tuple					, a__						:       Dict     ):
      __magic_name__		  =							[]
      __magic_name__		  =							[]
      __magic_name__		  =							element if element.name else element.parent
      for parent in child.parents:  # type: bs4.element.Tag
           __magic_name__		  =							parent.find_all(child.name					, recursive=a__     )
           xpath_tags.append(child.name     )
           xpath_subscripts.append(
               0 if 1 == len(a__     ) else next(i for i, s in enumerate(a__					, 1     ) if s is child     )     )
           __magic_name__		  =							parent
      xpath_tags.reverse()
      xpath_subscripts.reverse()
      return xpath_tags, xpath_subscripts



 def 					snake_case__    (       self						:       Union[str, Any]					, a__						:       str     ):
      __magic_name__		  =							BeautifulSoup(a__					, '''html.parser'''     )

      __magic_name__		  =							[]
      __magic_name__		  =							[]
      __magic_name__		  =							[]

      for element in html_code.descendants:
           if type(a__     ) == bsa.element.NavigableString:
                if type(element.parent     ) != bsa.element.Tag:
                     continue

                __magic_name__		  =							html.unescape(a__     ).strip()
                if not text_in_this_tag:
                     continue

                all_doc_strings.append(a__     )

                __magic_name__		  =							self.xpath_soup(a__     )
                stringaxtag_seq.append(a__     )
                stringaxsubs_seq.append(a__     )

      if len(a__     ) != len(a__     ):
           raise ValueError('''Number of doc strings and xtags does not correspond'''     )
      if len(a__     ) != len(a__     ):
           raise ValueError('''Number of doc strings and xsubs does not correspond'''     )

      return all_doc_strings, stringaxtag_seq, stringaxsubs_seq



 def 					snake_case__    (       self						:       str					, a__						:       int					, a__						:       Any     ):
      __magic_name__		  =							''
      for tagname, subs in zip(a__					, a__     ):
           xpath += F'''/{tagname}'''
           if subs != 0:
                xpath += F'''[{subs}]'''
      return xpath



 def __call__(       self						:       List[Any]					, a__						:       Dict     ):
      __magic_name__		  =							False

      # Check that strings has a valid type
      if isinstance(a__					, a__     ):
           __magic_name__		  =							True
      elif isinstance(a__					, (list, tuple)     ):
           if len(a__     ) == 0 or isinstance(html_strings[0]					, a__     ):
                __magic_name__		  =							True

      if not valid_strings:
           raise ValueError(
               '''HTML strings must of type `str`, `List[str]` (batch of examples), '''
               F'''but is of type {type(a__     )}.'''     )

      __magic_name__		  =							bool(isinstance(a__					, (list, tuple)     ) and (isinstance(html_strings[0]					, a__     ))     )

      if not is_batched:
           __magic_name__		  =							[html_strings]

      # Get nodes + xpaths
      __magic_name__		  =							[]
      __magic_name__		  =							[]
      for html_string in html_strings:
           __magic_name__		  =							self.get_three_from_single(a__     )
           nodes.append(a__     )
           __magic_name__		  =							[]
           for node, tag_list, sub_list in zip(a__					, a__					, a__     ):
                __magic_name__		  =							self.construct_xpath(a__					, a__     )
                xpath_strings.append(a__     )
           xpaths.append(a__     )

      # return as Dict
      __magic_name__		  =							{'nodes': nodes, 'xpaths': xpaths}
      __magic_name__		  =							BatchFeature(data=a__					, tensor_type=a__     )

      return encoded_inputs










'''simple docstring'''






import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyImgaImgPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
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    __magic_name__				(						lowerCAmelCase   ,unittest.TestCase							):
       UpperCAmelCase		 =KandinskyImgaImgPipeline
       UpperCAmelCase		 =["prompt", "image_embeds", "negative_image_embeds", "image"]
       UpperCAmelCase		 =[
           "prompt",
           "negative_prompt",
           "image_embeds",
           "negative_image_embeds",
           "image",
       ]
       UpperCAmelCase		 =[
           "generator",
           "height",
           "width",
           "strength",
           "guidance_scale",
           "negative_prompt",
           "num_inference_steps",
           "return_dict",
           "guidance_scale",
           "num_images_per_prompt",
           "output_type",
           "return_dict",
       ]
       UpperCAmelCase		 =False






       @property
       def      lowerCAmelCase						(     self)		->    str:




          '''simple docstring'''
          return 3_2






       @property
       def      lowerCAmelCase						(     self)		->    List[str]:




          '''simple docstring'''
          return 3_2






       @property
       def      lowerCAmelCase						(     self)		->    Union[str, Any]:




          '''simple docstring'''
          return self.time_input_dim






       @property
       def      lowerCAmelCase						(     self)		->    Optional[int]:




          '''simple docstring'''
          return self.time_input_dim * 4






       @property
       def      lowerCAmelCase						(     self)		->    Tuple:




          '''simple docstring'''
          return 1_0_0






       @property
       def      lowerCAmelCase						(     self)		->    Optional[Any]:




          '''simple docstring'''
          _UpperCAmelCase  :  int				       =XLMRobertaTokenizerFast.from_pretrained('YiYiXu/tiny-random-mclip-base')
          return tokenizer






       @property
       def      lowerCAmelCase						(     self)		->    Dict:




          '''simple docstring'''
          torch.manual_seed(0)
          _UpperCAmelCase  :  Optional[Any]				       =MCLIPConfig(
              numDims=self.cross_attention_dim    ,				transformerDimensions=self.text_embedder_hidden_size    ,				hidden_size=self.text_embedder_hidden_size    ,				intermediate_size=3_7    ,				num_attention_heads=4    ,				num_hidden_layers=5    ,				vocab_size=1_0_0_5    ,				)

          _UpperCAmelCase  :  List[Any]				       =MultilingualCLIP(snake_case)
          _UpperCAmelCase  :  List[Any]				       =text_encoder.eval()

          return text_encoder






       @property
       def      lowerCAmelCase						(     self)		->    Optional[Any]:




          '''simple docstring'''
          torch.manual_seed(0)

          _UpperCAmelCase  :  Union[str, Any]				       ={
              'in_channels': 4,
              # Out channels is double in channels because predicts mean and variance
              'out_channels': 8,
              'addition_embed_type': 'text_image',
              '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': 'text_image_proj',
              'cross_attention_dim': self.cross_attention_dim,
              'attention_head_dim': 4,
              'resnet_time_scale_shift': 'scale_shift',
              'class_embed_type': None,
          }

          _UpperCAmelCase  :  Tuple				       =UNetaDConditionModel(**snake_case)
          return model






       @property
       def      lowerCAmelCase						(     self)		->    List[Any]:




          '''simple docstring'''
          return {
              "block_out_channels": [3_2, 6_4],
              "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
              "in_channels": 3,
              "latent_channels": 4,
              "layers_per_block": 1,
              "norm_num_groups": 8,
              "norm_type": "spatial",
              "num_vq_embeddings": 1_2,
              "out_channels": 3,
              "up_block_types": [
                  "AttnUpDecoderBlock2D",
                  "UpDecoderBlock2D",
              ],
              "vq_embed_dim": 4,
          }






       @property
       def      lowerCAmelCase						(     self)		->    int:




          '''simple docstring'''
          torch.manual_seed(0)
          _UpperCAmelCase  :  List[Any]				       =VQModel(**self.dummy_movq_kwargs)
          return model






       def      lowerCAmelCase						(     self)		->    List[Any]:




          '''simple docstring'''
          _UpperCAmelCase  :  Tuple				       =self.dummy_text_encoder
          _UpperCAmelCase  :  Any				       =self.dummy_tokenizer
          _UpperCAmelCase  :  List[str]				       =self.dummy_unet
          _UpperCAmelCase  :  Union[str, Any]				       =self.dummy_movq

          _UpperCAmelCase  :  List[str]				       ={
              'num_train_timesteps': 1_0_0_0,
              'beta_schedule': 'linear',
              'beta_start': 0.0_00_85,
              'beta_end': 0.0_12,
              'clip_sample': False,
              'set_alpha_to_one': False,
              'steps_offset': 0,
              'prediction_type': 'epsilon',
              'thresholding': False,
          }

          _UpperCAmelCase  :  Dict				       =DDIMScheduler(**snake_case)

          _UpperCAmelCase  :  Optional[int]				       ={
              'text_encoder': text_encoder,
              'tokenizer': tokenizer,
              'unet': unet,
              'scheduler': scheduler,
              'movq': movq,
          }

          return components






       def      lowerCAmelCase						(     self    ,				snake_case    ,				snake_case=0)		->    int:




          '''simple docstring'''
          _UpperCAmelCase  :  List[Any]				       =floats_tensor((1, self.cross_attention_dim)    ,				rng=random.Random(snake_case)).to(snake_case)
          _UpperCAmelCase  :  Tuple				       =floats_tensor((1, self.cross_attention_dim)    ,				rng=random.Random(seed + 1)).to(snake_case)
          # create init_image
          _UpperCAmelCase  :  Dict				       =floats_tensor((1, 3, 6_4, 6_4)    ,				rng=random.Random(snake_case)).to(snake_case)
          _UpperCAmelCase  :  Optional[int]				       =image.cpu().permute(0    ,				2    ,				3    ,				1)[0]
          _UpperCAmelCase  :  List[Any]				       =Image.fromarray(np.uinta(snake_case)).convert('RGB').resize((2_5_6, 2_5_6))

          if str(snake_case).startswith('mps'):
             _UpperCAmelCase  :  str				       =torch.manual_seed(snake_case)
          else:
             _UpperCAmelCase  :  Optional[Any]				       =torch.Generator(device=snake_case).manual_seed(snake_case)
          _UpperCAmelCase  :  str				       ={
              'prompt': 'horse',
              'image': init_image,
              'image_embeds': image_embeds,
              'negative_image_embeds': negative_image_embeds,
              'generator': generator,
              'height': 6_4,
              'width': 6_4,
              'num_inference_steps': 1_0,
              'guidance_scale': 7.0,
              'strength': 0.2,
              'output_type': 'np',
          }
          return inputs






       def      lowerCAmelCase						(     self)		->    str:




          '''simple docstring'''
          _UpperCAmelCase  :  List[Any]				       ='cpu'

          _UpperCAmelCase  :  Dict				       =self.get_dummy_components()

          _UpperCAmelCase  :  Optional[Any]				       =self.pipeline_class(**snake_case)
          _UpperCAmelCase  :  List[str]				       =pipe.to(snake_case)

          pipe.set_progress_bar_config(disable=snake_case)

          _UpperCAmelCase  :  List[Any]				       =pipe(**self.get_dummy_inputs(snake_case))
          _UpperCAmelCase  :  List[str]				       =output.images

          _UpperCAmelCase  :  Any				       =pipe(
              **self.get_dummy_inputs(snake_case)    ,				return_dict=snake_case    ,				)[0]

          _UpperCAmelCase  :  Any				       =image[0, -3:, -3:, -1]
          _UpperCAmelCase  :  Tuple				       =image_from_tuple[0, -3:, -3:, -1]

          assert image.shape == (1, 6_4, 6_4, 3)

          _UpperCAmelCase  :  List[Any]				       =np.array(
              [0.61_47_49_43, 0.6_07_35_39, 0.43_30_85_44, 0.5_92_82_69, 0.47_49_35_95, 0.46_75_59_73, 0.4_61_38_38, 0.45_36_87_97, 0.50_11_92_33])
          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    __magic_name__				(						unittest.TestCase							):






       def      lowerCAmelCase						(     self)		->    Optional[int]:




          '''simple docstring'''
          # clean up the VRAM after each test
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()






       def      lowerCAmelCase						(     self)		->    Optional[int]:




          '''simple docstring'''
          _UpperCAmelCase  :  Tuple				       =load_numpy(
              'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
              '/kandinsky/kandinsky_img2img_frog.npy')

          _UpperCAmelCase  :  List[Any]				       =load_image(
              'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main' '/kandinsky/cat.png')
          _UpperCAmelCase  :  Union[str, Any]				       ='A red cartoon frog, 4k'

          _UpperCAmelCase  :  Union[str, Any]				       =KandinskyPriorPipeline.from_pretrained(
              'kandinsky-community/kandinsky-2-1-prior'    ,				torch_dtype=torch.floataa)
          pipe_prior.to(snake_case)

          _UpperCAmelCase  :  Optional[Any]				       =KandinskyImgaImgPipeline.from_pretrained(
              'kandinsky-community/kandinsky-2-1'    ,				torch_dtype=torch.floataa)
          _UpperCAmelCase  :  Tuple				       =pipeline.to(snake_case)

          pipeline.set_progress_bar_config(disable=snake_case)

          _UpperCAmelCase  :  List[str]				       =torch.Generator(device='cpu').manual_seed(0)
          _UpperCAmelCase       ,			_UpperCAmelCase  :  List[Any]				       =pipe_prior(
              snake_case    ,				generator=snake_case    ,				num_inference_steps=5    ,				negative_prompt=''    ,				).to_tuple()

          _UpperCAmelCase  :  str				       =pipeline(
              snake_case    ,				image=snake_case    ,				image_embeds=snake_case    ,				negative_image_embeds=snake_case    ,				generator=snake_case    ,				num_inference_steps=1_0_0    ,				height=7_6_8    ,				width=7_6_8    ,				strength=0.2    ,				output_type='np'    ,				)

          _UpperCAmelCase  :  Any				       =output.images[0]

          assert image.shape == (7_6_8, 7_6_8, 3)

          assert_mean_pixel_difference(snake_case    ,				snake_case)

