




import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    UNetaDConditionModel,
    VideoToVideoSDPipeline,
)
from diffusers.utils import floats_tensor, is_xformers_available, skip_mps
from diffusers.utils.testing_utils import enable_full_determinism, slow, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()




@skip_mps
class       _A				(		__lowercase       ,     unittest.TestCase     ):
		__a					    =       VideoToVideoSDPipeline
		__a					    =       TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"""video"""}     ) - {"""image""", """width""", """height"""}
		__a					    =       TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""video"""}     ) - {"""image"""}
		__a					    =       PipelineTesterMixin.required_optional_params - {"""latents"""}
		__a					    =       False

		# No `output_type`.
		__a					    =       frozenset(
		    [
		        """num_inference_steps""",
		        """generator""",
		        """latents""",
		        """return_dict""",
		        """callback""",
		        """callback_steps""",
		    ]     )






		def 				UpperCAmelCase (    self			):
								torch.manual_seed(0			)
								_UpperCAmelCase      =			UNetaDConditionModel(
								    block_out_channels=(32, 64, 64, 64)						,		layers_per_block=2						,		sample_size=32						,		in_channels=4						,		out_channels=4						,		down_block_types=("""CrossAttnDownBlock3D""", """CrossAttnDownBlock3D""", """CrossAttnDownBlock3D""", """DownBlock3D""")						,		up_block_types=("""UpBlock3D""", """CrossAttnUpBlock3D""", """CrossAttnUpBlock3D""", """CrossAttnUpBlock3D""")						,		cross_attention_dim=32						,		attention_head_dim=4						,		)
								_UpperCAmelCase      =			DDIMScheduler(
								    beta_start=0.0_0085						,		beta_end=0.012						,		beta_schedule="""scaled_linear"""						,		clip_sample=_SCREAMING_SNAKE_CASE						,		set_alpha_to_one=_SCREAMING_SNAKE_CASE						,		)
								torch.manual_seed(0			)
								_UpperCAmelCase      =			AutoencoderKL(
								    block_out_channels=[32, 64]						,		in_channels=3						,		out_channels=3						,		down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]						,		up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]						,		latent_channels=4						,		sample_size=128						,		)
								torch.manual_seed(0			)
								_UpperCAmelCase      =			CLIPTextConfig(
								    bos_token_id=0						,		eos_token_id=2						,		hidden_size=32						,		intermediate_size=37						,		layer_norm_eps=1e-05						,		num_attention_heads=4						,		num_hidden_layers=5						,		pad_token_id=1						,		vocab_size=1000						,		hidden_act="""gelu"""						,		projection_dim=512						,		)
								_UpperCAmelCase      =			CLIPTextModel(_SCREAMING_SNAKE_CASE			)
								_UpperCAmelCase      =			CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""			)

								_UpperCAmelCase      =			{
								    """unet""": unet,
								    """scheduler""": scheduler,
								    """vae""": vae,
								    """text_encoder""": text_encoder,
								    """tokenizer""": tokenizer,
								}
								return components






		def 				UpperCAmelCase (    self						,		_SCREAMING_SNAKE_CASE						,		_SCREAMING_SNAKE_CASE=0			):
								# 3 frames
								_UpperCAmelCase      =			floats_tensor((1, 3, 3, 32, 32)						,		rng=random.Random(_SCREAMING_SNAKE_CASE			)			).to(_SCREAMING_SNAKE_CASE			)

								if str(_SCREAMING_SNAKE_CASE			).startswith("""mps"""			):
														_UpperCAmelCase      =			torch.manual_seed(_SCREAMING_SNAKE_CASE			)
								else:
														_UpperCAmelCase      =			torch.Generator(device=_SCREAMING_SNAKE_CASE			).manual_seed(_SCREAMING_SNAKE_CASE			)
								_UpperCAmelCase      =			{
								    """prompt""": """A painting of a squirrel eating a burger""",
								    """video""": video,
								    """generator""": generator,
								    """num_inference_steps""": 2,
								    """guidance_scale""": 6.0,
								    """output_type""": """pt""",
								}
								return inputs






		def 				UpperCAmelCase (    self			):
								_UpperCAmelCase      =			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
								_UpperCAmelCase      =			self.get_dummy_components()
								_UpperCAmelCase      =			VideoToVideoSDPipeline(**_SCREAMING_SNAKE_CASE			)
								_UpperCAmelCase      =			sd_pipe.to(_SCREAMING_SNAKE_CASE			)
								sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE			)

								_UpperCAmelCase      =			self.get_dummy_inputs(_SCREAMING_SNAKE_CASE			)
								_UpperCAmelCase      =			"""np"""
								_UpperCAmelCase      =			sd_pipe(**_SCREAMING_SNAKE_CASE			).frames
								_UpperCAmelCase      =			frames[0][-3:, -3:, -1]

								assert frames[0].shape == (32, 32, 3)
								_UpperCAmelCase      =			np.array([106, 117, 113, 174, 137, 112, 148, 151, 131]			)

								assert np.abs(image_slice.flatten() - expected_slice			).max() < 1e-2






		@unittest.skipIf(
		    torch_device != """cuda""" or not is_xformers_available()						,		reason="""XFormers attention is only available with CUDA and `xformers` installed"""						,		)
		def 				UpperCAmelCase (    self			):
								self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=_SCREAMING_SNAKE_CASE						,		expected_max_diff=5e-3			)






		@unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""			)
		def 				UpperCAmelCase (    self			):
								pass






		@unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""			)
		def 				UpperCAmelCase (    self			):
								pass






		@unittest.skip(reason="""`num_images_per_prompt` argument is not supported for this pipeline."""			)
		def 				UpperCAmelCase (    self			):
								pass






		def 				UpperCAmelCase (    self			):
								return super().test_progress_bar()





@slow
@skip_mps
class       _A				(		unittest.TestCase     ):






		def 				UpperCAmelCase (    self			):
								_UpperCAmelCase      =			VideoToVideoSDPipeline.from_pretrained("""cerspense/zeroscope_v2_XL"""						,		torch_dtype=torch.floataa			)
								pipe.enable_model_cpu_offload()

								# 10 frames
								_UpperCAmelCase      =			torch.Generator(device="""cpu"""			).manual_seed(0			)
								_UpperCAmelCase      =			torch.randn((1, 10, 3, 1024, 576)						,		generator=_SCREAMING_SNAKE_CASE			)
								_UpperCAmelCase      =			video.to("""cuda"""			)

								_UpperCAmelCase      =			"""Spiderman is surfing"""

								_UpperCAmelCase      =			pipe(_SCREAMING_SNAKE_CASE						,		video=_SCREAMING_SNAKE_CASE						,		generator=_SCREAMING_SNAKE_CASE						,		num_inference_steps=3						,		output_type="""pt"""			).frames

								_UpperCAmelCase      =			np.array([-1.045_8984, -1.127_9297, -0.966_3086, -0.9150_3906, -0.7509_7656]			)
								assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array			).sum() < 1e-2





import argparse

import fairseq
import torch

from transformers import UniSpeechSatConfig, UniSpeechSatForCTC, UniSpeechSatForPreTraining, logging


logging.set_verbosity_info()
a													=			logging.get_logger(__name__)

a													=			{
    "post_extract_proj": "feature_projection.projection",
    "encoder.pos_conv.0": "encoder.pos_conv_embed.conv",
    "self_attn.k_proj": "encoder.layers.*.attention.k_proj",
    "self_attn.v_proj": "encoder.layers.*.attention.v_proj",
    "self_attn.q_proj": "encoder.layers.*.attention.q_proj",
    "self_attn.out_proj": "encoder.layers.*.attention.out_proj",
    "self_attn_layer_norm": "encoder.layers.*.layer_norm",
    "fc1": "encoder.layers.*.feed_forward.intermediate_dense",
    "fc2": "encoder.layers.*.feed_forward.output_dense",
    "final_layer_norm": "encoder.layers.*.final_layer_norm",
    "encoder.layer_norm": "encoder.layer_norm",
    "encoder.layer_norm_for_extract": "layer_norm_for_extract",
    "w2v_model.layer_norm": "feature_projection.layer_norm",
    "quantizer.weight_proj": "quantizer.weight_proj",
    "quantizer.vars": "quantizer.codevectors",
    "project_q": "project_q",
    "final_proj": "project_hid",
    "w2v_encoder.proj": "lm_head",
    "label_embs_concat": "label_embeddings_concat",
    "mask_emb": "masked_spec_embed",
    "spk_proj": "speaker_proj",
}
a													=			[
    "lm_head",
    "quantizer.weight_proj",
    "quantizer.codevectors",
    "project_q",
    "project_hid",
    "label_embeddings_concat",
    "speaker_proj",
    "layer_norm_for_extract",
]




def        _SCREAMING_SNAKE_CASE      (		snake_case			,     snake_case			,     snake_case			,     snake_case			,     snake_case       )						->				List[str]:
						for attribute in key.split("""."""       ):
												_UpperCAmelCase      =			getattr(snake_case			,     snake_case       )

						if weight_type is not None:
												_UpperCAmelCase      =			getattr(snake_case			,     snake_case       ).shape
						else:
												_UpperCAmelCase      =			hf_pointer.shape

						if hf_shape != value.shape:
												raise ValueError(
												    f"Shape of hf {key + '.' + weight_type if weight_type is not None else ''} is {hf_shape}, but should be"
												    f" {value.shape} for {full_name}"       )

						if weight_type == "weight":
												_UpperCAmelCase      =			value
						elif weight_type == "weight_g":
												_UpperCAmelCase      =			value
						elif weight_type == "weight_v":
												_UpperCAmelCase      =			value
						elif weight_type == "bias":
												_UpperCAmelCase      =			value
						else:
												_UpperCAmelCase      =			value

						logger.info(f"{key + '.' + weight_type if weight_type is not None else ''} was initialized from {full_name}."       )




def        _SCREAMING_SNAKE_CASE      (		snake_case			,     snake_case       )						->				Optional[int]:
						_UpperCAmelCase      =			[]
						_UpperCAmelCase      =			fairseq_model.state_dict()

						_UpperCAmelCase      =			hf_model.unispeech_sat.feature_extractor

						for name, value in fairseq_dict.items():
												_UpperCAmelCase      =			False
												if "conv_layers" in name:
																		load_conv_layer(
																		    snake_case			,     snake_case			,     snake_case			,     snake_case			,     hf_model.config.feat_extract_norm == """group"""			,     )
																		_UpperCAmelCase      =			True
												else:
																		for key, mapped_key in MAPPING.items():
																								_UpperCAmelCase      =			"""unispeech_sat.""" + mapped_key if mapped_key not in TOP_LEVEL_KEYS else mapped_key
																								if key in name or key.split("""w2v_model."""       )[-1] == name.split("""."""       )[0]:
																														if "layer_norm_for_extract" in name and (".".join(name.split("""."""       )[:-1]       ) != key):
																																				# special case since naming is very similar
																																				continue
																														_UpperCAmelCase      =			True
																														if "*" in mapped_key:
																																				_UpperCAmelCase      =			name.split(snake_case       )[0].split("""."""       )[-2]
																																				_UpperCAmelCase      =			mapped_key.replace("""*"""			,     snake_case       )
																														if "weight_g" in name:
																																				_UpperCAmelCase      =			"""weight_g"""
																														elif "weight_v" in name:
																																				_UpperCAmelCase      =			"""weight_v"""
																														elif "bias" in name:
																																				_UpperCAmelCase      =			"""bias"""
																														elif "weight" in name:
																																				# TODO: don't match quantizer.weight_proj
																																				_UpperCAmelCase      =			"""weight"""
																														else:
																																				_UpperCAmelCase      =			None
																														set_recursively(snake_case			,     snake_case			,     snake_case			,     snake_case			,     snake_case       )
																								continue
												if not is_used:
																		unused_weights.append(snake_case       )

						logger.warning(f"Unused weights: {unused_weights}"       )




def        _SCREAMING_SNAKE_CASE      (		snake_case			,     snake_case			,     snake_case			,     snake_case			,     snake_case       )						->				str:
						_UpperCAmelCase      =			full_name.split("""conv_layers."""       )[-1]
						_UpperCAmelCase      =			name.split("""."""       )
						_UpperCAmelCase      =			int(items[0]       )
						_UpperCAmelCase      =			int(items[1]       )

						if type_id == 0:
												if "bias" in name:
																		if value.shape != feature_extractor.conv_layers[layer_id].conv.bias.data.shape:
																								raise ValueError(
																								    f"{full_name} has size {value.shape}, but"
																								    f" {feature_extractor.conv_layers[layer_id].conv.bias.data.shape} was found."       )
																		_UpperCAmelCase      =			value
																		logger.info(f"Feat extract conv layer {layer_id} was initialized from {full_name}."       )
												elif "weight" in name:
																		if value.shape != feature_extractor.conv_layers[layer_id].conv.weight.data.shape:
																								raise ValueError(
																								    f"{full_name} has size {value.shape}, but"
																								    f" {feature_extractor.conv_layers[layer_id].conv.weight.data.shape} was found."       )
																		_UpperCAmelCase      =			value
																		logger.info(f"Feat extract conv layer {layer_id} was initialized from {full_name}."       )
						elif (type_id == 2 and not use_group_norm) or (type_id == 2 and layer_id == 0 and use_group_norm):
												if "bias" in name:
																		if value.shape != feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape:
																								raise ValueError(
																								    f"{full_name} has size {value.shape}, but"
																								    f" {feature_extractor[layer_id].layer_norm.bias.data.shape} was found."       )
																		_UpperCAmelCase      =			value
																		logger.info(f"Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."       )
												elif "weight" in name:
																		if value.shape != feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape:
																								raise ValueError(
																								    f"{full_name} has size {value.shape}, but"
																								    f" {feature_extractor[layer_id].layer_norm.weight.data.shape} was found."       )
																		_UpperCAmelCase      =			value
																		logger.info(f"Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."       )
						else:
												unused_weights.append(snake_case       )




@torch.no_grad()
def        _SCREAMING_SNAKE_CASE      (		snake_case			,     snake_case			,     snake_case=None			,     snake_case=None			,     snake_case=True       )						->				List[Any]:
						if config_path is not None:
												_UpperCAmelCase      =			UniSpeechSatConfig.from_pretrained(snake_case       )
						else:
												_UpperCAmelCase      =			UniSpeechSatConfig()

						_UpperCAmelCase      =			""""""

						if is_finetuned:
												_UpperCAmelCase      =			UniSpeechSatForCTC(snake_case       )
						else:
												_UpperCAmelCase      =			UniSpeechSatForPreTraining(snake_case       )

						_UpperCAmelCase    ,					_UpperCAmelCase    ,					_UpperCAmelCase      =			fairseq.checkpoint_utils.load_model_ensemble_and_task(
						    [checkpoint_path]			,     arg_overrides={"""data""": """/""".join(dict_path.split("""/"""       )[:-1]       )}       )
						_UpperCAmelCase      =			model[0].eval()

						recursively_load_weights(snake_case			,     snake_case       )

						hf_wavavec.save_pretrained(snake_case       )


if __name__ == "__main__":
				a													=			argparse.ArgumentParser()
				parser.add_argument("--pytorch_dump_folder_path", default=None, type=str, help="Path to the output PyTorch model.")
				parser.add_argument("--checkpoint_path", default=None, type=str, help="Path to fairseq checkpoint")
				parser.add_argument("--dict_path", default=None, type=str, help="Path to dict of fine-tuned model")
				parser.add_argument("--config_path", default=None, type=str, help="Path to hf config.json of model to convert")
				parser.add_argument(
				    "--not_finetuned", action="store_true", help="Whether the model to convert is a fine-tuned model or not"
				)
				a													=			parser.parse_args()
				convert_unispeech_sat_checkpoint(
				    args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
				)