




from math import acos, sin
from typing import List, Tuple, Union

import numpy as np
import torch
from PIL import Image

from ...models import AutoencoderKL, UNetaDConditionModel
from ...schedulers import DDIMScheduler, DDPMScheduler
from ...utils import randn_tensor
from ..pipeline_utils import AudioPipelineOutput, BaseOutput, DiffusionPipeline, ImagePipelineOutput
from .mel import Mel




class 		__A      (			UpperCamelCase__			):
						UpperCamelCase     						=					["""vqvae"""]

						def __init__(		self   :Union[str, Any]    , __snake_case   :List[Any]    , __snake_case   :List[str]    , __snake_case   :Optional[int]    , __snake_case   :Optional[Any]    , ):
										'''simple docstring'''

										super().__init__()
										self.register_modules(unet=__snake_case    , scheduler=__snake_case    , mel=__snake_case    , vqvae=__snake_case )

						def     A__						(		self   :str ):
										'''simple docstring'''

										return 50 if isinstance(self.scheduler    , __snake_case ) else 10_00

						@torch.no_grad()
						def __call__(		self   :List[Any]    , __snake_case   :Any = 1    , __snake_case   :Optional[int] = None    , __snake_case   :int = None    , __snake_case   :str = 0    , __snake_case   :Tuple = 0    , __snake_case   :Tuple = None    , __snake_case   :int = None    , __snake_case   :Optional[int] = 0    , __snake_case   :Any = 0    , __snake_case   :str = None    , __snake_case   :Optional[int] = 0    , __snake_case   :List[str] = None    , __snake_case   :str = None    , __snake_case   :Dict=True    , ):
										'''simple docstring'''

										__magic_name__	:	str								=steps or self.get_default_steps()
										self.scheduler.set_timesteps(__snake_case )
										__magic_name__	:	List[str]								=step_generator or generator
										# For backwards compatibility
										if type(self.unet.config.sample_size ) == int:
														__magic_name__	:	Tuple								=(self.unet.config.sample_size, self.unet.config.sample_size)
										if noise is None:
														__magic_name__	:	str								=randn_tensor(
														    (
														        batch_size,
														        self.unet.config.in_channels,
														        self.unet.config.sample_size[0],
														        self.unet.config.sample_size[1],
														    )    , generator=__snake_case    , device=self.device    , )
										__magic_name__	:	List[str]								=noise
										__magic_name__	:	Dict								=None

										if audio_file is not None or raw_audio is not None:
														self.mel.load_audio(__snake_case    , __snake_case )
														__magic_name__	:	List[Any]								=self.mel.audio_slice_to_image(__snake_case )
														__magic_name__	:	Optional[int]								=np.frombuffer(input_image.tobytes()    , dtype="""uint8""" ).reshape(
														    (input_image.height, input_image.width) )
														__magic_name__	:	Tuple								=(input_image / 2_55) * 2 - 1
														__magic_name__	:	str								=torch.tensor(input_image[np.newaxis, :, :]    , dtype=torch.float ).to(self.device )

														if self.vqvae is not None:
																		__magic_name__	:	Any								=self.vqvae.encode(torch.unsqueeze(__snake_case    , 0 ) ).latent_dist.sample(
																		    generator=__snake_case )[0]
																		__magic_name__	:	Optional[Any]								=self.vqvae.config.scaling_factor * input_images

														if start_step > 0:
																		__magic_name__	:	str								=self.scheduler.add_noise(__snake_case    , __snake_case    , self.scheduler.timesteps[start_step - 1] )

														__magic_name__	:	Any								=(
														    self.unet.config.sample_size[1] * self.mel.get_sample_rate() / self.mel.x_res / self.mel.hop_length
														)
														__magic_name__	:	str								=int(mask_start_secs * pixels_per_second )
														__magic_name__	:	Dict								=int(mask_end_secs * pixels_per_second )
														__magic_name__	:	Optional[int]								=self.scheduler.add_noise(__snake_case    , __snake_case    , torch.tensor(self.scheduler.timesteps[start_step:] ) )

										for step, t in enumerate(self.progress_bar(self.scheduler.timesteps[start_step:] ) ):
														if isinstance(self.unet    , __snake_case ):
																		__magic_name__	:	Tuple								=self.unet(__snake_case    , __snake_case    , __snake_case )["""sample"""]
														else:
																		__magic_name__	:	Any								=self.unet(__snake_case    , __snake_case )["""sample"""]

														if isinstance(self.scheduler    , __snake_case ):
																		__magic_name__	:	str								=self.scheduler.step(
																		    model_output=__snake_case    , timestep=__snake_case    , sample=__snake_case    , eta=__snake_case    , generator=__snake_case    , )["""prev_sample"""]
														else:
																		__magic_name__	:	Dict								=self.scheduler.step(
																		    model_output=__snake_case    , timestep=__snake_case    , sample=__snake_case    , generator=__snake_case    , )["""prev_sample"""]

														if mask is not None:
																		if mask_start > 0:
																						__magic_name__	:	List[str]								=mask[:, step, :, :mask_start]
																		if mask_end > 0:
																						__magic_name__	:	Optional[Any]								=mask[:, step, :, -mask_end:]

										if self.vqvae is not None:
														# 0.18215 was scaling factor used in training to ensure unit variance
														__magic_name__	:	Dict								=1 / self.vqvae.config.scaling_factor * images
														__magic_name__	:	str								=self.vqvae.decode(__snake_case )["""sample"""]

										__magic_name__	:	Tuple								=(images / 2 + 0.5).clamp(0    , 1 )
										__magic_name__	:	Any								=images.cpu().permute(0    , 2    , 3    , 1 ).numpy()
										__magic_name__	:	Any								=(images * 2_55).round().astype("""uint8""" )
										__magic_name__	:	int								=list(
										    (Image.fromarray(_[:, :, 0] ) for _ in images)
										    if images.shape[3] == 1
										    else (Image.fromarray(__snake_case    , mode="""RGB""" ).convert("""L""" ) for _ in images) )

										__magic_name__	:	int								=[self.mel.image_to_audio(__snake_case ) for _ in images]
										if not return_dict:
														return images, (self.mel.get_sample_rate(), audios)

										return BaseOutput(**AudioPipelineOutput(np.array(__snake_case )[:, np.newaxis, :] )    , **ImagePipelineOutput(__snake_case ) )

						@torch.no_grad()
						def     A__						(		self   :str    , __snake_case   :str    , __snake_case   :Optional[Any] = 50 ):
										'''simple docstring'''

										assert isinstance(self.scheduler    , __snake_case )
										self.scheduler.set_timesteps(__snake_case )
										__magic_name__	:	List[str]								=np.array(
										    [np.frombuffer(image.tobytes()    , dtype="""uint8""" ).reshape((1, image.height, image.width) ) for image in images] )
										__magic_name__	:	Tuple								=(sample / 2_55) * 2 - 1
										__magic_name__	:	str								=torch.Tensor(__snake_case ).to(self.device )

										for t in self.progress_bar(torch.flip(self.scheduler.timesteps    , (0,) ) ):
														__magic_name__	:	int								=t - self.scheduler.config.num_train_timesteps // self.scheduler.num_inference_steps
														__magic_name__	:	Union[str, Any]								=self.scheduler.alphas_cumprod[t]
														__magic_name__	:	Optional[int]								=(
														    self.scheduler.alphas_cumprod[prev_timestep]
														    if prev_timestep >= 0
														    else self.scheduler.final_alpha_cumprod
														)
														__magic_name__	:	Tuple								=1 - alpha_prod_t
														__magic_name__	:	int								=self.unet(__snake_case    , __snake_case )["""sample"""]
														__magic_name__	:	List[Any]								=(1 - alpha_prod_t_prev) ** 0.5 * model_output
														__magic_name__	:	Any								=(sample - pred_sample_direction) * alpha_prod_t_prev ** (-0.5)
														__magic_name__	:	Dict								=sample * alpha_prod_t ** 0.5 + beta_prod_t ** 0.5 * model_output

										return sample


						@staticmethod
						def     A__						(		__snake_case   :Union[str, Any]    , __snake_case   :Dict    , __snake_case   :str ):
										'''simple docstring'''

										__magic_name__	:	Dict								=acos(torch.dot(torch.flatten(__snake_case )    , torch.flatten(__snake_case ) ) / torch.norm(__snake_case ) / torch.norm(__snake_case ) )
										return sin((1 - alpha) * theta ) * xa / sin(__snake_case ) + sin(alpha * theta ) * xa / sin(__snake_case )











import copy
import os
from typing import Union

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


UpperCAmelCase_		:				Union[str, Any]									=     logging.get_logger(__name__)

UpperCAmelCase_		:				Dict									=     {
    "google/pix2struct-textcaps-base": (
        "https://huggingface.co/google/pix2struct-textcaps-base/resolve/main/config.json"
    ),
}




class 		__A      (			UpperCamelCase__			):
						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   :List[Any]    , __snake_case   :Dict=5_02_44    , __snake_case   :Dict=7_68    , __snake_case   :Union[str, Any]=64    , __snake_case   :Union[str, Any]=20_48    , __snake_case   :List[Any]=12    , __snake_case   :Any=12    , __snake_case   :str=32    , __snake_case   :Dict=1_28    , __snake_case   :Optional[int]=0.1    , __snake_case   :Any=1E-6    , __snake_case   :Optional[int]=1.0    , __snake_case   :List[Any]="gelu_new"    , __snake_case   :Any=0    , __snake_case   :Tuple=False    , __snake_case   :Any=0    , __snake_case   :int=1    , __snake_case   :Dict=False    , __snake_case   :Optional[Any]=True    , **__snake_case   :Dict    , ):
										'''simple docstring'''

										__magic_name__	:	List[str]								=vocab_size
										__magic_name__	:	List[str]								=hidden_size
										__magic_name__	:	Optional[int]								=d_kv
										__magic_name__	:	Dict								=d_ff
										__magic_name__	:	Tuple								=num_layers
										__magic_name__	:	str								=num_heads
										__magic_name__	:	Optional[Any]								=relative_attention_num_buckets
										__magic_name__	:	List[Any]								=relative_attention_max_distance
										__magic_name__	:	str								=dropout_rate
										__magic_name__	:	Optional[Any]								=layer_norm_epsilon
										__magic_name__	:	Union[str, Any]								=initializer_factor
										__magic_name__	:	Tuple								=use_cache

										__magic_name__	:	Any								=eos_token_id
										__magic_name__	:	int								=decoder_start_token_id

										# for backwards compatibility
										__magic_name__	:	Union[str, Any]								=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   :Union[str, Any]    , __snake_case   :Union[str, os.PathLike]    , **__snake_case   :Any ):
										'''simple docstring'''

										cls._set_token_in_kwargs(__snake_case )

										__magic_name__				, __magic_name__	:	Optional[Any]								=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":
														__magic_name__	:	Tuple								=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 		__A      (			UpperCamelCase__			):
						UpperCamelCase     						=					"""pix2struct_vision_model"""

						def __init__(		self   :Union[str, Any]    , __snake_case   :Tuple=7_68    , __snake_case   :Any=7_68    , __snake_case   :Tuple=20_48    , __snake_case   :Dict=64    , __snake_case   :List[str]=12    , __snake_case   :str=12    , __snake_case   :str="gelu_new"    , __snake_case   :Optional[int]=1E-6    , __snake_case   :Optional[int]=0.0    , __snake_case   :Any=0.0    , __snake_case   :Any=1E-10    , __snake_case   :List[str]=1.0    , __snake_case   :Tuple=40_96    , __snake_case   :Optional[int]=32    , __snake_case   :Union[str, Any]=1_28    , **__snake_case   :Optional[int]    , ):
										'''simple docstring'''

										super().__init__(**__snake_case )

										__magic_name__	:	Union[str, Any]								=hidden_size
										__magic_name__	:	Dict								=patch_embed_hidden_size
										__magic_name__	:	Tuple								=d_ff
										__magic_name__	:	str								=dropout_rate
										__magic_name__	:	str								=num_hidden_layers
										__magic_name__	:	Dict								=num_attention_heads
										__magic_name__	:	List[str]								=initializer_range
										__magic_name__	:	Optional[int]								=initializer_factor
										__magic_name__	:	str								=attention_dropout
										__magic_name__	:	Union[str, Any]								=layer_norm_eps
										__magic_name__	:	List[str]								=dense_act_fn
										__magic_name__	:	List[str]								=seq_len
										__magic_name__	:	str								=relative_attention_num_buckets
										__magic_name__	:	Optional[int]								=relative_attention_max_distance
										__magic_name__	:	List[str]								=d_kv


						@classmethod
						def     A__						(		cls   :Optional[Any]    , __snake_case   :Union[str, os.PathLike]    , **__snake_case   :Any ):
										'''simple docstring'''

										cls._set_token_in_kwargs(__snake_case )

										__magic_name__				, __magic_name__	:	Optional[int]								=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":
														__magic_name__	:	int								=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 		__A      (			UpperCamelCase__			):
						UpperCamelCase     						=					"""pix2struct"""
						UpperCamelCase     						=					True

						def __init__(		self   :Union[str, Any]    , __snake_case   :Optional[int]=None    , __snake_case   :List[Any]=None    , __snake_case   :Optional[Any]=1.0    , __snake_case   :Tuple=0.02    , __snake_case   :str=False    , __snake_case   :List[str]=False    , __snake_case   :str=True    , **__snake_case   :List[str]    , ):
										'''simple docstring'''

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

										if text_config is None:
														__magic_name__	:	int								={}
														logger.info("""text_config is None. Initializing the Pix2StructTextConfig with default values.""" )

										if vision_config is None:
														__magic_name__	:	str								={}
														logger.info("""vision_config is None. Initializing the Pix2StructVisionConfig with default values.""" )

										__magic_name__	:	Union[str, Any]								=PixaStructTextConfig(**__snake_case )
										__magic_name__	:	str								=PixaStructVisionConfig(**__snake_case )

										__magic_name__	:	int								=self.text_config.decoder_start_token_id
										__magic_name__	:	Optional[Any]								=self.text_config.pad_token_id
										__magic_name__	:	str								=self.text_config.eos_token_id

										__magic_name__	:	int								=initializer_factor
										__magic_name__	:	List[Any]								=initializer_range

										__magic_name__	:	Dict								=self.initializer_range
										__magic_name__	:	Union[str, Any]								=self.initializer_range

										__magic_name__	:	Tuple								=is_vqa

						@classmethod
						def     A__						(		cls   :Optional[int]    , __snake_case   :PixaStructTextConfig    , __snake_case   :PixaStructVisionConfig    , **__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   :Dict ):
										'''simple docstring'''

										__magic_name__	:	Union[str, Any]								=copy.deepcopy(self.__dict__ )
										__magic_name__	:	Any								=self.text_config.to_dict()
										__magic_name__	:	List[str]								=self.vision_config.to_dict()
										__magic_name__	:	Optional[int]								=self.__class__.model_type
										return output




