


'''simple docstring'''



import inspect
import os
from argparse import ArgumentParser, Namespace
from importlib import import_module

import huggingface_hub
import numpy as np
from packaging import version

from .. import (
    FEATURE_EXTRACTOR_MAPPING,
    IMAGE_PROCESSOR_MAPPING,
    PROCESSOR_MAPPING,
    TOKENIZER_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    AutoImageProcessor,
    AutoProcessor,
    AutoTokenizer,
    is_datasets_available,
    is_tf_available,
    is_torch_available,
)
from ..utils import TF2_WEIGHTS_INDEX_NAME, TF2_WEIGHTS_NAME, logging
from . import BaseTransformersCLICommand


if is_tf_available():
      import tensorflow as tf

      tf.config.experimental.enable_tensor_float_aa_execution(False)

if is_torch_available():
      import torch

if is_datasets_available():
      from datasets import load_dataset


A__		: Any												=  5e-5  # larger error tolerance than in our internal tests, to avoid flaky user-facing errors



def       UpperCAmelCase__ (     UpperCAmelCase_   :    Dict     )					->		List[Any]:
     return PTtoTFCommand(
         args.model_name				,   args.local_dir				,   args.max_error				,   args.new_weights				,   args.no_pr				,   args.push				,   args.extra_commit_description				,   args.override_model_class				,   )





class        UpperCAmelCase_  (snake_case_  ):





    """simple docstring"""






    @staticmethod
    def 				lowercase_    (							SCREAMING_SNAKE_CASE_     )    ->					Union[str, Any]:
         __lowerCamelCase       :     Any        =		parser.add_parser(
             'pt-to-tf'      ,					help=(
                 'CLI tool to run convert a transformers model from a PyTorch checkpoint to a TensorFlow checkpoint.'
                 ' Can also be used to validate existing weights without opening PRs, with --no-pr.'
             )      ,					)
         train_parser.add_argument(
             '--model-name'      ,					type=SCREAMING_SNAKE_CASE_      ,					required=SCREAMING_SNAKE_CASE_      ,					help='The model name, including owner/organization, as seen on the hub.'      ,					)
         train_parser.add_argument(
             '--local-dir'      ,					type=SCREAMING_SNAKE_CASE_      ,					default=''      ,					help='Optional local directory of the model repository. Defaults to /tmp/{model_name}'      ,					)
         train_parser.add_argument(
             '--max-error'      ,					type=SCREAMING_SNAKE_CASE_      ,					default=SCREAMING_SNAKE_CASE_      ,					help=(
                 f'Maximum error tolerance. Defaults to {MAX_ERROR}. This flag should be avoided, use at your own risk.'
             )      ,					)
         train_parser.add_argument(
             '--new-weights'      ,					action='store_true'      ,					help='Optional flag to create new TensorFlow weights, even if they already exist.'      ,					)
         train_parser.add_argument(
             '--no-pr'      ,					action='store_true'      ,					help='Optional flag to NOT open a PR with converted weights.'     )
         train_parser.add_argument(
             '--push'      ,					action='store_true'      ,					help='Optional flag to push the weights directly to `main` (requires permissions)'      ,					)
         train_parser.add_argument(
             '--extra-commit-description'      ,					type=SCREAMING_SNAKE_CASE_      ,					default=''      ,					help='Optional additional commit description to use when opening a PR (e.g. to tag the owner).'      ,					)
         train_parser.add_argument(
             '--override-model-class'      ,					type=SCREAMING_SNAKE_CASE_      ,					default=SCREAMING_SNAKE_CASE_      ,					help='If you think you know better than the auto-detector, you can specify the model class here. '
             'Can be either an AutoModel class or a specific model class like BertForSequenceClassification.'      ,					)
         train_parser.set_defaults(func=SCREAMING_SNAKE_CASE_     )







    @staticmethod
    def 				lowercase_    (							SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )    ->					str:
         __lowerCamelCase       :     Tuple        =		set(pt_outputs.keys()     )
         __lowerCamelCase       :     Dict        =		set(tf_outputs.keys()     )
         if pt_out_attrs != tf_out_attrs:
              raise ValueError(
                  f'The model outputs have different attributes, aborting. (Pytorch: {pt_out_attrs}, TensorFlow:'
                  f' {tf_out_attrs})'     )

         # 2. For each output attribute, computes the difference
         def _find_pt_tf_differences(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_=""     ):
              # If the current attribute is a tensor, it is a leaf and we make the comparison. Otherwise, we will dig in
              # recursivelly, keeping the name of the attribute.
              if isinstance(SCREAMING_SNAKE_CASE_      ,					torch.Tensor     ):
                   __lowerCamelCase       :     Optional[Any]        =		np.max(np.abs(pt_out.numpy() - tf_out.numpy()     )     )
                   __lowerCamelCase       :     List[Any]        =		tensor_difference
              else:
                   __lowerCamelCase       :     str        =		attr_name
                   for i, pt_item in enumerate(SCREAMING_SNAKE_CASE_     ):
                        # If it is a named attribute, we keep the name. Otherwise, just its index.
                        if isinstance(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     ):
                             __lowerCamelCase       :     int        =		root_name + pt_item
                             __lowerCamelCase       :     List[str]        =		tf_out[pt_item]
                             __lowerCamelCase       :     Any        =		pt_out[pt_item]
                        else:
                             __lowerCamelCase       :     Union[str, Any]        =		root_name + f'[{i}]'
                             __lowerCamelCase       :     Dict        =		tf_out[i]
                        __lowerCamelCase       :     str        =		_find_pt_tf_differences(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )

              return differences

         return _find_pt_tf_differences(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					{}     )







    def __init__(							self      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					*SCREAMING_SNAKE_CASE_      ,					)    ->					int:
         __lowerCamelCase       :     Optional[int]        =		logging.get_logger('transformers-cli/pt_to_tf'     )
         __lowerCamelCase       :     Optional[int]        =		model_name
         __lowerCamelCase       :     str        =		local_dir if local_dir else os.path.join('/tmp'      ,					SCREAMING_SNAKE_CASE_     )
         __lowerCamelCase       :     List[Any]        =		max_error
         __lowerCamelCase       :     str        =		new_weights
         __lowerCamelCase       :     Optional[Any]        =		no_pr
         __lowerCamelCase       :     List[str]        =		push
         __lowerCamelCase       :     Tuple        =		extra_commit_description
         __lowerCamelCase       :     Union[str, Any]        =		override_model_class







    def 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )    ->					Tuple:


         def _get_audio_input():
              __lowerCamelCase       :     Union[str, Any]        =		load_dataset('hf-internal-testing/librispeech_asr_dummy'      ,					'clean'      ,					split='validation'     )
              __lowerCamelCase       :     Dict        =		ds.sort('id'     ).select(range(2     )     )[:2]["audio"]
              __lowerCamelCase       :     Dict        =		[x["array"] for x in speech_samples]
              return raw_samples

         __lowerCamelCase       :     Union[str, Any]        =		type(pt_model.config     )
         if model_config_class in PROCESSOR_MAPPING:
              __lowerCamelCase       :     Optional[Any]        =		AutoProcessor.from_pretrained(self._local_dir     )
              if model_config_class in TOKENIZER_MAPPING and processor.tokenizer.pad_token is None:
                   __lowerCamelCase       :     List[str]        =		processor.tokenizer.eos_token
         elif model_config_class in IMAGE_PROCESSOR_MAPPING:
              __lowerCamelCase       :     Union[str, Any]        =		AutoImageProcessor.from_pretrained(self._local_dir     )
         elif model_config_class in FEATURE_EXTRACTOR_MAPPING:
              __lowerCamelCase       :     Optional[Any]        =		AutoFeatureExtractor.from_pretrained(self._local_dir     )
         elif model_config_class in TOKENIZER_MAPPING:
              __lowerCamelCase       :     Any        =		AutoTokenizer.from_pretrained(self._local_dir     )
              if processor.pad_token is None:
                   __lowerCamelCase       :     List[str]        =		processor.eos_token
         else:
              raise ValueError(f'Unknown data processing type (model config type: {model_config_class})'     )

         __lowerCamelCase       :     List[str]        =		set(inspect.signature(pt_model.forward     ).parameters.keys()     )
         __lowerCamelCase       :     List[Any]        =		{}
         if "input_ids" in model_forward_signature:
              processor_inputs.update(
                  {
                      'text': ['Hi there!', 'I am a batch with more than one row and different input lengths.'],
                      'padding': True,
                      'truncation': True,
                  }     )
         if "pixel_values" in model_forward_signature:
              __lowerCamelCase       :     List[str]        =		load_dataset('cifar10'      ,					'plain_text'      ,					split='test'     )[:2]["img"]
              processor_inputs.update({'images': sample_images}     )
         if "input_features" in model_forward_signature:
              __lowerCamelCase       :     Union[str, Any]        =		inspect.signature(processor.feature_extractor     ).parameters
              # Pad to the largest input length by default but take feature extractor default
              # padding value if it exists e.g. "max_length" and is not False or None
              if "padding" in feature_extractor_signature:
                   __lowerCamelCase       :     Union[str, Any]        =		feature_extractor_signature["padding"].default
                   if default_strategy is not False and default_strategy is not None:
                        __lowerCamelCase       :     Tuple        =		default_strategy
                   else:
                        __lowerCamelCase       :     List[Any]        =		True
              else:
                   __lowerCamelCase       :     Any        =		True
              processor_inputs.update({'audio': _get_audio_input(), 'padding': padding_strategy}     )
         if "input_values" in model_forward_signature:  # Wav2Vec2 audio input
              processor_inputs.update({'audio': _get_audio_input(), 'padding': True}     )
         __lowerCamelCase       :     List[str]        =		processor(**SCREAMING_SNAKE_CASE_      ,					return_tensors='pt'     )
         __lowerCamelCase       :     Dict        =		processor(**SCREAMING_SNAKE_CASE_      ,					return_tensors='tf'     )

         # Extra input requirements, in addition to the input modality
         if (
             config.is_encoder_decoder
             or (hasattr(SCREAMING_SNAKE_CASE_      ,					'encoder'     ) and hasattr(SCREAMING_SNAKE_CASE_      ,					'decoder'     ))
             or "decoder_input_ids" in tf_dummy_inputs
         ):
              __lowerCamelCase       :     List[Any]        =		np.asarray([[1], [1]]      ,					dtype=SCREAMING_SNAKE_CASE_     ) * (pt_model.config.decoder_start_token_id or 0)
              pt_input.update({'decoder_input_ids': torch.tensor(SCREAMING_SNAKE_CASE_     )}     )
              tf_input.update({'decoder_input_ids': tf.convert_to_tensor(SCREAMING_SNAKE_CASE_     )}     )

         return pt_input, tf_input







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


         if version.parse(huggingface_hub.__version__     ) < version.parse('0.9.0'     ):
              raise ImportError(
                  'The huggingface_hub version must be >= 0.9.0 to use this command. Please update your huggingface_hub'
                  ' installation.'     )
         else:
              from huggingface_hub import Repository, create_commit
              from huggingface_hub._commit_api import CommitOperationAdd

         # Fetch remote data
         __lowerCamelCase       :     Optional[int]        =		Repository(local_dir=self._local_dir      ,					clone_from=self._model_name     )

         # Load config and get the appropriate architecture -- the latter is needed to convert the head's weights
         __lowerCamelCase       :     List[Any]        =		AutoConfig.from_pretrained(self._local_dir     )
         __lowerCamelCase       :     Tuple        =		config.architectures
         if self._override_model_class is not None:
              if self._override_model_class.startswith('TF'     ):
                   __lowerCamelCase       :     List[Any]        =		[self._override_model_class[2:]]
              else:
                   __lowerCamelCase       :     Tuple        =		[self._override_model_class]
              try:
                   __lowerCamelCase       :     Dict        =		getattr(import_module('transformers'     )      ,					architectures[0]     )
              except AttributeError:
                   raise ValueError(f'Model class {self._override_model_class} not found in transformers.'     )
              try:
                   __lowerCamelCase       :     Optional[Any]        =		getattr(import_module('transformers'     )      ,					'TF' + architectures[0]     )
              except AttributeError:
                   raise ValueError(f'TF model class TF{self._override_model_class} not found in transformers.'     )
         elif architectures is None:  # No architecture defined -- use auto classes
              __lowerCamelCase       :     Optional[int]        =		getattr(import_module('transformers'     )      ,					'AutoModel'     )
              __lowerCamelCase       :     Tuple        =		getattr(import_module('transformers'     )      ,					'TFAutoModel'     )
              self._logger.warning('No detected architecture, using AutoModel/TFAutoModel'     )
         else:  # Architecture defined -- use it
              if len(SCREAMING_SNAKE_CASE_     ) > 1:
                   raise ValueError(f'More than one architecture was found, aborting. (architectures = {architectures})'     )
              self._logger.warning(f'Detected architecture: {architectures[0]}'     )
              __lowerCamelCase       :     Dict        =		getattr(import_module('transformers'     )      ,					architectures[0]     )
              try:
                   __lowerCamelCase       :     str        =		getattr(import_module('transformers'     )      ,					'TF' + architectures[0]     )
              except AttributeError:
                   raise AttributeError(f'The TensorFlow equivalent of {architectures[0]} doesn\'t exist in transformers.'     )

        # Check the TF dummy inputs to see what keys we need in the forward pass
         __lowerCamelCase       :     int        =		tf_class.from_config(SCREAMING_SNAKE_CASE_     )
         __lowerCamelCase       :     int        =		tf_from_pt_model.dummy_inputs

         del tf_from_pt_model  # Try to keep only one model in memory at a time

         # Load the model and get some basic inputs
         __lowerCamelCase       :     Tuple        =		pt_class.from_pretrained(self._local_dir     )
         pt_model.eval()

         __lowerCamelCase       :     Optional[int]        =		self.get_inputs(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )

         with torch.no_grad():
              __lowerCamelCase       :     Optional[Any]        =		pt_model(**SCREAMING_SNAKE_CASE_      ,					output_hidden_states=SCREAMING_SNAKE_CASE_     )
         del pt_model  # will no longer be used, and may have a large memory footprint

         __lowerCamelCase       :     Optional[Any]        =		tf_class.from_pretrained(self._local_dir      ,					from_pt=SCREAMING_SNAKE_CASE_     )
         __lowerCamelCase       :     Optional[int]        =		tf_from_pt_model(**SCREAMING_SNAKE_CASE_      ,					output_hidden_states=SCREAMING_SNAKE_CASE_      ,					training=SCREAMING_SNAKE_CASE_     )

         # Confirms that cross loading PT weights into TF worked.
         __lowerCamelCase       :     Any        =		self.find_pt_tf_differences(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )
         __lowerCamelCase       :     Union[str, Any]        =		{k: v for k, v in crossload_differences.items() if "hidden" not in k}
         __lowerCamelCase       :     int        =		{k: v for k, v in crossload_differences.items() if "hidden" in k}
         if len(SCREAMING_SNAKE_CASE_     ) == 0 and architectures is not None:
              raise ValueError(
                  f'Something went wrong -- the config file has architectures ({architectures}), but no model head'
                  ' output was found. All outputs start with \'hidden\''     )
         __lowerCamelCase       :     Optional[int]        =		max(output_differences.values()     ) if output_differences else 0.0
         __lowerCamelCase       :     Optional[int]        =		max(hidden_differences.values()     )
         if max_crossload_output_diff > self._max_error or max_crossload_hidden_diff > self._max_error:
              raise ValueError(
                  'The cross-loaded TensorFlow model has different outputs, something went wrong!\n'
                  + f'\nList of maximum output differences above the threshold ({self._max_error}):\n'
                  + '\n'.join([f'{k}: {v:.3e}' for k, v in output_differences.items() if v > self._max_error]     )
                  + f'\n\nList of maximum hidden layer differences above the threshold ({self._max_error}):\n'
                  + '\n'.join([f'{k}: {v:.3e}' for k, v in hidden_differences.items() if v > self._max_error]     )     )

         # Save the weights in a TF format (if needed) and confirms that the results are still good
         __lowerCamelCase       :     Optional[Any]        =		os.path.join(self._local_dir      ,					SCREAMING_SNAKE_CASE_     )
         __lowerCamelCase       :     Tuple        =		os.path.join(self._local_dir      ,					SCREAMING_SNAKE_CASE_     )
         if (not os.path.exists(SCREAMING_SNAKE_CASE_     ) and not os.path.exists(SCREAMING_SNAKE_CASE_     )) or self._new_weights:
              tf_from_pt_model.save_pretrained(self._local_dir     )
         del tf_from_pt_model  # will no longer be used, and may have a large memory footprint

         __lowerCamelCase       :     List[str]        =		tf_class.from_pretrained(self._local_dir     )
         __lowerCamelCase       :     Tuple        =		tf_model(**SCREAMING_SNAKE_CASE_      ,					output_hidden_states=SCREAMING_SNAKE_CASE_     )

         __lowerCamelCase       :     List[str]        =		self.find_pt_tf_differences(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )
         __lowerCamelCase       :     Any        =		{k: v for k, v in conversion_differences.items() if "hidden" not in k}
         __lowerCamelCase       :     Optional[Any]        =		{k: v for k, v in conversion_differences.items() if "hidden" in k}
         if len(SCREAMING_SNAKE_CASE_     ) == 0 and architectures is not None:
              raise ValueError(
                  f'Something went wrong -- the config file has architectures ({architectures}), but no model head'
                  ' output was found. All outputs start with \'hidden\''     )
         __lowerCamelCase       :     Optional[Any]        =		max(output_differences.values()     ) if output_differences else 0.0
         __lowerCamelCase       :     str        =		max(hidden_differences.values()     )
         if max_conversion_output_diff > self._max_error or max_conversion_hidden_diff > self._max_error:
              raise ValueError(
                  'The converted TensorFlow model has different outputs, something went wrong!\n'
                  + f'\nList of maximum output differences above the threshold ({self._max_error}):\n'
                  + '\n'.join([f'{k}: {v:.3e}' for k, v in output_differences.items() if v > self._max_error]     )
                  + f'\n\nList of maximum hidden layer differences above the threshold ({self._max_error}):\n'
                  + '\n'.join([f'{k}: {v:.3e}' for k, v in hidden_differences.items() if v > self._max_error]     )     )

         __lowerCamelCase       :     Optional[int]        =		"Update TF weights" if self._new_weights else "Add TF weights"
         if self._push:
              repo.git_add(auto_lfs_track=SCREAMING_SNAKE_CASE_     )
              repo.git_commit(SCREAMING_SNAKE_CASE_     )
              repo.git_push(blocking=SCREAMING_SNAKE_CASE_     )  # this prints a progress bar with the upload
              self._logger.warning(f'TF weights pushed into {self._model_name}'     )
         elif not self._no_pr:
              self._logger.warning('Uploading the weights into a new PR...'     )
              __lowerCamelCase       :     Union[str, Any]        =		(
                  "Model converted by the [`transformers`' `pt_to_tf`"
                  " CLI](https://github.com/huggingface/transformers/blob/main/src/transformers/commands/pt_to_tf.py). "
                  "All converted model outputs and hidden layers were validated against its PyTorch counterpart.\n\n"
                  f'Maximum crossload output difference={max_crossload_output_diff:.3e}; '
                  f'Maximum crossload hidden layer difference={max_crossload_hidden_diff:.3e};\n'
                  f'Maximum conversion output difference={max_conversion_output_diff:.3e}; '
                  f'Maximum conversion hidden layer difference={max_conversion_hidden_diff:.3e};\n'
              )
              if self._max_error > MAX_ERROR:
                   commit_descrition += (
                       f'\n\nCAUTION: The maximum admissible error was manually increased to {self._max_error}!'
                   )
              if self._extra_commit_description:
                   commit_descrition += "\n\n" + self._extra_commit_description

              # sharded model -> adds all related files (index and .h5 shards)
              if os.path.exists(SCREAMING_SNAKE_CASE_     ):
                   __lowerCamelCase       :     str        =		[
                       CommitOperationAdd(path_in_repo=SCREAMING_SNAKE_CASE_      ,					path_or_fileobj=SCREAMING_SNAKE_CASE_     )
                   ]
                   for shard_path in tf.io.gfile.glob(self._local_dir + '/tf_model-*.h5'     ):
                        operations += [
                            CommitOperationAdd(path_in_repo=os.path.basename(SCREAMING_SNAKE_CASE_     )      ,					path_or_fileobj=SCREAMING_SNAKE_CASE_     )
                        ]
              else:
                   __lowerCamelCase       :     List[str]        =		[CommitOperationAdd(path_in_repo=SCREAMING_SNAKE_CASE_      ,					path_or_fileobj=SCREAMING_SNAKE_CASE_     )]

              __lowerCamelCase       :     Dict        =		create_commit(
                  repo_id=self._model_name      ,					operations=SCREAMING_SNAKE_CASE_      ,					commit_message=SCREAMING_SNAKE_CASE_      ,					commit_description=SCREAMING_SNAKE_CASE_      ,					repo_type='model'      ,					create_pr=SCREAMING_SNAKE_CASE_      ,					).pr_url
              self._logger.warning(f'PR open in {hub_pr_url}'     )










'''simple docstring'''



import inspect
from typing import List, Optional, Union

import torch
from torch import nn
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPImageProcessor, CLIPModel, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    DPMSolverMultistepScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput





class        UpperCAmelCase_  (nn.Module  ):





    """simple docstring"""






    def __init__(							self      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_=1.0     )    ->					List[str]:
         super().__init__()

         __lowerCamelCase       :     int        =		cut_size
         __lowerCamelCase       :     Optional[int]        =		cut_power







    def 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )    ->					List[Any]:
         __lowerCamelCase    ,	__lowerCamelCase       :     List[str]        =		pixel_values.shape[2:4]
         __lowerCamelCase       :     Union[str, Any]        =		min(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )
         __lowerCamelCase       :     Tuple        =		min(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					self.cut_size     )
         __lowerCamelCase       :     str        =		[]
         for _ in range(SCREAMING_SNAKE_CASE_     ):
              __lowerCamelCase       :     Union[str, Any]        =		int(torch.rand([]     ) ** self.cut_power * (max_size - min_size) + min_size     )
              __lowerCamelCase       :     List[Any]        =		torch.randint(0      ,					sideX - size + 1      ,					()     )
              __lowerCamelCase       :     int        =		torch.randint(0      ,					sideY - size + 1      ,					()     )
              __lowerCamelCase       :     str        =		pixel_values[:, :, offsety : offsety + size, offsetx : offsetx + size]
              cutouts.append(F.adaptive_avg_poolad(SCREAMING_SNAKE_CASE_      ,					self.cut_size     )     )
         return torch.cat(SCREAMING_SNAKE_CASE_     )







def       UpperCAmelCase__ (     UpperCAmelCase_   :    List[Any]				,   UpperCAmelCase_   :    Tuple							)					->		str:
     __lowerCamelCase       :     List[str]        =		F.normalize(UpperCAmelCase_				,   dim=-1							)
     __lowerCamelCase       :     Optional[Any]        =		F.normalize(UpperCAmelCase_				,   dim=-1							)
     return (x - y).norm(dim=-1							).div(2							).arcsin().pow(2							).mul(2							)



def       UpperCAmelCase__ (     UpperCAmelCase_   :    Optional[Any]				,   UpperCAmelCase_   :    Any							)					->		List[str]:
     for param in model.parameters():
          __lowerCamelCase       :     Dict        =		value





class        UpperCAmelCase_  (_UpperCAmelCase  ):





    """simple docstring"""






    def __init__(							self      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					)    ->					List[str]:
         super().__init__()
         self.register_modules(
             vae=SCREAMING_SNAKE_CASE_      ,					text_encoder=SCREAMING_SNAKE_CASE_      ,					clip_model=SCREAMING_SNAKE_CASE_      ,					tokenizer=SCREAMING_SNAKE_CASE_      ,					unet=SCREAMING_SNAKE_CASE_      ,					scheduler=SCREAMING_SNAKE_CASE_      ,					feature_extractor=SCREAMING_SNAKE_CASE_      ,					)

         __lowerCamelCase       :     int        =		transforms.Normalize(mean=feature_extractor.image_mean      ,					std=feature_extractor.image_std     )
         __lowerCamelCase       :     Union[str, Any]        =		(
             feature_extractor.size
             if isinstance(feature_extractor.size      ,					SCREAMING_SNAKE_CASE_     )
             else feature_extractor.size['shortest_edge']
         )
         __lowerCamelCase       :     List[Any]        =		MakeCutouts(self.cut_out_size     )

         set_requires_grad(self.text_encoder      ,					SCREAMING_SNAKE_CASE_     )
         set_requires_grad(self.clip_model      ,					SCREAMING_SNAKE_CASE_     )







    def 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_ = "auto"     )    ->					str:
         if slice_size == "auto":
              # half the attention head size is usually a good trade-off between
              # speed and memory
              __lowerCamelCase       :     str        =		self.unet.config.attention_head_dim // 2
         self.unet.set_attention_slice(SCREAMING_SNAKE_CASE_     )







    def 				lowercase_    (							self     )    ->					Union[str, Any]:
         self.enable_attention_slicing(SCREAMING_SNAKE_CASE_     )







    def 				lowercase_    (							self     )    ->					Union[str, Any]:
         set_requires_grad(self.vae      ,					SCREAMING_SNAKE_CASE_     )







    def 				lowercase_    (							self     )    ->					List[str]:
         set_requires_grad(self.vae      ,					SCREAMING_SNAKE_CASE_     )







    def 				lowercase_    (							self     )    ->					Optional[int]:
         set_requires_grad(self.unet      ,					SCREAMING_SNAKE_CASE_     )







    def 				lowercase_    (							self     )    ->					Union[str, Any]:
         set_requires_grad(self.unet      ,					SCREAMING_SNAKE_CASE_     )







    @torch.enable_grad()
    def 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_=True      ,					)    ->					Union[str, Any]:
         __lowerCamelCase       :     Union[str, Any]        =		latents.detach().requires_grad_()

         __lowerCamelCase       :     List[str]        =		self.scheduler.scale_model_input(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )

         # predict the noise residual
         __lowerCamelCase       :     Tuple        =		self.unet(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					encoder_hidden_states=SCREAMING_SNAKE_CASE_     ).sample

         if isinstance(self.scheduler      ,					(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)     ):
              __lowerCamelCase       :     List[str]        =		self.scheduler.alphas_cumprod[timestep]
              __lowerCamelCase       :     str        =		1 - alpha_prod_t
              # compute predicted original sample from predicted noise also called
              # "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
              __lowerCamelCase       :     Union[str, Any]        =		(latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

              __lowerCamelCase       :     Union[str, Any]        =		torch.sqrt(SCREAMING_SNAKE_CASE_     )
              __lowerCamelCase       :     Optional[int]        =		pred_original_sample * (fac) + latents * (1 - fac)
         elif isinstance(self.scheduler      ,					SCREAMING_SNAKE_CASE_     ):
              __lowerCamelCase       :     List[str]        =		self.scheduler.sigmas[index]
              __lowerCamelCase       :     List[Any]        =		latents - sigma * noise_pred
         else:
              raise ValueError(f'scheduler type {type(self.scheduler     )} not supported'     )

         __lowerCamelCase       :     Optional[Any]        =		1 / self.vae.config.scaling_factor * sample
         __lowerCamelCase       :     Optional[Any]        =		self.vae.decode(SCREAMING_SNAKE_CASE_     ).sample
         __lowerCamelCase       :     List[Any]        =		(image / 2 + 0.5).clamp(0      ,					1     )

         if use_cutouts:
              __lowerCamelCase       :     Union[str, Any]        =		self.make_cutouts(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )
         else:
              __lowerCamelCase       :     Any        =		transforms.Resize(self.cut_out_size     )(SCREAMING_SNAKE_CASE_     )
         __lowerCamelCase       :     Any        =		self.normalize(SCREAMING_SNAKE_CASE_     ).to(latents.dtype     )

         __lowerCamelCase       :     Optional[Any]        =		self.clip_model.get_image_features(SCREAMING_SNAKE_CASE_     )
         __lowerCamelCase       :     Optional[Any]        =		image_embeddings_clip / image_embeddings_clip.norm(p=2      ,					dim=-1      ,					keepdim=SCREAMING_SNAKE_CASE_     )

         if use_cutouts:
              __lowerCamelCase       :     Optional[Any]        =		spherical_dist_loss(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )
              __lowerCamelCase       :     int        =		dists.view([num_cutouts, sample.shape[0], -1]     )
              __lowerCamelCase       :     List[Any]        =		dists.sum(2     ).mean(0     ).sum() * clip_guidance_scale
         else:
              __lowerCamelCase       :     Any        =		spherical_dist_loss(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     ).mean() * clip_guidance_scale

         __lowerCamelCase       :     Union[str, Any]        =		-torch.autograd.grad(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     )[0]

         if isinstance(self.scheduler      ,					SCREAMING_SNAKE_CASE_     ):
              __lowerCamelCase       :     Optional[int]        =		latents.detach() + grads * (sigma**2)
              __lowerCamelCase       :     int        =		noise_pred_original
         else:
              __lowerCamelCase       :     Dict        =		noise_pred_original - torch.sqrt(SCREAMING_SNAKE_CASE_     ) * grads
         return noise_pred, latents







    @torch.no_grad()
    def __call__(							self      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_ = 5_12      ,					SCREAMING_SNAKE_CASE_ = 5_12      ,					SCREAMING_SNAKE_CASE_ = 50      ,					SCREAMING_SNAKE_CASE_ = 7.5      ,					SCREAMING_SNAKE_CASE_ = 1      ,					SCREAMING_SNAKE_CASE_ = 0.0      ,					SCREAMING_SNAKE_CASE_ = 1_00      ,					SCREAMING_SNAKE_CASE_ = None      ,					SCREAMING_SNAKE_CASE_ = 4      ,					SCREAMING_SNAKE_CASE_ = True      ,					SCREAMING_SNAKE_CASE_ = None      ,					SCREAMING_SNAKE_CASE_ = None      ,					SCREAMING_SNAKE_CASE_ = "pil"      ,					SCREAMING_SNAKE_CASE_ = True      ,					)    ->					Tuple:
         if isinstance(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     ):
              __lowerCamelCase       :     List[str]        =		1
         elif isinstance(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     ):
              __lowerCamelCase       :     Tuple        =		len(SCREAMING_SNAKE_CASE_     )
         else:
              raise ValueError(f'`prompt` has to be of type `str` or `list` but is {type(SCREAMING_SNAKE_CASE_     )}'     )

         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}.'     )

         # get prompt text embeddings
         __lowerCamelCase       :     str        =		self.tokenizer(
             SCREAMING_SNAKE_CASE_      ,					padding='max_length'      ,					max_length=self.tokenizer.model_max_length      ,					truncation=SCREAMING_SNAKE_CASE_      ,					return_tensors='pt'      ,					)
         __lowerCamelCase       :     List[str]        =		self.text_encoder(text_input.input_ids.to(self.device     )     )[0]
         # duplicate text embeddings for each generation per prompt
         __lowerCamelCase       :     Optional[Any]        =		text_embeddings.repeat_interleave(SCREAMING_SNAKE_CASE_      ,					dim=0     )

         if clip_guidance_scale > 0:
              if clip_prompt is not None:
                   __lowerCamelCase       :     int        =		self.tokenizer(
                       SCREAMING_SNAKE_CASE_      ,					padding='max_length'      ,					max_length=self.tokenizer.model_max_length      ,					truncation=SCREAMING_SNAKE_CASE_      ,					return_tensors='pt'      ,					).input_ids.to(self.device     )
              else:
                   __lowerCamelCase       :     str        =		text_input.input_ids.to(self.device     )
              __lowerCamelCase       :     Optional[Any]        =		self.clip_model.get_text_features(SCREAMING_SNAKE_CASE_     )
              __lowerCamelCase       :     Dict        =		text_embeddings_clip / text_embeddings_clip.norm(p=2      ,					dim=-1      ,					keepdim=SCREAMING_SNAKE_CASE_     )
              # duplicate text embeddings clip for each generation per prompt
              __lowerCamelCase       :     Optional[int]        =		text_embeddings_clip.repeat_interleave(SCREAMING_SNAKE_CASE_      ,					dim=0     )

         # 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.
         __lowerCamelCase       :     Tuple        =		guidance_scale > 1.0
         # get unconditional embeddings for classifier free guidance
         if do_classifier_free_guidance:
              __lowerCamelCase       :     List[Any]        =		text_input.input_ids.shape[-1]
              __lowerCamelCase       :     Union[str, Any]        =		self.tokenizer(['']      ,					padding='max_length'      ,					max_length=SCREAMING_SNAKE_CASE_      ,					return_tensors='pt'     )
              __lowerCamelCase       :     Any        =		self.text_encoder(uncond_input.input_ids.to(self.device     )     )[0]
              # duplicate unconditional embeddings for each generation per prompt
              __lowerCamelCase       :     Optional[int]        =		uncond_embeddings.repeat_interleave(SCREAMING_SNAKE_CASE_      ,					dim=0     )

              # 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
              __lowerCamelCase       :     Tuple        =		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`.
         __lowerCamelCase       :     int        =		(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
         __lowerCamelCase       :     Optional[int]        =		text_embeddings.dtype
         if latents is None:
              if self.device.type == "mps":
                   # randn does not work reproducibly on mps
                   __lowerCamelCase       :     List[str]        =		torch.randn(SCREAMING_SNAKE_CASE_      ,					generator=SCREAMING_SNAKE_CASE_      ,					device='cpu'      ,					dtype=SCREAMING_SNAKE_CASE_     ).to(
                       self.device     )
              else:
                   __lowerCamelCase       :     Dict        =		torch.randn(SCREAMING_SNAKE_CASE_      ,					generator=SCREAMING_SNAKE_CASE_      ,					device=self.device      ,					dtype=SCREAMING_SNAKE_CASE_     )
         else:
              if latents.shape != latents_shape:
                   raise ValueError(f'Unexpected latents shape, got {latents.shape}, expected {latents_shape}'     )
              __lowerCamelCase       :     Union[str, Any]        =		latents.to(self.device     )

         # set timesteps
         __lowerCamelCase       :     Any        =		'offset' in set(inspect.signature(self.scheduler.set_timesteps     ).parameters.keys()     )
         __lowerCamelCase       :     Optional[int]        =		{}
         if accepts_offset:
              __lowerCamelCase       :     List[str]        =		1

         self.scheduler.set_timesteps(SCREAMING_SNAKE_CASE_      ,					**SCREAMING_SNAKE_CASE_     )

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

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

         # check if the scheduler accepts generator
         __lowerCamelCase       :     Optional[int]        =		'generator' in set(inspect.signature(self.scheduler.step     ).parameters.keys()     )
         if accepts_generator:
              __lowerCamelCase       :     List[Any]        =		generator

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

              # predict the noise residual
              __lowerCamelCase       :     Tuple        =		self.unet(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					encoder_hidden_states=SCREAMING_SNAKE_CASE_     ).sample

              # perform classifier free guidance
              if do_classifier_free_guidance:
                   __lowerCamelCase    ,	__lowerCamelCase       :     Optional[int]        =		noise_pred.chunk(2     )
                   __lowerCamelCase       :     Any        =		noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

              # perform clip guidance
              if clip_guidance_scale > 0:
                   __lowerCamelCase       :     int        =		(
                       text_embeddings.chunk(2     )[1] if do_classifier_free_guidance else text_embeddings
                   )
                   __lowerCamelCase    ,	__lowerCamelCase       :     int        =		self.cond_fn(
                       SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					)

              # compute the previous noisy sample x_t -> x_t-1
              __lowerCamelCase       :     List[Any]        =		self.scheduler.step(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					**SCREAMING_SNAKE_CASE_     ).prev_sample

         # scale and decode the image latents with vae
         __lowerCamelCase       :     Any        =		1 / self.vae.config.scaling_factor * latents
         __lowerCamelCase       :     Tuple        =		self.vae.decode(SCREAMING_SNAKE_CASE_     ).sample

         __lowerCamelCase       :     str        =		(image / 2 + 0.5).clamp(0      ,					1     )
         __lowerCamelCase       :     Optional[Any]        =		image.cpu().permute(0      ,					2      ,					3      ,					1     ).numpy()

         if output_type == "pil":
              __lowerCamelCase       :     Optional[Any]        =		self.numpy_to_pil(SCREAMING_SNAKE_CASE_     )

         if not return_dict:
              return (image, None)

         return StableDiffusionPipelineOutput(images=SCREAMING_SNAKE_CASE_      ,					nsfw_content_detected=SCREAMING_SNAKE_CASE_     )






