
"""simple docstring"""




import os
import re
import sys
import traceback
import warnings
from pathlib import Path
from typing import Dict, Optional, Union
from uuid import uuida

from huggingface_hub import HfFolder, ModelCard, ModelCardData, hf_hub_download, whoami
from huggingface_hub.file_download import REGEX_COMMIT_HASH
from huggingface_hub.utils import (
    EntryNotFoundError,
    RepositoryNotFoundError,
    RevisionNotFoundError,
    is_jinja_available,
)
from packaging import version
from requests import HTTPError

from .. import __version__
from .constants import (
    DEPRECATED_REVISION_ARGS,
    DIFFUSERS_CACHE,
    HUGGINGFACE_CO_RESOLVE_ENDPOINT,
    SAFETENSORS_WEIGHTS_NAME,
    WEIGHTS_NAME,
)
from .import_utils import (
    ENV_VARS_TRUE_VALUES,
    _flax_version,
    _jax_version,
    _onnxruntime_version,
    _torch_version,
    is_flax_available,
    is_onnx_available,
    is_torch_available,
)
from .logging import get_logger


lowerCAmelCase__							=				get_logger(__name__)


lowerCAmelCase__							=				Path(__file__).parent / 'model_card_template.md'
lowerCAmelCase__							=				uuida().hex
lowerCAmelCase__							=				os.getenv('HF_HUB_OFFLINE', '').upper() in ENV_VARS_TRUE_VALUES
lowerCAmelCase__							=				os.getenv('DISABLE_TELEMETRY', '').upper() in ENV_VARS_TRUE_VALUES
lowerCAmelCase__							=				HUGGINGFACE_CO_RESOLVE_ENDPOINT + '/api/telemetry/'



def        _lowerCamelCase ( __a = None    ):
		SCREAMING_SNAKE_CASE_						 = F'diffusers/{__version__}; python/{sys.version.split()[0]}; session_id/{SESSION_ID}'
		if DISABLE_TELEMETRY or HF_HUB_OFFLINE:
				return ua + "; telemetry/off"
		if is_torch_available():
				ua += F'; torch/{_torch_version}'
		if is_flax_available():
				ua += F'; jax/{_jax_version}'
				ua += F'; flax/{_flax_version}'
		if is_onnx_available():
				ua += F'; onnxruntime/{_onnxruntime_version}'
		# CI will set this value to True
		if os.environ.get('''DIFFUSERS_IS_CI''',							''''''    ).upper() in ENV_VARS_TRUE_VALUES:
				ua += "; is_ci/true"
		if isinstance(__a,							__a    ):
				ua += "; " + "; ".join(F'{k}/{v}' for k, v in user_agent.items()    )
		elif isinstance(__a,							__a    ):
				ua += "; " + user_agent
		return ua



def        _lowerCamelCase ( __a,							__a = None,							__a = None    ):
		if token is None:
				SCREAMING_SNAKE_CASE_						 = HfFolder.get_token()
		if organization is None:
				SCREAMING_SNAKE_CASE_						 = whoami(__a    )['''name''']
				return F'{username}/{model_id}'
		else:
				return F'{organization}/{model_id}'



def        _lowerCamelCase ( __a,							__a    ):
		if not is_jinja_available():
				raise ValueError(
				    '''Modelcard rendering is based on Jinja templates.'''
				    ''' Please make sure to have `jinja` installed before using `create_model_card`.'''
				    ''' To install it, please run `pip install Jinja2`.'''    )

		if hasattr(__a,							'''local_rank'''    ) and args.local_rank not in [-1, 0]:
				return

		SCREAMING_SNAKE_CASE_						 = args.hub_token if hasattr(__a,							'''hub_token'''    ) else None
		SCREAMING_SNAKE_CASE_						 = get_full_repo_name(__a,							token=__a    )

		SCREAMING_SNAKE_CASE_						 = ModelCard.from_template(
		    card_data=ModelCardData(  # Card metadata object that will be converted to YAML block
		        language='''en''',							license='''apache-2.0''',							library_name='''diffusers''',							tags=[],							datasets=args.dataset_name,							metrics=[],							),							template_path=__a,							model_name=__a,							repo_name=__a,							dataset_name=args.dataset_name if hasattr(__a,							'''dataset_name'''    ) else None,							learning_rate=args.learning_rate,							train_batch_size=args.train_batch_size,							eval_batch_size=args.eval_batch_size,							gradient_accumulation_steps=(
		        args.gradient_accumulation_steps if hasattr(__a,							'''gradient_accumulation_steps'''    ) else None
		    ),							adam_betaa=args.adam_betaa if hasattr(__a,							'''adam_beta1'''    ) else None,							adam_betaa=args.adam_betaa if hasattr(__a,							'''adam_beta2'''    ) else None,							adam_weight_decay=args.adam_weight_decay if hasattr(__a,							'''adam_weight_decay'''    ) else None,							adam_epsilon=args.adam_epsilon if hasattr(__a,							'''adam_epsilon'''    ) else None,							lr_scheduler=args.lr_scheduler if hasattr(__a,							'''lr_scheduler'''    ) else None,							lr_warmup_steps=args.lr_warmup_steps if hasattr(__a,							'''lr_warmup_steps'''    ) else None,							ema_inv_gamma=args.ema_inv_gamma if hasattr(__a,							'''ema_inv_gamma'''    ) else None,							ema_power=args.ema_power if hasattr(__a,							'''ema_power'''    ) else None,							ema_max_decay=args.ema_max_decay if hasattr(__a,							'''ema_max_decay'''    ) else None,							mixed_precision=args.mixed_precision,							)

		SCREAMING_SNAKE_CASE_						 = os.path.join(args.output_dir,							'''README.md'''    )
		model_card.save(__a    )



def        _lowerCamelCase ( __a,							__a = None    ):
		if resolved_file is None or commit_hash is not None:
				return commit_hash
		SCREAMING_SNAKE_CASE_						 = str(Path(__a    ).as_posix()    )
		SCREAMING_SNAKE_CASE_						 = re.search(r'''snapshots/([^/]+)/''',							__a    )
		if search is None:
				return None
		SCREAMING_SNAKE_CASE_						 = search.groups()[0]
		return commit_hash if REGEX_COMMIT_HASH.match(__a    ) else None


# Old default cache path, potentially to be migrated.
# This logic was more or less taken from `transformers`, with the following differences:
# - Diffusers doesn't use custom environment variables to specify the cache path.
# - There is no need to migrate the cache format, just move the files to the new location.
lowerCAmelCase__							=				os.path.expanduser(
    os.getenv('HF_HOME', os.path.join(os.getenv('XDG_CACHE_HOME', '~/.cache'), 'huggingface'))
)
lowerCAmelCase__							=				os.path.join(hf_cache_home, 'diffusers')



def        _lowerCamelCase ( __a = None,							__a = None    ):
		if new_cache_dir is None:
				SCREAMING_SNAKE_CASE_						 = DIFFUSERS_CACHE
		if old_cache_dir is None:
				SCREAMING_SNAKE_CASE_						 = old_diffusers_cache

		SCREAMING_SNAKE_CASE_						 = Path(__a    ).expanduser()
		SCREAMING_SNAKE_CASE_						 = Path(__a    ).expanduser()
		for old_blob_path in old_cache_dir.glob('''**/blobs/*'''    ):
				if old_blob_path.is_file() and not old_blob_path.is_symlink():
						SCREAMING_SNAKE_CASE_						 = new_cache_dir / old_blob_path.relative_to(__a    )
						new_blob_path.parent.mkdir(parents=__a,							exist_ok=__a    )
						os.replace(__a,							__a    )
						try:
								os.symlink(__a,							__a    )
						except OSError:
								logger.warning(
								    '''Could not create symlink between old cache and new cache. If you use an older version of diffusers again, files will be re-downloaded.'''    )
		# At this point, old_cache_dir contains symlinks to the new cache (it can still be used).


lowerCAmelCase__							=				os.path.join(DIFFUSERS_CACHE, 'version_diffusers_cache.txt')
if not os.path.isfile(cache_version_file):
						lowerCAmelCase__							=				0
else:
						with open(cache_version_file) as f:
												try:
																		lowerCAmelCase__							=				int(f.read())
												except ValueError:
																		lowerCAmelCase__							=				0

if cache_version < 1:
						lowerCAmelCase__							=				os.path.isdir(old_diffusers_cache) and len(os.listdir(old_diffusers_cache)) > 0
						if old_cache_is_not_empty:
												logger.warning(
												    'The cache for model files in Diffusers v0.14.0 has moved to a new location. Moving your '
												    'existing cached models. This is a one-time operation, you can interrupt it or run it '
												    'later by calling `diffusers.utils.hub_utils.move_cache()`.'
												)
												try:
																		move_cache()
												except Exception as e:
																		lowerCAmelCase__							=				'\n'.join(traceback.format_tb(e.__traceback__))
																		logger.error(
																		    f'''There was a problem when trying to move your cache:\n\n{trace}\n{e.__class__.__name__}: {e}\n\nPlease '''
																		    'file an issue at https://github.com/huggingface/diffusers/issues/new/choose, copy paste this whole '
																		    'message and we will do our best to help.'
																		)

if cache_version < 1:
						try:
												os.makedirs(DIFFUSERS_CACHE, exist_ok=True)
												with open(cache_version_file, 'w') as f:
																		f.write('1')
						except Exception:
												logger.warning(
												    f'''There was a problem when trying to write in your cache folder ({DIFFUSERS_CACHE}). Please, ensure '''
												    'the directory exists and can be written to.'
												)



def        _lowerCamelCase ( __a,							__a = None    ):
		if variant is not None:
				SCREAMING_SNAKE_CASE_						 = weights_name.split('''.'''    )
				SCREAMING_SNAKE_CASE_						 = splits[:-1] + [variant] + splits[-1:]
				SCREAMING_SNAKE_CASE_						 = '''.'''.join(__a    )

		return weights_name



def        _lowerCamelCase ( __a,							*,
    __a,							__a,							__a,							__a,							__a,							__a,							__a,							__a,							__a,							__a,							__a=None,							):
		SCREAMING_SNAKE_CASE_						 = str(__a    )
		if os.path.isfile(__a    ):
				return pretrained_model_name_or_path
		elif os.path.isdir(__a    ):
				if os.path.isfile(os.path.join(__a,							__a    )    ):
						# Load from a PyTorch checkpoint
						SCREAMING_SNAKE_CASE_						 = os.path.join(__a,							__a    )
						return model_file
				elif subfolder is not None and os.path.isfile(
				    os.path.join(__a,							__a,							__a    )    ):
						SCREAMING_SNAKE_CASE_						 = os.path.join(__a,							__a,							__a    )
						return model_file
				else:
						raise EnvironmentError(
						    F'Error no file named {weights_name} found in directory {pretrained_model_name_or_path}.'    )
		else:
				# 1. First check if deprecated way of loading from branches is used
				if (
				    revision in DEPRECATED_REVISION_ARGS
				    and (weights_name == WEIGHTS_NAME or weights_name == SAFETENSORS_WEIGHTS_NAME)
				    and version.parse(version.parse(__a    ).base_version    ) >= version.parse('''0.20.0'''    )
				):
						try:
								SCREAMING_SNAKE_CASE_						 = hf_hub_download(
								    __a,							filename=_add_variant(__a,							__a    ),							cache_dir=__a,							force_download=__a,							proxies=__a,							resume_download=__a,							local_files_only=__a,							use_auth_token=__a,							user_agent=__a,							subfolder=__a,							revision=revision or commit_hash,							)
								warnings.warn(
								    F'Loading the variant {revision} from {pretrained_model_name_or_path} via `revision=\'{revision}\'` is deprecated. Loading instead from `revision=\'main\'` with `variant={revision}`. Loading model variants via `revision=\'{revision}\'` will be removed in diffusers v1. Please use `variant=\'{revision}\'` instead.',							__a,							)
								return model_file
						except:  # noqa: E722
								warnings.warn(
								    F'You are loading the variant {revision} from {pretrained_model_name_or_path} via `revision=\'{revision}\'`. This behavior is deprecated and will be removed in diffusers v1. One should use `variant=\'{revision}\'` instead. However, it appears that {pretrained_model_name_or_path} currently does not have a {_add_variant(__a,							__a    )} file in the \'main\' branch of {pretrained_model_name_or_path}. \n The Diffusers team and community would be very grateful if you could open an issue: https://github.com/huggingface/diffusers/issues/new with the title \'{pretrained_model_name_or_path} is missing {_add_variant(__a,							__a    )}\' so that the correct variant file can be added.',							__a,							)
				try:
						# 2. Load model file as usual
						SCREAMING_SNAKE_CASE_						 = hf_hub_download(
						    __a,							filename=__a,							cache_dir=__a,							force_download=__a,							proxies=__a,							resume_download=__a,							local_files_only=__a,							use_auth_token=__a,							user_agent=__a,							subfolder=__a,							revision=revision or commit_hash,							)
						return model_file

				except RepositoryNotFoundError:
						raise EnvironmentError(
						    F'{pretrained_model_name_or_path} is not a local folder and is not a valid model identifier '
						    '''listed on \'https://huggingface.co/models\'\nIf this is a private repository, make sure to pass a '''
						    '''token having permission to this repo with `use_auth_token` or log in with `huggingface-cli '''
						    '''login`.'''    )
				except RevisionNotFoundError:
						raise EnvironmentError(
						    F'{revision} is not a valid git identifier (branch name, tag name or commit id) that exists for '
						    '''this model name. Check the model page at '''
						    F'\'https://huggingface.co/{pretrained_model_name_or_path}\' for available revisions.'    )
				except EntryNotFoundError:
						raise EnvironmentError(
						    F'{pretrained_model_name_or_path} does not appear to have a file named {weights_name}.'    )
				except HTTPError as err:
						raise EnvironmentError(
						    F'There was a specific connection error when trying to load {pretrained_model_name_or_path}:\n{err}'    )
				except ValueError:
						raise EnvironmentError(
						    F'We couldn\'t connect to \'{HUGGINGFACE_CO_RESOLVE_ENDPOINT}\' to load this model, couldn\'t find it'
						    F' in the cached files and it looks like {pretrained_model_name_or_path} is not the path to a'
						    F' directory containing a file named {weights_name} or'
						    ''' \nCheckout your internet connection or see how to run the library in'''
						    ''' offline mode at \'https://huggingface.co/docs/diffusers/installation#offline-mode\'.'''    )
				except EnvironmentError:
						raise EnvironmentError(
						    F'Can\'t load the model for \'{pretrained_model_name_or_path}\'. If you were trying to load it from '
						    '''\'https://huggingface.co/models\', make sure you don\'t have a local directory with the same name. '''
						    F'Otherwise, make sure \'{pretrained_model_name_or_path}\' is the correct path to a directory '
						    F'containing a file named {weights_name}'    )

"""simple docstring"""




import gc
import tempfile
import unittest

import numpy as np
import torch

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


lowerCAmelCase__							=				False





class  snake_case     (							unittest.TestCase   ):
							pass







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

							def _lowercase					(self    ):

									"""simple docstring"""







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

							def _lowercase					(self    ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_						 = VersatileDiffusionPipeline.from_pretrained('''shi-labs/versatile-diffusion'''       ,       torch_dtype=torch.floataa    )
									pipe.to(SCREAMING_SNAKE_CASE_    )
									pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_    )

									SCREAMING_SNAKE_CASE_						 = load_image(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'''    )

									SCREAMING_SNAKE_CASE_						 = torch.manual_seed(0    )
									SCREAMING_SNAKE_CASE_						 = pipe.dual_guided(
									    prompt='''first prompt'''       ,       image=SCREAMING_SNAKE_CASE_       ,       text_to_image_strength=0.75       ,       generator=SCREAMING_SNAKE_CASE_       ,       guidance_scale=7.5       ,       num_inference_steps=2       ,       output_type='''numpy'''       ,       ).images

									with tempfile.TemporaryDirectory() as tmpdirname:
											pipe.save_pretrained(SCREAMING_SNAKE_CASE_    )
											SCREAMING_SNAKE_CASE_						 = VersatileDiffusionPipeline.from_pretrained(SCREAMING_SNAKE_CASE_       ,       torch_dtype=torch.floataa    )
									pipe.to(SCREAMING_SNAKE_CASE_    )
									pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_    )

									SCREAMING_SNAKE_CASE_						 = generator.manual_seed(0    )
									SCREAMING_SNAKE_CASE_						 = pipe.dual_guided(
									    prompt='''first prompt'''       ,       image=SCREAMING_SNAKE_CASE_       ,       text_to_image_strength=0.75       ,       generator=SCREAMING_SNAKE_CASE_       ,       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 _lowercase					(self    ):

									"""simple docstring"""







									SCREAMING_SNAKE_CASE_						 = VersatileDiffusionPipeline.from_pretrained('''shi-labs/versatile-diffusion'''       ,       torch_dtype=torch.floataa    )
									pipe.to(SCREAMING_SNAKE_CASE_    )
									pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_    )

									SCREAMING_SNAKE_CASE_						 = '''cyberpunk 2077'''
									SCREAMING_SNAKE_CASE_						 = load_image(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'''    )
									SCREAMING_SNAKE_CASE_						 = torch.manual_seed(0    )
									SCREAMING_SNAKE_CASE_						 = pipe.dual_guided(
									    prompt=SCREAMING_SNAKE_CASE_       ,       image=SCREAMING_SNAKE_CASE_       ,       text_to_image_strength=0.75       ,       generator=SCREAMING_SNAKE_CASE_       ,       guidance_scale=7.5       ,       num_inference_steps=50       ,       output_type='''numpy'''       ,       ).images

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

									assert image.shape == (1, 5_12, 5_12, 3)
									SCREAMING_SNAKE_CASE_						 = np.array([0.14_48, 0.16_19, 0.17_41, 0.10_86, 0.11_47, 0.11_28, 0.11_99, 0.11_65, 0.10_01]    )

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

									SCREAMING_SNAKE_CASE_						 = '''A painting of a squirrel eating a burger '''
									SCREAMING_SNAKE_CASE_						 = torch.manual_seed(0    )
									SCREAMING_SNAKE_CASE_						 = pipe.text_to_image(
									    prompt=SCREAMING_SNAKE_CASE_       ,       generator=SCREAMING_SNAKE_CASE_       ,       guidance_scale=7.5       ,       num_inference_steps=50       ,       output_type='''numpy'''    ).images

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

									assert image.shape == (1, 5_12, 5_12, 3)
									SCREAMING_SNAKE_CASE_						 = np.array([0.33_67, 0.31_69, 0.26_56, 0.38_70, 0.47_90, 0.37_96, 0.40_09, 0.48_78, 0.47_78]    )

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

									SCREAMING_SNAKE_CASE_						 = pipe.image_variation(SCREAMING_SNAKE_CASE_       ,       generator=SCREAMING_SNAKE_CASE_       ,       output_type='''numpy'''    ).images

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

									assert image.shape == (1, 5_12, 5_12, 3)
									SCREAMING_SNAKE_CASE_						 = np.array([0.30_76, 0.31_23, 0.32_84, 0.37_82, 0.37_70, 0.38_94, 0.42_97, 0.43_31, 0.44_56]    )

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