



'''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:      Optional[int]		 =		get_logger(__name__)


lowerCAmelCase:      Optional[int]		 =		Path(__file__).parent / 'model_card_template.md'
lowerCAmelCase:      Dict		 =		uuida().hex
lowerCAmelCase:      int		 =		os.getenv('HF_HUB_OFFLINE', '').upper() in ENV_VARS_TRUE_VALUES
lowerCAmelCase:      Tuple		 =		os.getenv('DISABLE_TELEMETRY', '').upper() in ENV_VARS_TRUE_VALUES
lowerCAmelCase:      Optional[int]		 =		HUGGINGFACE_CO_RESOLVE_ENDPOINT + '/api/telemetry/'
def    lowerCamelCase__ (		_A = None       ):
  a		:      List[Any]       =  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:
    a		:      List[Any]       =  HfFolder.get_token()
  if organization is None:
    a		:      List[str]       =  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

  a		:      Optional[Any]       =  args.hub_token if hasattr(_A				,    'hub_token'       ) else None
  a		:      Optional[int]       =  get_full_repo_name(_A				,    token=_A       )

  a		:      int       =  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		:      Any       =  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
  a		:      Any       =  str(Path(_A       ).as_posix()       )
  a		:      List[str]       =  re.search(r'snapshots/([^/]+)/'				,    _A       )
  if search is None:
    return None
  a		:      List[str]       =  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:      Tuple		 =		os.path.expanduser(
    os.getenv('HF_HOME', os.path.join(os.getenv('XDG_CACHE_HOME', '~/.cache'), 'huggingface'))
)
lowerCAmelCase:      Optional[Any]		 =		os.path.join(hf_cache_home, 'diffusers')
def    lowerCamelCase__ (		_A = None				,    _A = None       ):
  if new_cache_dir is None:
    a		:      Optional[int]       =  DIFFUSERS_CACHE
  if old_cache_dir is None:
    a		:      str       =  old_diffusers_cache

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

if cache_version < 1:
     lowerCAmelCase:      Optional[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:
               lowerCAmelCase:      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    lowerCamelCase__ (		_A				,    _A = None       ):
  if variant is not None:
    a		:      Any       =  weights_name.split('.'       )
    a		:      List[str]       =  splits[:-1] + [variant] + splits[-1:]
    a		:      Tuple       =  '.'.join(_A       )

  return weights_name

def    lowerCamelCase__ (		_A				,    *,
    _A				,    _A				,    _A				,    _A				,    _A				,    _A				,    _A				,    _A				,    _A				,    _A				,    _A=None				,    ):
  a		:      Tuple       =  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		:      Optional[int]       =  os.path.join(_A				,    _A       )
      return model_file
    elif subfolder is not None and os.path.isfile(
        os.path.join(_A				,    _A				,    _A       )       ):
      a		:      Union[str, Any]       =  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		:      Optional[Any]       =  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		:      List[str]       =  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'''






from typing import Dict, List, Optional, Union

import numpy as np

from .feature_extraction_utils import BatchFeature, FeatureExtractionMixin
from .utils import PaddingStrategy, TensorType, is_tf_tensor, is_torch_tensor, logging, to_numpy


lowerCAmelCase:      int		 =		logging.get_logger(__name__)

class 							a__(      lowerCamelCase__			):







    def __init__(   self				:						Union[str, Any]		,    __snake_case				:						int		,    __snake_case				:						int		,    __snake_case				:						float		,    **__snake_case				:						Optional[int]			):
      a		:      Optional[Any]       =  feature_size
      a		:      Tuple       =  sampling_rate
      a		:      str       =  padding_value

      a		:      Any       =  kwargs.pop('padding_side'		,    'right'			)
      a		:      Tuple       =  kwargs.pop('return_attention_mask'		,    __snake_case			)

      super().__init__(**__snake_case			)







    def     lowercase_ (   self				:						Tuple		,    __snake_case				:						Union[
            BatchFeature,
            List[BatchFeature],
            Dict[str, BatchFeature],
            Dict[str, List[BatchFeature]],
            List[Dict[str, BatchFeature]],
        ]		,    __snake_case				:						Union[bool, str, PaddingStrategy] = True		,    __snake_case				:						Optional[int] = None		,    __snake_case				:						bool = False		,    __snake_case				:						Optional[int] = None		,    __snake_case				:						Optional[bool] = None		,    __snake_case				:						Optional[Union[str, TensorType]] = None		,    ):
      # If we have a list of dicts, let's convert it in a dict of lists
      # We do this to allow using this method as a collate_fn function in PyTorch Dataloader
      if isinstance(__snake_case		,    (list, tuple)			) and isinstance(processed_features[0]		,    (dict, BatchFeature)			):
        a		:      Union[str, Any]       =  {
            key: [example[key] for example in processed_features] for key in processed_features[0].keys()
        }

      # The model's main input name, usually `input_values`, has be passed for padding
      if self.model_input_names[0] not in processed_features:
        raise ValueError(
            'You should supply an instance of `transformers.BatchFeature` or list of `transformers.BatchFeature`'
            F""" to this method that includes {self.model_input_names[0]}, but you provided"""
            F""" {list(processed_features.keys()			)}"""			)

      a		:      Tuple       =  processed_features[self.model_input_names[0]]
      a		:      str       =  (
          return_attention_mask if return_attention_mask is not None else self.return_attention_mask
      )

      if len(__snake_case			) == 0:
        if return_attention_mask:
          a		:      Dict       =  []
        return processed_features

      # If we have PyTorch/TF tensors or lists as inputs, we cast them as Numpy arrays
      # and rebuild them afterwards if no return_tensors is specified
      # Note that we lose the specific device the tensor may be on for PyTorch

      a		:      Optional[int]       =  required_input[0]
      if isinstance(__snake_case		,    (list, tuple)			):
        # first_element might be an empty list/tuple in some edge cases so we grab the first non empty element.
        a		:      Dict       =  0
        while len(required_input[index]			) == 0:
          index += 1
        if index < len(__snake_case			):
          a		:      Tuple       =  required_input[index][0]

      if return_tensors is None:
        if is_tf_tensor(__snake_case			):
          a		:      Any       =  'tf'
        elif is_torch_tensor(__snake_case			):
          a		:      Optional[Any]       =  'pt'
        elif isinstance(__snake_case		,    (int, float, list, tuple, np.ndarray)			):
          a		:      List[Any]       =  'np'
        else:
          raise ValueError(
              F"""type of {first_element} unknown: {type(__snake_case			)}. """
              'Should be one of a python, numpy, pytorch or tensorflow object.'			)

      for key, value in processed_features.items():
        if isinstance(value[0]		,    (int, float)			):
          a		:      List[Any]       =  to_numpy(__snake_case			)
        else:
          a		:      Tuple       =  [to_numpy(__snake_case			) for v in value]

        # Convert padding_strategy in PaddingStrategy
      a		:      Any       =  self._get_padding_strategies(padding=__snake_case		,    max_length=__snake_case			)

      a		:      int       =  processed_features[self.model_input_names[0]]

      a		:      Dict       =  len(__snake_case			)
      if not all(len(__snake_case			) == batch_size for v in processed_features.values()			):
        raise ValueError('Some items in the output dictionary have a different batch size than others.'			)

      a		:      Optional[int]       =  []
      for i in range(__snake_case			):
        a		:      int       =  {k: v[i] for k, v in processed_features.items()}
        # truncation
        a		:      List[str]       =  self._truncate(
            __snake_case		,    max_length=__snake_case		,    pad_to_multiple_of=__snake_case		,    truncation=__snake_case		,    )
        truncated_inputs.append(__snake_case			)

      if padding_strategy == PaddingStrategy.LONGEST:
        # make sure that `max_length` cannot be longer than the longest truncated length
        a		:      List[str]       =  max(len(input_slice[self.model_input_names[0]]			) for input_slice in truncated_inputs			)
        a		:      Tuple       =  PaddingStrategy.MAX_LENGTH

      a		:      int       =  {}
      for i in range(__snake_case			):
        # padding
        a		:      int       =  self._pad(
            truncated_inputs[i]		,    max_length=__snake_case		,    padding_strategy=__snake_case		,    pad_to_multiple_of=__snake_case		,    return_attention_mask=__snake_case		,    )

        for key, value in outputs.items():
          if key not in batch_outputs:
            a		:      int       =  []
          if value.dtype is np.dtype(np.floataa			):
            a		:      List[str]       =  value.astype(np.floataa			)
          batch_outputs[key].append(__snake_case			)

      return BatchFeature(__snake_case		,    tensor_type=__snake_case			)







    def     lowercase_ (   self				:						List[str]		,    __snake_case				:						Union[Dict[str, np.ndarray], BatchFeature]		,    __snake_case				:						Optional[int] = None		,    __snake_case				:						PaddingStrategy = PaddingStrategy.DO_NOT_PAD		,    __snake_case				:						Optional[int] = None		,    __snake_case				:						Optional[bool] = None		,    ):
      a		:      List[str]       =  processed_features[self.model_input_names[0]]

      if padding_strategy == PaddingStrategy.LONGEST:
        a		:      Any       =  len(__snake_case			)

      if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
        a		:      Dict       =  ((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

      a		:      List[Any]       =  padding_strategy != PaddingStrategy.DO_NOT_PAD and len(__snake_case			) < max_length

      if return_attention_mask and "attention_mask" not in processed_features:
        a		:      Union[str, Any]       =  np.ones(len(__snake_case			)		,    dtype=np.intaa			)

      if needs_to_be_padded:
        a		:      Tuple       =  max_length - len(__snake_case			)
        if self.padding_side == "right":
          if return_attention_mask:
            a		:      Any       =  np.pad(
                processed_features['attention_mask']		,    (0, difference)			)
          a		:      Optional[Any]       =  ((0, difference), (0, 0)) if self.feature_size > 1 else (0, difference)
          a		:      Optional[int]       =  np.pad(
              __snake_case		,    __snake_case		,    'constant'		,    constant_values=self.padding_value			)
        elif self.padding_side == "left":
          if return_attention_mask:
            a		:      Optional[int]       =  np.pad(
                processed_features['attention_mask']		,    (difference, 0)			)
          a		:      Optional[Any]       =  ((difference, 0), (0, 0)) if self.feature_size > 1 else (difference, 0)
          a		:      List[Any]       =  np.pad(
              __snake_case		,    __snake_case		,    'constant'		,    constant_values=self.padding_value			)
        else:
          raise ValueError('Invalid padding strategy:' + str(self.padding_side			)			)

      return processed_features







    def     lowercase_ (   self				:						List[str]		,    __snake_case				:						Union[Dict[str, np.ndarray], BatchFeature]		,    __snake_case				:						Optional[int] = None		,    __snake_case				:						Optional[int] = None		,    __snake_case				:						Optional[bool] = None		,    ):
      if not truncation:
        return processed_features
      elif truncation and max_length is None:
        raise ValueError('When setting ``truncation=True``, make sure that ``max_length`` is defined.'			)

      a		:      Dict       =  processed_features[self.model_input_names[0]]

      # find `max_length` that fits `pad_to_multiple_of`
      if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
        a		:      Optional[Any]       =  ((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

      a		:      str       =  len(__snake_case			) > max_length

      if needs_to_be_truncated:
        a		:      str       =  processed_features[self.model_input_names[0]][:max_length]
        if "attention_mask" in processed_features:
          a		:      List[str]       =  processed_features['attention_mask'][:max_length]

      return processed_features







    def     lowercase_ (   self				:						Dict		,    __snake_case				:						Optional[Any]=False		,    __snake_case				:						Optional[int]=None			):

      # Get padding strategy
      if padding is not False:
        if padding is True:
          a		:      Union[str, Any]       =  PaddingStrategy.LONGEST  # Default to pad to the longest sequence in the batch
        elif not isinstance(__snake_case		,    __snake_case			):
          a		:      List[Any]       =  PaddingStrategy(__snake_case			)
        elif isinstance(__snake_case		,    __snake_case			):
          a		:      List[str]       =  padding
      else:
        a		:      Dict       =  PaddingStrategy.DO_NOT_PAD

      # Set max length if needed
      if max_length is None:
        if padding_strategy == PaddingStrategy.MAX_LENGTH:
          raise ValueError(
              F"""When setting ``padding={PaddingStrategy.MAX_LENGTH}``, make sure that max_length is defined"""			)

        # Test if we have a padding value
      if padding_strategy != PaddingStrategy.DO_NOT_PAD and (self.padding_value is None):
        raise ValueError(
            'Asking to pad but the feature_extractor does not have a padding value. Please select a value to use'
            ' as `padding_value`. For example: `feature_extractor.padding_value = 0.0`.'			)

      return padding_strategy