




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


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





class 			lowerCamelCase_				( _SCREAMING_SNAKE_CASE   ):


 '''simple docstring'''







 def __init__(				self    :						int							,							_lowerCAmelCase    :						WhisperForConditionalGeneration							,							_lowerCAmelCase    :						WhisperProcessor							,							_lowerCAmelCase    :						AutoencoderKL							,							_lowerCAmelCase    :						CLIPTextModel							,							_lowerCAmelCase    :						CLIPTokenizer							,							_lowerCAmelCase    :						UNetaDConditionModel							,							_lowerCAmelCase    :						Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]							,							_lowerCAmelCase    :						StableDiffusionSafetyChecker							,							_lowerCAmelCase    :						CLIPImageProcessor							,							):
       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  lowerCAmelCase_    (				self    :						Optional[Any]							,							_lowerCAmelCase    :						Optional[Union[str, int]] = "auto"		):
       if slice_size == "auto":
             SCREAMING_SNAKE_CASE_      		=     self.unet.config.attention_head_dim // 2
       self.unet.set_attention_slice(_lowerCAmelCase		)



 def  lowerCAmelCase_    (				self    :						Tuple		):
       self.enable_attention_slicing(_lowerCAmelCase		)




 @torch.no_grad()
 def __call__(				self    :						Dict							,							_lowerCAmelCase    :						Optional[int]							,							_lowerCAmelCase    :						Dict=16_000							,							_lowerCAmelCase    :						int = 512							,							_lowerCAmelCase    :						int = 512							,							_lowerCAmelCase    :						int = 50							,							_lowerCAmelCase    :						float = 7.5							,							_lowerCAmelCase    :						Optional[Union[str, List[str]]] = None							,							_lowerCAmelCase    :						Optional[int] = 1							,							_lowerCAmelCase    :						float = 0.0							,							_lowerCAmelCase    :						Optional[torch.Generator] = None							,							_lowerCAmelCase    :						Optional[torch.FloatTensor] = None							,							_lowerCAmelCase    :						Optional[str] = "pil"							,							_lowerCAmelCase    :						bool = True							,							_lowerCAmelCase    :						Optional[Callable[[int, int, torch.FloatTensor], None]] = None							,							_lowerCAmelCase    :						int = 1							,							**_lowerCAmelCase    :						List[str]							,							):
       SCREAMING_SNAKE_CASE_      		=     self.speech_processor.feature_extractor(
           _lowerCAmelCase							,							return_tensors='pt'							,							sampling_rate=_lowerCAmelCase		).input_features.to(self.device		)
       SCREAMING_SNAKE_CASE_      		=     self.speech_model.generate(_lowerCAmelCase							,							max_length=480_000		)

       SCREAMING_SNAKE_CASE_      		=     self.speech_processor.tokenizer.batch_decode(_lowerCAmelCase							,							skip_special_tokens=_lowerCAmelCase							,							normalize=_lowerCAmelCase		)[
           0
       ]

       if isinstance(_lowerCAmelCase							,							_lowerCAmelCase		):
             SCREAMING_SNAKE_CASE_      		=     1
       elif isinstance(_lowerCAmelCase							,							_lowerCAmelCase		):
             SCREAMING_SNAKE_CASE_      		=     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
       SCREAMING_SNAKE_CASE_      		=     self.tokenizer(
           _lowerCAmelCase							,							padding='max_length'							,							max_length=self.tokenizer.model_max_length							,							return_tensors='pt'							,							)
       SCREAMING_SNAKE_CASE_      		=     text_inputs.input_ids

       if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
             SCREAMING_SNAKE_CASE_      		=     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_      		=     text_input_ids[:, : self.tokenizer.model_max_length]
       SCREAMING_SNAKE_CASE_      		=     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_      		=     text_embeddings.shape
       SCREAMING_SNAKE_CASE_      		=     text_embeddings.repeat(1							,							_lowerCAmelCase							,							1		)
       SCREAMING_SNAKE_CASE_      		=     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.
       SCREAMING_SNAKE_CASE_      		=     guidance_scale > 1.0
       # get unconditional embeddings for classifier free guidance
       if do_classifier_free_guidance:
             SCREAMING_SNAKE_CASE_      		=     42
             if negative_prompt is None:
                   SCREAMING_SNAKE_CASE_      		=     [''] * 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		):
                   SCREAMING_SNAKE_CASE_      		=     [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:
                   SCREAMING_SNAKE_CASE_      		=     negative_prompt

             SCREAMING_SNAKE_CASE_      		=     text_input_ids.shape[-1]
             SCREAMING_SNAKE_CASE_      		=     self.tokenizer(
                 _lowerCAmelCase							,							padding='max_length'							,							max_length=_lowerCAmelCase							,							truncation=_lowerCAmelCase							,							return_tensors='pt'							,							)
             SCREAMING_SNAKE_CASE_      		=     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_      		=     uncond_embeddings.shape[1]
             SCREAMING_SNAKE_CASE_      		=     uncond_embeddings.repeat(1							,							_lowerCAmelCase							,							1		)
             SCREAMING_SNAKE_CASE_      		=     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
             SCREAMING_SNAKE_CASE_      		=     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_      		=     (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
       SCREAMING_SNAKE_CASE_      		=     text_embeddings.dtype
       if latents is None:
             if self.device.type == "mps":
                   # randn does not exist on mps
                   SCREAMING_SNAKE_CASE_      		=     torch.randn(_lowerCAmelCase							,							generator=_lowerCAmelCase							,							device='cpu'							,							dtype=_lowerCAmelCase		).to(
                       self.device		)
             else:
                   SCREAMING_SNAKE_CASE_      		=     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}"		)
             SCREAMING_SNAKE_CASE_      		=     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
       SCREAMING_SNAKE_CASE_      		=     self.scheduler.timesteps.to(self.device		)

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

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

             # predict the noise residual
             SCREAMING_SNAKE_CASE_      		=     self.unet(_lowerCAmelCase							,							_lowerCAmelCase							,							encoder_hidden_states=_lowerCAmelCase		).sample

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

             # compute the previous noisy sample x_t -> x_t-1
             SCREAMING_SNAKE_CASE_      		=     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		)

       SCREAMING_SNAKE_CASE_      		=     1 / 0.1_8215 * latents
       SCREAMING_SNAKE_CASE_      		=     self.vae.decode(_lowerCAmelCase		).sample

       SCREAMING_SNAKE_CASE_      		=     (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_      		=     image.cpu().permute(0							,							2							,							3							,							1		).float().numpy()

       if output_type == "pil":
             SCREAMING_SNAKE_CASE_      		=     self.numpy_to_pil(_lowerCAmelCase		)

       if not return_dict:
             return image

       return StableDiffusionPipelineOutput(images=_lowerCAmelCase							,							nsfw_content_detected=_lowerCAmelCase		)





import numpy as np
import torch
from torch.utils.data import DataLoader

from accelerate.utils.dataclasses import DistributedType





class 			lowerCamelCase_				:


 '''simple docstring'''







 def __init__(				self    :						Any							,							_lowerCAmelCase    :						Optional[int]=2							,							_lowerCAmelCase    :						Any=3							,							_lowerCAmelCase    :						Tuple=64							,							_lowerCAmelCase    :						List[str]=None		):
       SCREAMING_SNAKE_CASE_      		=     np.random.default_rng(_lowerCAmelCase		)
       SCREAMING_SNAKE_CASE_      		=     length
       SCREAMING_SNAKE_CASE_      		=     rng.normal(size=(length,)		).astype(np.floataa		)
       SCREAMING_SNAKE_CASE_      		=     a * self.x + b + rng.normal(scale=0.1							,							size=(length,)		).astype(np.floataa		)



 def __len__(				self    :						Optional[int]		):
       return self.length




 def __getitem__(				self    :						str							,							_lowerCAmelCase    :						Union[str, Any]		):
       return {"x": self.x[i], "y": self.y[i]}





class 			lowerCamelCase_				( torch.nn.Module   ):


 '''simple docstring'''







 def __init__(				self    :						Tuple							,							_lowerCAmelCase    :						Dict=0							,							_lowerCAmelCase    :						List[str]=0							,							_lowerCAmelCase    :						str=False		):
       super().__init__()
       SCREAMING_SNAKE_CASE_      		=     torch.nn.Parameter(torch.tensor([2, 3]		).float()		)
       SCREAMING_SNAKE_CASE_      		=     torch.nn.Parameter(torch.tensor([2, 3]		).float()		)
       SCREAMING_SNAKE_CASE_      		=     True




 def  lowerCAmelCase_    (				self    :						Dict							,							_lowerCAmelCase    :						Union[str, Any]=None		):
       if self.first_batch:
             print(F"Model dtype: {self.a.dtype}, {self.b.dtype}. Input dtype: {x.dtype}"		)
             SCREAMING_SNAKE_CASE_      		=     False
       return x * self.a[0] + self.b[0]





class 			lowerCamelCase_				( torch.nn.Module   ):


 '''simple docstring'''







 def __init__(				self    :						Optional[int]							,							_lowerCAmelCase    :						Any=0							,							_lowerCAmelCase    :						Any=0							,							_lowerCAmelCase    :						Optional[Any]=False		):
       super().__init__()
       SCREAMING_SNAKE_CASE_      		=     torch.nn.Parameter(torch.tensor(_lowerCAmelCase		).float()		)
       SCREAMING_SNAKE_CASE_      		=     torch.nn.Parameter(torch.tensor(_lowerCAmelCase		).float()		)
       SCREAMING_SNAKE_CASE_      		=     True




 def  lowerCAmelCase_    (				self    :						Optional[Any]							,							_lowerCAmelCase    :						Optional[int]=None		):
       if self.first_batch:
             print(F"Model dtype: {self.a.dtype}, {self.b.dtype}. Input dtype: {x.dtype}"		)
             SCREAMING_SNAKE_CASE_      		=     False
       return x * self.a + self.b


def    UpperCAmelCase_				( __UpperCAmelCase		:      Dict     ,     __UpperCAmelCase		:      int = 16				)   ->					Union[str, Any]:
      from datasets import load_dataset
      from transformers import AutoTokenizer

      SCREAMING_SNAKE_CASE_      		=     AutoTokenizer.from_pretrained('bert-base-cased'				)
      SCREAMING_SNAKE_CASE_      		=     {'train': 'tests/test_samples/MRPC/train.csv', 'validation': 'tests/test_samples/MRPC/dev.csv'}
      SCREAMING_SNAKE_CASE_      		=     load_dataset('csv'     ,     data_files=__UpperCAmelCase				)
      SCREAMING_SNAKE_CASE_      		=     datasets['train'].unique('label'				)

      SCREAMING_SNAKE_CASE_      		=     {v: i for i, v in enumerate(__UpperCAmelCase				)}

      def tokenize_function(__UpperCAmelCase		:      Optional[int]				):
            # max_length=None => use the model max length (it's actually the default)
            SCREAMING_SNAKE_CASE_      		=     tokenizer(
                examples['sentence1']     ,     examples['sentence2']     ,     truncation=__UpperCAmelCase     ,     max_length=__UpperCAmelCase     ,     padding='max_length'				)
            if "label" in examples:
                  SCREAMING_SNAKE_CASE_      		=     [label_to_id[l] for l in examples['label']]
            return outputs

      # Apply the method we just defined to all the examples in all the splits of the dataset
      SCREAMING_SNAKE_CASE_      		=     datasets.map(
          __UpperCAmelCase     ,     batched=__UpperCAmelCase     ,     remove_columns=['sentence1', 'sentence2', 'label']     ,     )

      def collate_fn(__UpperCAmelCase		:      Dict				):
            # On TPU it's best to pad everything to the same length or training will be very slow.
            if accelerator.distributed_type == DistributedType.TPU:
                  return tokenizer.pad(__UpperCAmelCase     ,     padding='max_length'     ,     max_length=1_28     ,     return_tensors='pt'				)
            return tokenizer.pad(__UpperCAmelCase     ,     padding='longest'     ,     return_tensors='pt'				)

      # Instantiate dataloaders.
      SCREAMING_SNAKE_CASE_      		=     DataLoader(tokenized_datasets['train']     ,     shuffle=__UpperCAmelCase     ,     collate_fn=__UpperCAmelCase     ,     batch_size=2				)
      SCREAMING_SNAKE_CASE_      		=     DataLoader(tokenized_datasets['validation']     ,     shuffle=__UpperCAmelCase     ,     collate_fn=__UpperCAmelCase     ,     batch_size=1				)

      return train_dataloader, eval_dataloader