


# DISCLAIMER: This code is strongly influenced by https://github.com/pesser/pytorch_diffusion
# and https://github.com/hojonathanho/diffusion
import math
from dataclasses import dataclass
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from diffusers.configuration_utils import ConfigMixin, register_to_config
from diffusers.schedulers.scheduling_utils import SchedulerMixin
from diffusers.utils import BaseOutput, deprecate
@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->DDIM
class 						lowerCamelCase_							(					lowerCAmelCase__					):



					'''simple docstring'''







					__UpperCAmelCase							  =					42
					__UpperCAmelCase							  =					None




def 		lowercase_	(  _UpperCamelCase							,      _UpperCamelCase=0.999							,      _UpperCamelCase="cosine"							,      ):


			'''simple docstring'''







			if alpha_transform_type == "cosine":

						def alpha_bar_fn(_UpperCamelCase	):
									return math.cos((t + 0.008) / 1.008 * math.pi / 2	) ** 2

			elif alpha_transform_type == "exp":

						def alpha_bar_fn(_UpperCamelCase	):
									return math.exp(t * -12.0	)

			else:
						raise ValueError(F'Unsupported alpha_tranform_type: {alpha_transform_type}'	)

			__lowercase			       = []
			for i in range(_UpperCamelCase	):
						__lowercase			       = i / num_diffusion_timesteps
						__lowercase			       = (i + 1) / num_diffusion_timesteps
						betas.append(min(1 - alpha_bar_fn(_UpperCamelCase	) / alpha_bar_fn(_UpperCamelCase	)							,      _UpperCamelCase	)	)
			return torch.tensor(_UpperCamelCase							,      dtype=torch.floataa	)





class 						lowerCamelCase_							(					lowerCAmelCase__  ,  lowerCAmelCase__					):



					'''simple docstring'''







					__UpperCAmelCase							  =					1






					@register_to_config
					def __init__(					self			,							snake_case_ = 1_0_0_0			,							snake_case_ = 0.0_0_0_1			,							snake_case_ = 0.0_2			,							snake_case_ = "linear"			,							snake_case_ = None			,							snake_case_ = True			,							snake_case_ = True			,							snake_case_ = 0			,							snake_case_ = "epsilon"			,							snake_case_ = 1.0			,							**snake_case_			,							) ->      Optional[Any]:







								'''simple docstring'''

								if kwargs.get('''set_alpha_to_one'''			,							snake_case_    ) is not None:
											__lowercase			       = (
											    '''The `set_alpha_to_one` argument is deprecated. Please use `set_alpha_to_zero` instead.'''
											)
											deprecate('''set_alpha_to_one'''			,							'''1.0.0'''			,							snake_case_			,							standard_warn=snake_case_    )
											__lowercase			       = kwargs['''set_alpha_to_one''']
								if trained_betas is not None:
											__lowercase			       = torch.tensor(snake_case_			,							dtype=torch.floataa    )
								elif beta_schedule == "linear":
											__lowercase			       = torch.linspace(snake_case_			,							snake_case_			,							snake_case_			,							dtype=torch.floataa    )
								elif beta_schedule == "scaled_linear":
											# this schedule is very specific to the latent diffusion model.
											__lowercase			       = (
											    torch.linspace(beta_start**0.5			,							beta_end**0.5			,							snake_case_			,							dtype=torch.floataa    ) ** 2
											)
								elif beta_schedule == "squaredcos_cap_v2":
											# Glide cosine schedule
											__lowercase			       = betas_for_alpha_bar(snake_case_    )
								else:
											raise NotImplementedError(F'{beta_schedule} does is not implemented for {self.__class__}'    )

								__lowercase			       = 1.0 - self.betas
								__lowercase			       = torch.cumprod(self.alphas			,							dim=0    )

								# At every step in inverted ddim, we are looking into the next alphas_cumprod
								# For the final step, there is no next alphas_cumprod, and the index is out of bounds
								# `set_alpha_to_zero` decides whether we set this parameter simply to zero
								# in this case, self.step() just output the predicted noise
								# or whether we use the final alpha of the "non-previous" one.
								__lowercase			       = torch.tensor(0.0    ) if set_alpha_to_zero else self.alphas_cumprod[-1]

								# standard deviation of the initial noise distribution
								__lowercase			       = 1.0

								# setable values
								__lowercase			       = None
								__lowercase			       = torch.from_numpy(np.arange(0			,							snake_case_    ).copy().astype(np.intaa    )    )






					def A   (					self			,							snake_case_			,							snake_case_ = None    ) ->      torch.FloatTensor:







								'''simple docstring'''

								return sample






					def A   (					self			,							snake_case_			,							snake_case_ = None    ) ->      Optional[Any]:







								'''simple docstring'''

								if num_inference_steps > self.config.num_train_timesteps:
											raise ValueError(
											    F'`num_inference_steps`: {num_inference_steps} cannot be larger than `self.config.train_timesteps`:'
											    F' {self.config.num_train_timesteps} as the unet model trained with this scheduler can only handle'
											    F' maximal {self.config.num_train_timesteps} timesteps.'    )

								__lowercase			       = num_inference_steps
								__lowercase			       = 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			       = (np.arange(0			,							snake_case_    ) * step_ratio).round().copy().astype(np.intaa    )
								__lowercase			       = torch.from_numpy(snake_case_    ).to(snake_case_    )
								self.timesteps += self.config.steps_offset






					def A   (					self			,							snake_case_			,							snake_case_			,							snake_case_			,							snake_case_ = 0.0			,							snake_case_ = False			,							snake_case_ = None			,							snake_case_ = True			,							) ->      Union[DDIMSchedulerOutput, Tuple]:







								'''simple docstring'''

								__lowercase			       = timestep + self.config.num_train_timesteps // self.num_inference_steps

								# 2. compute alphas, betas
								# change original implementation to exactly match noise levels for analogous forward process
								__lowercase			       = self.alphas_cumprod[timestep]
								__lowercase			       = (
								    self.alphas_cumprod[prev_timestep]
								    if prev_timestep < self.config.num_train_timesteps
								    else self.final_alpha_cumprod
								)

								__lowercase			       = 1 - alpha_prod_t

								# 3. compute predicted original sample from predicted noise also called
								# "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
								if self.config.prediction_type == "epsilon":
											__lowercase			       = (sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
											__lowercase			       = model_output
								elif self.config.prediction_type == "sample":
											__lowercase			       = model_output
											__lowercase			       = (sample - alpha_prod_t ** 0.5 * pred_original_sample) / beta_prod_t ** 0.5
								elif self.config.prediction_type == "v_prediction":
											__lowercase			       = (alpha_prod_t**0.5) * sample - (beta_prod_t**0.5) * model_output
											__lowercase			       = (alpha_prod_t**0.5) * model_output + (beta_prod_t**0.5) * sample
								else:
											raise ValueError(
											    F'prediction_type given as {self.config.prediction_type} must be one of `epsilon`, `sample`, or'
											    ''' `v_prediction`'''    )

								# 4. Clip or threshold "predicted x_0"
								if self.config.clip_sample:
											__lowercase			       = pred_original_sample.clamp(
											    -self.config.clip_sample_range			,							self.config.clip_sample_range    )

								# 5. compute "direction pointing to x_t" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
								__lowercase			       = (1 - alpha_prod_t_prev) ** 0.5 * pred_epsilon

								# 6. compute x_t without "random noise" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
								__lowercase			       = alpha_prod_t_prev ** 0.5 * pred_original_sample + pred_sample_direction

								if not return_dict:
											return (prev_sample, pred_original_sample)
								return DDIMSchedulerOutput(prev_sample=snake_case_			,							pred_original_sample=snake_case_    )






					def __len__(					self    ) ->      Any:







								'''simple docstring'''

								return self.config.num_train_timesteps






import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...utils import logging


a			:			Optional[int]												=					logging.get_logger(__name__)

a			:			Dict												=					{
    '''google/pix2struct-textcaps-base''': (
        '''https://huggingface.co/google/pix2struct-textcaps-base/resolve/main/config.json'''
    ),
}
class 						lowerCamelCase_							(					lowerCAmelCase__					):



					'''simple docstring'''







					__UpperCAmelCase							  =					"pix2struct_text_model"
					__UpperCAmelCase							  =					["past_key_values"]
					__UpperCAmelCase							  =					{
					    "hidden_size": "hidden_size",
					    "num_attention_heads": "num_heads",
					    "num_hidden_layers": "num_layers",
					}






					def __init__(					self			,							snake_case_=5_0_2_4_4			,							snake_case_=7_6_8			,							snake_case_=6_4			,							snake_case_=2_0_4_8			,							snake_case_=1_2			,							snake_case_=1_2			,							snake_case_=3_2			,							snake_case_=1_2_8			,							snake_case_=0.1			,							snake_case_=1e-6			,							snake_case_=1.0			,							snake_case_="gelu_new"			,							snake_case_=0			,							snake_case_=False			,							snake_case_=0			,							snake_case_=1			,							snake_case_=False			,							snake_case_=True			,							**snake_case_			,							) ->      Optional[int]:







								'''simple docstring'''

								__lowercase			       = vocab_size
								__lowercase			       = hidden_size
								__lowercase			       = d_kv
								__lowercase			       = d_ff
								__lowercase			       = num_layers
								__lowercase			       = num_heads
								__lowercase			       = relative_attention_num_buckets
								__lowercase			       = relative_attention_max_distance
								__lowercase			       = dropout_rate
								__lowercase			       = layer_norm_epsilon
								__lowercase			       = initializer_factor
								__lowercase			       = use_cache

								__lowercase			       = eos_token_id
								__lowercase			       = decoder_start_token_id

								# for backwards compatibility
								__lowercase			       = dense_act_fn

								super().__init__(
								    pad_token_id=snake_case_			,							eos_token_id=snake_case_			,							decoder_start_token_id=snake_case_			,							tie_word_embeddings=snake_case_			,							is_decoder=snake_case_			,							**snake_case_			,							)






					@classmethod
					def A   (					cls			,							snake_case_			,							**snake_case_    ) ->      "PretrainedConfig":







								'''simple docstring'''

								cls._set_token_in_kwargs(snake_case_    )

								__lowercase       ,			__lowercase			       = cls.get_config_dict(snake_case_			,							**snake_case_    )

								# get the text config dict if we are loading from Pix2StructConfig
								if config_dict.get('''model_type'''    ) == "pix2struct":
											__lowercase			       = config_dict['''text_config''']

								if "model_type" in config_dict and hasattr(cls			,							'''model_type'''    ) and config_dict["model_type"] != cls.model_type:
											logger.warning(
											    F'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
											    F'{cls.model_type}. This is not supported for all configurations of models and can yield errors.'    )

								return cls.from_dict(snake_case_			,							**snake_case_    )




class 						lowerCamelCase_							(					lowerCAmelCase__					):



					'''simple docstring'''







					__UpperCAmelCase							  =					"pix2struct_vision_model"






					def __init__(					self			,							snake_case_=7_6_8			,							snake_case_=7_6_8			,							snake_case_=2_0_4_8			,							snake_case_=6_4			,							snake_case_=1_2			,							snake_case_=1_2			,							snake_case_="gelu_new"			,							snake_case_=1e-6			,							snake_case_=0.0			,							snake_case_=0.0			,							snake_case_=1e-1_0			,							snake_case_=1.0			,							snake_case_=4_0_9_6			,							snake_case_=3_2			,							snake_case_=1_2_8			,							**snake_case_			,							) ->      Union[str, Any]:







								'''simple docstring'''

								super().__init__(**snake_case_    )

								__lowercase			       = hidden_size
								__lowercase			       = patch_embed_hidden_size
								__lowercase			       = d_ff
								__lowercase			       = dropout_rate
								__lowercase			       = num_hidden_layers
								__lowercase			       = num_attention_heads
								__lowercase			       = initializer_range
								__lowercase			       = initializer_factor
								__lowercase			       = attention_dropout
								__lowercase			       = layer_norm_eps
								__lowercase			       = dense_act_fn
								__lowercase			       = seq_len
								__lowercase			       = relative_attention_num_buckets
								__lowercase			       = relative_attention_max_distance
								__lowercase			       = d_kv






					@classmethod
					def A   (					cls			,							snake_case_			,							**snake_case_    ) ->      "PretrainedConfig":







								'''simple docstring'''

								cls._set_token_in_kwargs(snake_case_    )

								__lowercase       ,			__lowercase			       = cls.get_config_dict(snake_case_			,							**snake_case_    )

								# get the vision config dict if we are loading from Pix2StructConfig
								if config_dict.get('''model_type'''    ) == "pix2struct":
											__lowercase			       = config_dict['''vision_config''']

								if "model_type" in config_dict and hasattr(cls			,							'''model_type'''    ) and config_dict["model_type"] != cls.model_type:
											logger.warning(
											    F'You are using a model of type {config_dict["model_type"]} to instantiate a model of type '
											    F'{cls.model_type}. This is not supported for all configurations of models and can yield errors.'    )

								return cls.from_dict(snake_case_			,							**snake_case_    )




class 						lowerCamelCase_							(					lowerCAmelCase__					):



					'''simple docstring'''







					__UpperCAmelCase							  =					"pix2struct"
					__UpperCAmelCase							  =					True






					def __init__(					self			,							snake_case_=None			,							snake_case_=None			,							snake_case_=1.0			,							snake_case_=0.0_2			,							snake_case_=False			,							snake_case_=False			,							snake_case_=True			,							**snake_case_			,							) ->      Optional[int]:







								'''simple docstring'''

								super().__init__(tie_word_embeddings=snake_case_			,							is_encoder_decoder=snake_case_			,							**snake_case_    )

								if text_config is None:
											__lowercase			       = {}
											logger.info('''text_config is None. Initializing the Pix2StructTextConfig with default values.'''    )

								if vision_config is None:
											__lowercase			       = {}
											logger.info('''vision_config is None. Initializing the Pix2StructVisionConfig with default values.'''    )

								__lowercase			       = PixaStructTextConfig(**snake_case_    )
								__lowercase			       = PixaStructVisionConfig(**snake_case_    )

								__lowercase			       = self.text_config.decoder_start_token_id
								__lowercase			       = self.text_config.pad_token_id
								__lowercase			       = self.text_config.eos_token_id

								__lowercase			       = initializer_factor
								__lowercase			       = initializer_range

								__lowercase			       = self.initializer_range
								__lowercase			       = self.initializer_range

								__lowercase			       = is_vqa






					@classmethod
					def A   (					cls			,							snake_case_			,							snake_case_			,							**snake_case_    ) ->      List[str]:







								'''simple docstring'''

								return cls(text_config=text_config.to_dict()			,							vision_config=vision_config.to_dict()			,							**snake_case_    )






					def A   (					self    ) ->      Tuple:







								'''simple docstring'''

								__lowercase			       = copy.deepcopy(self.__dict__    )
								__lowercase			       = self.text_config.to_dict()
								__lowercase			       = self.vision_config.to_dict()
								__lowercase			       = self.__class__.model_type
								return output


