



"""simple docstring"""




# 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


_lowerCAmelCase							:   str				=				get_logger(__name__)






class       UpperCAmelCase_    :
      __SCREAMING_SNAKE_CASE		:  Tuple						=					'dummy_data'
      __SCREAMING_SNAKE_CASE		:  str						=					'datasets'
      __SCREAMING_SNAKE_CASE		:  List[Any]						=					False



      def __init__(	self	:      Optional[Any]	,      A	:      str	,      A	:      str	,      A	:      Union[Version, str]	,      A	:      Optional[str] = None	,      A	:      bool = False	,      A	:      bool = True	,      A	:      Optional[List[Callable]] = None	,      ):
           _UpperCAmelCase     :      Any														=       0
           _UpperCAmelCase     :      Tuple														=       dataset_name
           _UpperCAmelCase     :      Any														=       cache_dir
           _UpperCAmelCase     :      List[Any]														=       use_local_dummy_data
           _UpperCAmelCase     :      str														=       config
           # download_callbacks take a single url as input
           _UpperCAmelCase     :      List[Callable]														=       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
           _UpperCAmelCase     :      Optional[Any]														=       load_existing_dummy_data

           # TODO(PVP, QL) might need to make this more general
           _UpperCAmelCase     :      List[str]														=       str(A		)
           # to be downloaded
           _UpperCAmelCase     :      Optional[int]														=       None
           _UpperCAmelCase     :      List[str]														=       None



      @property
      def 		snake_case_   (	self	:      Tuple		):
           if self._dummy_file is None:
                _UpperCAmelCase     :      int														=       self.download_dummy_data()
           return self._dummy_file



      @property
      def 		snake_case_   (	self	:      Any		):
           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 		snake_case_   (	self	:      Tuple		):
           return os.path.join(self.dummy_data_folder	,      "dummy_data.zip"		)



      def 		snake_case_   (	self	:      Optional[Any]		):
           _UpperCAmelCase     :      Union[str, Any]														=       (
               self.local_path_to_dummy_data if self.use_local_dummy_data is True else self.github_path_to_dummy_data
           )

           _UpperCAmelCase     :      List[Any]														=       cached_path(
               A	,      cache_dir=self.cache_dir	,      extract_compressed_file=A	,      force_extract=A		)

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



      @property
      def 		snake_case_   (	self	:      Dict		):
           return os.path.join(self.datasets_scripts_dir	,      self.dataset_name	,      self.dummy_zip_file		)



      @property
      def 		snake_case_   (	self	:      Optional[int]		):
           if self._bucket_url is None:
                _UpperCAmelCase     :      Tuple														=       hf_github_url(self.dataset_name	,      self.dummy_zip_file.replace(os.sep	,      "/"		)		)
           return self._bucket_url



      @property
      def 		snake_case_   (	self	:      Tuple		):
           # 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 		snake_case_   (	self	:      Tuple	,      A	:      Any	,      *A	:      Dict		):
           if self.load_existing_dummy_data:
                # dummy data is downloaded and tested
                _UpperCAmelCase     :      int														=       self.dummy_file
           else:
                # dummy data cannot be downloaded and only the path to dummy file is returned
                _UpperCAmelCase     :      Dict														=       self.dummy_file_name

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



      def 		snake_case_   (	self	:      str	,      A	:      int	,      *A	:      Optional[int]		):
           return self.download_and_extract(A		)



      def 		snake_case_   (	self	:      List[str]	,      A	:      Optional[Any]	,      A	:      Tuple		):
           return self.download_and_extract(A		)



      def 		snake_case_   (	self	:      Any	,      A	:      List[Any]	,      *A	:      Optional[int]	,      **A	:      Optional[Any]		):
           return path



      def 		snake_case_   (	self	:      Optional[int]		):
           return {}



      def 		snake_case_   (	self	:      List[str]	,      A	:      List[str]	,      A	:      Tuple		):
           _UpperCAmelCase     :      str														=       {}
           for key, single_urls in data_url.items():
                for download_callback in self.download_callbacks:
                     if isinstance(A	,      A		):
                          for single_url in single_urls:
                               download_callback(A		)
                     else:
                          _UpperCAmelCase     :      int														=       single_urls
                          download_callback(A		)
            # 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(A	,      A		):
                     _UpperCAmelCase     :      Optional[int]														=       [os.path.join(A	,      urllib.parse.quote_plus(Path(A		).name		)		) for x in single_urls]
                else:
                     _UpperCAmelCase     :      Union[str, Any]														=       single_urls
                     _UpperCAmelCase     :      Any														=       os.path.join(A	,      urllib.parse.quote_plus(Path(A		).name		)		)
                _UpperCAmelCase     :      Dict														=       value

           # make sure that values are unique
           if all(isinstance(A	,      A		) 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
                _UpperCAmelCase     :      Tuple														=       {key: value + key for key, value in dummy_data_dict.items()}

           return dummy_data_dict



      def 		snake_case_   (	self	:      Tuple	,      A	:      Optional[Any]	,      A	:      Union[str, Any]		):
           _UpperCAmelCase     :      Union[str, Any]														=       []
           # trick: if there are many shards named like `data.txt-000001-of-00300`, only use the first one
           _UpperCAmelCase     :      List[Any]														=       all(bool(re.findall("[0-9]{3,}-of-[0-9]{3,}"	,      A		)		) for url in data_url		)
           _UpperCAmelCase     :      str														=       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):
                _UpperCAmelCase     :      Optional[int]														=       [data_url[0]] * len(A		)
           for single_url in data_url:
                for download_callback in self.download_callbacks:
                     download_callback(A		)
                # 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
                _UpperCAmelCase     :      Optional[Any]														=       os.path.join(A	,      urllib.parse.quote_plus(single_url.split("/"		)[-1]		)		)
                dummy_data_list.append(A		)
           return dummy_data_list



      def 		snake_case_   (	self	:      Optional[int]	,      A	:      str	,      A	:      List[Any]		):
           for download_callback in self.download_callbacks:
                download_callback(A		)
           # 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
           _UpperCAmelCase     :      Dict														=       os.path.join(A	,      urllib.parse.quote_plus(data_url.split("/"		)[-1]		)		)
           if os.path.exists(A		) 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 		snake_case_   (	self	:      Dict		):
           pass



      def 		snake_case_   (	self	:      Tuple		):
           pass



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

           _UpperCAmelCase     :      str														=       Path(A		)
           _UpperCAmelCase     :      Optional[Any]														=       _iter_archive_members(A		) 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(A		).as_posix(), file_path.open("rb"		)



      def 		snake_case_   (	self	:      List[str]	,      A	:      Optional[Any]		):
           if not isinstance(A	,      A		):
                _UpperCAmelCase     :      str														=       [paths]
           for path in paths:
                if os.path.isfile(A		):
                     if os.path.basename(A		).startswith((".", "__")		):
                          return
                     yield path
                else:
                     for dirpath, dirnames, filenames in os.walk(A		):
                          if os.path.basename(A		).startswith((".", "__")		):
                               continue
                          dirnames.sort()
                          for filename in sorted(A		):
                               if filename.startswith((".", "__")		):
                                    continue
                               yield os.path.join(A	,      A		)





"""simple docstring"""




import os
import tempfile
import unittest

import numpy as np

from diffusers.utils import is_flax_available
from diffusers.utils.testing_utils import require_flax, slow


if is_flax_available():
 import jax
 import jax.numpy as jnp
 from flax.jax_utils import replicate
 from flax.training.common_utils import shard

 from diffusers import FlaxDDIMScheduler, FlaxDiffusionPipeline, FlaxStableDiffusionPipeline






@require_flax
class       UpperCAmelCase_    (							unittest.TestCase		):



      def 		snake_case_   (	self	:      Optional[int]		):
           with tempfile.TemporaryDirectory() as tmpdirname:
                # pipeline has Flax weights
                _UpperCAmelCase     :      Union[str, Any]														=       FlaxDiffusionPipeline.from_pretrained(
                    "hf-internal-testing/tiny-stable-diffusion-pipe"	,      safety_checker=A	,      cache_dir=A		)

                _UpperCAmelCase     :      List[str]														=       [t[-1] for t in os.walk(os.path.join(A	,      os.listdir(A		)[0]	,      "snapshots"		)		)]
                _UpperCAmelCase     :      Union[str, Any]														=       [item for sublist in all_root_files for item in sublist]

                # None of the downloaded files should be a PyTorch file even if we have some here:
                # https://huggingface.co/hf-internal-testing/tiny-stable-diffusion-pipe/blob/main/unet/diffusion_pytorch_model.bin
                assert not any(f.endswith(".bin"		) for f in files		)







@slow
@require_flax
class       UpperCAmelCase_    (							unittest.TestCase		):



      def 		snake_case_   (	self	:      List[str]		):
           _UpperCAmelCase							,			_UpperCAmelCase     :      int														=       FlaxStableDiffusionPipeline.from_pretrained(
               "hf-internal-testing/tiny-stable-diffusion-pipe"	,      safety_checker=A		)

           _UpperCAmelCase     :      str														=       (
               "A cinematic film still of Morgan Freeman starring as Jimi Hendrix, portrait, 40mm lens, shallow depth of"
               " field, close up, split lighting, cinematic"
           )

           _UpperCAmelCase     :      int														=       jax.random.PRNGKey(0		)
           _UpperCAmelCase     :      List[str]														=       4

           _UpperCAmelCase     :      Any														=       jax.device_count()
           _UpperCAmelCase     :      Tuple														=       num_samples * [prompt]
           _UpperCAmelCase     :      Optional[int]														=       pipeline.prepare_inputs(A		)

           # shard inputs and rng
           _UpperCAmelCase     :      Union[str, Any]														=       replicate(A		)
           _UpperCAmelCase     :      List[Any]														=       jax.random.split(A	,      A		)
           _UpperCAmelCase     :      List[Any]														=       shard(A		)

           _UpperCAmelCase     :      Tuple														=       pipeline(A	,      A	,      A	,      A	,      jit=A		).images

           assert images.shape == (num_samples, 1, 6_4, 6_4, 3)
           if jax.device_count() == 8:
                assert np.abs(np.abs(images[0, 0, :2, :2, -2:]	,      dtype=np.floataa		).sum() - 4.1_514_745		) < 1e-3
                assert np.abs(np.abs(A	,      dtype=np.floataa		).sum() - 49_947.875		) < 5e-1

           _UpperCAmelCase     :      Optional[int]														=       pipeline.numpy_to_pil(np.asarray(images.reshape((num_samples,) + images.shape[-3:]		)		)		)
           assert len(A		) == num_samples



      def 		snake_case_   (	self	:      Union[str, Any]		):
           _UpperCAmelCase							,			_UpperCAmelCase     :      Optional[Any]														=       FlaxStableDiffusionPipeline.from_pretrained(
               "CompVis/stable-diffusion-v1-4"	,      revision="flax"	,      safety_checker=A		)

           _UpperCAmelCase     :      Dict														=       (
               "A cinematic film still of Morgan Freeman starring as Jimi Hendrix, portrait, 40mm lens, shallow depth of"
               " field, close up, split lighting, cinematic"
           )

           _UpperCAmelCase     :      Dict														=       jax.random.PRNGKey(0		)
           _UpperCAmelCase     :      str														=       5_0

           _UpperCAmelCase     :      List[str]														=       jax.device_count()
           _UpperCAmelCase     :      Any														=       num_samples * [prompt]
           _UpperCAmelCase     :      Dict														=       pipeline.prepare_inputs(A		)

           # shard inputs and rng
           _UpperCAmelCase     :      Dict														=       replicate(A		)
           _UpperCAmelCase     :      str														=       jax.random.split(A	,      A		)
           _UpperCAmelCase     :      Optional[Any]														=       shard(A		)

           _UpperCAmelCase     :      List[Any]														=       pipeline(A	,      A	,      A	,      A	,      jit=A		).images

           assert images.shape == (num_samples, 1, 5_1_2, 5_1_2, 3)
           if jax.device_count() == 8:
                assert np.abs((np.abs(images[0, 0, :2, :2, -2:]	,      dtype=np.floataa		).sum() - 0.05_652_401)		) < 1e-3
                assert np.abs((np.abs(A	,      dtype=np.floataa		).sum() - 2_383_808.2)		) < 5e-1



      def 		snake_case_   (	self	:      List[Any]		):
           _UpperCAmelCase							,			_UpperCAmelCase     :      List[Any]														=       FlaxStableDiffusionPipeline.from_pretrained(
               "CompVis/stable-diffusion-v1-4"	,      revision="bf16"	,      dtype=jnp.bfloataa	,      safety_checker=A		)

           _UpperCAmelCase     :      Any														=       (
               "A cinematic film still of Morgan Freeman starring as Jimi Hendrix, portrait, 40mm lens, shallow depth of"
               " field, close up, split lighting, cinematic"
           )

           _UpperCAmelCase     :      List[Any]														=       jax.random.PRNGKey(0		)
           _UpperCAmelCase     :      List[Any]														=       5_0

           _UpperCAmelCase     :      Union[str, Any]														=       jax.device_count()
           _UpperCAmelCase     :      Optional[int]														=       num_samples * [prompt]
           _UpperCAmelCase     :      Union[str, Any]														=       pipeline.prepare_inputs(A		)

           # shard inputs and rng
           _UpperCAmelCase     :      int														=       replicate(A		)
           _UpperCAmelCase     :      Optional[int]														=       jax.random.split(A	,      A		)
           _UpperCAmelCase     :      str														=       shard(A		)

           _UpperCAmelCase     :      Union[str, Any]														=       pipeline(A	,      A	,      A	,      A	,      jit=A		).images

           assert images.shape == (num_samples, 1, 5_1_2, 5_1_2, 3)
           if jax.device_count() == 8:
                assert np.abs((np.abs(images[0, 0, :2, :2, -2:]	,      dtype=np.floataa		).sum() - 0.04_003_906)		) < 1e-3
                assert np.abs((np.abs(A	,      dtype=np.floataa		).sum() - 2_373_516.75)		) < 5e-1



      def 		snake_case_   (	self	:      int		):
           _UpperCAmelCase							,			_UpperCAmelCase     :      Dict														=       FlaxStableDiffusionPipeline.from_pretrained(
               "CompVis/stable-diffusion-v1-4"	,      revision="bf16"	,      dtype=jnp.bfloataa		)

           _UpperCAmelCase     :      str														=       (
               "A cinematic film still of Morgan Freeman starring as Jimi Hendrix, portrait, 40mm lens, shallow depth of"
               " field, close up, split lighting, cinematic"
           )

           _UpperCAmelCase     :      Optional[int]														=       jax.random.PRNGKey(0		)
           _UpperCAmelCase     :      Dict														=       5_0

           _UpperCAmelCase     :      Union[str, Any]														=       jax.device_count()
           _UpperCAmelCase     :      str														=       num_samples * [prompt]
           _UpperCAmelCase     :      int														=       pipeline.prepare_inputs(A		)

           # shard inputs and rng
           _UpperCAmelCase     :      Optional[int]														=       replicate(A		)
           _UpperCAmelCase     :      Tuple														=       jax.random.split(A	,      A		)
           _UpperCAmelCase     :      Dict														=       shard(A		)

           _UpperCAmelCase     :      Optional[Any]														=       pipeline(A	,      A	,      A	,      A	,      jit=A		).images

           assert images.shape == (num_samples, 1, 5_1_2, 5_1_2, 3)
           if jax.device_count() == 8:
                assert np.abs((np.abs(images[0, 0, :2, :2, -2:]	,      dtype=np.floataa		).sum() - 0.04_003_906)		) < 1e-3
                assert np.abs((np.abs(A	,      dtype=np.floataa		).sum() - 2_373_516.75)		) < 5e-1



      def 		snake_case_   (	self	:      Optional[Any]		):
           _UpperCAmelCase     :      Dict														=       FlaxDDIMScheduler(
               beta_start=0.00_085	,      beta_end=0.012	,      beta_schedule="scaled_linear"	,      set_alpha_to_one=A	,      steps_offset=1	,      )

           _UpperCAmelCase							,			_UpperCAmelCase     :      Dict														=       FlaxStableDiffusionPipeline.from_pretrained(
               "CompVis/stable-diffusion-v1-4"	,      revision="bf16"	,      dtype=jnp.bfloataa	,      scheduler=A	,      safety_checker=A	,      )
           _UpperCAmelCase     :      Union[str, Any]														=       scheduler.create_state()

           _UpperCAmelCase     :      str														=       scheduler_state

           _UpperCAmelCase     :      Union[str, Any]														=       (
               "A cinematic film still of Morgan Freeman starring as Jimi Hendrix, portrait, 40mm lens, shallow depth of"
               " field, close up, split lighting, cinematic"
           )

           _UpperCAmelCase     :      str														=       jax.random.PRNGKey(0		)
           _UpperCAmelCase     :      List[str]														=       5_0

           _UpperCAmelCase     :      Optional[int]														=       jax.device_count()
           _UpperCAmelCase     :      int														=       num_samples * [prompt]
           _UpperCAmelCase     :      Optional[int]														=       pipeline.prepare_inputs(A		)

           # shard inputs and rng
           _UpperCAmelCase     :      List[Any]														=       replicate(A		)
           _UpperCAmelCase     :      Tuple														=       jax.random.split(A	,      A		)
           _UpperCAmelCase     :      Union[str, Any]														=       shard(A		)

           _UpperCAmelCase     :      Dict														=       pipeline(A	,      A	,      A	,      A	,      jit=A		).images

           assert images.shape == (num_samples, 1, 5_1_2, 5_1_2, 3)
           if jax.device_count() == 8:
                assert np.abs((np.abs(images[0, 0, :2, :2, -2:]	,      dtype=np.floataa		).sum() - 0.045_043_945)		) < 1e-3
                assert np.abs((np.abs(A	,      dtype=np.floataa		).sum() - 2_347_693.5)		) < 5e-1



      def 		snake_case_   (	self	:      int		):
           _UpperCAmelCase     :      Optional[Any]														=       (
               "A cinematic film still of Morgan Freeman starring as Jimi Hendrix, portrait, 40mm lens, shallow depth of"
               " field, close up, split lighting, cinematic"
           )

           _UpperCAmelCase     :      Optional[Any]														=       jax.device_count()
           _UpperCAmelCase     :      Union[str, Any]														=       num_samples * [prompt]
           _UpperCAmelCase     :      Optional[Any]														=       jax.random.split(jax.random.PRNGKey(0		)	,      A		)

           _UpperCAmelCase							,			_UpperCAmelCase     :      str														=       FlaxStableDiffusionPipeline.from_pretrained(
               "CompVis/stable-diffusion-v1-4"	,      revision="bf16"	,      dtype=jnp.bfloataa	,      safety_checker=A	,      )

           _UpperCAmelCase     :      int														=       replicate(A		)
           _UpperCAmelCase     :      Union[str, Any]														=       pipeline.prepare_inputs(A		)
           _UpperCAmelCase     :      int														=       shard(A		)
           _UpperCAmelCase     :      Dict														=       pipeline(A	,      A	,      A	,      jit=A		).images
           assert images.shape == (num_samples, 1, 5_1_2, 5_1_2, 3)
           _UpperCAmelCase     :      List[str]														=       images[2, 0, 2_5_6, 1_0:1_7, 1]

           # With memory efficient attention
           _UpperCAmelCase							,			_UpperCAmelCase     :      str														=       FlaxStableDiffusionPipeline.from_pretrained(
               "CompVis/stable-diffusion-v1-4"	,      revision="bf16"	,      dtype=jnp.bfloataa	,      safety_checker=A	,      use_memory_efficient_attention=A	,      )

           _UpperCAmelCase     :      Any														=       replicate(A		)
           _UpperCAmelCase     :      str														=       pipeline.prepare_inputs(A		)
           _UpperCAmelCase     :      Tuple														=       shard(A		)
           _UpperCAmelCase     :      Tuple														=       pipeline(A	,      A	,      A	,      jit=A		).images
           assert images_eff.shape == (num_samples, 1, 5_1_2, 5_1_2, 3)
           _UpperCAmelCase     :      str														=       images[2, 0, 2_5_6, 1_0:1_7, 1]

           # I checked the results visually and they are very similar. However, I saw that the max diff is `1` and the `sum`
           # over the 8 images is exactly `256`, which is very suspicious. Testing a random slice for now.
           assert abs(slice_eff - slice		).max() < 1e-2
