



# Lint as: python3
import os
import re
import urllib.parse
from pathlib import Path
from typing import Callable, List, Optional, Union
from zipfile import ZipFile

from ..utils.file_utils import cached_path, hf_github_url
from ..utils.logging import get_logger
from ..utils.version import Version


a_      :      Tuple    =     get_logger(__name__)
class    __UpperCamelCase			:

 """simple docstring"""



 _lowercase   :    str      						=					'''dummy_data'''
 _lowercase   :    Tuple      						=					'''datasets'''
 _lowercase   :    Any      						=					False

 def __init__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE = None		,				SCREAMING_SNAKE_CASE = False		,				SCREAMING_SNAKE_CASE = True		,				SCREAMING_SNAKE_CASE = None		,				) ->			List[str]:
  a__												= 0
  a__												= dataset_name
  a__												= cache_dir
  a__												= use_local_dummy_data
  a__												= config
  # download_callbacks take a single url as input
  a__												= download_callbacks or []
  # if False, it doesn't load existing files and it returns the paths of the dummy files relative
  # to the dummy_data zip file root
  a__												= load_existing_dummy_data

  # TODO(PVP, QL) might need to make this more general
  a__												= str(SCREAMING_SNAKE_CASE      )
  # to be downloaded
  a__												= None
  a__												= None

 @property
 def 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  if self._dummy_file is None:
   a__												= self.download_dummy_data()
  return self._dummy_file

 @property
 def 						_UpperCAmelCase       (	self      ) ->			str:
  if self.config is not None:
   # structure is dummy / config_name / version_name
   return os.path.join('''dummy'''		,				self.config.name		,				self.version_name      )
  # structure is dummy / version_name
  return os.path.join('''dummy'''		,				self.version_name      )

 @property
 def 						_UpperCAmelCase       (	self      ) ->			Union[str, Any]:
  return os.path.join(self.dummy_data_folder		,				'''dummy_data.zip'''      )

 def 						_UpperCAmelCase       (	self      ) ->			Optional[Any]:
  a__												= (
      self.local_path_to_dummy_data if self.use_local_dummy_data is True else self.github_path_to_dummy_data
  )

  a__												= cached_path(
      SCREAMING_SNAKE_CASE		,				cache_dir=self.cache_dir		,				extract_compressed_file=SCREAMING_SNAKE_CASE		,				force_extract=SCREAMING_SNAKE_CASE      )

  return os.path.join(SCREAMING_SNAKE_CASE		,				self.dummy_file_name      )

 @property
 def 						_UpperCAmelCase       (	self      ) ->			str:
  return os.path.join(self.datasets_scripts_dir		,				self.dataset_name		,				self.dummy_zip_file      )

 @property
 def 						_UpperCAmelCase       (	self      ) ->			Any:
  if self._bucket_url is None:
   a__												= hf_github_url(self.dataset_name		,				self.dummy_zip_file.replace(os.sep		,				'''/'''      )      )
  return self._bucket_url

 @property
 def 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  # return full path if its a dir
  if os.path.isdir(self.dummy_file      ):
   return self.dummy_file
  # else cut off path to file -> example `xsum`.
  return "/".join(self.dummy_file.replace(os.sep		,				'''/'''      ).split('''/'''      )[:-1]      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				*SCREAMING_SNAKE_CASE      ) ->			str:
  if self.load_existing_dummy_data:
   # dummy data is downloaded and tested
   a__												= self.dummy_file
  else:
   # dummy data cannot be downloaded and only the path to dummy file is returned
   a__												= self.dummy_file_name

  # special case when data_url is a dict
  if isinstance(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ):
   return self.create_dummy_data_dict(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      )
  elif isinstance(SCREAMING_SNAKE_CASE		,				(list, tuple)      ):
   return self.create_dummy_data_list(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      )
  else:
   return self.create_dummy_data_single(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				*SCREAMING_SNAKE_CASE      ) ->			Tuple:
  return self.download_and_extract(SCREAMING_SNAKE_CASE      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ) ->			Union[str, Any]:
  return self.download_and_extract(SCREAMING_SNAKE_CASE      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				*SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE      ) ->			Any:
  return path

 def 						_UpperCAmelCase       (	self      ) ->			Any:
  return {}

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ) ->			str:
  a__												= {}
  for key, single_urls in data_url.items():
   for download_callback in self.download_callbacks:
    if isinstance(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ):
     for single_url in single_urls:
      download_callback(SCREAMING_SNAKE_CASE      )
    else:
     a__												= single_urls
     download_callback(SCREAMING_SNAKE_CASE      )
            # we force the name of each key to be the last file / folder name of the url path
            # if the url has arguments, we need to encode them with urllib.parse.quote_plus
   if isinstance(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ):
    a__												= [os.path.join(SCREAMING_SNAKE_CASE		,				urllib.parse.quote_plus(Path(SCREAMING_SNAKE_CASE      ).name      )      ) for x in single_urls]
   else:
    a__												= single_urls
    a__												= os.path.join(SCREAMING_SNAKE_CASE		,				urllib.parse.quote_plus(Path(SCREAMING_SNAKE_CASE      ).name      )      )
   a__												= value

  # make sure that values are unique
  if all(isinstance(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ) for i in dummy_data_dict.values()      ) and len(set(dummy_data_dict.values()      )      ) < len(
      dummy_data_dict.values()      ):
   # append key to value to make its name unique
   a__												= {key: value + key for key, value in dummy_data_dict.items()}

  return dummy_data_dict

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ) ->			List[str]:
  a__												= []
  # trick: if there are many shards named like `data.txt-000001-of-00300`, only use the first one
  a__												= all(bool(re.findall('''[0-9]{3,}-of-[0-9]{3,}'''		,				SCREAMING_SNAKE_CASE      )      ) for url in data_url      )
  a__												= all(
      url.startswith('''https://ftp.ncbi.nlm.nih.gov/pubmed/baseline/pubmed'''      ) for url in data_url      )
  if data_url and (is_tf_records or is_pubmed_records):
   a__												= [data_url[0]] * len(SCREAMING_SNAKE_CASE      )
  for single_url in data_url:
   for download_callback in self.download_callbacks:
    download_callback(SCREAMING_SNAKE_CASE      )
   # we force the name of each key to be the last file / folder name of the url path
   # if the url has arguments, we need to encode them with urllib.parse.quote_plus
   a__												= os.path.join(SCREAMING_SNAKE_CASE		,				urllib.parse.quote_plus(single_url.split('''/'''      )[-1]      )      )
   dummy_data_list.append(SCREAMING_SNAKE_CASE      )
  return dummy_data_list

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ) ->			Any:
  for download_callback in self.download_callbacks:
   download_callback(SCREAMING_SNAKE_CASE      )
  # we force the name of each key to be the last file / folder name of the url path
  # if the url has arguments, we need to encode them with urllib.parse.quote_plus
  a__												= os.path.join(SCREAMING_SNAKE_CASE		,				urllib.parse.quote_plus(data_url.split('''/'''      )[-1]      )      )
  if os.path.exists(SCREAMING_SNAKE_CASE      ) or not self.load_existing_dummy_data:
   return value
  else:
   # Backward compatibility, maybe deprecate at one point.
   # For many datasets with single url calls to dl_manager.download_and_extract,
   # the dummy_data.zip file is actually the zipped downloaded file
   # while now we expected the dummy_data.zip file to be a directory containing
   # the downloaded file.
   return path_to_dummy_data

 def 						_UpperCAmelCase       (	self      ) ->			Tuple:
  pass

 def 						_UpperCAmelCase       (	self      ) ->			int:
  pass

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			List[str]:
  def _iter_archive_members(SCREAMING_SNAKE_CASE      ):
   # this preserves the order of the members inside the ZIP archive
   a__												= Path(self.dummy_file      ).parent
   a__												= path.relative_to(SCREAMING_SNAKE_CASE      )
   with ZipFile(self.local_path_to_dummy_data      ) as zip_file:
    a__												= zip_file.namelist()
   for member in members:
    if member.startswith(relative_path.as_posix()      ):
     yield dummy_parent_path.joinpath(SCREAMING_SNAKE_CASE      )

  a__												= Path(SCREAMING_SNAKE_CASE      )
  a__												= _iter_archive_members(SCREAMING_SNAKE_CASE      ) if self.use_local_dummy_data else path.rglob('''*'''      )
  for file_path in file_paths:
   if file_path.is_file() and not file_path.name.startswith(('''.''', '''__''')      ):
    yield file_path.relative_to(SCREAMING_SNAKE_CASE      ).as_posix(), file_path.open('''rb'''      )



 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			Union[str, Any]:
  if not isinstance(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ):
   a__												= [paths]
  for path in paths:
   if os.path.isfile(SCREAMING_SNAKE_CASE      ):
    if os.path.basename(SCREAMING_SNAKE_CASE      ).startswith(('''.''', '''__''')      ):
     return
    yield path
   else:
    for dirpath, dirnames, filenames in os.walk(SCREAMING_SNAKE_CASE      ):
     if os.path.basename(SCREAMING_SNAKE_CASE      ).startswith(('''.''', '''__''')      ):
      continue
     dirnames.sort()
     for filename in sorted(SCREAMING_SNAKE_CASE      ):
      if filename.startswith(('''.''', '''__''')      ):
       continue
      yield os.path.join(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      )










import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


a_      :      Optional[int]    =     False
@skip_mps
class    __UpperCamelCase			(       _lowercase	,	_lowercase	,	_lowercase	,	unittest.TestCase				):

 """simple docstring"""



 _lowercase   :    List[Any]      						=					StableDiffusionAttendAndExcitePipeline
 _lowercase   :    str      						=					False
 _lowercase   :    List[str]      						=					TEXT_TO_IMAGE_PARAMS
 _lowercase   :    Optional[int]      						=					TEXT_TO_IMAGE_BATCH_PARAMS.union({'''token_indices'''}				)
 _lowercase   :    Union[str, Any]      						=					TEXT_TO_IMAGE_IMAGE_PARAMS
 _lowercase   :    List[Any]      						=					TEXT_TO_IMAGE_IMAGE_PARAMS

 @classmethod
 def 						_UpperCAmelCase       (	cls      ) ->			Dict:
  super().setUpClass()
  torch.use_deterministic_algorithms(SCREAMING_SNAKE_CASE      )

 @classmethod
 def 						_UpperCAmelCase       (	cls      ) ->			Tuple:
  super().tearDownClass()
  torch.use_deterministic_algorithms(SCREAMING_SNAKE_CASE      )

 def 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  torch.manual_seed(0      )
  a__												= UNetaDConditionModel(
      block_out_channels=(3_2, 6_4)		,				layers_per_block=1		,				sample_size=3_2		,				in_channels=4		,				out_channels=4		,				down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')		,				up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')		,				cross_attention_dim=3_2		,				attention_head_dim=(2, 4)		,				use_linear_projection=SCREAMING_SNAKE_CASE		,				)
  a__												= DDIMScheduler(
      beta_start=0.0_00_85		,				beta_end=0.0_12		,				beta_schedule='''scaled_linear'''		,				clip_sample=SCREAMING_SNAKE_CASE		,				set_alpha_to_one=SCREAMING_SNAKE_CASE		,				)
  torch.manual_seed(0      )
  a__												= AutoencoderKL(
      block_out_channels=[3_2, 6_4]		,				in_channels=3		,				out_channels=3		,				down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']		,				up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']		,				latent_channels=4		,				sample_size=1_2_8		,				)
  torch.manual_seed(0      )
  a__												= CLIPTextConfig(
      bos_token_id=0		,				eos_token_id=2		,				hidden_size=3_2		,				intermediate_size=3_7		,				layer_norm_eps=1e-0_5		,				num_attention_heads=4		,				num_hidden_layers=5		,				pad_token_id=1		,				vocab_size=1_0_0_0		,				hidden_act='''gelu'''		,				projection_dim=5_1_2		,				)
  a__												= CLIPTextModel(SCREAMING_SNAKE_CASE      )
  a__												= CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''      )

  a__												= {
      '''unet''': unet,
      '''scheduler''': scheduler,
      '''vae''': vae,
      '''text_encoder''': text_encoder,
      '''tokenizer''': tokenizer,
      '''safety_checker''': None,
      '''feature_extractor''': None,
  }

  return components

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE=0      ) ->			Dict:
  if str(SCREAMING_SNAKE_CASE      ).startswith('''mps'''      ):
   a__												= torch.manual_seed(SCREAMING_SNAKE_CASE      )
  else:
   a__												= torch.Generator(device=SCREAMING_SNAKE_CASE      ).manual_seed(SCREAMING_SNAKE_CASE      )
  a__												= a__												= {
      '''prompt''': '''a cat and a frog''',
      '''token_indices''': [2, 5],
      '''generator''': generator,
      '''num_inference_steps''': 1,
      '''guidance_scale''': 6.0,
      '''output_type''': '''numpy''',
      '''max_iter_to_alter''': 2,
      '''thresholds''': {0: 0.7},
  }
  return inputs

 def 						_UpperCAmelCase       (	self      ) ->			str:
  a__												= '''cpu'''

  a__												= self.get_dummy_components()
  a__												= self.pipeline_class(**SCREAMING_SNAKE_CASE      )
  pipe.to(SCREAMING_SNAKE_CASE      )
  pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE      )

  a__												= self.get_dummy_inputs(SCREAMING_SNAKE_CASE      )
  a__												= pipe(**SCREAMING_SNAKE_CASE      ).images
  a__												= image[0, -3:, -3:, -1]

  self.assertEqual(image.shape		,				(1, 6_4, 6_4, 3)      )
  a__												= np.array(
      [0.63_90_53_64, 0.62_89_73_07, 0.48_59_90_17, 0.5_13_36_24, 0.5_55_00_48, 0.45_76_95_16, 0.50_32_69_73, 0.5_02_31_39, 0.45_38_44_96]      )
  a__												= np.abs(image_slice.flatten() - expected_slice      ).max()
  self.assertLessEqual(SCREAMING_SNAKE_CASE		,				1e-3      )

 def 						_UpperCAmelCase       (	self      ) ->			str:
  super().test_cpu_offload_forward_pass(expected_max_diff=5e-4      )

 def 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  # NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
  self._test_inference_batch_consistent(batch_sizes=[1, 2]      )

 def 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  self._test_inference_batch_single_identical(batch_size=2		,				expected_max_diff=7e-4      )

 def 						_UpperCAmelCase       (	self      ) ->			int:
  super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3      )

 def 						_UpperCAmelCase       (	self      ) ->			Optional[Any]:
  super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4      )

 def 						_UpperCAmelCase       (	self      ) ->			Dict:
  super().test_save_load_local(expected_max_difference=5e-4      )



 def 						_UpperCAmelCase       (	self      ) ->			str:
  super().test_save_load_optional_components(expected_max_difference=4e-4      )





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

 """simple docstring"""



 @classmethod
 def 						_UpperCAmelCase       (	cls      ) ->			Optional[int]:
  super().setUpClass()
  torch.use_deterministic_algorithms(SCREAMING_SNAKE_CASE      )

 @classmethod
 def 						_UpperCAmelCase       (	cls      ) ->			str:
  super().tearDownClass()
  torch.use_deterministic_algorithms(SCREAMING_SNAKE_CASE      )

 def 						_UpperCAmelCase       (	self      ) ->			List[str]:
  super().tearDown()
  gc.collect()
  torch.cuda.empty_cache()



 def 						_UpperCAmelCase       (	self      ) ->			str:
  a__												= torch.manual_seed(5_1      )

  a__												= StableDiffusionAttendAndExcitePipeline.from_pretrained(
      '''CompVis/stable-diffusion-v1-4'''		,				safety_checker=SCREAMING_SNAKE_CASE		,				torch_dtype=torch.floataa      )
  pipe.to('''cuda'''      )

  a__												= '''a painting of an elephant with glasses'''
  a__												= [5, 7]

  a__												= pipe(
      prompt=SCREAMING_SNAKE_CASE		,				token_indices=SCREAMING_SNAKE_CASE		,				guidance_scale=7.5		,				generator=SCREAMING_SNAKE_CASE		,				num_inference_steps=5		,				max_iter_to_alter=5		,				output_type='''numpy'''		,				).images[0]

  a__												= load_numpy(
      '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy'''      )
  assert np.abs((expected_image - image).max()      ) < 5e-1





