





"""simple docstring"""





import gc
import random
import unittest

import numpy as np
import torch
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModel

from diffusers import HeunDiscreteScheduler, PriorTransformer, ShapEImgaImgPipeline
from diffusers.pipelines.shap_e import ShapERenderer
from diffusers.utils import floats_tensor, load_image, load_numpy, slow
from diffusers.utils.testing_utils import require_torch_gpu, torch_device

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference

class 						lowercase   (   UpperCAmelCase_	,    unittest.TestCase   ):
      _SCREAMING_SNAKE_CASE							      = ShapEImgaImgPipeline
      _SCREAMING_SNAKE_CASE							      = ['image']
      _SCREAMING_SNAKE_CASE							      = ['image']
      _SCREAMING_SNAKE_CASE							      = [
          'num_images_per_prompt',
          'num_inference_steps',
          'generator',
          'latents',
          'guidance_scale',
          'frame_size',
          'output_type',
          'return_dict',
      ]
      _SCREAMING_SNAKE_CASE							      = False





      @property
      def        _snake_case     (			self					)						->	List[Any]:
            return 32





      @property
      def        _snake_case     (			self					)						->	Union[str, Any]:
            return 32





      @property
      def        _snake_case     (			self					)						->	Optional[Any]:
            return self.time_input_dim * 4





      @property
      def        _snake_case     (			self					)						->	List[Any]:
            return 8





      @property
      def        _snake_case     (			self					)						->	Union[str, Any]:
            torch.manual_seed(0					)
            lowerCAmelCase    		=			CLIPVisionConfig(
                hidden_size=self.text_embedder_hidden_size ,		image_size=64 ,		projection_dim=self.text_embedder_hidden_size ,		intermediate_size=37 ,		num_attention_heads=4 ,		num_channels=3 ,		num_hidden_layers=5 ,		patch_size=1 ,		)

            lowerCAmelCase    		=			CLIPVisionModel(__lowerCamelCase					)
            return model





      @property
      def        _snake_case     (			self					)						->	str:
            lowerCAmelCase    		=			CLIPImageProcessor(
                crop_size=224 ,		do_center_crop=__lowerCamelCase ,		do_normalize=__lowerCamelCase ,		do_resize=__lowerCamelCase ,		image_mean=[0.48_145_466, 0.4_578_275, 0.40_821_073] ,		image_std=[0.26_862_954, 0.26_130_258, 0.27_577_711] ,		resample=3 ,		size=224 ,		)

            return image_processor





      @property
      def        _snake_case     (			self					)						->	List[str]:
            torch.manual_seed(0					)

            lowerCAmelCase    		=			{
                """num_attention_heads""": 2,
                """attention_head_dim""": 16,
                """embedding_dim""": self.time_input_dim,
                """num_embeddings""": 32,
                """embedding_proj_dim""": self.text_embedder_hidden_size,
                """time_embed_dim""": self.time_embed_dim,
                """num_layers""": 1,
                """clip_embed_dim""": self.time_input_dim * 2,
                """additional_embeddings""": 0,
                """time_embed_act_fn""": """gelu""",
                """norm_in_type""": """layer""",
                """embedding_proj_norm_type""": """layer""",
                """encoder_hid_proj_type""": None,
                """added_emb_type""": None,
            }

            lowerCAmelCase    		=			PriorTransformer(**__lowerCamelCase					)
            return model





      @property
      def        _snake_case     (			self					)						->	Union[str, Any]:
            torch.manual_seed(0					)

            lowerCAmelCase    		=			{
                """param_shapes""": (
                    (self.renderer_dim, 93),
                    (self.renderer_dim, 8),
                    (self.renderer_dim, 8),
                    (self.renderer_dim, 8),
                ),
                """d_latent""": self.time_input_dim,
                """d_hidden""": self.renderer_dim,
                """n_output""": 12,
                """background""": (
                    0.1,
                    0.1,
                    0.1,
                ),
            }
            lowerCAmelCase    		=			ShapERenderer(**__lowerCamelCase					)
            return model





      def        _snake_case     (			self					)						->	Tuple:
            lowerCAmelCase    		=			self.dummy_prior
            lowerCAmelCase    		=			self.dummy_image_encoder
            lowerCAmelCase    		=			self.dummy_image_processor
            lowerCAmelCase    		=			self.dummy_renderer

            lowerCAmelCase    		=			HeunDiscreteScheduler(
                beta_schedule="""exp""" ,		num_train_timesteps=1_024 ,		prediction_type="""sample""" ,		use_karras_sigmas=__lowerCamelCase ,		clip_sample=__lowerCamelCase ,		clip_sample_range=1.0 ,		)
            lowerCAmelCase    		=			{
                """prior""": prior,
                """image_encoder""": image_encoder,
                """image_processor""": image_processor,
                """renderer""": renderer,
                """scheduler""": scheduler,
            }

            return components





      def        _snake_case     (			self ,		lowercase ,		lowercase=0					)						->	Any:
            lowerCAmelCase    		=			floats_tensor((1, 3, 64, 64) ,		rng=random.Random(__lowerCamelCase					)					).to(__lowerCamelCase					)

            if str(__lowerCamelCase					).startswith("""mps"""					):
                  lowerCAmelCase    		=			torch.manual_seed(__lowerCamelCase					)
            else:
                  lowerCAmelCase    		=			torch.Generator(device=__lowerCamelCase					).manual_seed(__lowerCamelCase					)
            lowerCAmelCase    		=			{
                """image""": input_image,
                """generator""": generator,
                """num_inference_steps""": 1,
                """frame_size""": 32,
                """output_type""": """np""",
            }
            return inputs





      def        _snake_case     (			self					)						->	Optional[Any]:
            lowerCAmelCase    		=			"""cpu"""

            lowerCAmelCase    		=			self.get_dummy_components()

            lowerCAmelCase    		=			self.pipeline_class(**__lowerCamelCase					)
            lowerCAmelCase    		=			pipe.to(__lowerCamelCase					)

            pipe.set_progress_bar_config(disable=__lowerCamelCase					)

            lowerCAmelCase    		=			pipe(**self.get_dummy_inputs(__lowerCamelCase					)					)
            lowerCAmelCase    		=			output.images[0]
            lowerCAmelCase    		=			image[0, -3:, -3:, -1]

            assert image.shape == (20, 32, 32, 3)

            lowerCAmelCase    		=			np.array(
                [
                    0.00_039_216,
                    0.00_039_216,
                    0.00_039_216,
                    0.00_039_216,
                    0.00_039_216,
                    0.00_039_216,
                    0.00_039_216,
                    0.00_039_216,
                    0.00_039_216,
                ]					)

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





      def        _snake_case     (			self					)						->	Tuple:
            self._test_inference_batch_consistent(batch_sizes=[1, 2]					)





      def        _snake_case     (			self					)						->	Union[str, Any]:
            lowerCAmelCase    		=			torch_device == """cpu"""
            lowerCAmelCase    		=			True
            self._test_inference_batch_single_identical(
                batch_size=2 ,		test_max_difference=__lowerCamelCase ,		relax_max_difference=__lowerCamelCase ,		)





      def        _snake_case     (			self					)						->	Dict:
            lowerCAmelCase    		=			self.get_dummy_components()
            lowerCAmelCase    		=			self.pipeline_class(**__lowerCamelCase					)
            lowerCAmelCase    		=			pipe.to(__lowerCamelCase					)
            pipe.set_progress_bar_config(disable=__lowerCamelCase					)

            lowerCAmelCase    		=			1
            lowerCAmelCase    		=			2

            lowerCAmelCase    		=			self.get_dummy_inputs(__lowerCamelCase					)

            for key in inputs.keys():
                  if key in self.batch_params:
                        lowerCAmelCase    		=			batch_size * [inputs[key]]

            lowerCAmelCase    		=			pipe(**__lowerCamelCase ,		num_images_per_prompt=__lowerCamelCase					)[0]

            assert images.shape[0] == batch_size * num_images_per_prompt







@slow
@require_torch_gpu
class 						lowercase   (   unittest.TestCase   ):





      def        _snake_case     (			self					)						->	Dict:
            super().tearDown()
            gc.collect()
            torch.cuda.empty_cache()





      def        _snake_case     (			self					)						->	str:
            lowerCAmelCase    		=			load_image(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main""" """/shap_e/corgi.png"""					)
            lowerCAmelCase    		=			load_numpy(
                """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
                """/shap_e/test_shap_e_img2img_out.npy"""					)
            lowerCAmelCase    		=			ShapEImgaImgPipeline.from_pretrained("""openai/shap-e-img2img"""					)
            lowerCAmelCase    		=			pipe.to(__lowerCamelCase					)
            pipe.set_progress_bar_config(disable=__lowerCamelCase					)

            lowerCAmelCase    		=			torch.Generator(device=__lowerCamelCase					).manual_seed(0					)

            lowerCAmelCase    		=			pipe(
                __lowerCamelCase ,		generator=__lowerCamelCase ,		guidance_scale=3.0 ,		num_inference_steps=64 ,		frame_size=64 ,		output_type="""np""" ,		).images[0]

            assert images.shape == (20, 64, 64, 3)

            assert_mean_pixel_difference(__lowerCamelCase ,		__lowerCamelCase					)













import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
		from .features import FeatureType


SCREAMING_SNAKE_CASE_	,SCREAMING_SNAKE_CASE_	,SCREAMING_SNAKE_CASE_   :      Any			=   False, False, False

@dataclass
class 		snake_case_   :







	'''simple docstring'''
	__UpperCamelCase   =		None
	__UpperCamelCase   =		True
	__UpperCamelCase   =		True
	__UpperCamelCase   =		None
	# Automatically constructed
	__UpperCamelCase   =		"dict"
	__UpperCamelCase   =		pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}						)
	__UpperCamelCase   =		field(default='''Audio'''     ,     init=UpperCAmelCase_     ,     repr=UpperCAmelCase_						)




	def __call__(						self							:     Optional[Any]   )       -> List[str]:

			'''simple docstring'''




			return self.pa_type




	def    UpperCAmelCase			(						self							:     List[Any]					,      __lowerCamelCase							:     Union[str, bytes, dict]   )       -> dict:

			'''simple docstring'''




			try:
					import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
			except ImportError as err:
					raise ImportError('To support encoding audio data, please install \'soundfile\'.'   ) from err
			if isinstance(__lowerCamelCase					,      __lowerCamelCase   ):
					return {"bytes": None, "path": value}
			elif isinstance(__lowerCamelCase					,      __lowerCamelCase   ):
					return {"bytes": value, "path": None}
			elif "array" in value:
					# convert the audio array to wav bytes
					__lowercase											=     BytesIO()
					sf.write(__lowerCamelCase					,      value['array']					,      value['sampling_rate']					,      format='wav'   )
					return {"bytes": buffer.getvalue(), "path": None}
			elif value.get('path'   ) is not None and os.path.isfile(value['path']   ):
					# we set "bytes": None to not duplicate the data if they're already available locally
					if value["path"].endswith('pcm'   ):
							# "PCM" only has raw audio bytes
							if value.get('sampling_rate'   ) is None:
									# At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
									raise KeyError('To use PCM files, please specify a \'sampling_rate\' in Audio object'   )
							if value.get('bytes'   ):
									# If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
									__lowercase											=     np.frombuffer(value['bytes']					,      dtype=np.intaa   ).astype(np.floataa   ) / 32_767
							else:
									__lowercase											=     np.memmap(value['path']					,      dtype='h'					,      mode='r'   ).astype(np.floataa   ) / 32_767

							__lowercase											=     BytesIO(bytes()   )
							sf.write(__lowerCamelCase					,      __lowerCamelCase					,      value['sampling_rate']					,      format='wav'   )
							return {"bytes": buffer.getvalue(), "path": None}
					else:
							return {"bytes": None, "path": value.get('path'   )}
			elif value.get('bytes'   ) is not None or value.get('path'   ) is not None:
					# store the audio bytes, and path is used to infer the audio format using the file extension
					return {"bytes": value.get('bytes'   ), "path": value.get('path'   )}
			else:
					raise ValueError(
					    F"An audio sample should have one of 'path' or 'bytes' but they are missing or None in {value}."   )




	def    UpperCAmelCase			(						self							:     Any					,      __lowerCamelCase							:     dict					,      __lowerCamelCase							:     Optional[Dict[str, Union[str, bool, None]]] = None   )       -> dict:

			'''simple docstring'''




			if not self.decode:
					raise RuntimeError('Decoding is disabled for this feature. Please use Audio(decode=True) instead.'   )

			__lowercase						,      __lowercase											=     (value['path'], BytesIO(value['bytes']   )) if value['bytes'] is not None else (value['path'], None)
			if path is None and file is None:
					raise ValueError(F"An audio sample should have one of 'path' or 'bytes' but both are None in {value}."   )

			try:
					import librosa
					import soundfile as sf
			except ImportError as err:
					raise ImportError('To support decoding audio files, please install \'librosa\' and \'soundfile\'.'   ) from err

			__lowercase											=     xsplitext(__lowerCamelCase   )[1][1:].lower() if path is not None else None
			if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
					raise RuntimeError(
					    'Decoding \'opus\' files requires system library \'libsndfile\'>=1.0.31, '
					    'You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. '   )
			elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
					raise RuntimeError(
					    'Decoding \'mp3\' files requires system library \'libsndfile\'>=1.1.0, '
					    'You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. '   )

			if file is None:
					__lowercase											=     token_per_repo_id or {}
					__lowercase											=     path.split('::'   )[-1]
					try:
							__lowercase											=     string_to_dict(__lowerCamelCase					,      config.HUB_DATASETS_URL   )['repo_id']
							__lowercase											=     token_per_repo_id[repo_id]
					except (ValueError, KeyError):
							__lowercase											=     None

					with xopen(__lowerCamelCase					,      'rb'					,      use_auth_token=__lowerCamelCase   ) as f:
							__lowercase						,      __lowercase											=     sf.read(__lowerCamelCase   )

			else:
					__lowercase						,      __lowercase											=     sf.read(__lowerCamelCase   )

			__lowercase											=     array.T
			if self.mono:
					__lowercase											=     librosa.to_mono(__lowerCamelCase   )
			if self.sampling_rate and self.sampling_rate != sampling_rate:
					__lowercase											=     librosa.resample(__lowerCamelCase					,      orig_sr=__lowerCamelCase					,      target_sr=self.sampling_rate   )
					__lowercase											=     self.sampling_rate

			return {"path": path, "array": array, "sampling_rate": sampling_rate}




	def    UpperCAmelCase			(						self							:     int   )       -> Union["FeatureType", Dict[str, "FeatureType"]]:

			'''simple docstring'''




			from .features import Value

			if self.decode:
					raise ValueError('Cannot flatten a decoded Audio feature.'   )
			return {
			    "bytes": Value('binary'   ),
			    "path": Value('string'   ),
			}




	def    UpperCAmelCase			(						self							:     List[Any]					,      __lowerCamelCase							:     Union[pa.StringArray, pa.StructArray]   )       -> pa.StructArray:

			'''simple docstring'''




			if pa.types.is_string(storage.type   ):
					__lowercase											=     pa.array([None] * len(__lowerCamelCase   )					,      type=pa.binary()   )
					__lowercase											=     pa.StructArray.from_arrays([bytes_array, storage]					,      ['bytes', 'path']					,      mask=storage.is_null()   )
			elif pa.types.is_binary(storage.type   ):
					__lowercase											=     pa.array([None] * len(__lowerCamelCase   )					,      type=pa.string()   )
					__lowercase											=     pa.StructArray.from_arrays([storage, path_array]					,      ['bytes', 'path']					,      mask=storage.is_null()   )
			elif pa.types.is_struct(storage.type   ) and storage.type.get_all_field_indices('array'   ):
					__lowercase											=     pa.array([Audio().encode_example(__lowerCamelCase   ) if x is not None else None for x in storage.to_pylist()]   )
			elif pa.types.is_struct(storage.type   ):
					if storage.type.get_field_index('bytes'   ) >= 0:
							__lowercase											=     storage.field('bytes'   )
					else:
							__lowercase											=     pa.array([None] * len(__lowerCamelCase   )					,      type=pa.binary()   )
					if storage.type.get_field_index('path'   ) >= 0:
							__lowercase											=     storage.field('path'   )
					else:
							__lowercase											=     pa.array([None] * len(__lowerCamelCase   )					,      type=pa.string()   )
					__lowercase											=     pa.StructArray.from_arrays([bytes_array, path_array]					,      ['bytes', 'path']					,      mask=storage.is_null()   )
			return array_cast(__lowerCamelCase					,      self.pa_type   )






	def    UpperCAmelCase			(						self							:     Union[str, Any]					,      __lowerCamelCase							:     pa.StructArray   )       -> pa.StructArray:

			'''simple docstring'''




			@no_op_if_value_is_null
			def path_to_bytes(__lowerCamelCase							:     Any   ):
					with xopen(__lowerCamelCase					,      'rb'   ) as f:
							__lowercase											=     f.read()
					return bytes_

			__lowercase											=     pa.array(
			    [
			        (path_to_bytes(x['path']   ) if x['bytes'] is None else x['bytes']) if x is not None else None
			        for x in storage.to_pylist()
			    ]					,      type=pa.binary()					,      )
			__lowercase											=     pa.array(
			    [os.path.basename(__lowerCamelCase   ) if path is not None else None for path in storage.field('path'   ).to_pylist()]					,      type=pa.string()					,      )
			__lowercase											=     pa.StructArray.from_arrays([bytes_array, path_array]					,      ['bytes', 'path']					,      mask=bytes_array.is_null()   )
			return array_cast(__lowerCamelCase					,      self.pa_type   )




