'''simple docstring'''






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

import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextModel,
    CLIPTokenizer,
    WhisperForConditionalGeneration,
    WhisperProcessor,
)

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.utils import logging


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







class lowercase_       (		A__			):







   """simple docstring"""

   def __init__(      self      :			Dict, UpperCamelCase__      :			WhisperForConditionalGeneration, UpperCamelCase__      :			WhisperProcessor, UpperCamelCase__      :			AutoencoderKL, UpperCamelCase__      :			CLIPTextModel, UpperCamelCase__      :			CLIPTokenizer, UpperCamelCase__      :			UNetaDConditionModel, UpperCamelCase__      :			Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler], UpperCamelCase__      :			StableDiffusionSafetyChecker, UpperCamelCase__      :			CLIPImageProcessor, )  ->			Dict:
        super().__init__()

        if safety_checker is None:
             logger.warning(
                 f'You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure'
                 ' that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered'
                 ' results in services or applications open to the public. Both the diffusers team and Hugging Face'
                 ' strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling'
                 ' it only for use-cases that involve analyzing network behavior or auditing its results. For more'
                 ' information, please have a look at https://github.com/huggingface/diffusers/pull/254 .'     )

        self.register_modules(
            speech_model=_lowerCamelCase, speech_processor=_lowerCamelCase, vae=_lowerCamelCase, text_encoder=_lowerCamelCase, tokenizer=_lowerCamelCase, unet=_lowerCamelCase, scheduler=_lowerCamelCase, feature_extractor=_lowerCamelCase, )
   def 	__UpperCAmelCase   (      self      :			Union[str, Any], UpperCamelCase__      :			Optional[Union[str, int]] = "auto"     )  ->			int:



        if slice_size == "auto":
             _A    		=  self.unet.config.attention_head_dim // 2
        self.unet.set_attention_slice(_lowerCamelCase     )
   def 	__UpperCAmelCase   (      self      :			str     )  ->			List[str]:
        self.enable_attention_slicing(_lowerCamelCase     )

   @torch.no_grad()
   def __call__(      self      :			Any, UpperCamelCase__      :			Optional[int], UpperCamelCase__      :			Dict=1_60_00, UpperCamelCase__      :			int = 5_12, UpperCamelCase__      :			int = 5_12, UpperCamelCase__      :			int = 50, 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__      :			int, )  ->			Dict:
        _A    		=  self.speech_processor.feature_extractor(
            _lowerCamelCase, return_tensors='pt', sampling_rate=_lowerCamelCase     ).input_features.to(self.device     )
        _A    		=  self.speech_model.generate(_lowerCamelCase, max_length=48_00_00     )

        _A    		=  self.speech_processor.tokenizer.batch_decode(_lowerCamelCase, skip_special_tokens=_lowerCamelCase, normalize=_lowerCamelCase     )[
            0
        ]

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

        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(_lowerCamelCase, _lowerCamelCase     ) or callback_steps <= 0)
        ):
             raise ValueError(
                 f'`callback_steps` has to be a positive integer but is {callback_steps} of type'
                 f' {type(_lowerCamelCase     )}.'     )

        # get prompt text embeddings
        _A    		=  self.tokenizer(
            _lowerCamelCase, padding='max_length', max_length=self.tokenizer.model_max_length, return_tensors='pt', )
        _A    		=  text_inputs.input_ids

        if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
             _A    		=  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}'     )
             _A    		=  text_input_ids[:, : self.tokenizer.model_max_length]
        _A    		=  self.text_encoder(text_input_ids.to(self.device     )     )[0]

        # duplicate text embeddings for each generation per prompt, using mps friendly method
        _A    		=  text_embeddings.shape
        _A    		=  text_embeddings.repeat(1, _lowerCamelCase, 1     )
        _A    		=  text_embeddings.view(bs_embed * num_images_per_prompt, _lowerCamelCase, -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.
        _A    		=  guidance_scale > 1.0
        # get unconditional embeddings for classifier free guidance
        if do_classifier_free_guidance:
             _A    		=  42
             if negative_prompt is None:
                  _A    		=  [''] * batch_size
             elif type(_lowerCamelCase     ) is not type(_lowerCamelCase     ):
                  raise TypeError(
                      f'`negative_prompt` should be the same type to `prompt`, but got {type(_lowerCamelCase     )} !='
                      f' {type(_lowerCamelCase     )}.'     )
             elif isinstance(_lowerCamelCase, _lowerCamelCase     ):
                  _A    		=  [negative_prompt]
             elif batch_size != len(_lowerCamelCase     ):
                  raise ValueError(
                      f'`negative_prompt`: {negative_prompt} has batch size {len(_lowerCamelCase     )}, but `prompt`:'
                      f' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'
                      ' the batch size of `prompt`.'     )
             else:
                  _A    		=  negative_prompt

             _A    		=  text_input_ids.shape[-1]
             _A    		=  self.tokenizer(
                 _lowerCamelCase, padding='max_length', max_length=_lowerCamelCase, truncation=_lowerCamelCase, return_tensors='pt', )
             _A    		=  self.text_encoder(uncond_input.input_ids.to(self.device     )     )[0]

             # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
             _A    		=  uncond_embeddings.shape[1]
             _A    		=  uncond_embeddings.repeat(1, _lowerCamelCase, 1     )
             _A    		=  uncond_embeddings.view(batch_size * num_images_per_prompt, _lowerCamelCase, -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
             _A    		=  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`.
        _A    		=  (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
        _A    		=  text_embeddings.dtype
        if latents is None:
             if self.device.type == "mps":
                  # randn does not exist on mps
                  _A    		=  torch.randn(_lowerCamelCase, generator=_lowerCamelCase, device='cpu', dtype=_lowerCamelCase     ).to(
                      self.device     )
             else:
                  _A    		=  torch.randn(_lowerCamelCase, generator=_lowerCamelCase, device=self.device, dtype=_lowerCamelCase     )
        else:
             if latents.shape != latents_shape:
                  raise ValueError(f'Unexpected latents shape, got {latents.shape}, expected {latents_shape}'     )
             _A    		=  latents.to(self.device     )

        # set timesteps
        self.scheduler.set_timesteps(_lowerCamelCase     )

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

        # scale the initial noise by the standard deviation required by the scheduler
        _A    		=  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]
        _A    		=  'eta' in set(inspect.signature(self.scheduler.step     ).parameters.keys()     )
        _A    		=  {}
        if accepts_eta:
             _A    		=  eta

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

             # predict the noise residual
             _A    		=  self.unet(_lowerCamelCase, _lowerCamelCase, encoder_hidden_states=_lowerCamelCase     ).sample

             # perform guidance
             if do_classifier_free_guidance:
                  _A    		=  noise_pred.chunk(2     )
                  _A    		=  noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

             # compute the previous noisy sample x_t -> x_t-1
             _A    		=  self.scheduler.step(_lowerCamelCase, _lowerCamelCase, _lowerCamelCase, **_lowerCamelCase     ).prev_sample

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

        _A    		=  1 / 0.18_215 * latents
        _A    		=  self.vae.decode(_lowerCamelCase     ).sample

        _A    		=  (image / 2 + 0.5).clamp(0, 1     )

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

        if output_type == "pil":
             _A    		=  self.numpy_to_pil(_lowerCamelCase     )

        if not return_dict:
             return image

        return StableDiffusionPipelineOutput(images=_lowerCamelCase, nsfw_content_detected=_lowerCamelCase     )





"""simple docstring"""






from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ...utils.backbone_utils import BackboneConfigMixin, get_aligned_output_features_output_indices


__SCREAMING_SNAKE_CASE							=     logging.get_logger(__name__)

__SCREAMING_SNAKE_CASE							=     {
    'microsoft/focalnet-tiny': 'https://huggingface.co/microsoft/focalnet-tiny/resolve/main/config.json',
}





class 				a__					(						A__   ,							A__       ):
     UpperCAmelCase__         =      '''focalnet'''



     def __init__(							self :Any      ,  _lowerCamelCase :List[Any]=224      ,  _lowerCamelCase :str=4      ,  _lowerCamelCase :List[str]=3      ,  _lowerCamelCase :Optional[Any]=96      ,  _lowerCamelCase :Dict=False      ,  _lowerCamelCase :Optional[Any]=[192, 384, 768, 768]      ,  _lowerCamelCase :Tuple=[2, 2, 6, 2]      ,  _lowerCamelCase :List[Any]=[2, 2, 2, 2]      ,  _lowerCamelCase :Optional[int]=[3, 3, 3, 3]      ,  _lowerCamelCase :Optional[Any]="gelu"      ,  _lowerCamelCase :Tuple=4.0      ,  _lowerCamelCase :Optional[Any]=0.0      ,  _lowerCamelCase :int=0.1      ,  _lowerCamelCase :List[str]=False      ,  _lowerCamelCase :str=1E-4      ,  _lowerCamelCase :Optional[int]=False      ,  _lowerCamelCase :List[str]=False      ,  _lowerCamelCase :str=False      ,  _lowerCamelCase :Optional[int]=0.02      ,  _lowerCamelCase :int=1E-5      ,  _lowerCamelCase :Tuple=32      ,  _lowerCamelCase :List[str]=None      ,  _lowerCamelCase :Dict=None      ,  **_lowerCamelCase :List[Any]      ,  ):







      '''simple docstring'''



      super().__init__(**_lowerCamelCase   )

      UpperCamelCase_     :					Any   	=image_size
      UpperCamelCase_     :					int   	=patch_size
      UpperCamelCase_     :					int   	=num_channels
      UpperCamelCase_     :					Union[str, Any]   	=embed_dim
      UpperCamelCase_     :					int   	=use_conv_embed
      UpperCamelCase_     :					Optional[int]   	=hidden_sizes
      UpperCamelCase_     :					str   	=depths
      UpperCamelCase_     :					Any   	=focal_levels
      UpperCamelCase_     :					List[str]   	=focal_windows
      UpperCamelCase_     :					str   	=hidden_act
      UpperCamelCase_     :					Dict   	=mlp_ratio
      UpperCamelCase_     :					List[str]   	=hidden_dropout_prob
      UpperCamelCase_     :					Optional[int]   	=drop_path_rate
      UpperCamelCase_     :					str   	=use_layerscale
      UpperCamelCase_     :					List[Any]   	=layerscale_value
      UpperCamelCase_     :					Optional[int]   	=use_post_layernorm
      UpperCamelCase_     :					Dict   	=use_post_layernorm_in_modulation
      UpperCamelCase_     :					Optional[Any]   	=normalize_modulator
      UpperCamelCase_     :					List[Any]   	=initializer_range
      UpperCamelCase_     :					List[str]   	=layer_norm_eps
      UpperCamelCase_     :					List[Any]   	=encoder_stride
      UpperCamelCase_     :					str   	=['stem'] + [f'''stage{idx}''' for idx in range(1      ,  len(self.depths   ) + 1   )]
      UpperCamelCase_	,     UpperCamelCase_     :					int   	=get_aligned_output_features_output_indices(
          out_features=_lowerCamelCase      ,  out_indices=_lowerCamelCase      ,  stage_names=self.stage_names   )
