






"""simple docstring"""





import types
from typing import List, Optional, Tuple, Union

import torch
from transformers import CLIPTextModelWithProjection, CLIPTokenizer
from transformers.models.clip.modeling_clip import CLIPTextModelOutput

from diffusers.models import PriorTransformer
from diffusers.pipelines import DiffusionPipeline, StableDiffusionImageVariationPipeline
from diffusers.schedulers import UnCLIPScheduler
from diffusers.utils import logging, randn_tensor


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






def     A(  self						,       snake_case_						,       snake_case_						,       snake_case_						,       snake_case_		):
       """simple docstring"""





       lowercase__:    str        =      image.to(device=snake_case_		)
       lowercase__:    Union[str, Any]        =      image  # take image as image_embeddings
       lowercase__:    Optional[int]        =      image_embeddings.unsqueeze(1		)

       # duplicate image embeddings for each generation per prompt, using mps friendly method
       lowercase__  ,     lowercase__  ,     lowercase__:    Optional[Any]        =      image_embeddings.shape
       lowercase__:    Tuple        =      image_embeddings.repeat(1						,       snake_case_						,       1		)
       lowercase__:    List[Any]        =      image_embeddings.view(bs_embed * num_images_per_prompt						,       snake_case_						,       -1		)

       if do_classifier_free_guidance:
              lowercase__:    Dict        =      torch.zeros_like(snake_case_		)

              # 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
              lowercase__:    Any        =      torch.cat([uncond_embeddings, image_embeddings]		)

       return image_embeddings







class 		_a       (						lowercase_   ):





   '''simple docstring'''






   def __init__(   self			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_ = None			,				)	-> List[str]:




          '''simple docstring'''




          super().__init__()

          lowercase__:    List[Any]        =      {"image_encoder": None} if decoder_pipe_kwargs is None else decoder_pipe_kwargs

          lowercase__:    Tuple        =      decoder_pipe_kwargs.get("torch_dtype"			,				UpperCAmelCase_) or prior.dtype

          lowercase__:    Optional[Any]        =      StableDiffusionImageVariationPipeline.from_pretrained(
              "lambdalabs/sd-image-variations-diffusers"			,				**UpperCAmelCase_)

          # replace `_encode_image` method
          lowercase__:    int        =      types.MethodType(_encode_image			,				self.decoder_pipe)

          self.register_modules(
              prior=UpperCAmelCase_			,				tokenizer=UpperCAmelCase_			,				text_encoder=UpperCAmelCase_			,				prior_scheduler=UpperCAmelCase_			,				)





   def __lowercase    (   self			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_ = None			,				UpperCAmelCase_ = None			,				)	-> Any:




          '''simple docstring'''




          if text_model_output is None:
                 lowercase__:    List[Any]        =      len(UpperCAmelCase_) if isinstance(UpperCAmelCase_			,				UpperCAmelCase_) else 1
                 # get prompt text embeddings
                 lowercase__:    List[Any]        =      self.tokenizer(
                     UpperCAmelCase_			,				padding="max_length"			,				max_length=self.tokenizer.model_max_length			,				return_tensors="pt"			,				)
                 lowercase__:    Optional[int]        =      text_inputs.input_ids
                 lowercase__:    Optional[int]        =      text_inputs.attention_mask.bool().to(UpperCAmelCase_)

                 if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
                        lowercase__:    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}""")
                        lowercase__:    List[Any]        =      text_input_ids[:, : self.tokenizer.model_max_length]

                 lowercase__:    List[Any]        =      self.text_encoder(text_input_ids.to(UpperCAmelCase_))

                 lowercase__:    Any        =      text_encoder_output.text_embeds
                 lowercase__:    str        =      text_encoder_output.last_hidden_state

          else:
                 lowercase__:    Optional[Any]        =      text_model_output[0].shape[0]
                 lowercase__  ,     lowercase__:    Union[str, Any]        =      text_model_output[0], text_model_output[1]
                 lowercase__:    Union[str, Any]        =      text_attention_mask

          lowercase__:    int        =      text_embeddings.repeat_interleave(UpperCAmelCase_			,				dim=0)
          lowercase__:    List[Any]        =      text_encoder_hidden_states.repeat_interleave(UpperCAmelCase_			,				dim=0)
          lowercase__:    Any        =      text_mask.repeat_interleave(UpperCAmelCase_			,				dim=0)

          if do_classifier_free_guidance:
                 lowercase__:    Optional[Any]        =      [""] * batch_size

                 lowercase__:    Any        =      self.tokenizer(
                     UpperCAmelCase_			,				padding="max_length"			,				max_length=self.tokenizer.model_max_length			,				truncation=UpperCAmelCase_			,				return_tensors="pt"			,				)
                 lowercase__:    Union[str, Any]        =      uncond_input.attention_mask.bool().to(UpperCAmelCase_)
                 lowercase__:    str        =      self.text_encoder(uncond_input.input_ids.to(UpperCAmelCase_))

                 lowercase__:    int        =      uncond_embeddings_text_encoder_output.text_embeds
                 lowercase__:    Tuple        =      uncond_embeddings_text_encoder_output.last_hidden_state

                 # duplicate unconditional embeddings for each generation per prompt, using mps friendly method

                 lowercase__:    str        =      uncond_embeddings.shape[1]
                 lowercase__:    int        =      uncond_embeddings.repeat(1			,				UpperCAmelCase_)
                 lowercase__:    str        =      uncond_embeddings.view(batch_size * num_images_per_prompt			,				UpperCAmelCase_)

                 lowercase__:    List[Any]        =      uncond_text_encoder_hidden_states.shape[1]
                 lowercase__:    Optional[Any]        =      uncond_text_encoder_hidden_states.repeat(1			,				UpperCAmelCase_			,				1)
                 lowercase__:    List[str]        =      uncond_text_encoder_hidden_states.view(
                     batch_size * num_images_per_prompt			,				UpperCAmelCase_			,				-1)
                 lowercase__:    List[str]        =      uncond_text_mask.repeat_interleave(UpperCAmelCase_			,				dim=0)

                 # done duplicates

                 # 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
                 lowercase__:    Any        =      torch.cat([uncond_embeddings, text_embeddings])
                 lowercase__:    Dict        =      torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states])

                 lowercase__:    Optional[int]        =      torch.cat([uncond_text_mask, text_mask])

          return text_embeddings, text_encoder_hidden_states, text_mask





   @property
   def __lowercase    (   self)	-> Any:




          '''simple docstring'''




          if self.device != torch.device("meta") or not hasattr(self.prior			,				"_hf_hook"):
                 return self.device
          for module in self.prior.modules():
                 if (
                     hasattr(UpperCAmelCase_			,				"_hf_hook")
                     and hasattr(module._hf_hook			,				"execution_device")
                     and module._hf_hook.execution_device is not None
                 ):
                        return torch.device(module._hf_hook.execution_device)
          return self.device





   def __lowercase    (   self			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_)	-> List[Any]:




          '''simple docstring'''




          if latents is None:
                 lowercase__:    Union[str, Any]        =      randn_tensor(UpperCAmelCase_			,				generator=UpperCAmelCase_			,				device=UpperCAmelCase_			,				dtype=UpperCAmelCase_)
          else:
                 if latents.shape != shape:
                        raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {shape}""")
                 lowercase__:    Optional[Any]        =      latents.to(UpperCAmelCase_)

          lowercase__:    List[str]        =      latents * scheduler.init_noise_sigma
          return latents





   def __lowercase    (   self			,				UpperCAmelCase_ = None)	-> Any:




          '''simple docstring'''




          self.decoder_pipe.to(UpperCAmelCase_)
          super().to(UpperCAmelCase_)





   @torch.no_grad()
   def __call__(   self			,				UpperCAmelCase_ = None			,				UpperCAmelCase_ = None			,				UpperCAmelCase_ = None			,				UpperCAmelCase_ = 1			,				UpperCAmelCase_ = 25			,				UpperCAmelCase_ = None			,				UpperCAmelCase_ = None			,				UpperCAmelCase_ = None			,				UpperCAmelCase_ = None			,				UpperCAmelCase_ = 4.0			,				UpperCAmelCase_ = 8.0			,				UpperCAmelCase_ = 50			,				UpperCAmelCase_ = 1			,				UpperCAmelCase_ = 0.0			,				UpperCAmelCase_ = "pil"			,				UpperCAmelCase_ = True			,				)	-> Dict:




          '''simple docstring'''




          if prompt is not None:
                 if isinstance(UpperCAmelCase_			,				UpperCAmelCase_):
                        lowercase__:    Dict        =      1
                 elif isinstance(UpperCAmelCase_			,				UpperCAmelCase_):
                        lowercase__:    Union[str, Any]        =      len(UpperCAmelCase_)
                 else:
                        raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(UpperCAmelCase_)}""")
          else:
                 lowercase__:    Tuple        =      text_model_output[0].shape[0]

          lowercase__:    Dict        =      self._execution_device

          lowercase__:    int        =      batch_size * num_images_per_prompt

          lowercase__:    Union[str, Any]        =      prior_guidance_scale > 1.0 or decoder_guidance_scale > 1.0

          lowercase__  ,     lowercase__  ,     lowercase__:    Union[str, Any]        =      self._encode_prompt(
              UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_)

          # prior

          self.prior_scheduler.set_timesteps(UpperCAmelCase_			,				device=UpperCAmelCase_)
          lowercase__:    Dict        =      self.prior_scheduler.timesteps

          lowercase__:    List[str]        =      self.prior.config.embedding_dim

          lowercase__:    Optional[int]        =      self.prepare_latents(
              (batch_size, embedding_dim)			,				text_embeddings.dtype			,				UpperCAmelCase_			,				UpperCAmelCase_			,				UpperCAmelCase_			,				self.prior_scheduler			,				)

          for i, t in enumerate(self.progress_bar(UpperCAmelCase_)):
                 # expand the latents if we are doing classifier free guidance
                 lowercase__:    Optional[int]        =      torch.cat([prior_latents] * 2) if do_classifier_free_guidance else prior_latents

                 lowercase__:    Any        =      self.prior(
                     UpperCAmelCase_			,				timestep=UpperCAmelCase_			,				proj_embedding=UpperCAmelCase_			,				encoder_hidden_states=UpperCAmelCase_			,				attention_mask=UpperCAmelCase_			,				).predicted_image_embedding

                 if do_classifier_free_guidance:
                        lowercase__  ,     lowercase__:    int        =      predicted_image_embedding.chunk(2)
                        lowercase__:    Optional[Any]        =      predicted_image_embedding_uncond + prior_guidance_scale * (
                            predicted_image_embedding_text - predicted_image_embedding_uncond
                        )

                 if i + 1 == prior_timesteps_tensor.shape[0]:
                        lowercase__:    Dict        =      None
                 else:
                        lowercase__:    List[Any]        =      prior_timesteps_tensor[i + 1]

                 lowercase__:    Any        =      self.prior_scheduler.step(
                     UpperCAmelCase_			,				timestep=UpperCAmelCase_			,				sample=UpperCAmelCase_			,				generator=UpperCAmelCase_			,				prev_timestep=UpperCAmelCase_			,				).prev_sample

          lowercase__:    List[Any]        =      self.prior.post_process_latents(UpperCAmelCase_)

          lowercase__:    List[Any]        =      prior_latents

          lowercase__:    List[str]        =      self.decoder_pipe(
              image=UpperCAmelCase_			,				height=UpperCAmelCase_			,				width=UpperCAmelCase_			,				num_inference_steps=UpperCAmelCase_			,				guidance_scale=UpperCAmelCase_			,				generator=UpperCAmelCase_			,				output_type=UpperCAmelCase_			,				return_dict=UpperCAmelCase_			,				num_images_per_prompt=UpperCAmelCase_			,				eta=UpperCAmelCase_			,				)
          return output














"""simple docstring"""





from __future__ import annotations






def     A(  snake_case_						,       snake_case_		):
       """simple docstring"""





       lowercase__  ,     lowercase__:    Optional[int]        =      0, len(snake_case_		) - 1
       while start_ind <= end_ind:
              if array[start_ind] == search_item:
                     return start_ind
              elif array[end_ind] == search_item:
                     return end_ind
              else:
                     start_ind += 1
                     end_ind -= 1
    # returns -1 if search_item is not found in array
       return -1


if __name__ == "__main__":
  print(double_linear_search(list(range(1_00)), 40))






