


import importlib
import inspect
from functools import wraps
from typing import TYPE_CHECKING, Optional, Union

from .download.streaming_download_manager import (
    xbasename,
    xdirname,
    xet_parse,
    xexists,
    xgetsize,
    xglob,
    xgzip_open,
    xisdir,
    xisfile,
    xjoin,
    xlistdir,
    xnumpy_load,
    xopen,
    xpandas_read_csv,
    xpandas_read_excel,
    xPath,
    xrelpath,
    xsio_loadmat,
    xsplit,
    xsplitext,
    xwalk,
    xxml_dom_minidom_parse,
)
from .utils.logging import get_logger
from .utils.patching import patch_submodule
from .utils.py_utils import get_imports


a					:				str       =      get_logger(__name__)


if TYPE_CHECKING:
  from .builder import DatasetBuilder






def  _UpperCAmelCase						( __lowerCamelCase			,  __lowerCamelCase = None							)		->Any:


 '''simple docstring'''
 snake_case :			Tuple							=   importlib.import_module(__lowerCamelCase							)

 # TODO(QL): always update the module to add subsequent new authentication
 if hasattr(__lowerCamelCase			,  '''_patched_for_streaming'''							) and module._patched_for_streaming:
  return

 def wrap_auth(__lowerCamelCase							):
  @wraps(__lowerCamelCase							)
  def wrapper(*__lowerCamelCase			,  **__lowerCamelCase							):
   return function(*__lowerCamelCase			,  use_auth_token=__lowerCamelCase			,  **__lowerCamelCase							)

  snake_case :			Dict							=   '''wrap_auth'''
  return wrapper

 # open files in a streaming fashion
 patch_submodule(__lowerCamelCase			,  '''open'''			,  wrap_auth(__lowerCamelCase							)							).start()
 patch_submodule(__lowerCamelCase			,  '''os.listdir'''			,  wrap_auth(__lowerCamelCase							)							).start()
 patch_submodule(__lowerCamelCase			,  '''os.walk'''			,  wrap_auth(__lowerCamelCase							)							).start()
 patch_submodule(__lowerCamelCase			,  '''glob.glob'''			,  wrap_auth(__lowerCamelCase							)							).start()
 # allow to navigate in remote zip files
 patch_submodule(__lowerCamelCase			,  '''os.path.join'''			,  __lowerCamelCase							).start()
 patch_submodule(__lowerCamelCase			,  '''os.path.dirname'''			,  __lowerCamelCase							).start()
 patch_submodule(__lowerCamelCase			,  '''os.path.basename'''			,  __lowerCamelCase							).start()
 patch_submodule(__lowerCamelCase			,  '''os.path.relpath'''			,  __lowerCamelCase							).start()
 patch_submodule(__lowerCamelCase			,  '''os.path.split'''			,  __lowerCamelCase							).start()
 patch_submodule(__lowerCamelCase			,  '''os.path.splitext'''			,  __lowerCamelCase							).start()
 # allow checks on paths
 patch_submodule(__lowerCamelCase			,  '''os.path.exists'''			,  wrap_auth(__lowerCamelCase							)							).start()
 patch_submodule(__lowerCamelCase			,  '''os.path.isdir'''			,  wrap_auth(__lowerCamelCase							)							).start()
 patch_submodule(__lowerCamelCase			,  '''os.path.isfile'''			,  wrap_auth(__lowerCamelCase							)							).start()
 patch_submodule(__lowerCamelCase			,  '''os.path.getsize'''			,  wrap_auth(__lowerCamelCase							)							).start()
 patch_submodule(__lowerCamelCase			,  '''pathlib.Path'''			,  __lowerCamelCase							).start()
 # file readers
 patch_submodule(__lowerCamelCase			,  '''gzip.open'''			,  wrap_auth(__lowerCamelCase							)							).start()
 patch_submodule(__lowerCamelCase			,  '''numpy.load'''			,  wrap_auth(__lowerCamelCase							)							).start()
 patch_submodule(__lowerCamelCase			,  '''pandas.read_csv'''			,  wrap_auth(__lowerCamelCase							)			,  attrs=['''__version__''']							).start()
 patch_submodule(__lowerCamelCase			,  '''pandas.read_excel'''			,  wrap_auth(__lowerCamelCase							)			,  attrs=['''__version__''']							).start()
 patch_submodule(__lowerCamelCase			,  '''scipy.io.loadmat'''			,  wrap_auth(__lowerCamelCase							)			,  attrs=['''__version__''']							).start()
 patch_submodule(__lowerCamelCase			,  '''xml.etree.ElementTree.parse'''			,  wrap_auth(__lowerCamelCase							)							).start()
 patch_submodule(__lowerCamelCase			,  '''xml.dom.minidom.parse'''			,  wrap_auth(__lowerCamelCase							)							).start()
 snake_case :			Optional[Any]							=   True






def  _UpperCAmelCase						( __lowerCamelCase							)		->List[Any]:


 '''simple docstring'''
 extend_module_for_streaming(builder.__module__			,  use_auth_token=builder.use_auth_token							)
 # if needed, we also have to extend additional internal imports (like wmt14 -> wmt_utils)
 if not builder.__module__.startswith('''datasets.'''							):  # check that it's not a packaged builder like csv
  for imports in get_imports(inspect.getfile(builder.__class__							)							):
   if imports[0] == "internal":
    snake_case :			Optional[Any]							=   imports[1]
    snake_case :			Any							=   '''.'''.join(builder.__module__.split('''.'''							)[:-1] + [internal_import_name]							)
    extend_module_for_streaming(__lowerCamelCase			,  use_auth_token=builder.use_auth_token							)

    # builders can inherit from other builders that might use streaming functionality
    # (for example, ImageFolder and AudioFolder inherit from FolderBuilder which implements examples generation)
    # but these parents builders are not patched automatically as they are not instantiated, so we patch them here
 from .builder import DatasetBuilder

 snake_case :			str							=   [
     cls.__module__
     for cls in type(__lowerCamelCase							).__mro__[1:]  # make sure it's not the same module we've already patched
     if issubclass(cls			,  __lowerCamelCase							) and cls.__module__ != DatasetBuilder.__module__
 ]  # check it's not a standard builder from datasets.builder
 for module in parent_builder_modules:
  extend_module_for_streaming(__lowerCamelCase			,  use_auth_token=builder.use_auth_token							)










import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionDualGuidedPipeline
from diffusers.utils.testing_utils import load_image, nightly, require_torch_gpu, torch_device


a					:				Optional[int]       =      False

@nightly
@require_torch_gpu
class   lowerCamelCase						(	unittest.TestCase  ):

       def 					UpperCAmelCase					(      self       ):






        '''simple docstring'''






        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()

       def 					UpperCAmelCase					(      self       ):






        '''simple docstring'''






        snake_case :			Optional[int]							=   VersatileDiffusionDualGuidedPipeline.from_pretrained('''shi-labs/versatile-diffusion'''       )
        # remove text_unet
        pipe.remove_unused_weights()
        pipe.to(a__       )
        pipe.set_progress_bar_config(disable=a__       )

        snake_case :			Union[str, Any]							=   load_image(
            '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'''       )

        snake_case :			str							=   torch.manual_seed(0       )
        snake_case :			Tuple							=   pipe(
            prompt='''first prompt'''						,    image=a__						,    text_to_image_strength=0.75						,    generator=a__						,    guidance_scale=7.5						,    num_inference_steps=2						,    output_type='''numpy'''						,    ).images

        with tempfile.TemporaryDirectory() as tmpdirname:
         pipe.save_pretrained(a__       )
         snake_case :			str							=   VersatileDiffusionDualGuidedPipeline.from_pretrained(a__       )

        pipe.to(a__       )
        pipe.set_progress_bar_config(disable=a__       )

        snake_case :			Optional[Any]							=   generator.manual_seed(0       )
        snake_case :			List[str]							=   pipe(
            prompt='''first prompt'''						,    image=a__						,    text_to_image_strength=0.75						,    generator=a__						,    guidance_scale=7.5						,    num_inference_steps=2						,    output_type='''numpy'''						,    ).images

        assert np.abs(image - new_image       ).sum() < 1e-5, "Models don't have the same forward pass"



       def 					UpperCAmelCase					(      self       ):






        '''simple docstring'''






        snake_case :			Tuple							=   VersatileDiffusionDualGuidedPipeline.from_pretrained('''shi-labs/versatile-diffusion'''       )
        pipe.remove_unused_weights()
        pipe.to(a__       )
        pipe.set_progress_bar_config(disable=a__       )

        snake_case :			int							=   '''cyberpunk 2077'''
        snake_case :			Dict							=   load_image(
            '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'''       )
        snake_case :			Optional[int]							=   torch.manual_seed(0       )
        snake_case :			Union[str, Any]							=   pipe(
            prompt=a__						,    image=a__						,    text_to_image_strength=0.75						,    generator=a__						,    guidance_scale=7.5						,    num_inference_steps=50						,    output_type='''numpy'''						,    ).images

        snake_case :			str							=   image[0, 2_53:2_56, 2_53:2_56, -1]

        assert image.shape == (1, 5_12, 5_12, 3)
        snake_case :			int							=   np.array([0.0787, 0.0849, 0.0826, 0.0812, 0.0807, 0.0795, 0.0818, 0.0798, 0.0779]       )

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






