
"""simple docstring"""





import importlib
import inspect
import json
import os
from collections import OrderedDict

# Build the list of all feature extractors
from ...configuration_utils import PretrainedConfig
from ...dynamic_module_utils import get_class_from_dynamic_module, resolve_trust_remote_code
from ...feature_extraction_utils import FeatureExtractionMixin
from ...image_processing_utils import ImageProcessingMixin
from ...tokenization_utils import TOKENIZER_CONFIG_FILE
from ...utils import FEATURE_EXTRACTOR_NAME, get_file_from_repo, logging
from .auto_factory import _LazyAutoMapping
from .configuration_auto import (
    CONFIG_MAPPING_NAMES,
    AutoConfig,
    model_type_to_module_name,
    replace_list_option_in_docstrings,
)
from .feature_extraction_auto import AutoFeatureExtractor
from .image_processing_auto import AutoImageProcessor
from .tokenization_auto import AutoTokenizer


_a									=				logging.get_logger(__name__)

_a									=				OrderedDict(
    [
        ('align', 'AlignProcessor'),
        ('altclip', 'AltCLIPProcessor'),
        ('bark', 'BarkProcessor'),
        ('blip', 'BlipProcessor'),
        ('blip-2', 'Blip2Processor'),
        ('bridgetower', 'BridgeTowerProcessor'),
        ('chinese_clip', 'ChineseCLIPProcessor'),
        ('clap', 'ClapProcessor'),
        ('clip', 'CLIPProcessor'),
        ('clipseg', 'CLIPSegProcessor'),
        ('flava', 'FlavaProcessor'),
        ('git', 'GitProcessor'),
        ('groupvit', 'CLIPProcessor'),
        ('hubert', 'Wav2Vec2Processor'),
        ('instructblip', 'InstructBlipProcessor'),
        ('layoutlmv2', 'LayoutLMv2Processor'),
        ('layoutlmv3', 'LayoutLMv3Processor'),
        ('markuplm', 'MarkupLMProcessor'),
        ('mctct', 'MCTCTProcessor'),
        ('mgp-str', 'MgpstrProcessor'),
        ('oneformer', 'OneFormerProcessor'),
        ('owlvit', 'OwlViTProcessor'),
        ('pix2struct', 'Pix2StructProcessor'),
        ('sam', 'SamProcessor'),
        ('sew', 'Wav2Vec2Processor'),
        ('sew-d', 'Wav2Vec2Processor'),
        ('speech_to_text', 'Speech2TextProcessor'),
        ('speech_to_text_2', 'Speech2Text2Processor'),
        ('speecht5', 'SpeechT5Processor'),
        ('trocr', 'TrOCRProcessor'),
        ('tvlt', 'TvltProcessor'),
        ('unispeech', 'Wav2Vec2Processor'),
        ('unispeech-sat', 'Wav2Vec2Processor'),
        ('vilt', 'ViltProcessor'),
        ('vision-text-dual-encoder', 'VisionTextDualEncoderProcessor'),
        ('wav2vec2', 'Wav2Vec2Processor'),
        ('wav2vec2-conformer', 'Wav2Vec2Processor'),
        ('wavlm', 'Wav2Vec2Processor'),
        ('whisper', 'WhisperProcessor'),
        ('xclip', 'XCLIPProcessor'),
    ]
)

_a									=				_LazyAutoMapping(CONFIG_MAPPING_NAMES, PROCESSOR_MAPPING_NAMES)







def       __a	(						__lowerCamelCase    ):
   for module_name, processors in PROCESSOR_MAPPING_NAMES.items():
      if class_name in processors:
         UpperCAmelCase_   :   Optional[Any]     	=       model_type_to_module_name(__lowerCamelCase    )

         UpperCAmelCase_   :   Tuple     	=       importlib.import_module(f""".{module_name}""",						"transformers.models"    )
         try:
            return getattr(__lowerCamelCase,						__lowerCamelCase    )
         except AttributeError:
            continue

   for processor in PROCESSOR_MAPPING._extra_content.values():
      if getattr(__lowerCamelCase,						"__name__",						__lowerCamelCase    ) == class_name:
         return processor

    # We did not fine the class, but maybe it's because a dep is missing. In that case, the class will be in the main
    # init and we return the proper dummy to get an appropriate error message.
   UpperCAmelCase_   :   str     	=       importlib.import_module("transformers"    )
   if hasattr(__lowerCamelCase,						__lowerCamelCase    ):
      return getattr(__lowerCamelCase,						__lowerCamelCase    )

   return None


class A_      :




  '''simple docstring'''







  def __init__(						self    ):

     """simple docstring"""
     raise EnvironmentError(
         "AutoProcessor is designed to be instantiated "
         "using the `AutoProcessor.from_pretrained(pretrained_model_name_or_path)` method."    )







  @classmethod
  @replace_list_option_in_docstrings(_lowercase    )
  def 			UpperCamelCase__   (						cls							,  lowercase_							,  **lowercase_    ):

     """simple docstring"""
     UpperCAmelCase_   :   str     	=       kwargs.pop("config"							,  _lowercase    )
     UpperCAmelCase_   :   Union[str, Any]     	=       kwargs.pop("trust_remote_code"							,  _lowercase    )
     UpperCAmelCase_   :   int     	=       True

     UpperCAmelCase_   :   Union[str, Any]     	=       None
     UpperCAmelCase_   :   Any     	=       None

     # First, let's see if we have a preprocessor config.
     # Filter the kwargs for `get_file_from_repo`.
     UpperCAmelCase_   :   Any     	=       {
         key: kwargs[key] for key in inspect.signature(_lowercase    ).parameters.keys() if key in kwargs
     }
     # Let's start by checking whether the processor class is saved in an image processor
     UpperCAmelCase_   :   List[str]     	=       get_file_from_repo(
         _lowercase							,  _lowercase							,  **_lowercase    )
     if preprocessor_config_file is not None:
        UpperCAmelCase_   :   Any     	=       ImageProcessingMixin.get_image_processor_dict(_lowercase							,  **_lowercase    )
        UpperCAmelCase_   :   Tuple     	=       config_dict.get("processor_class"							,  _lowercase    )
        if "AutoProcessor" in config_dict.get("auto_map"							,  {}    ):
           UpperCAmelCase_   :   Any     	=       config_dict['auto_map']['AutoProcessor']

        # If not found, let's check whether the processor class is saved in a feature extractor config
     if preprocessor_config_file is not None and processor_class is None:
        UpperCAmelCase_   :   Dict     	=       FeatureExtractionMixin.get_feature_extractor_dict(_lowercase							,  **_lowercase    )
        UpperCAmelCase_   :   List[str]     	=       config_dict.get("processor_class"							,  _lowercase    )
        if "AutoProcessor" in config_dict.get("auto_map"							,  {}    ):
           UpperCAmelCase_   :   int     	=       config_dict['auto_map']['AutoProcessor']

     if processor_class is None:
        # Next, let's check whether the processor class is saved in a tokenizer
        UpperCAmelCase_   :   Dict     	=       get_file_from_repo(
            _lowercase							,  _lowercase							,  **_lowercase    )
        if tokenizer_config_file is not None:
           with open(_lowercase							,  encoding="utf-8"    ) as reader:
              UpperCAmelCase_   :   int     	=       json.load(_lowercase    )

           UpperCAmelCase_   :   Dict     	=       config_dict.get("processor_class"							,  _lowercase    )
           if "AutoProcessor" in config_dict.get("auto_map"							,  {}    ):
              UpperCAmelCase_   :   Optional[int]     	=       config_dict['auto_map']['AutoProcessor']

     if processor_class is None:
        # Otherwise, load config, if it can be loaded.
        if not isinstance(_lowercase							,  _lowercase    ):
           UpperCAmelCase_   :   Optional[Any]     	=       AutoConfig.from_pretrained(
               _lowercase							,  trust_remote_code=_lowercase							,  **_lowercase    )

        # And check if the config contains the processor class.
        UpperCAmelCase_   :   int     	=       getattr(_lowercase							,  "processor_class"							,  _lowercase    )
        if hasattr(_lowercase							,  "auto_map"    ) and "AutoProcessor" in config.auto_map:
           UpperCAmelCase_   :   List[str]     	=       config.auto_map['AutoProcessor']

     if processor_class is not None:
        UpperCAmelCase_   :   Tuple     	=       processor_class_from_name(_lowercase    )

     UpperCAmelCase_   :   List[str]     	=       processor_auto_map is not None
     UpperCAmelCase_   :   Union[str, Any]     	=       processor_class is not None or type(_lowercase    ) in PROCESSOR_MAPPING
     UpperCAmelCase_   :   int     	=       resolve_trust_remote_code(
         _lowercase							,  _lowercase							,  _lowercase							,  _lowercase    )

     if has_remote_code and trust_remote_code:
        UpperCAmelCase_   :   str     	=       get_class_from_dynamic_module(
            _lowercase							,  _lowercase							,  **_lowercase    )
        UpperCAmelCase_   :   Dict     	=       kwargs.pop("code_revision"							,  _lowercase    )
        if os.path.isdir(_lowercase    ):
           processor_class.register_for_auto_class()
        return processor_class.from_pretrained(
            _lowercase							,  trust_remote_code=_lowercase							,  **_lowercase    )
     elif processor_class is not None:
        return processor_class.from_pretrained(
            _lowercase							,  trust_remote_code=_lowercase							,  **_lowercase    )
     # Last try: we use the PROCESSOR_MAPPING.
     elif type(_lowercase    ) in PROCESSOR_MAPPING:
        return PROCESSOR_MAPPING[type(_lowercase    )].from_pretrained(_lowercase							,  **_lowercase    )

     # At this stage, there doesn't seem to be a `Processor` class available for this model, so let's try a
     # tokenizer.
     try:
        return AutoTokenizer.from_pretrained(
            _lowercase							,  trust_remote_code=_lowercase							,  **_lowercase    )
     except Exception:
        try:
           return AutoImageProcessor.from_pretrained(
               _lowercase							,  trust_remote_code=_lowercase							,  **_lowercase    )
        except Exception:
           pass

        try:
           return AutoFeatureExtractor.from_pretrained(
               _lowercase							,  trust_remote_code=_lowercase							,  **_lowercase    )
        except Exception:
           pass

     raise ValueError(
         F"""Unrecognized processing class in {pretrained_model_name_or_path}. Can't instantiate a processor, a """
         "tokenizer, an image processor or a feature extractor for this model. Make sure the repository contains"
         "the files of at least one of those processing classes."    )







  @staticmethod
  def 			UpperCamelCase__   (						lowercase_							,  lowercase_    ):

     """simple docstring"""
     PROCESSOR_MAPPING.register(_lowercase							,  _lowercase    )










'''simple docstring'''

import math
from dataclasses import dataclass
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, logging, randn_tensor
from .scheduling_utils import KarrasDiffusionSchedulers, SchedulerMixin


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






@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->EulerAncestralDiscrete
class 				__magic_name__  (						UpperCAmelCase_					):







	"""simple docstring"""



	SCREAMING_SNAKE_CASE_       :   torch.FloatTensor
	SCREAMING_SNAKE_CASE_       :   Optional[torch.FloatTensor]   =	None






def 						_UpperCAmelCase     (	a		:						Any    ,      a		:						Dict=0.9_99    ,      a		:						Optional[int]="cosine"    ,      )      ->      Tuple:





						"""simple docstring"""
						if alpha_transform_type == "cosine":

												def alpha_bar_fn(a		:						Optional[int]				):
																		return math.cos((t + 0.0_08) / 1.0_08 * math.pi / 2				) ** 2

						elif alpha_transform_type == "exp":

												def alpha_bar_fn(a		:						Optional[Any]				):
																		return math.exp(t * -12.0				)

						else:
												raise ValueError(f"Unsupported alpha_tranform_type: {alpha_transform_type}"				)

						lowercase_   : Optional[int]   	=      []
						for i in range(a				):
												lowercase_   : List[Any]   	=      i / num_diffusion_timesteps
												lowercase_   : Union[str, Any]   	=      (i + 1) / num_diffusion_timesteps
												betas.append(min(1 - alpha_bar_fn(a				) / alpha_bar_fn(a				)    ,      a				)				)
						return torch.tensor(a    ,      dtype=torch.floataa				)






class 				__magic_name__  (						UpperCAmelCase_,    UpperCAmelCase_					):







	"""simple docstring"""



	SCREAMING_SNAKE_CASE_       :   Union[str, Any]   =	[e.name for e in KarrasDiffusionSchedulers]
	SCREAMING_SNAKE_CASE_       :   Tuple   =	1
	@register_to_config
	def __init__(				self	,   _lowercase = 1000	,   _lowercase = 0.00_01	,   _lowercase = 0.02	,   _lowercase = "linear"	,   _lowercase = None	,   _lowercase = "epsilon"	,   _lowercase = "linspace"	,   _lowercase = 0	,   )    ->	List[str]:
							if trained_betas is not None:
													lowercase_   : Any   	=      torch.tensor(_lowercase	,   dtype=torch.floataa							)
							elif beta_schedule == "linear":
													lowercase_   : Optional[int]   	=      torch.linspace(_lowercase	,   _lowercase	,   _lowercase	,   dtype=torch.floataa							)
							elif beta_schedule == "scaled_linear":
													# this schedule is very specific to the latent diffusion model.
													lowercase_   : Dict   	=      (
													    torch.linspace(beta_start**0.5	,   beta_end**0.5	,   _lowercase	,   dtype=torch.floataa							) ** 2
													)
							elif beta_schedule == "squaredcos_cap_v2":
													# Glide cosine schedule
													lowercase_   : Any   	=      betas_for_alpha_bar(_lowercase							)
							else:
													raise NotImplementedError(f"{beta_schedule} does is not implemented for {self.__class__}"							)

							lowercase_   : Dict   	=      1.0 - self.betas
							lowercase_   : int   	=      torch.cumprod(self.alphas	,   dim=0							)

							lowercase_   : Optional[int]   	=      np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5							)
							lowercase_   : str   	=      np.concatenate([sigmas[::-1], [0.0]]							).astype(np.floataa							)
							lowercase_   : Tuple   	=      torch.from_numpy(_lowercase							)

							# setable values
							lowercase_   : Union[str, Any]   	=      None
							lowercase_   : Optional[Any]   	=      np.linspace(0	,   num_train_timesteps - 1	,   _lowercase	,   dtype=_lowercase							)[::-1].copy()
							lowercase_   : Dict   	=      torch.from_numpy(_lowercase							)
							lowercase_   : int   	=      False
	@property
	def 							lowerCamelCase__		(				self							)    ->	Union[str, Any]:
							# standard deviation of the initial noise distribution
							if self.config.timestep_spacing in ["linspace", "trailing"]:
													return self.sigmas.max()

							return (self.sigmas.max() ** 2 + 1) ** 0.5
	def 							lowerCamelCase__		(				self	,   _lowercase	,   _lowercase							)    ->	torch.FloatTensor:
							if isinstance(_lowercase	,   torch.Tensor							):
													lowercase_   : List[str]   	=      timestep.to(self.timesteps.device							)
							lowercase_   : List[str]   	=      (self.timesteps == timestep).nonzero().item()
							lowercase_   : Any   	=      self.sigmas[step_index]
							lowercase_   : Tuple   	=      sample / ((sigma**2 + 1) ** 0.5)
							lowercase_   : List[str]   	=      True
							return sample
	def 							lowerCamelCase__		(				self	,   _lowercase	,   _lowercase = None							)    ->	Tuple:
							lowercase_   : Any   	=      num_inference_steps

							# "linspace", "leading", "trailing" corresponds to annotation of Table 2. of https://arxiv.org/abs/2305.08891
							if self.config.timestep_spacing == "linspace":
													lowercase_   : Tuple   	=      np.linspace(0	,   self.config.num_train_timesteps - 1	,   _lowercase	,   dtype=_lowercase							)[
													    ::-1
													].copy()
							elif self.config.timestep_spacing == "leading":
													lowercase_   : int   	=      self.config.num_train_timesteps // self.num_inference_steps
													# creates integer timesteps by multiplying by ratio
													# casting to int to avoid issues when num_inference_step is power of 3
													lowercase_   : Optional[int]   	=      (np.arange(0	,   _lowercase							) * step_ratio).round()[::-1].copy().astype(_lowercase							)
													timesteps += self.config.steps_offset
							elif self.config.timestep_spacing == "trailing":
													lowercase_   : Optional[Any]   	=      self.config.num_train_timesteps / self.num_inference_steps
													# creates integer timesteps by multiplying by ratio
													# casting to int to avoid issues when num_inference_step is power of 3
													lowercase_   : Dict   	=      (np.arange(self.config.num_train_timesteps	,   0	,   -step_ratio							)).round().copy().astype(_lowercase							)
													timesteps -= 1
							else:
													raise ValueError(
													    f"{self.config.timestep_spacing} is not supported. Please make sure to choose one of 'linspace', 'leading' or 'trailing'."							)

							lowercase_   : str   	=      np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5							)
							lowercase_   : Dict   	=      np.interp(_lowercase	,   np.arange(0	,   len(_lowercase							)							)	,   _lowercase							)
							lowercase_   : Any   	=      np.concatenate([sigmas, [0.0]]							).astype(np.floataa							)
							lowercase_   : Dict   	=      torch.from_numpy(_lowercase							).to(device=_lowercase							)
							if str(_lowercase							).startswith('mps'							):
													# mps does not support float64
													lowercase_   : Union[str, Any]   	=      torch.from_numpy(_lowercase							).to(_lowercase	,   dtype=torch.floataa							)
							else:
													lowercase_   : Optional[int]   	=      torch.from_numpy(_lowercase							).to(device=_lowercase							)
	def 							lowerCamelCase__		(				self	,   _lowercase	,   _lowercase	,   _lowercase	,   _lowercase = None	,   _lowercase = True	,   )    ->	Union[EulerAncestralDiscreteSchedulerOutput, Tuple]:

							if (
							    isinstance(_lowercase	,   _lowercase							)
							    or isinstance(_lowercase	,   torch.IntTensor							)
							    or isinstance(_lowercase	,   torch.LongTensor							)
							):
													raise ValueError(
													    (
													        'Passing integer indices (e.g. from `enumerate(timesteps)`) as timesteps to'
													        ' `EulerDiscreteScheduler.step()` is not supported. Make sure to pass'
													        ' one of the `scheduler.timesteps` as a timestep.'
													    )	,   )

							if not self.is_scale_input_called:
													logger.warning(
													    'The `scale_model_input` function should be called before `step` to ensure correct denoising. '
													    'See `StableDiffusionPipeline` for a usage example.'							)

							if isinstance(_lowercase	,   torch.Tensor							):
													lowercase_   : Optional[int]   	=      timestep.to(self.timesteps.device							)

							lowercase_   : List[Any]   	=      (self.timesteps == timestep).nonzero().item()
							lowercase_   : List[str]   	=      self.sigmas[step_index]

							# 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
							if self.config.prediction_type == "epsilon":
													lowercase_   : List[str]   	=      sample - sigma * model_output
							elif self.config.prediction_type == "v_prediction":
													# * c_out + input * c_skip
													lowercase_   : int   	=      model_output * (-sigma / (sigma**2 + 1) ** 0.5) + (sample / (sigma**2 + 1))
							elif self.config.prediction_type == "sample":
													raise NotImplementedError('prediction_type not implemented yet: sample'							)
							else:
													raise ValueError(
													    f"prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`"							)

							lowercase_   : Tuple   	=      self.sigmas[step_index]
							lowercase_   : str   	=      self.sigmas[step_index + 1]
							lowercase_   : Dict   	=      (sigma_to**2 * (sigma_from**2 - sigma_to**2) / sigma_from**2) ** 0.5
							lowercase_   : Optional[Any]   	=      (sigma_to**2 - sigma_up**2) ** 0.5

							# 2. Convert to an ODE derivative
							lowercase_   : Any   	=      (sample - pred_original_sample) / sigma

							lowercase_   : Tuple   	=      sigma_down - sigma

							lowercase_   : str   	=      sample + derivative * dt

							lowercase_   : Optional[int]   	=      model_output.device
							lowercase_   : Union[str, Any]   	=      randn_tensor(model_output.shape	,   dtype=model_output.dtype	,   device=_lowercase	,   generator=_lowercase							)

							lowercase_   : int   	=      prev_sample + noise * sigma_up

							if not return_dict:
													return (prev_sample,)

							return EulerAncestralDiscreteSchedulerOutput(
							    prev_sample=_lowercase	,   pred_original_sample=_lowercase							)
	def 							lowerCamelCase__		(				self	,   _lowercase	,   _lowercase	,   _lowercase	,   )    ->	torch.FloatTensor:
							# Make sure sigmas and timesteps have the same device and dtype as original_samples
							lowercase_   : List[str]   	=      self.sigmas.to(device=original_samples.device	,   dtype=original_samples.dtype							)
							if original_samples.device.type == "mps" and torch.is_floating_point(_lowercase							):
													# mps does not support float64
													lowercase_   : Dict   	=      self.timesteps.to(original_samples.device	,   dtype=torch.floataa							)
													lowercase_   : List[str]   	=      timesteps.to(original_samples.device	,   dtype=torch.floataa							)
							else:
													lowercase_   : List[Any]   	=      self.timesteps.to(original_samples.device							)
													lowercase_   : List[Any]   	=      timesteps.to(original_samples.device							)

							lowercase_   : Optional[Any]   	=      [(schedule_timesteps == t).nonzero().item() for t in timesteps]

							lowercase_   : Any   	=      sigmas[step_indices].flatten()
							while len(sigma.shape							) < len(original_samples.shape							):
													lowercase_   : Optional[int]   	=      sigma.unsqueeze(-1							)

							lowercase_   : str   	=      original_samples + noise * sigma
							return noisy_samples







	def __len__(				self							)    ->	Dict:
							return self.config.num_train_timesteps
