
import collections.abc
from typing import Optional, Tuple, Union

import torch
import torch.utils.checkpoint
from torch import nn
from torch.nn import BCEWithLogitsLoss, CrossEntropyLoss, MSELoss

from ...activations import ACTaFN
from ...modeling_outputs import BaseModelOutputWithNoAttention, ImageClassifierOutputWithNoAttention
from ...modeling_utils import PreTrainedModel
from ...utils import add_code_sample_docstrings, add_start_docstrings, add_start_docstrings_to_model_forward, logging
from .configuration_poolformer import PoolFormerConfig


lowercase  : Union[str, Any]						   =							logging.get_logger(__name__)

# General docstring
lowercase  : Optional[int]						   =							"""PoolFormerConfig"""

# Base docstring
lowercase  : Union[str, Any]						   =							"""sail/poolformer_s12"""
lowercase  : int						   =							[1, 512, 7, 7]

# Image classification docstring
lowercase  : Any						   =							"""sail/poolformer_s12"""
lowercase  : Any						   =							"""tabby, tabby cat"""

lowercase  : List[Any]						   =							[
    """sail/poolformer_s12""",
    # See all PoolFormer models at https://huggingface.co/models?filter=poolformer
]







def 						_snake_case(  SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ = 0.0 ,     SCREAMING_SNAKE_CASE__ = False    )			->   List[Any]:
       if drop_prob == 0.0 or not training:
              return input
       lowercase							:					Any	     =				1 - drop_prob
       lowercase							:					Union[str, Any]	     =				(input.shape[0],) + (1,) * (input.ndim - 1)  # work with diff dim tensors, not just 2D ConvNets
       lowercase							:					Union[str, Any]	     =				keep_prob + torch.rand(SCREAMING_SNAKE_CASE__ ,     dtype=input.dtype ,     device=input.device    )
       random_tensor.floor_()  # binarize
       lowercase							:					Tuple	     =				input.div(SCREAMING_SNAKE_CASE__    ) * random_tensor
       return output

class __snake_case		(				nn.Module					):
    def __init__(  self		,snake_case = None				):



           '''simple docstring'''







           super().__init__()
           lowercase							:					List[Any]	     =				drop_prob
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case				):



           '''simple docstring'''







           return drop_path(snake_case		,self.drop_prob		,self.training				)
    def 				_SCREAMING_SNAKE_CASE	(  self				):



           '''simple docstring'''







           return "p={}".format(self.drop_prob				)






class __snake_case		(				nn.Module					):
    def __init__(  self		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case=None				):



           '''simple docstring'''







           super().__init__()
           lowercase							:					Dict	     =				patch_size if isinstance(snake_case		,collections.abc.Iterable				) else (patch_size, patch_size)
           lowercase							:					Tuple	     =				stride if isinstance(snake_case		,collections.abc.Iterable				) else (stride, stride)
           lowercase							:					Optional[Any]	     =				padding if isinstance(snake_case		,collections.abc.Iterable				) else (padding, padding)

           lowercase							:					Dict	     =				nn.Convad(snake_case		,snake_case		,kernel_size=snake_case		,stride=snake_case		,padding=snake_case				)
           lowercase							:					Tuple	     =				norm_layer(snake_case				) if norm_layer else nn.Identity()
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case				):



           '''simple docstring'''







           lowercase							:					List[str]	     =				self.projection(snake_case				)
           lowercase							:					List[str]	     =				self.norm(snake_case				)
           return embeddings






class __snake_case		(				nn.GroupNorm					):
    def __init__(  self		,snake_case		,**snake_case				):



           '''simple docstring'''







           super().__init__(1		,snake_case		,**snake_case				)






class __snake_case		(				nn.Module					):
    def __init__(  self		,snake_case				):



           '''simple docstring'''







           super().__init__()
           lowercase							:					Any	     =				nn.AvgPoolad(snake_case		,stride=1		,padding=pool_size // 2		,count_include_pad=snake_case				)
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case				):



           '''simple docstring'''







           return self.pool(snake_case				) - hidden_states






class __snake_case		(				nn.Module					):
    def __init__(  self		,snake_case		,snake_case		,snake_case		,snake_case				):



           '''simple docstring'''







           super().__init__()
           lowercase							:					List[Any]	     =				nn.Convad(snake_case		,snake_case		,1				)
           lowercase							:					Tuple	     =				nn.Convad(snake_case		,snake_case		,1				)
           lowercase							:					List[Any]	     =				PoolFormerDropPath(snake_case				)
           if isinstance(config.hidden_act		,snake_case				):
                  lowercase							:					int	     =				ACTaFN[config.hidden_act]
           else:
                  lowercase							:					str	     =				config.hidden_act
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case				):



           '''simple docstring'''







           lowercase							:					List[Any]	     =				self.conva(snake_case				)
           lowercase							:					Any	     =				self.act_fn(snake_case				)
           lowercase							:					int	     =				self.drop(snake_case				)
           lowercase							:					Tuple	     =				self.conva(snake_case				)
           lowercase							:					int	     =				self.drop(snake_case				)

           return hidden_states






class __snake_case		(				nn.Module					):
    def __init__(  self		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case		,snake_case				):



           '''simple docstring'''







           super().__init__()
           lowercase							:					Tuple	     =				PoolFormerPooling(snake_case				)
           lowercase							:					Optional[int]	     =				PoolFormerOutput(snake_case		,snake_case		,snake_case		,snake_case				)
           lowercase							:					List[str]	     =				PoolFormerGroupNorm(snake_case				)
           lowercase							:					Union[str, Any]	     =				PoolFormerGroupNorm(snake_case				)

           # Useful for training neural nets
           lowercase							:					Optional[int]	     =				PoolFormerDropPath(snake_case				) if drop_path > 0.0 else nn.Identity()
           lowercase							:					List[str]	     =				config.use_layer_scale
           if config.use_layer_scale:
                  lowercase							:					Tuple	     =				nn.Parameter(
                      config.layer_scale_init_value * torch.ones((snake_case)				)		,requires_grad=snake_case				)
                  lowercase							:					Dict	     =				nn.Parameter(
                      config.layer_scale_init_value * torch.ones((snake_case)				)		,requires_grad=snake_case				)
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case				):



           '''simple docstring'''







           if self.use_layer_scale:
                  lowercase							:					List[Any]	     =				self.pooling(self.before_norm(snake_case				)				)
                  lowercase							:					List[str]	     =				self.layer_scale_a.unsqueeze(-1				).unsqueeze(-1				) * pooling_output
                  # First residual connection
                  lowercase							:					str	     =				hidden_states + self.drop_path(snake_case				)
                  lowercase							:					List[Any]	     =				()

                  lowercase							:					Any	     =				self.output(self.after_norm(snake_case				)				)
                  lowercase							:					int	     =				self.layer_scale_a.unsqueeze(-1				).unsqueeze(-1				) * layer_output
                  # Second residual connection
                  lowercase							:					str	     =				hidden_states + self.drop_path(snake_case				)

                  lowercase							:					Union[str, Any]	     =				(output,) + outputs
                  return outputs

           else:
                  lowercase							:					Optional[int]	     =				self.drop_path(self.pooling(self.before_norm(snake_case				)				)				)
                  # First residual connection
                  lowercase							:					List[str]	     =				pooling_output + hidden_states
                  lowercase							:					List[Any]	     =				()

                  # Second residual connection inside the PoolFormerOutput block
                  lowercase							:					Tuple	     =				self.drop_path(self.output(self.after_norm(snake_case				)				)				)
                  lowercase							:					Tuple	     =				hidden_states + layer_output

                  lowercase							:					Any	     =				(output,) + outputs
                  return outputs






class __snake_case		(				nn.Module					):
    def __init__(  self		,snake_case				):



           '''simple docstring'''







           super().__init__()
           lowercase							:					str	     =				config
           # stochastic depth decay rule
           lowercase							:					Union[str, Any]	     =				[x.item() for x in torch.linspace(0		,config.drop_path_rate		,sum(config.depths				)				)]

           # patch embeddings
           lowercase							:					Union[str, Any]	     =				[]
           for i in range(config.num_encoder_blocks				):
                  embeddings.append(
                      PoolFormerEmbeddings(
                          patch_size=config.patch_sizes[i]		,stride=config.strides[i]		,padding=config.padding[i]		,num_channels=config.num_channels if i == 0 else config.hidden_sizes[i - 1]		,hidden_size=config.hidden_sizes[i]		,)				)
           lowercase							:					Dict	     =				nn.ModuleList(snake_case				)

           # Transformer blocks
           lowercase							:					Any	     =				[]
           lowercase							:					Optional[int]	     =				0
           for i in range(config.num_encoder_blocks				):
                  # each block consists of layers
                  lowercase							:					Tuple	     =				[]
                  if i != 0:
                         cur += config.depths[i - 1]
                  for j in range(config.depths[i]				):
                         layers.append(
                             PoolFormerLayer(
                                 snake_case		,num_channels=config.hidden_sizes[i]		,pool_size=config.pool_size		,hidden_size=config.hidden_sizes[i]		,intermediate_size=int(config.hidden_sizes[i] * config.mlp_ratio				)		,drop_path=dpr[cur + j]		,)				)
                  blocks.append(nn.ModuleList(snake_case				)				)

           lowercase							:					List[Any]	     =				nn.ModuleList(snake_case				)
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case		,snake_case=False		,snake_case=True				):



           '''simple docstring'''







           lowercase							:					int	     =				() if output_hidden_states else None

           lowercase							:					str	     =				pixel_values
           for idx, layers in enumerate(zip(self.patch_embeddings		,self.block				)				):
                  lowercase     ,      lowercase							:					str	     =				layers
                  # Get patch embeddings from hidden_states
                  lowercase							:					str	     =				embedding_layer(snake_case				)
                  # Send the embeddings through the blocks
                  for _, blk in enumerate(snake_case				):
                         lowercase							:					Union[str, Any]	     =				blk(snake_case				)
                         lowercase							:					Optional[Any]	     =				layer_outputs[0]

                  if output_hidden_states:
                         lowercase							:					Any	     =				all_hidden_states + (hidden_states,)

           if not return_dict:
                  return tuple(v for v in [hidden_states, all_hidden_states] if v is not None				)

           return BaseModelOutputWithNoAttention(last_hidden_state=snake_case		,hidden_states=snake_case				)






class __snake_case		(				lowerCAmelCase					):
    _a		:						int=    PoolFormerConfig
    _a		:						Dict=    "poolformer"
    _a		:						Dict=    "pixel_values"
    _a		:						int=    True
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case				):



           '''simple docstring'''







           if isinstance(snake_case		,(nn.Linear, nn.Convad)				):
                  module.weight.data.normal_(mean=0.0		,std=self.config.initializer_range				)
                  if module.bias is not None:
                         module.bias.data.zero_()
           elif isinstance(snake_case		,nn.LayerNorm				):
                  module.bias.data.zero_()
                  module.weight.data.fill_(1.0				)
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case		,snake_case=False				):



           '''simple docstring'''







           if isinstance(snake_case		,snake_case				):
                  lowercase							:					Optional[int]	     =				value






lowercase  : Optional[int]						   =							R"""
    This model is a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) sub-class. Use
    it as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage and
    behavior.

    Parameters:
        config ([`PoolFormerConfig`]): Model configuration class with all the parameters of the model.
            Initializing with a config file does not load the weights associated with the model, only the
            configuration. Check out the [`~PreTrainedModel.from_pretrained`] method to load the model weights.
"""

lowercase  : Any						   =							R"""
    Args:
        pixel_values (`torch.FloatTensor` of shape `(batch_size, num_channels, height, width)`):
            Pixel values. Pixel values can be obtained using [`AutoImageProcessor`]. See
            [`PoolFormerImageProcessor.__call__`] for details.
"""

@add_start_docstrings(
    "The bare PoolFormer Model transformer outputting raw hidden-states without any specific head on top." ,     lowerCAmelCase ,     )
class __snake_case		(				lowerCAmelCase					):
    def __init__(  self		,snake_case				):



           '''simple docstring'''







           super().__init__(snake_case				)
           lowercase							:					List[Any]	     =				config

           lowercase							:					Tuple	     =				PoolFormerEncoder(snake_case				)

           # Initialize weights and apply final processing
           self.post_init()
    def 				_SCREAMING_SNAKE_CASE	(  self				):



           '''simple docstring'''







           return self.embeddings.patch_embeddings
    @add_start_docstrings_to_model_forward(snake_case				)
    @add_code_sample_docstrings(
        checkpoint=_CHECKPOINT_FOR_DOC		,output_type=snake_case		,config_class=_CONFIG_FOR_DOC		,modality="""vision"""		,expected_output=_EXPECTED_OUTPUT_SHAPE		,)
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case = None		,snake_case = None		,snake_case = None		,):



           '''simple docstring'''







           lowercase							:					Tuple	     =				(
               output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
           )
           lowercase							:					Any	     =				return_dict if return_dict is not None else self.config.use_return_dict

           if pixel_values is None:
                  raise ValueError("""You have to specify pixel_values"""				)

           lowercase							:					Optional[int]	     =				self.encoder(
               snake_case		,output_hidden_states=snake_case		,return_dict=snake_case		,)
           lowercase							:					Optional[Any]	     =				encoder_outputs[0]

           if not return_dict:
                  return (sequence_output, None) + encoder_outputs[1:]

           return BaseModelOutputWithNoAttention(
               last_hidden_state=snake_case		,hidden_states=encoder_outputs.hidden_states		,)






class __snake_case		(				nn.Module					):
    def __init__(  self		,snake_case				):



           '''simple docstring'''







           super().__init__()
           lowercase							:					Tuple	     =				nn.Linear(config.hidden_size		,config.hidden_size				)
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case				):



           '''simple docstring'''







           lowercase							:					Any	     =				self.dense(snake_case				)
           return output






@add_start_docstrings(
    "\n    PoolFormer Model transformer with an image classification head on top\n    " ,     lowerCAmelCase ,     )
class __snake_case		(				lowerCAmelCase					):
    def __init__(  self		,snake_case				):



           '''simple docstring'''







           super().__init__(snake_case				)
           lowercase							:					List[str]	     =				config.num_labels
           lowercase							:					Optional[Any]	     =				PoolFormerModel(snake_case				)

           # Final norm
           lowercase							:					str	     =				PoolFormerGroupNorm(config.hidden_sizes[-1]				)
           # Classifier head
           lowercase							:					Dict	     =				(
               nn.Linear(config.hidden_sizes[-1]		,config.num_labels				) if config.num_labels > 0 else nn.Identity()
           )

           # Initialize weights and apply final processing
           self.post_init()
    @add_start_docstrings_to_model_forward(snake_case				)
    @add_code_sample_docstrings(
        checkpoint=_IMAGE_CLASS_CHECKPOINT		,output_type=snake_case		,config_class=_CONFIG_FOR_DOC		,expected_output=_IMAGE_CLASS_EXPECTED_OUTPUT		,)
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case = None		,snake_case = None		,snake_case = None		,snake_case = None		,):



           '''simple docstring'''







           lowercase							:					str	     =				return_dict if return_dict is not None else self.config.use_return_dict

           lowercase							:					Optional[Any]	     =				self.poolformer(
               snake_case		,output_hidden_states=snake_case		,return_dict=snake_case		,)

           lowercase							:					Union[str, Any]	     =				outputs[0]

           lowercase							:					Optional[int]	     =				self.classifier(self.norm(snake_case				).mean([-2, -1]				)				)

           lowercase							:					str	     =				None
           if labels is not None:
                  if self.config.problem_type is None:
                         if self.num_labels == 1:
                                lowercase							:					Optional[Any]	     =				"""regression"""
                         elif self.num_labels > 1 and (labels.dtype == torch.long or labels.dtype == torch.int):
                                lowercase							:					List[str]	     =				"""single_label_classification"""
                         else:
                                lowercase							:					int	     =				"""multi_label_classification"""

                  if self.config.problem_type == "regression":
                         lowercase							:					List[Any]	     =				MSELoss()
                         if self.num_labels == 1:
                                lowercase							:					Dict	     =				loss_fct(logits.squeeze()		,labels.squeeze()				)
                         else:
                                lowercase							:					Union[str, Any]	     =				loss_fct(snake_case		,snake_case				)
                  elif self.config.problem_type == "single_label_classification":
                         lowercase							:					Union[str, Any]	     =				CrossEntropyLoss()
                         lowercase							:					str	     =				loss_fct(logits.view(-1		,self.num_labels				)		,labels.view(-1				)				)
                  elif self.config.problem_type == "multi_label_classification":
                         lowercase							:					int	     =				BCEWithLogitsLoss()
                         lowercase							:					List[Any]	     =				loss_fct(snake_case		,snake_case				)

           if not return_dict:
                  lowercase							:					Union[str, Any]	     =				(logits,) + outputs[2:]
                  return ((loss,) + output) if loss is not None else output

           return ImageClassifierOutputWithNoAttention(loss=snake_case		,logits=snake_case		,hidden_states=outputs.hidden_states				)


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


lowercase  : Tuple						   =							get_logger(__name__)


lowercase  : Optional[int]						   =							Path(__file__).parent / """model_card_template.md"""
lowercase  : Dict						   =							uuida().hex
lowercase  : Tuple						   =							os.getenv("""HF_HUB_OFFLINE""", """""").upper() in ENV_VARS_TRUE_VALUES
lowercase  : str						   =							os.getenv("""DISABLE_TELEMETRY""", """""").upper() in ENV_VARS_TRUE_VALUES
lowercase  : Tuple						   =							HUGGINGFACE_CO_RESOLVE_ENDPOINT + """/api/telemetry/"""







def 						_snake_case(  SCREAMING_SNAKE_CASE__ = None    )			->   str:
       lowercase							:					str	     =				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(SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    ):
              ua += "; " + "; ".join(f"{k}/{v}" for k, v in user_agent.items()    )
       elif isinstance(SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    ):
              ua += "; " + user_agent
       return ua







def 						_snake_case(  SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ = None ,     SCREAMING_SNAKE_CASE__ = None    )			->   Dict:
       if token is None:
              lowercase							:					Optional[int]	     =				HfFolder.get_token()
       if organization is None:
              lowercase							:					int	     =				whoami(SCREAMING_SNAKE_CASE__    )["""name"""]
              return f"{username}/{model_id}"
       else:
              return f"{organization}/{model_id}"







def 						_snake_case(  SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    )			->   Tuple:
       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(SCREAMING_SNAKE_CASE__ ,     """local_rank"""    ) and args.local_rank not in [-1, 0]:
              return

       lowercase							:					str	     =				args.hub_token if hasattr(SCREAMING_SNAKE_CASE__ ,     """hub_token"""    ) else None
       lowercase							:					int	     =				get_full_repo_name(SCREAMING_SNAKE_CASE__ ,     token=SCREAMING_SNAKE_CASE__    )

       lowercase							:					Dict	     =				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=SCREAMING_SNAKE_CASE__ ,     model_name=SCREAMING_SNAKE_CASE__ ,     repo_name=SCREAMING_SNAKE_CASE__ ,     dataset_name=args.dataset_name if hasattr(SCREAMING_SNAKE_CASE__ ,     """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(SCREAMING_SNAKE_CASE__ ,     """gradient_accumulation_steps"""    ) else None
           ) ,     adam_betaa=args.adam_betaa if hasattr(SCREAMING_SNAKE_CASE__ ,     """adam_beta1"""    ) else None ,     adam_betaa=args.adam_betaa if hasattr(SCREAMING_SNAKE_CASE__ ,     """adam_beta2"""    ) else None ,     adam_weight_decay=args.adam_weight_decay if hasattr(SCREAMING_SNAKE_CASE__ ,     """adam_weight_decay"""    ) else None ,     adam_epsilon=args.adam_epsilon if hasattr(SCREAMING_SNAKE_CASE__ ,     """adam_epsilon"""    ) else None ,     lr_scheduler=args.lr_scheduler if hasattr(SCREAMING_SNAKE_CASE__ ,     """lr_scheduler"""    ) else None ,     lr_warmup_steps=args.lr_warmup_steps if hasattr(SCREAMING_SNAKE_CASE__ ,     """lr_warmup_steps"""    ) else None ,     ema_inv_gamma=args.ema_inv_gamma if hasattr(SCREAMING_SNAKE_CASE__ ,     """ema_inv_gamma"""    ) else None ,     ema_power=args.ema_power if hasattr(SCREAMING_SNAKE_CASE__ ,     """ema_power"""    ) else None ,     ema_max_decay=args.ema_max_decay if hasattr(SCREAMING_SNAKE_CASE__ ,     """ema_max_decay"""    ) else None ,     mixed_precision=args.mixed_precision ,     )

       lowercase							:					str	     =				os.path.join(args.output_dir ,     """README.md"""    )
       model_card.save(SCREAMING_SNAKE_CASE__    )







def 						_snake_case(  SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ = None    )			->   Optional[Any]:
       if resolved_file is None or commit_hash is not None:
              return commit_hash
       lowercase							:					List[Any]	     =				str(Path(SCREAMING_SNAKE_CASE__    ).as_posix()    )
       lowercase							:					Any	     =				re.search(R"""snapshots/([^/]+)/""" ,     SCREAMING_SNAKE_CASE__    )
       if search is None:
              return None
       lowercase							:					List[Any]	     =				search.groups()[0]
       return commit_hash if REGEX_COMMIT_HASH.match(SCREAMING_SNAKE_CASE__    ) 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.
lowercase  : Optional[Any]						   =							os.path.expanduser(
    os.getenv("""HF_HOME""", os.path.join(os.getenv("""XDG_CACHE_HOME""", """~/.cache"""), """huggingface"""))
)
lowercase  : Optional[int]						   =							os.path.join(hf_cache_home, """diffusers""")







def 						_snake_case(  SCREAMING_SNAKE_CASE__ = None ,     SCREAMING_SNAKE_CASE__ = None    )			->   None:
       if new_cache_dir is None:
              lowercase							:					Union[str, Any]	     =				DIFFUSERS_CACHE
       if old_cache_dir is None:
              lowercase							:					List[str]	     =				old_diffusers_cache

       lowercase							:					Dict	     =				Path(SCREAMING_SNAKE_CASE__    ).expanduser()
       lowercase							:					int	     =				Path(SCREAMING_SNAKE_CASE__    ).expanduser()
       for old_blob_path in old_cache_dir.glob("""**/blobs/*"""    ):
              if old_blob_path.is_file() and not old_blob_path.is_symlink():
                     lowercase							:					Any	     =				new_cache_dir / old_blob_path.relative_to(SCREAMING_SNAKE_CASE__    )
                     new_blob_path.parent.mkdir(parents=SCREAMING_SNAKE_CASE__ ,     exist_ok=SCREAMING_SNAKE_CASE__    )
                     os.replace(SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    )
                     try:
                            os.symlink(SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    )
                     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).


lowercase  : Dict						   =							os.path.join(DIFFUSERS_CACHE, """version_diffusers_cache.txt""")
if not os.path.isfile(cache_version_file):
  lowercase  : Any						   =							0
else:
  with open(cache_version_file) as f:
    try:
      lowercase  : List[Any]						   =							int(f.read())
    except ValueError:
      lowercase  : int						   =							0

if cache_version < 1:
  lowercase  : Union[str, Any]						   =							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:
      lowercase  : int						   =							"""\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 						_snake_case(  SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ = None    )			->   str:
       if variant is not None:
              lowercase							:					List[str]	     =				weights_name.split("""."""    )
              lowercase							:					Optional[Any]	     =				splits[:-1] + [variant] + splits[-1:]
              lowercase							:					int	     =				""".""".join(SCREAMING_SNAKE_CASE__    )

       return weights_name







def 						_snake_case(  SCREAMING_SNAKE_CASE__ ,     *,
    SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__=None ,     )			->   Optional[Any]:
       lowercase							:					Optional[int]	     =				str(SCREAMING_SNAKE_CASE__    )
       if os.path.isfile(SCREAMING_SNAKE_CASE__    ):
              return pretrained_model_name_or_path
       elif os.path.isdir(SCREAMING_SNAKE_CASE__    ):
              if os.path.isfile(os.path.join(SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    )    ):
                     # Load from a PyTorch checkpoint
                     lowercase							:					List[Any]	     =				os.path.join(SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    )
                     return model_file
              elif subfolder is not None and os.path.isfile(
                  os.path.join(SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    )    ):
                     lowercase							:					Any	     =				os.path.join(SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    )
                     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(SCREAMING_SNAKE_CASE__    ).base_version    ) >= version.parse("""0.20.0"""    )
              ):
                     try:
                            lowercase							:					Any	     =				hf_hub_download(
                                SCREAMING_SNAKE_CASE__ ,     filename=_add_variant(SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    ) ,     cache_dir=SCREAMING_SNAKE_CASE__ ,     force_download=SCREAMING_SNAKE_CASE__ ,     proxies=SCREAMING_SNAKE_CASE__ ,     resume_download=SCREAMING_SNAKE_CASE__ ,     local_files_only=SCREAMING_SNAKE_CASE__ ,     use_auth_token=SCREAMING_SNAKE_CASE__ ,     user_agent=SCREAMING_SNAKE_CASE__ ,     subfolder=SCREAMING_SNAKE_CASE__ ,     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." ,     SCREAMING_SNAKE_CASE__ ,     )
                            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(SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    )} 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(SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    )}' so that the correct variant file can be added." ,     SCREAMING_SNAKE_CASE__ ,     )
              try:
                     # 2. Load model file as usual
                     lowercase							:					int	     =				hf_hub_download(
                         SCREAMING_SNAKE_CASE__ ,     filename=SCREAMING_SNAKE_CASE__ ,     cache_dir=SCREAMING_SNAKE_CASE__ ,     force_download=SCREAMING_SNAKE_CASE__ ,     proxies=SCREAMING_SNAKE_CASE__ ,     resume_download=SCREAMING_SNAKE_CASE__ ,     local_files_only=SCREAMING_SNAKE_CASE__ ,     use_auth_token=SCREAMING_SNAKE_CASE__ ,     user_agent=SCREAMING_SNAKE_CASE__ ,     subfolder=SCREAMING_SNAKE_CASE__ ,     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}"    )
