






from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block
@dataclass
class  _UpperCAmelCase   (			A__      ):

				"""simple docstring"""




				lowercase__											=			42



class  _UpperCAmelCase   (			nn.Module      ):

				"""simple docstring"""




				def __init__(	self		:       Union[str, Any],    lowerCamelCase		:       Any=3,    lowerCamelCase		:       Optional[Any]=3,    lowerCamelCase		:       Dict=("DownEncoderBlock2D",),    lowerCamelCase		:       Any=(64,),    lowerCamelCase		:       Optional[int]=2,    lowerCamelCase		:       str=32,    lowerCamelCase		:       Union[str, Any]="silu",    lowerCamelCase		:       List[Any]=True,    ):

								'''simple docstring'''
								super().__init__()
								lowercase__  					=     layers_per_block

								lowercase__  					=     torch.nn.Convad(
								    lowerCamelCase,    block_out_channels[0],    kernel_size=3,    stride=1,    padding=1,    )

								lowercase__  					=     None
								lowercase__  					=     nn.ModuleList([]    )

								# down
								lowercase__  					=     block_out_channels[0]
								for i, down_block_type in enumerate(lowerCamelCase    ):
												lowercase__  					=     output_channel
												lowercase__  					=     block_out_channels[i]
												lowercase__  					=     i == len(lowerCamelCase    ) - 1

												lowercase__  					=     get_down_block(
												    lowerCamelCase,    num_layers=self.layers_per_block,    in_channels=lowerCamelCase,    out_channels=lowerCamelCase,    add_downsample=not is_final_block,    resnet_eps=1E-6,    downsample_padding=0,    resnet_act_fn=lowerCamelCase,    resnet_groups=lowerCamelCase,    attention_head_dim=lowerCamelCase,    temb_channels=lowerCamelCase,    )
												self.down_blocks.append(lowerCamelCase    )

								# mid
								lowercase__  					=     UNetMidBlockaD(
								    in_channels=block_out_channels[-1],    resnet_eps=1E-6,    resnet_act_fn=lowerCamelCase,    output_scale_factor=1,    resnet_time_scale_shift='''default''',    attention_head_dim=block_out_channels[-1],    resnet_groups=lowerCamelCase,    temb_channels=lowerCamelCase,    )

								# out
								lowercase__  					=     nn.GroupNorm(num_channels=block_out_channels[-1],    num_groups=lowerCamelCase,    eps=1E-6    )
								lowercase__  					=     nn.SiLU()

								lowercase__  					=     2 * out_channels if double_z else out_channels
								lowercase__  					=     nn.Convad(block_out_channels[-1],    lowerCamelCase,    3,    padding=1    )

								lowercase__  					=     False




				def 							lowercase__  (	self		:       Optional[Any],    lowerCamelCase		:       Optional[int]    ):

								'''simple docstring'''
								lowercase__  					=     x
								lowercase__  					=     self.conv_in(lowerCamelCase    )

								if self.training and self.gradient_checkpointing:

												def create_custom_forward(lowerCamelCase		:       Dict    ):
																def custom_forward(*lowerCamelCase		:       Optional[Any]    ):
																				return module(*lowerCamelCase    )

																return custom_forward

												# down
												if is_torch_version('''>=''',    '''1.11.0'''    ):
																for down_block in self.down_blocks:
																				lowercase__  					=     torch.utils.checkpoint.checkpoint(
																				    create_custom_forward(lowerCamelCase    ),    lowerCamelCase,    use_reentrant=lowerCamelCase    )
																# middle
																lowercase__  					=     torch.utils.checkpoint.checkpoint(
																    create_custom_forward(self.mid_block    ),    lowerCamelCase,    use_reentrant=lowerCamelCase    )
												else:
																for down_block in self.down_blocks:
																				lowercase__  					=     torch.utils.checkpoint.checkpoint(create_custom_forward(lowerCamelCase    ),    lowerCamelCase    )
																# middle
																lowercase__  					=     torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block    ),    lowerCamelCase    )

								else:
												# down
												for down_block in self.down_blocks:
																lowercase__  					=     down_block(lowerCamelCase    )

												# middle
												lowercase__  					=     self.mid_block(lowerCamelCase    )

								# post-process
								lowercase__  					=     self.conv_norm_out(lowerCamelCase    )
								lowercase__  					=     self.conv_act(lowerCamelCase    )
								lowercase__  					=     self.conv_out(lowerCamelCase    )

								return sample



class  _UpperCAmelCase   (			nn.Module      ):

				"""simple docstring"""




				def __init__(	self		:       Tuple,    lowerCamelCase		:       Optional[int]=3,    lowerCamelCase		:       Union[str, Any]=3,    lowerCamelCase		:       Union[str, Any]=("UpDecoderBlock2D",),    lowerCamelCase		:       Tuple=(64,),    lowerCamelCase		:       Union[str, Any]=2,    lowerCamelCase		:       Optional[int]=32,    lowerCamelCase		:       Optional[int]="silu",    lowerCamelCase		:       str="group",    ):

								'''simple docstring'''
								super().__init__()
								lowercase__  					=     layers_per_block

								lowercase__  					=     nn.Convad(
								    lowerCamelCase,    block_out_channels[-1],    kernel_size=3,    stride=1,    padding=1,    )

								lowercase__  					=     None
								lowercase__  					=     nn.ModuleList([]    )

								lowercase__  					=     in_channels if norm_type == '''spatial''' else None

								# mid
								lowercase__  					=     UNetMidBlockaD(
								    in_channels=block_out_channels[-1],    resnet_eps=1E-6,    resnet_act_fn=lowerCamelCase,    output_scale_factor=1,    resnet_time_scale_shift='''default''' if norm_type == '''group''' else norm_type,    attention_head_dim=block_out_channels[-1],    resnet_groups=lowerCamelCase,    temb_channels=lowerCamelCase,    )

								# up
								lowercase__  					=     list(reversed(lowerCamelCase    )    )
								lowercase__  					=     reversed_block_out_channels[0]
								for i, up_block_type in enumerate(lowerCamelCase    ):
												lowercase__  					=     output_channel
												lowercase__  					=     reversed_block_out_channels[i]

												lowercase__  					=     i == len(lowerCamelCase    ) - 1

												lowercase__  					=     get_up_block(
												    lowerCamelCase,    num_layers=self.layers_per_block + 1,    in_channels=lowerCamelCase,    out_channels=lowerCamelCase,    prev_output_channel=lowerCamelCase,    add_upsample=not is_final_block,    resnet_eps=1E-6,    resnet_act_fn=lowerCamelCase,    resnet_groups=lowerCamelCase,    attention_head_dim=lowerCamelCase,    temb_channels=lowerCamelCase,    resnet_time_scale_shift=lowerCamelCase,    )
												self.up_blocks.append(lowerCamelCase    )
												lowercase__  					=     output_channel

								# out
								if norm_type == "spatial":
												lowercase__  					=     SpatialNorm(block_out_channels[0],    lowerCamelCase    )
								else:
												lowercase__  					=     nn.GroupNorm(num_channels=block_out_channels[0],    num_groups=lowerCamelCase,    eps=1E-6    )
								lowercase__  					=     nn.SiLU()
								lowercase__  					=     nn.Convad(block_out_channels[0],    lowerCamelCase,    3,    padding=1    )

								lowercase__  					=     False




				def 							lowercase__  (	self		:       List[Any],    lowerCamelCase		:       Any,    lowerCamelCase		:       str=None    ):

								'''simple docstring'''
								lowercase__  					=     z
								lowercase__  					=     self.conv_in(lowerCamelCase    )

								lowercase__  					=     next(iter(self.up_blocks.parameters()    )    ).dtype
								if self.training and self.gradient_checkpointing:

												def create_custom_forward(lowerCamelCase		:       Optional[Any]    ):
																def custom_forward(*lowerCamelCase		:       Tuple    ):
																				return module(*lowerCamelCase    )

																return custom_forward

												if is_torch_version('''>=''',    '''1.11.0'''    ):
																# middle
																lowercase__  					=     torch.utils.checkpoint.checkpoint(
																    create_custom_forward(self.mid_block    ),    lowerCamelCase,    lowerCamelCase,    use_reentrant=lowerCamelCase    )
																lowercase__  					=     sample.to(lowerCamelCase    )

																# up
																for up_block in self.up_blocks:
																				lowercase__  					=     torch.utils.checkpoint.checkpoint(
																				    create_custom_forward(lowerCamelCase    ),    lowerCamelCase,    lowerCamelCase,    use_reentrant=lowerCamelCase    )
												else:
																# middle
																lowercase__  					=     torch.utils.checkpoint.checkpoint(
																    create_custom_forward(self.mid_block    ),    lowerCamelCase,    lowerCamelCase    )
																lowercase__  					=     sample.to(lowerCamelCase    )

																# up
																for up_block in self.up_blocks:
																				lowercase__  					=     torch.utils.checkpoint.checkpoint(create_custom_forward(lowerCamelCase    ),    lowerCamelCase,    lowerCamelCase    )
								else:
												# middle
												lowercase__  					=     self.mid_block(lowerCamelCase,    lowerCamelCase    )
												lowercase__  					=     sample.to(lowerCamelCase    )

												# up
												for up_block in self.up_blocks:
																lowercase__  					=     up_block(lowerCamelCase,    lowerCamelCase    )

        # post-process
								if latent_embeds is None:
												lowercase__  					=     self.conv_norm_out(lowerCamelCase    )
								else:
												lowercase__  					=     self.conv_norm_out(lowerCamelCase,    lowerCamelCase    )
								lowercase__  					=     self.conv_act(lowerCamelCase    )
								lowercase__  					=     self.conv_out(lowerCamelCase    )

								return sample



class  _UpperCAmelCase   (			nn.Module      ):

				"""simple docstring"""




				def __init__(	self		:       Any,    lowerCamelCase		:       Any,    lowerCamelCase		:       List[str],    lowerCamelCase		:       Tuple,    lowerCamelCase		:       List[Any]=None,    lowerCamelCase		:       List[Any]="random",    lowerCamelCase		:       Any=False,    lowerCamelCase		:       Tuple=True    ):

								'''simple docstring'''
								super().__init__()
								lowercase__  					=     n_e
								lowercase__  					=     vq_embed_dim
								lowercase__  					=     beta
								lowercase__  					=     legacy

								lowercase__  					=     nn.Embedding(self.n_e,    self.vq_embed_dim    )
								self.embedding.weight.data.uniform_(-1.0 / self.n_e,    1.0 / self.n_e    )

								lowercase__  					=     remap
								if self.remap is not None:
												self.register_buffer('''used''',    torch.tensor(np.load(self.remap    )    )    )
												lowercase__  					=     self.used.shape[0]
												lowercase__  					=     unknown_index  # "random" or "extra" or integer
												if self.unknown_index == "extra":
																lowercase__  					=     self.re_embed
																lowercase__  					=     self.re_embed + 1
												print(
												    F"""Remapping {self.n_e} indices to {self.re_embed} indices. """
												    F"""Using {self.unknown_index} for unknown indices."""    )
								else:
												lowercase__  					=     n_e

								lowercase__  					=     sane_index_shape




				def 							lowercase__  (	self		:       int,    lowerCamelCase		:       Optional[int]    ):

								'''simple docstring'''
								lowercase__  					=     inds.shape
								assert len(lowerCamelCase    ) > 1
								lowercase__  					=     inds.reshape(ishape[0],    -1    )
								lowercase__  					=     self.used.to(lowerCamelCase    )
								lowercase__  					=     (inds[:, :, None] == used[None, None, ...]).long()
								lowercase__  					=     match.argmax(-1    )
								lowercase__  					=     match.sum(2    ) < 1
								if self.unknown_index == "random":
												lowercase__  					=     torch.randint(0,    self.re_embed,    size=new[unknown].shape    ).to(device=new.device    )
								else:
												lowercase__  					=     self.unknown_index
								return new.reshape(lowerCamelCase    )




				def 							lowercase__  (	self		:       List[str],    lowerCamelCase		:       Any    ):

								'''simple docstring'''
								lowercase__  					=     inds.shape
								assert len(lowerCamelCase    ) > 1
								lowercase__  					=     inds.reshape(ishape[0],    -1    )
								lowercase__  					=     self.used.to(lowerCamelCase    )
								if self.re_embed > self.used.shape[0]:  # extra token
												lowercase__  					=     0  # simply set to zero
								lowercase__  					=     torch.gather(used[None, :][inds.shape[0] * [0], :],    1,    lowerCamelCase    )
								return back.reshape(lowerCamelCase    )




				def 							lowercase__  (	self		:       Optional[Any],    lowerCamelCase		:       Optional[Any]    ):

								'''simple docstring'''
								# reshape z -> (batch, height, width, channel) and flatten
								lowercase__  					=     z.permute(0,    2,    3,    1    ).contiguous()
								lowercase__  					=     z.view(-1,    self.vq_embed_dim    )

								# distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
								lowercase__  					=     torch.argmin(torch.cdist(lowerCamelCase,    self.embedding.weight    ),    dim=1    )

								lowercase__  					=     self.embedding(lowerCamelCase    ).view(z.shape    )
								lowercase__  					=     None
								lowercase__  					=     None

								# compute loss for embedding
								if not self.legacy:
												lowercase__  					=     self.beta * torch.mean((z_q.detach() - z) ** 2    ) + torch.mean((z_q - z.detach()) ** 2    )
								else:
												lowercase__  					=     torch.mean((z_q.detach() - z) ** 2    ) + self.beta * torch.mean((z_q - z.detach()) ** 2    )

								# preserve gradients
								lowercase__  					=     z + (z_q - z).detach()

								# reshape back to match original input shape
								lowercase__  					=     z_q.permute(0,    3,    1,    2    ).contiguous()

								if self.remap is not None:
												lowercase__  					=     min_encoding_indices.reshape(z.shape[0],    -1    )  # add batch axis
												lowercase__  					=     self.remap_to_used(lowerCamelCase    )
												lowercase__  					=     min_encoding_indices.reshape(-1,    1    )  # flatten

								if self.sane_index_shape:
												lowercase__  					=     min_encoding_indices.reshape(z_q.shape[0],    z_q.shape[2],    z_q.shape[3]    )

								return z_q, loss, (perplexity, min_encodings, min_encoding_indices)




				def 							lowercase__  (	self		:       Optional[Any],    lowerCamelCase		:       str,    lowerCamelCase		:       int    ):

								'''simple docstring'''
								# shape specifying (batch, height, width, channel)
								if self.remap is not None:
												lowercase__  					=     indices.reshape(shape[0],    -1    )  # add batch axis
												lowercase__  					=     self.unmap_to_all(lowerCamelCase    )
												lowercase__  					=     indices.reshape(-1    )  # flatten again

								# get quantized latent vectors
								lowercase__  					=     self.embedding(lowerCamelCase    )

								if shape is not None:
												lowercase__  					=     z_q.view(lowerCamelCase    )
												# reshape back to match original input shape
												lowercase__  					=     z_q.permute(0,    3,    1,    2    ).contiguous()

								return z_q



class  _UpperCAmelCase   (			A__      ):

				"""simple docstring"""




				def __init__(	self		:       Optional[int],    lowerCamelCase		:       Optional[Any],    lowerCamelCase		:       Dict=False    ):

								'''simple docstring'''
								lowercase__  					=     parameters
								lowercase__					,		lowercase__  					=     torch.chunk(lowerCamelCase,    2,    dim=1    )
								lowercase__  					=     torch.clamp(self.logvar,    -30.0,    20.0    )
								lowercase__  					=     deterministic
								lowercase__  					=     torch.exp(0.5 * self.logvar    )
								lowercase__  					=     torch.exp(self.logvar    )
								if self.deterministic:
												lowercase__  					=     lowercase__  					=     torch.zeros_like(
												    self.mean,    device=self.parameters.device,    dtype=self.parameters.dtype    )




				def 							lowercase__  (	self		:       Dict,    lowerCamelCase		:       Optional[torch.Generator] = None    ):

								'''simple docstring'''
								# make sure sample is on the same device as the parameters and has same dtype
								lowercase__  					=     randn_tensor(
								    self.mean.shape,    generator=lowerCamelCase,    device=self.parameters.device,    dtype=self.parameters.dtype    )
								lowercase__  					=     self.mean + self.std * sample
								return x




				def 							lowercase__  (	self		:       str,    lowerCamelCase		:       int=None    ):

								'''simple docstring'''
								if self.deterministic:
												return torch.Tensor([0.0]    )
								else:
												if other is None:
																return 0.5 * torch.sum(torch.pow(self.mean,    2    ) + self.var - 1.0 - self.logvar,    dim=[1, 2, 3]    )
												else:
																return 0.5 * torch.sum(
																    torch.pow(self.mean - other.mean,    2    ) / other.var
																    + self.var / other.var
																    - 1.0
																    - self.logvar
																    + other.logvar,    dim=[1, 2, 3],    )




				def 							lowercase__  (	self		:       int,    lowerCamelCase		:       int,    lowerCamelCase		:       int=[1, 2, 3]    ):

								'''simple docstring'''
								if self.deterministic:
												return torch.Tensor([0.0]    )
								lowercase__  					=     np.log(2.0 * np.pi    )
								return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean,    2    ) / self.var,    dim=lowerCamelCase    )




				def 							lowercase__  (	self		:       Any    ):

								'''simple docstring'''
								return self.mean












import argparse
from typing import Dict

import tensorflow as tf
import torch
from tqdm import tqdm

from transformers import BigBirdPegasusConfig, BigBirdPegasusForConditionalGeneration


A__      :     Union[str, Any]       =					[
    # tf -> hf
    ('/', '.'),
    ('layer_', 'layers.'),
    ('kernel', 'weight'),
    ('beta', 'bias'),
    ('gamma', 'weight'),
    ('pegasus', 'model'),
]
A__      :     Optional[Any]       =					[
    ('.output.dense', '.fc2'),
    ('intermediate.LayerNorm', 'final_layer_norm'),
    ('intermediate.dense', 'fc1'),
]

A__      :     List[str]       =					(
    INIT_COMMON
    + [
        ('attention.self.LayerNorm', 'self_attn_layer_norm'),
        ('attention.output.dense', 'self_attn.out_proj'),
        ('attention.self', 'self_attn'),
        ('attention.encdec.LayerNorm', 'encoder_attn_layer_norm'),
        ('attention.encdec_output.dense', 'encoder_attn.out_proj'),
        ('attention.encdec', 'encoder_attn'),
        ('key', 'k_proj'),
        ('value', 'v_proj'),
        ('query', 'q_proj'),
        ('decoder.LayerNorm', 'decoder.layernorm_embedding'),
    ]
    + END_COMMON
)

A__      :     List[str]       =					(
    INIT_COMMON
    + [
        ('embeddings.word_embeddings', 'shared.weight'),
        ('embeddings.position_embeddings', 'embed_positions.weight'),
        ('attention.self.LayerNorm', 'self_attn_layer_norm'),
        ('attention.output.dense', 'self_attn.output'),
        ('attention.self', 'self_attn.self'),
        ('encoder.LayerNorm', 'encoder.layernorm_embedding'),
    ]
    + END_COMMON
)

A__      :     Dict       =					[
    'encdec/key/bias',
    'encdec/query/bias',
    'encdec/value/bias',
    'self/key/bias',
    'self/query/bias',
    'self/value/bias',
    'encdec_output/dense/bias',
    'attention/output/dense/bias',
]






def     a     (		lowerCamelCase_						,  lowerCamelCase_							):






				'''simple docstring'''
				for tf_name, hf_name in patterns:
								lowercase__  					=     k.replace(lowerCamelCase_						,  lowerCamelCase_							)
				return k






def     a     (		lowerCamelCase_						,  lowerCamelCase_							):






				'''simple docstring'''
				lowercase__  					=     BigBirdPegasusConfig(**lowerCamelCase_							)
				lowercase__  					=     BigBirdPegasusForConditionalGeneration(lowerCamelCase_							)
				lowercase__  					=     torch_model.state_dict()
				lowercase__  					=     {}

				# separating decoder weights
				lowercase__  					=     {k: tf_weights[k] for k in tf_weights if k.startswith('''pegasus/decoder'''							)}
				lowercase__  					=     {k: tf_weights[k] for k in tf_weights if not k.startswith('''pegasus/decoder'''							)}

				for k, v in tqdm(decoder_weights.items()						,  '''tf -> hf conversion'''							):
								lowercase__  					=     [k.endswith(lowerCamelCase_							) for ending in KEYS_TO_IGNORE]
								if any(lowerCamelCase_							):
												continue
								lowercase__  					=     DECODER_PATTERNS
								lowercase__  					=     rename_state_dict_key(lowerCamelCase_						,  lowerCamelCase_							)
								if new_k not in state_dict:
												raise ValueError(F"""could not find new key {new_k} in state dict. (converted from {k})"""							)
								if any(True if i in k else False for i in ['''dense''', '''query''', '''key''', '''value''']							):
												lowercase__  					=     v.T
								lowercase__  					=     torch.from_numpy(lowerCamelCase_							)
								assert v.shape == state_dict[new_k].shape, F"""{new_k}, {k}, {v.shape}, {state_dict[new_k].shape}"""

				for k, v in tqdm(remaining_weights.items()						,  '''tf -> hf conversion'''							):
								lowercase__  					=     [k.endswith(lowerCamelCase_							) for ending in KEYS_TO_IGNORE]
								if any(lowerCamelCase_							):
												continue
								lowercase__  					=     REMAINING_PATTERNS
								lowercase__  					=     rename_state_dict_key(lowerCamelCase_						,  lowerCamelCase_							)
								if new_k not in state_dict and k != "pegasus/embeddings/position_embeddings":
												raise ValueError(F"""could not find new key {new_k} in state dict. (converted from {k})"""							)
								if any(True if i in k else False for i in ['''dense''', '''query''', '''key''', '''value''']							):
												lowercase__  					=     v.T
								lowercase__  					=     torch.from_numpy(lowerCamelCase_							)
								if k != "pegasus/embeddings/position_embeddings":
												assert v.shape == state_dict[new_k].shape, F"""{new_k}, {k}, {v.shape}, {state_dict[new_k].shape}"""

				lowercase__  					=     mapping['''model.embed_positions.weight''']
				lowercase__  					=     mapping.pop('''model.embed_positions.weight'''							)
				lowercase__					,		lowercase__  					=     torch_model.load_state_dict(lowerCamelCase_						,  strict=lowerCamelCase_							)
				lowercase__  					=     [
				    k
				    for k in missing
				    if k
				    not in [
				        '''final_logits_bias''',
				        '''model.encoder.embed_tokens.weight''',
				        '''model.decoder.embed_tokens.weight''',
				        '''lm_head.weight''',
				    ]
				]
				assert unexpected_missing == [], F"""no matches found for the following torch keys {unexpected_missing}"""
				assert extra == [], F"""no matches found for the following tf keys {extra}"""
				return torch_model






def     a     (		lowerCamelCase_							):






				'''simple docstring'''
				lowercase__  					=     tf.train.list_variables(lowerCamelCase_							)
				lowercase__  					=     {}
				lowercase__  					=     ['''global_step''']
				for name, shape in tqdm(lowerCamelCase_						,  desc='''converting tf checkpoint to dict'''							):
								lowercase__  					=     any(pat in name for pat in ignore_name							)
								if skip_key:
												continue
								lowercase__  					=     tf.train.load_variable(lowerCamelCase_						,  lowerCamelCase_							)
								lowercase__  					=     array
				return tf_weights






def     a     (		lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_							):






				'''simple docstring'''
				lowercase__  					=     get_tf_weights_as_numpy(lowerCamelCase_							)
				lowercase__  					=     convert_bigbird_pegasus(lowerCamelCase_						,  lowerCamelCase_							)
				torch_model.save_pretrained(lowerCamelCase_							)


if __name__ == "__main__":
							A__      :     Optional[Any]       =					argparse.ArgumentParser()
							parser.add_argument('--tf_ckpt_path', type=str, help='passed to tf.train.list_variables')
							parser.add_argument('--save_dir', default=None, type=str, help='Path to the output PyTorch model.')
							A__      :     Tuple       =					parser.parse_args()
							A__      :     int       =					{}
							convert_bigbird_pegasus_ckpt_to_pytorch(args.tf_ckpt_path, args.save_dir, config_update=config_update)




