


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:						Tuple   				=       get_logger(__name__)


_lowerCAmelCase:						List[str]   				=       Path(__file__).parent / 'model_card_template.md'
_lowerCAmelCase:						Any   				=       uuida().hex
_lowerCAmelCase:						List[Any]   				=       os.getenv('HF_HUB_OFFLINE', '').upper() in ENV_VARS_TRUE_VALUES
_lowerCAmelCase:						int   				=       os.getenv('DISABLE_TELEMETRY', '').upper() in ENV_VARS_TRUE_VALUES
_lowerCAmelCase:						Tuple   				=       HUGGINGFACE_CO_RESOLVE_ENDPOINT + '/api/telemetry/'






def   _lowercase(					__a				:						Union[Dict, str, None] = None     ):
							a__      =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   _lowercase(					__a				:						str    ,     __a				:						Optional[str] = None    ,     __a				:						Optional[str] = None     ):
							if token is None:
														a__      =HfFolder.get_token()
							if organization is None:
														a__      =whoami(__a     )['name']
														return f"""{username}/{model_id}"""
							else:
														return f"""{organization}/{model_id}"""






def   _lowercase(					__a				:						Union[str, Any]    ,     __a				:						Dict     ):
							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

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

							a__      =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    ,     )

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






def   _lowercase(					__a				:						Optional[str]    ,     __a				:						Optional[str] = None     ):
							if resolved_file is None or commit_hash is not None:
														return commit_hash
							a__      =str(Path(__a     ).as_posix()     )
							a__      =re.search(r'snapshots/([^/]+)/'    ,     __a     )
							if search is None:
														return None
							a__      =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:						List[str]   				=       os.path.expanduser(
    os.getenv('HF_HOME', os.path.join(os.getenv('XDG_CACHE_HOME', '~/.cache'), 'huggingface'))
)
_lowerCAmelCase:						List[str]   				=       os.path.join(hf_cache_home, 'diffusers')






def   _lowercase(					__a				:						Optional[str] = None    ,     __a				:						Optional[str] = None     ):
							if new_cache_dir is None:
														a__      =DIFFUSERS_CACHE
							if old_cache_dir is None:
														a__      =old_diffusers_cache

							a__      =Path(__a     ).expanduser()
							a__      =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():
																					a__      =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:						Dict   				=       os.path.join(DIFFUSERS_CACHE, 'version_diffusers_cache.txt')
if not os.path.isfile(cache_version_file):
	_lowerCAmelCase:						int   				=       0
else:
	with open(cache_version_file) as f:
		try:
			_lowerCAmelCase:						List[Any]   				=       int(f.read())
		except ValueError:
			_lowerCAmelCase:						Any   				=       0

if cache_version < 1:
	_lowerCAmelCase:						str   				=       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:						Optional[Any]   				=       '\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   _lowercase(					__a				:						str    ,     __a				:						Optional[str] = None     ):
							if variant is not None:
														a__      =weights_name.split('.'     )
														a__      =splits[:-1] + [variant] + splits[-1:]
														a__      ='.'.join(__a     )

							return weights_name







def   _lowercase(					__a				:						Union[str, Any]    ,     *,
    __a				:						Optional[Any]    ,     __a				:						Optional[Any]    ,     __a				:						List[Any]    ,     __a				:						Tuple    ,     __a				:						Optional[Any]    ,     __a				:						Dict    ,     __a				:						str    ,     __a				:						int    ,     __a				:						Tuple    ,     __a				:						Union[str, Any]    ,     __a				:						int=None    ,     ):
							a__      =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
																					a__      =os.path.join(__a    ,     __a     )
																					return model_file
														elif subfolder is not None and os.path.isfile(
														    os.path.join(__a    ,     __a    ,     __a     )     ):
																					a__      =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:
																												a__      =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
																					a__      =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}"""     )





import copy
import inspect
import unittest

from transformers import AutoBackbone
from transformers.configuration_utils import PretrainedConfig
from transformers.testing_utils import require_timm, require_torch, torch_device
from transformers.utils.import_utils import is_torch_available

from ...test_backbone_common import BackboneTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor


if is_torch_available():
	import torch

	from transformers import TimmBackbone, TimmBackboneConfig

from ...test_pipeline_mixin import PipelineTesterMixin




class       lowercase_					:





				def __init__(				self       ,  lowercase_       ,  lowercase_=None       ,  lowercase_=None       ,  lowercase_=None       ,  lowercase_="resnet50"       ,  lowercase_=3       ,  lowercase_=32       ,  lowercase_=3       ,  lowercase_=True       ,  lowercase_=True       ,  )  -> Union[str, Any]:
											a__      =parent
											a__      =out_indices if out_indices is not None else [4]
											a__      =stage_names
											a__      =out_features
											a__      =backbone
											a__      =batch_size
											a__      =image_size
											a__      =num_channels
											a__      =use_pretrained_backbone
											a__      =is_training





				def 						__UpperCamelCase    (				self)  -> Optional[Any]:
											a__      =floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size])
											a__      =self.get_config()

											return config, pixel_values





				def 						__UpperCamelCase    (				self)  -> Tuple:
											return TimmBackboneConfig(
											    image_size=self.image_size       ,  num_channels=self.num_channels       ,  out_features=self.out_features       ,  out_indices=self.out_indices       ,  stage_names=self.stage_names       ,  use_pretrained_backbone=self.use_pretrained_backbone       ,  backbone=self.backbone       ,  )





				def 						__UpperCamelCase    (				self       ,  lowercase_       ,  lowercase_)  -> str:
											a__      =TimmBackbone(config=lowercase_)
											model.to(lowercase_)
											model.eval()
											with torch.no_grad():
																		a__      =model(lowercase_)
											self.parent.assertEqual(
											    result.feature_map[-1].shape       ,  (self.batch_size, model.channels[-1], 14, 14)       ,  )





				def 						__UpperCamelCase    (				self)  -> str:
											a__      =self.prepare_config_and_inputs()
											a__    ,     a__      =config_and_inputs
											a__      ={'pixel_values': pixel_values}
											return config, inputs_dict







@require_torch
@require_timm
class       lowercase_					(lowercase__      ,					lowercase__      ,					lowercase__      ,					unittest.TestCase   ):
				snake_case	    =(TimmBackbone,) if is_torch_available() else ()
				snake_case	    ={'feature-extraction': TimmBackbone} if is_torch_available() else {}
				snake_case	    =False
				snake_case	    =False
				snake_case	    =False
				snake_case	    =False





				def 						__UpperCamelCase    (				self)  -> Optional[Any]:
											a__      =TimmBackboneModelTester(self)
											a__      =ConfigTester(self       ,  config_class=lowercase_       ,  has_text_modality=lowercase_)





				def 						__UpperCamelCase    (				self)  -> Dict:
											self.config_tester.create_and_test_config_to_json_string()
											self.config_tester.create_and_test_config_to_json_file()
											self.config_tester.create_and_test_config_from_and_save_pretrained()
											self.config_tester.create_and_test_config_with_num_labels()
											self.config_tester.check_config_can_be_init_without_params()
											self.config_tester.check_config_arguments_init()





				def 						__UpperCamelCase    (				self)  -> str:
											a__      ='resnet18'
											a__      ='microsoft/resnet-18'

											a__      =AutoBackbone.from_pretrained(lowercase_       ,  use_timm_backbone=lowercase_)
											a__      =AutoBackbone.from_pretrained(lowercase_)

											self.assertEqual(len(timm_model.out_features)       ,  len(transformers_model.out_features))
											self.assertEqual(len(timm_model.stage_names)       ,  len(transformers_model.stage_names))
											self.assertEqual(timm_model.channels       ,  transformers_model.channels)
											# Out indices are set to the last layer by default. For timm models, we don't know
											# the number of layers in advance, so we set it to (-1,), whereas for transformers
											# models, we set it to [len(stage_names) - 1] (kept for backward compatibility).
											self.assertEqual(timm_model.out_indices       ,  (-1,))
											self.assertEqual(transformers_model.out_indices       ,  [len(timm_model.stage_names) - 1])

											a__      =AutoBackbone.from_pretrained(lowercase_       ,  use_timm_backbone=lowercase_       ,  out_indices=[1, 2, 3])
											a__      =AutoBackbone.from_pretrained(lowercase_       ,  out_indices=[1, 2, 3])

											self.assertEqual(timm_model.out_indices       ,  transformers_model.out_indices)
											self.assertEqual(len(timm_model.out_features)       ,  len(transformers_model.out_features))
											self.assertEqual(timm_model.channels       ,  transformers_model.channels)





				@unittest.skip('TimmBackbone doesn\'t support feed forward chunking')
				def 						__UpperCamelCase    (				self)  -> int:
											pass





				@unittest.skip('TimmBackbone doesn\'t have num_hidden_layers attribute')
				def 						__UpperCamelCase    (				self)  -> List[str]:
											pass





				@unittest.skip('TimmBackbone initialization is managed on the timm side')
				def 						__UpperCamelCase    (				self)  -> Any:
											pass





				@unittest.skip('TimmBackbone models doesn\'t have inputs_embeds')
				def 						__UpperCamelCase    (				self)  -> Any:
											pass





				@unittest.skip('TimmBackbone models doesn\'t have inputs_embeds')
				def 						__UpperCamelCase    (				self)  -> List[str]:
											pass





				@unittest.skip('TimmBackbone model cannot be created without specifying a backbone checkpoint')
				def 						__UpperCamelCase    (				self)  -> Optional[int]:
											pass





				@unittest.skip('Only checkpoints on timm can be loaded into TimmBackbone')
				def 						__UpperCamelCase    (				self)  -> Union[str, Any]:
											pass





				@unittest.skip('model weights aren\'t tied in TimmBackbone.')
				def 						__UpperCamelCase    (				self)  -> Dict:
											pass





				@unittest.skip('model weights aren\'t tied in TimmBackbone.')
				def 						__UpperCamelCase    (				self)  -> List[Any]:
											pass





				@unittest.skip('Only checkpoints on timm can be loaded into TimmBackbone')
				def 						__UpperCamelCase    (				self)  -> List[str]:
											pass





				@unittest.skip('Only checkpoints on timm can be loaded into TimmBackbone')
				def 						__UpperCamelCase    (				self)  -> Union[str, Any]:
											pass





				@unittest.skip('TimmBackbone doesn\'t have hidden size info in its configuration.')
				def 						__UpperCamelCase    (				self)  -> int:
											pass





				@unittest.skip('TimmBackbone doesn\'t support output_attentions.')
				def 						__UpperCamelCase    (				self)  -> str:
											pass





				@unittest.skip('Safetensors is not supported by timm.')
				def 						__UpperCamelCase    (				self)  -> Optional[int]:
											pass





				@unittest.skip('Will be fixed soon by reducing the size of the model used for common tests.')
				def 						__UpperCamelCase    (				self)  -> Optional[Any]:
											pass





				def 						__UpperCamelCase    (				self)  -> Any:
											a__    ,     a__      =self.model_tester.prepare_config_and_inputs_for_common()

											for model_class in self.all_model_classes:
																		a__      =model_class(lowercase_)
																		a__      =inspect.signature(model.forward)
																		# signature.parameters is an OrderedDict => so arg_names order is deterministic
																		a__      =[*signature.parameters.keys()]

																		a__      =['pixel_values']
																		self.assertListEqual(arg_names[:1]       ,  lowercase_)





				def 						__UpperCamelCase    (				self)  -> Any:
											a__    ,     a__      =self.model_tester.prepare_config_and_inputs_for_common()
											a__      =True
											a__      =self.has_attentions

											# no need to test all models as different heads yield the same functionality
											a__      =self.all_model_classes[0]
											a__      =model_class(lowercase_)
											model.to(lowercase_)

											a__      =self._prepare_for_class(lowercase_       ,  lowercase_)
											a__      =model(**lowercase_)
											a__      =outputs[0][-1]

											# Encoder-/Decoder-only models
											a__      =outputs.hidden_states[0]
											hidden_states.retain_grad()

											if self.has_attentions:
																		a__      =outputs.attentions[0]
																		attentions.retain_grad()

											output.flatten()[0].backward(retain_graph=lowercase_)

											self.assertIsNotNone(hidden_states.grad)

											if self.has_attentions:
																		self.assertIsNotNone(attentions.grad)





				def 						__UpperCamelCase    (				self)  -> List[str]:
											a__    ,     a__      =self.model_tester.prepare_config_and_inputs_for_common()

											for model_class in self.all_model_classes:
																		a__      =model_class(lowercase_)
																		model.to(lowercase_)
																		model.eval()
																		a__      =model(**lowercase_)

																		self.assertEqual(len(result.feature_maps)       ,  len(config.out_indices))
																		self.assertEqual(len(model.channels)       ,  len(config.out_indices))

																		# Check output of last stage is taken if out_features=None, out_indices=None
																		a__      =copy.deepcopy(lowercase_)
																		a__      =None
																		a__      =model_class(lowercase_)
																		model.to(lowercase_)
																		model.eval()
																		a__      =model(**lowercase_)

																		self.assertEqual(len(result.feature_maps)       ,  1)
																		self.assertEqual(len(model.channels)       ,  1)

																		# Check backbone can be initialized with fresh weights
																		a__      =copy.deepcopy(lowercase_)
																		a__      =False
																		a__      =model_class(lowercase_)
																		model.to(lowercase_)
																		model.eval()
																		a__      =model(**lowercase_)

