
"""simple docstring"""




import gc
import unittest

import numpy as np
import torch

from diffusers import AutoencoderKL, DDIMScheduler, DiTPipeline, DPMSolverMultistepScheduler, TransformeraDModel
from diffusers.utils import is_xformers_available, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    CLASS_CONDITIONED_IMAGE_GENERATION_BATCH_PARAMS,
    CLASS_CONDITIONED_IMAGE_GENERATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


class 			__magic_name__       (       _UpperCamelCase   ,					unittest.TestCase			):
       UpperCamelCase_   =		DiTPipeline
       UpperCamelCase_   =		CLASS_CONDITIONED_IMAGE_GENERATION_PARAMS
       UpperCamelCase_   =		PipelineTesterMixin.required_optional_params - {
           '''latents''',
           '''num_images_per_prompt''',
           '''callback''',
           '''callback_steps''',
       }
       UpperCamelCase_   =		CLASS_CONDITIONED_IMAGE_GENERATION_BATCH_PARAMS
       UpperCamelCase_   =		False
       def 							lowercase_							(     self       )     ->	str:





         """simple docstring"""




         torch.manual_seed(0       )
         _lowercase:			Optional[Any]       =     TransformeraDModel(
             sample_size=16 ,				num_layers=2 ,				patch_size=4 ,				attention_head_dim=8 ,				num_attention_heads=2 ,				in_channels=4 ,				out_channels=8 ,				attention_bias=A_ ,				activation_fn='''gelu-approximate''' ,				num_embeds_ada_norm=1000 ,				norm_type='''ada_norm_zero''' ,				norm_elementwise_affine=A_ ,				)
         _lowercase:			Dict       =     AutoencoderKL()
         _lowercase:			Any       =     DDIMScheduler()
         _lowercase:			int       =     {"transformer": transformer.eval(), "vae": vae.eval(), "scheduler": scheduler}
         return components
       def 							lowercase_							(     self ,				A_ ,				A_=0       )     ->	Optional[Any]:





         """simple docstring"""




         if str(A_       ).startswith('''mps'''       ):
           _lowercase:			Union[str, Any]       =     torch.manual_seed(A_       )
         else:
           _lowercase:			List[str]       =     torch.Generator(device=A_       ).manual_seed(A_       )
         _lowercase:			int       =     {
             "class_labels": [1],
             "generator": generator,
             "num_inference_steps": 2,
             "output_type": "numpy",
         }
         return inputs
       def 							lowercase_							(     self       )     ->	List[str]:





         """simple docstring"""




         _lowercase:			List[Any]       =     "cpu"

         _lowercase:			str       =     self.get_dummy_components()
         _lowercase:			str       =     self.pipeline_class(**A_       )
         pipe.to(A_       )
         pipe.set_progress_bar_config(disable=A_       )

         _lowercase:			Union[str, Any]       =     self.get_dummy_inputs(A_       )
         _lowercase:			str       =     pipe(**A_       ).images
         _lowercase:			Tuple       =     image[0, -3:, -3:, -1]

         self.assertEqual(image.shape ,				(1, 16, 16, 3)       )
         _lowercase:			Tuple       =     np.array([0.29_46, 0.66_01, 0.43_29, 0.32_96, 0.41_44, 0.53_19, 0.72_73, 0.50_13, 0.44_57]       )
         _lowercase:			List[Any]       =     np.abs(image_slice.flatten() - expected_slice       ).max()
         self.assertLessEqual(A_ ,				1E-3       )
       def 							lowercase_							(     self       )     ->	Any:





         """simple docstring"""




         self._test_inference_batch_single_identical(relax_max_difference=A_ ,				expected_max_diff=1E-3       )



       @unittest.skipIf(
           torch_device != '''cuda''' or not is_xformers_available() ,				reason='''XFormers attention is only available with CUDA and `xformers` installed''' ,				)
       def 							lowercase_							(     self       )     ->	str:





         """simple docstring"""




         self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3       )


@require_torch_gpu
@slow
class 			__magic_name__       (       unittest.TestCase			):
       def 							lowercase_							(     self       )     ->	str:





         """simple docstring"""




         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()
       def 							lowercase_							(     self       )     ->	Optional[int]:





         """simple docstring"""




         _lowercase:			List[Any]       =     torch.manual_seed(0       )

         _lowercase:			Dict       =     DiTPipeline.from_pretrained('''facebook/DiT-XL-2-256'''       )
         pipe.to('''cuda'''       )

         _lowercase:			Tuple       =     ["vase", "umbrella", "white shark", "white wolf"]
         _lowercase:			Optional[Any]       =     pipe.get_label_ids(A_       )

         _lowercase:			Dict       =     pipe(A_ ,				generator=A_ ,				num_inference_steps=40 ,				output_type='''np'''       ).images

         for word, image in zip(A_ ,				A_       ):
           _lowercase:			Optional[Any]       =     load_numpy(
               f'''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/dit/{word}.npy'''       )
           assert np.abs((expected_image - image).max()       ) < 1E-2



       def 							lowercase_							(     self       )     ->	List[str]:





         """simple docstring"""




         _lowercase:			int       =     DiTPipeline.from_pretrained('''facebook/DiT-XL-2-512'''       )
         _lowercase:			Dict       =     DPMSolverMultistepScheduler.from_config(pipe.scheduler.config       )
         pipe.to('''cuda'''       )

         _lowercase:			Dict       =     ["vase", "umbrella"]
         _lowercase:			Any       =     pipe.get_label_ids(A_       )

         _lowercase:			List[str]       =     torch.manual_seed(0       )
         _lowercase:			str       =     pipe(A_ ,				generator=A_ ,				num_inference_steps=25 ,				output_type='''np'''       ).images

         for word, image in zip(A_ ,				A_       ):
           _lowercase:			Optional[Any]       =     load_numpy(
               '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
               f'''/dit/{word}_512.npy'''       )

           assert np.abs((expected_image - image).max()       ) < 1E-1


"""simple docstring"""




import os
import tempfile
from functools import partial
from unittest import TestCase
from unittest.mock import patch

import numpy as np
import pytest

from datasets.arrow_dataset import Dataset
from datasets.search import ElasticSearchIndex, FaissIndex, MissingIndex

from .utils import require_elasticsearch, require_faiss


A__   :					Any							=   pytest.mark.integration


@require_faiss
class 			__magic_name__       (       SCREAMING_SNAKE_CASE__ ):
       def 							lowercase_							(     self       )     ->	List[Any]:





         """simple docstring"""




         _lowercase:			str       =     Dataset.from_dict({'''filename''': ['''my_name-train''' + '''_''' + str(A_       ) for x in np.arange(30       ).tolist()]}       )
         return dset
       def 							lowercase_							(     self       )     ->	int:





         """simple docstring"""




         import faiss

         _lowercase:			Dataset       =     self._create_dummy_dataset()
         _lowercase:			List[str]       =     dset.map(
             lambda A_ ,				A_       : {"vecs": i * np.ones(5 ,				dtype=np.floataa       )} ,				with_indices=A_ ,				keep_in_memory=A_       )
         _lowercase:			Dict       =     dset.add_faiss_index('''vecs''' ,				batch_size=100 ,				metric_type=faiss.METRIC_INNER_PRODUCT       )
         _lowercase ,				_lowercase:			Union[str, Any]       =     dset.get_nearest_examples('''vecs''' ,				np.ones(5 ,				dtype=np.floataa       )       )
         self.assertEqual(examples['''filename'''][0] ,				'''my_name-train_29'''       )
         dset.drop_index('''vecs'''       )
       def 							lowercase_							(     self       )     ->	List[Any]:





         """simple docstring"""




         import faiss

         _lowercase:			Dataset       =     self._create_dummy_dataset()
         dset.add_faiss_index_from_external_arrays(
             external_arrays=np.ones((30, 5)       ) * np.arange(30       ).reshape(-1 ,				1       ) ,				index_name='''vecs''' ,				batch_size=100 ,				metric_type=faiss.METRIC_INNER_PRODUCT ,				)
         _lowercase ,				_lowercase:			Dict       =     dset.get_nearest_examples('''vecs''' ,				np.ones(5 ,				dtype=np.floataa       )       )
         self.assertEqual(examples['''filename'''][0] ,				'''my_name-train_29'''       )
       def 							lowercase_							(     self       )     ->	List[str]:





         """simple docstring"""




         import faiss

         _lowercase:			Dataset       =     self._create_dummy_dataset()
         dset.add_faiss_index_from_external_arrays(
             external_arrays=np.ones((30, 5)       ) * np.arange(30       ).reshape(-1 ,				1       ) ,				index_name='''vecs''' ,				metric_type=faiss.METRIC_INNER_PRODUCT ,				)

         # Setting delete=False and unlinking manually is not pretty... but it is required on Windows to
         # ensure somewhat stable behaviour. If we don't, we get PermissionErrors. This is an age-old issue.
         # see https://bugs.python.org/issue14243 and
         # https://stackoverflow.com/questions/23212435/permission-denied-to-write-to-my-temporary-file/23212515
         with tempfile.NamedTemporaryFile(delete=A_       ) as tmp_file:
           dset.save_faiss_index('''vecs''' ,				tmp_file.name       )
           dset.load_faiss_index('''vecs2''' ,				tmp_file.name       )
         os.unlink(tmp_file.name       )

         _lowercase ,				_lowercase:			Dict       =     dset.get_nearest_examples('''vecs2''' ,				np.ones(5 ,				dtype=np.floataa       )       )
         self.assertEqual(examples['''filename'''][0] ,				'''my_name-train_29'''       )
       def 							lowercase_							(     self       )     ->	int:





         """simple docstring"""




         _lowercase:			Dataset       =     self._create_dummy_dataset()
         dset.add_faiss_index_from_external_arrays(
             external_arrays=np.ones((30, 5)       ) * np.arange(30       ).reshape(-1 ,				1       ) ,				index_name='''vecs'''       )
         dset.drop_index('''vecs'''       )
         self.assertRaises(A_ ,				partial(dset.get_nearest_examples ,				'''vecs2''' ,				np.ones(5 ,				dtype=np.floataa       )       )       )



       def 							lowercase_							(     self       )     ->	List[Any]:





         """simple docstring"""




         from elasticsearch import Elasticsearch

         _lowercase:			Dataset       =     self._create_dummy_dataset()
         with patch('''elasticsearch.Elasticsearch.search'''       ) as mocked_search, patch(
             '''elasticsearch.client.IndicesClient.create'''       ) as mocked_index_create, patch('''elasticsearch.helpers.streaming_bulk'''       ) as mocked_bulk:
           _lowercase:			List[Any]       =     {'''acknowledged''': True}
           mocked_bulk.return_value([(True, None)] * 30       )
           _lowercase:			List[str]       =     {'''hits''': {'''hits''': [{'''_score''': 1, '''_id''': 29}]}}
           _lowercase:			int       =     Elasticsearch()

           dset.add_elasticsearch_index('''filename''' ,				es_client=A_       )
           _lowercase ,				_lowercase:			Dict       =     dset.get_nearest_examples('''filename''' ,				'''my_name-train_29'''       )
           self.assertEqual(examples['''filename'''][0] ,				'''my_name-train_29'''       )


@require_faiss
class 			__magic_name__       (       SCREAMING_SNAKE_CASE__ ):
       def 							lowercase_							(     self       )     ->	Any:





         """simple docstring"""




         import faiss

         _lowercase:			str       =     FaissIndex(metric_type=faiss.METRIC_INNER_PRODUCT       )

         # add vectors
         index.add_vectors(np.eye(5 ,				dtype=np.floataa       )       )
         self.assertIsNotNone(index.faiss_index       )
         self.assertEqual(index.faiss_index.ntotal ,				5       )
         index.add_vectors(np.zeros((5, 5) ,				dtype=np.floataa       )       )
         self.assertEqual(index.faiss_index.ntotal ,				10       )

         # single query
         _lowercase:			List[Any]       =     np.zeros(5 ,				dtype=np.floataa       )
         _lowercase:			int       =     1
         _lowercase ,				_lowercase:			Optional[Any]       =     index.search(A_       )
         self.assertRaises(A_ ,				index.search ,				query.reshape(-1 ,				1       )       )
         self.assertGreater(scores[0] ,				0       )
         self.assertEqual(indices[0] ,				1       )

         # batched queries
         _lowercase:			Tuple       =     np.eye(5 ,				dtype=np.floataa       )[::-1]
         _lowercase ,				_lowercase:			str       =     index.search_batch(A_       )
         self.assertRaises(A_ ,				index.search_batch ,				queries[0]       )
         _lowercase:			Tuple       =     [scores[0] for scores in total_scores]
         _lowercase:			Union[str, Any]       =     [indices[0] for indices in total_indices]
         self.assertGreater(np.min(A_       ) ,				0       )
         self.assertListEqual([4, 3, 2, 1, 0] ,				A_       )
       def 							lowercase_							(     self       )     ->	Union[str, Any]:





         """simple docstring"""




         import faiss

         _lowercase:			Union[str, Any]       =     FaissIndex(string_factory='''Flat'''       )
         index.add_vectors(np.eye(5 ,				dtype=np.floataa       )       )
         self.assertIsInstance(index.faiss_index ,				faiss.IndexFlat       )
         _lowercase:			Dict       =     FaissIndex(string_factory='''LSH'''       )
         index.add_vectors(np.eye(5 ,				dtype=np.floataa       )       )
         self.assertIsInstance(index.faiss_index ,				faiss.IndexLSH       )
         with self.assertRaises(A_       ):
           _lowercase:			List[str]       =     FaissIndex(string_factory='''Flat''' ,				custom_index=faiss.IndexFlat(5       )       )
       def 							lowercase_							(     self       )     ->	List[str]:





         """simple docstring"""




         import faiss

         _lowercase:			Any       =     faiss.IndexFlat(5       )
         _lowercase:			List[Any]       =     FaissIndex(custom_index=A_       )
         index.add_vectors(np.eye(5 ,				dtype=np.floataa       )       )
         self.assertIsInstance(index.faiss_index ,				faiss.IndexFlat       )



       def 							lowercase_							(     self       )     ->	str:





         """simple docstring"""




         import faiss

         _lowercase:			Tuple       =     FaissIndex(metric_type=faiss.METRIC_INNER_PRODUCT       )
         index.add_vectors(np.eye(5 ,				dtype=np.floataa       )       )

         # Setting delete=False and unlinking manually is not pretty... but it is required on Windows to
         # ensure somewhat stable behaviour. If we don't, we get PermissionErrors. This is an age-old issue.
         # see https://bugs.python.org/issue14243 and
         # https://stackoverflow.com/questions/23212435/permission-denied-to-write-to-my-temporary-file/23212515
         with tempfile.NamedTemporaryFile(delete=A_       ) as tmp_file:
           index.save(tmp_file.name       )
           _lowercase:			Optional[Any]       =     FaissIndex.load(tmp_file.name       )
         os.unlink(tmp_file.name       )

         _lowercase:			Optional[Any]       =     np.zeros(5 ,				dtype=np.floataa       )
         _lowercase:			Union[str, Any]       =     1
         _lowercase ,				_lowercase:			Tuple       =     index.search(A_       )
         self.assertGreater(scores[0] ,				0       )
         self.assertEqual(indices[0] ,				1       )




@require_faiss
def  _lowerCAmelCase		(		_UpperCamelCase						):






  """simple docstring"""



  import faiss

  _lowercase:			Dict       =     FaissIndex(metric_type=faiss.METRIC_INNER_PRODUCT						)
  index.add_vectors(np.eye(5 ,				dtype=np.floataa						)						)

  _lowercase:			Tuple       =     '''index.faiss'''
  _lowercase:			str       =     f'''mock://{index_name}'''
  index.save(_UpperCamelCase ,				storage_options=mockfs.storage_options						)
  _lowercase:			List[Any]       =     FaissIndex.load(_UpperCamelCase ,				storage_options=mockfs.storage_options						)

  _lowercase:			Union[str, Any]       =     np.zeros(5 ,				dtype=np.floataa						)
  _lowercase:			Dict       =     1
  _lowercase ,				_lowercase:			str       =     index.search(_UpperCamelCase						)
  assert scores[0] > 0
  assert indices[0] == 1







@require_elasticsearch
class 			__magic_name__       (       SCREAMING_SNAKE_CASE__ ):
       def 							lowercase_							(     self       )     ->	int:





         """simple docstring"""




         from elasticsearch import Elasticsearch

         with patch('''elasticsearch.Elasticsearch.search'''       ) as mocked_search, patch(
             '''elasticsearch.client.IndicesClient.create'''       ) as mocked_index_create, patch('''elasticsearch.helpers.streaming_bulk'''       ) as mocked_bulk:
           _lowercase:			int       =     Elasticsearch()
           _lowercase:			Tuple       =     {'''acknowledged''': True}
           _lowercase:			Tuple       =     ElasticSearchIndex(es_client=A_       )
           mocked_bulk.return_value([(True, None)] * 3       )
           index.add_documents(['''foo''', '''bar''', '''foobar''']       )

           # single query
           _lowercase:			Dict       =     '''foo'''
           _lowercase:			Union[str, Any]       =     {'''hits''': {'''hits''': [{'''_score''': 1, '''_id''': 0}]}}
           _lowercase ,				_lowercase:			Optional[Any]       =     index.search(A_       )
           self.assertEqual(scores[0] ,				1       )
           self.assertEqual(indices[0] ,				0       )

           # single query with timeout
           _lowercase:			Optional[int]       =     '''foo'''
           _lowercase:			Union[str, Any]       =     {'''hits''': {'''hits''': [{'''_score''': 1, '''_id''': 0}]}}
           _lowercase ,				_lowercase:			List[Any]       =     index.search(A_ ,				request_timeout=30       )
           self.assertEqual(scores[0] ,				1       )
           self.assertEqual(indices[0] ,				0       )

           # batched queries
           _lowercase:			Union[str, Any]       =     ['''foo''', '''bar''', '''foobar''']
           _lowercase:			str       =     {'''hits''': {'''hits''': [{'''_score''': 1, '''_id''': 1}]}}
           _lowercase ,				_lowercase:			Optional[int]       =     index.search_batch(A_       )
           _lowercase:			Any       =     [scores[0] for scores in total_scores]
           _lowercase:			List[str]       =     [indices[0] for indices in total_indices]
           self.assertGreater(np.min(A_       ) ,				0       )
           self.assertListEqual([1, 1, 1] ,				A_       )

           # batched queries with timeout
           _lowercase:			List[str]       =     ['''foo''', '''bar''', '''foobar''']
           _lowercase:			Dict       =     {'''hits''': {'''hits''': [{'''_score''': 1, '''_id''': 1}]}}
           _lowercase ,				_lowercase:			Optional[int]       =     index.search_batch(A_ ,				request_timeout=30       )
           _lowercase:			Optional[Any]       =     [scores[0] for scores in total_scores]
           _lowercase:			Dict       =     [indices[0] for indices in total_indices]
           self.assertGreater(np.min(A_       ) ,				0       )
           self.assertListEqual([1, 1, 1] ,				A_       )
