

import numpy as np

from transformers import BatchFeature
from transformers.testing_utils import require_tf, require_torch

from .test_feature_extraction_common import FeatureExtractionSavingTestMixin







class        a__       (   UpperCAmelCase       ):







  """simple docstring"""




  UpperCAmelCase__      : Optional[Any]			       =None
  UpperCAmelCase__      : str			       =None


  @property
  def  _lowercase      (			self    : Tuple		)					->Dict:
        """simple docstring"""







        return self.feat_extract_tester.prepare_feat_extract_dict()


  def  _lowercase      (			self    : Union[str, Any]		)					->Optional[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Optional[int]       =   self.feature_extraction_class(**self.feat_extract_dict		)
        self.assertTrue(hasattr(UpperCAmelCase__				,				"""feature_size"""		)		)
        self.assertTrue(hasattr(UpperCAmelCase__				,				"""sampling_rate"""		)		)
        self.assertTrue(hasattr(UpperCAmelCase__				,				"""padding_value"""		)		)


  def  _lowercase      (			self    : List[str]		)					->Optional[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Any       =   self.feat_extract_tester.prepare_inputs_for_common()
        SCREAMING_SNAKE_CASE :      List[Any]       =   self.feature_extraction_class(**self.feat_extract_dict		)
        SCREAMING_SNAKE_CASE :      int       =   feat_extract.model_input_names[0]

        SCREAMING_SNAKE_CASE :      List[str]       =   BatchFeature({input_name: speech_inputs}		)

        self.assertTrue(all(len(UpperCAmelCase__		) == len(UpperCAmelCase__		) for x, y in zip(UpperCAmelCase__				,				processed_features[input_name]		)		)		)

        SCREAMING_SNAKE_CASE :      List[str]       =   self.feat_extract_tester.prepare_inputs_for_common(equal_length=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Tuple       =   BatchFeature({input_name: speech_inputs}				,				tensor_type="""np"""		)

        SCREAMING_SNAKE_CASE :      Optional[Any]       =   processed_features[input_name]

        if len(batch_features_input.shape		) < 3:
              SCREAMING_SNAKE_CASE :      Any       =   batch_features_input[:, :, None]

        self.assertTrue(
            batch_features_input.shape
            == (self.feat_extract_tester.batch_size, len(speech_inputs[0]		), self.feat_extract_tester.feature_size)		)


  @require_torch
  def  _lowercase      (			self    : Any		)					->Optional[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      int       =   self.feat_extract_tester.prepare_inputs_for_common(equal_length=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      List[str]       =   self.feature_extraction_class(**self.feat_extract_dict		)
        SCREAMING_SNAKE_CASE :      List[Any]       =   feat_extract.model_input_names[0]

        SCREAMING_SNAKE_CASE :      str       =   BatchFeature({input_name: speech_inputs}				,				tensor_type="""pt"""		)

        SCREAMING_SNAKE_CASE :      int       =   processed_features[input_name]

        if len(batch_features_input.shape		) < 3:
              SCREAMING_SNAKE_CASE :      Tuple       =   batch_features_input[:, :, None]

        self.assertTrue(
            batch_features_input.shape
            == (self.feat_extract_tester.batch_size, len(speech_inputs[0]		), self.feat_extract_tester.feature_size)		)


  @require_tf
  def  _lowercase      (			self    : Tuple		)					->Union[str, Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      str       =   self.feat_extract_tester.prepare_inputs_for_common(equal_length=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Optional[int]       =   self.feature_extraction_class(**self.feat_extract_dict		)
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   feat_extract.model_input_names[0]

        SCREAMING_SNAKE_CASE :      Optional[int]       =   BatchFeature({input_name: speech_inputs}				,				tensor_type="""tf"""		)

        SCREAMING_SNAKE_CASE :      str       =   processed_features[input_name]

        if len(batch_features_input.shape		) < 3:
              SCREAMING_SNAKE_CASE :      Optional[int]       =   batch_features_input[:, :, None]

        self.assertTrue(
            batch_features_input.shape
            == (self.feat_extract_tester.batch_size, len(speech_inputs[0]		), self.feat_extract_tester.feature_size)		)


  def  _lowercase      (			self    : Union[str, Any]				,				UpperCAmelCase__    : Optional[int]=False		)					->Dict:
        """simple docstring"""







        def _inputs_have_equal_length(UpperCAmelCase__    : Tuple		):
              SCREAMING_SNAKE_CASE :      Any       =   len(input[0]		)
              for input_slice in input[1:]:
                    if len(UpperCAmelCase__		) != length:
                          return False
              return True

        def _inputs_are_equal(UpperCAmelCase__    : str				,				UpperCAmelCase__    : str		):
              if len(UpperCAmelCase__		) != len(UpperCAmelCase__		):
                    return False

              for input_slice_a, input_slice_a in zip(UpperCAmelCase__				,				UpperCAmelCase__		):
                    if not np.allclose(np.asarray(UpperCAmelCase__		)				,				np.asarray(UpperCAmelCase__		)				,				atol=1e-3		):
                          return False
              return True

        SCREAMING_SNAKE_CASE :      List[str]       =   self.feature_extraction_class(**self.feat_extract_dict		)
        SCREAMING_SNAKE_CASE :      List[str]       =   self.feat_extract_tester.prepare_inputs_for_common(numpify=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      int       =   feat_extract.model_input_names[0]

        SCREAMING_SNAKE_CASE :      int       =   BatchFeature({input_name: speech_inputs}		)

        SCREAMING_SNAKE_CASE :      str       =   self.feat_extract_tester.seq_length_diff
        SCREAMING_SNAKE_CASE :      Any       =   self.feat_extract_tester.max_seq_length + pad_diff
        SCREAMING_SNAKE_CASE :      Any       =   self.feat_extract_tester.min_seq_length
        SCREAMING_SNAKE_CASE :      Any       =   self.feat_extract_tester.batch_size
        SCREAMING_SNAKE_CASE :      int       =   self.feat_extract_tester.feature_size

        # test padding for List[int] + numpy
        SCREAMING_SNAKE_CASE :      int       =   feat_extract.pad(UpperCAmelCase__				,				padding=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   input_a[input_name]

        SCREAMING_SNAKE_CASE :      int       =   feat_extract.pad(UpperCAmelCase__				,				padding="""longest"""		)
        SCREAMING_SNAKE_CASE :      int       =   input_a[input_name]

        SCREAMING_SNAKE_CASE :      int       =   feat_extract.pad(UpperCAmelCase__				,				padding="""max_length"""				,				max_length=len(speech_inputs[-1]		)		)
        SCREAMING_SNAKE_CASE :      Tuple       =   input_a[input_name]

        SCREAMING_SNAKE_CASE :      List[Any]       =   feat_extract.pad(UpperCAmelCase__				,				padding="""longest"""				,				return_tensors="""np"""		)
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   input_a[input_name]

        # max_length parameter has to be provided when setting `padding="max_length"`
        with self.assertRaises(UpperCAmelCase__		):
              feat_extract.pad(UpperCAmelCase__				,				padding="""max_length"""		)[input_name]

        SCREAMING_SNAKE_CASE :      Optional[Any]       =   feat_extract.pad(
            UpperCAmelCase__				,				padding="""max_length"""				,				max_length=UpperCAmelCase__				,				return_tensors="""np"""		)
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   input_a[input_name]

        self.assertFalse(_inputs_have_equal_length(UpperCAmelCase__		)		)
        self.assertTrue(_inputs_have_equal_length(UpperCAmelCase__		)		)
        self.assertTrue(_inputs_have_equal_length(UpperCAmelCase__		)		)
        self.assertTrue(_inputs_are_equal(UpperCAmelCase__				,				UpperCAmelCase__		)		)
        self.assertTrue(len(input_a[0]		) == pad_min_length		)
        self.assertTrue(len(input_a[1]		) == pad_min_length + pad_diff		)
        self.assertTrue(input_a.shape[:2] == (batch_size, len(input_a[0]		))		)
        self.assertTrue(input_a.shape[:2] == (batch_size, pad_max_length)		)

        if feature_size > 1:
              self.assertTrue(input_a.shape[2] == input_a.shape[2] == feature_size		)

        # test padding for `pad_to_multiple_of` for List[int] + numpy
        SCREAMING_SNAKE_CASE :      str       =   feat_extract.pad(UpperCAmelCase__				,				pad_to_multiple_of=1_0		)
        SCREAMING_SNAKE_CASE :      List[str]       =   input_a[input_name]

        SCREAMING_SNAKE_CASE :      List[Any]       =   feat_extract.pad(UpperCAmelCase__				,				padding="""longest"""				,				pad_to_multiple_of=1_0		)
        SCREAMING_SNAKE_CASE :      Tuple       =   input_a[input_name]

        SCREAMING_SNAKE_CASE :      str       =   feat_extract.pad(
            UpperCAmelCase__				,				padding="""max_length"""				,				pad_to_multiple_of=1_0				,				max_length=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Optional[int]       =   input_a[input_name]

        SCREAMING_SNAKE_CASE :      Any       =   feat_extract.pad(
            UpperCAmelCase__				,				padding="""max_length"""				,				pad_to_multiple_of=1_0				,				max_length=UpperCAmelCase__				,				return_tensors="""np"""				,				)
        SCREAMING_SNAKE_CASE :      Optional[int]       =   input_a[input_name]

        self.assertTrue(all(len(UpperCAmelCase__		) % 1_0 == 0 for x in input_a		)		)
        self.assertTrue(_inputs_are_equal(UpperCAmelCase__				,				UpperCAmelCase__		)		)

        SCREAMING_SNAKE_CASE :      str       =   pad_max_length if pad_max_length % 1_0 == 0 else (pad_max_length // 1_0 + 1) * 1_0
        self.assertTrue(all(len(UpperCAmelCase__		) == expected_mult_pad_length for x in input_a		)		)
        self.assertEqual(input_a.shape[:2]				,				(batch_size, expected_mult_pad_length)		)

        if feature_size > 1:
              self.assertTrue(input_a.shape[2] == feature_size		)

        # Check padding value is correct
        SCREAMING_SNAKE_CASE :      List[Any]       =   (np.ones(self.feat_extract_tester.feature_size		) * feat_extract.padding_value).sum()
        self.assertTrue(
            abs(np.asarray(input_a[0]		)[pad_min_length:].sum() - padding_vector_sum * (pad_max_length - pad_min_length)		)
            < 1e-3		)
        self.assertTrue(
            abs(
                np.asarray(input_a[1]		)[pad_min_length + pad_diff :].sum()
                - padding_vector_sum * (pad_max_length - pad_min_length - pad_diff)		)
            < 1e-3		)
        self.assertTrue(
            abs(
                np.asarray(input_a[2]		)[pad_min_length + 2 * pad_diff :].sum()
                - padding_vector_sum * (pad_max_length - pad_min_length - 2 * pad_diff)		)
            < 1e-3		)
        self.assertTrue(
            abs(input_a[0, pad_min_length:].sum() - padding_vector_sum * (pad_max_length - pad_min_length)		) < 1e-3		)
        self.assertTrue(
            abs(input_a[0, pad_min_length:].sum() - padding_vector_sum * (expected_mult_pad_length - pad_min_length)		)
            < 1e-3		)


  def  _lowercase      (			self    : Tuple				,				UpperCAmelCase__    : Optional[Any]=False		)					->Dict:
        """simple docstring"""







        def _inputs_have_equal_length(UpperCAmelCase__    : Dict		):
              SCREAMING_SNAKE_CASE :      Any       =   len(input[0]		)
              for input_slice in input[1:]:
                    if len(UpperCAmelCase__		) != length:
                          return False
              return True

        def _inputs_are_equal(UpperCAmelCase__    : Any				,				UpperCAmelCase__    : Optional[int]		):
              if len(UpperCAmelCase__		) != len(UpperCAmelCase__		):
                    return False

              for input_slice_a, input_slice_a in zip(UpperCAmelCase__				,				UpperCAmelCase__		):
                    if not np.allclose(np.asarray(UpperCAmelCase__		)				,				np.asarray(UpperCAmelCase__		)				,				atol=1e-3		):
                          return False
              return True

        SCREAMING_SNAKE_CASE :      int       =   self.feature_extraction_class(**self.feat_extract_dict		)
        SCREAMING_SNAKE_CASE :      Dict       =   self.feat_extract_tester.prepare_inputs_for_common(numpify=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Dict       =   feat_extract.model_input_names[0]

        SCREAMING_SNAKE_CASE :      List[str]       =   BatchFeature({input_name: speech_inputs}		)

        # truncate to smallest
        SCREAMING_SNAKE_CASE :      List[str]       =   feat_extract.pad(
            UpperCAmelCase__				,				padding="""max_length"""				,				max_length=len(speech_inputs[0]		)				,				truncation=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   input_a[input_name]

        SCREAMING_SNAKE_CASE :      Optional[Any]       =   feat_extract.pad(UpperCAmelCase__				,				padding="""max_length"""				,				max_length=len(speech_inputs[0]		)		)
        SCREAMING_SNAKE_CASE :      List[str]       =   input_a[input_name]

        self.assertTrue(_inputs_have_equal_length(UpperCAmelCase__		)		)
        self.assertFalse(_inputs_have_equal_length(UpperCAmelCase__		)		)

        # truncate to smallest with np
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   feat_extract.pad(
            UpperCAmelCase__				,				padding="""max_length"""				,				max_length=len(speech_inputs[0]		)				,				return_tensors="""np"""				,				truncation=UpperCAmelCase__				,				)
        SCREAMING_SNAKE_CASE :      Dict       =   input_a[input_name]

        SCREAMING_SNAKE_CASE :      Dict       =   feat_extract.pad(
            UpperCAmelCase__				,				padding="""max_length"""				,				max_length=len(speech_inputs[0]		)				,				return_tensors="""np"""		)
        SCREAMING_SNAKE_CASE :      str       =   input_a[input_name]

        self.assertTrue(_inputs_have_equal_length(UpperCAmelCase__		)		)
        self.assertTrue(input_a.shape[1] == len(speech_inputs[0]		)		)

        # since truncation forces padding to be smaller than longest input
        # function can't return `np.ndarray`, but has to return list
        self.assertFalse(_inputs_have_equal_length(UpperCAmelCase__		)		)

        # truncate to middle
        SCREAMING_SNAKE_CASE :      str       =   feat_extract.pad(
            UpperCAmelCase__				,				padding="""max_length"""				,				max_length=len(speech_inputs[1]		)				,				truncation=UpperCAmelCase__				,				return_tensors="""np"""				,				)
        SCREAMING_SNAKE_CASE :      Optional[int]       =   input_a[input_name]

        SCREAMING_SNAKE_CASE :      Optional[int]       =   feat_extract.pad(
            UpperCAmelCase__				,				padding="""max_length"""				,				max_length=len(speech_inputs[1]		)				,				truncation=UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Any       =   input_a[input_name]

        SCREAMING_SNAKE_CASE :      List[str]       =   feat_extract.pad(
            UpperCAmelCase__				,				padding="""max_length"""				,				max_length=len(speech_inputs[1]		)				,				return_tensors="""np"""		)
        SCREAMING_SNAKE_CASE :      Any       =   input_a[input_name]

        self.assertTrue(input_a.shape[1] == len(speech_inputs[1]		)		)
        self.assertTrue(_inputs_have_equal_length(UpperCAmelCase__		)		)
        self.assertTrue(_inputs_have_equal_length(UpperCAmelCase__		)		)
        self.assertTrue(_inputs_are_equal(UpperCAmelCase__				,				UpperCAmelCase__		)		)

        # since truncation forces padding to be smaller than longest input
        # function can't return `np.ndarray`, but has to return list
        self.assertFalse(_inputs_have_equal_length(UpperCAmelCase__		)		)
        self.assertTrue(len(input_a[-1]		) == len(speech_inputs[-1]		)		)

        # padding has to be max_length when setting `truncation=True`
        with self.assertRaises(UpperCAmelCase__		):
              feat_extract.pad(UpperCAmelCase__				,				truncation=UpperCAmelCase__		)[input_name]

        # padding has to be max_length when setting `truncation=True`
        with self.assertRaises(UpperCAmelCase__		):
              feat_extract.pad(UpperCAmelCase__				,				padding="""longest"""				,				truncation=UpperCAmelCase__		)[input_name]

        # padding has to be max_length when setting `truncation=True`
        with self.assertRaises(UpperCAmelCase__		):
              feat_extract.pad(UpperCAmelCase__				,				padding="""longest"""				,				truncation=UpperCAmelCase__		)[input_name]

        # max_length parameter has to be provided when setting `truncation=True` and padding="max_length"
        with self.assertRaises(UpperCAmelCase__		):
              feat_extract.pad(UpperCAmelCase__				,				padding="""max_length"""				,				truncation=UpperCAmelCase__		)[input_name]

        # test truncation for `pad_to_multiple_of` for List[int] + numpy
        SCREAMING_SNAKE_CASE :      List[str]       =   1_2
        SCREAMING_SNAKE_CASE :      Tuple       =   feat_extract.pad(
            UpperCAmelCase__				,				padding="""max_length"""				,				max_length=len(speech_inputs[0]		)				,				pad_to_multiple_of=UpperCAmelCase__				,				truncation=UpperCAmelCase__				,				)
        SCREAMING_SNAKE_CASE :      List[Any]       =   input_a[input_name]

        SCREAMING_SNAKE_CASE :      List[Any]       =   feat_extract.pad(
            UpperCAmelCase__				,				padding="""max_length"""				,				max_length=len(speech_inputs[0]		)				,				pad_to_multiple_of=UpperCAmelCase__				,				)
        SCREAMING_SNAKE_CASE :      List[Any]       =   input_a[input_name]

        # retrieve expected_length as multiple of pad_to_multiple_of
        SCREAMING_SNAKE_CASE :      str       =   len(speech_inputs[0]		)
        if expected_length % pad_to_multiple_of != 0:
              SCREAMING_SNAKE_CASE :      Any       =   ((len(speech_inputs[0]		) // pad_to_multiple_of) + 1) * pad_to_multiple_of

        self.assertTrue(len(input_a[0]		) == expected_length		)
        self.assertTrue(_inputs_have_equal_length(UpperCAmelCase__		)		)
        self.assertFalse(_inputs_have_equal_length(UpperCAmelCase__		)		)


  def  _lowercase      (			self    : Tuple		)					->Dict:
        """simple docstring"""







        self._check_padding(numpify=UpperCAmelCase__		)


  def  _lowercase      (			self    : Any		)					->int:
        """simple docstring"""







        self._check_padding(numpify=UpperCAmelCase__		)


  def  _lowercase      (			self    : Optional[int]		)					->List[Any]:
        """simple docstring"""







        self._check_truncation(numpify=UpperCAmelCase__		)


  def  _lowercase      (			self    : str		)					->int:
        """simple docstring"""







        self._check_truncation(numpify=UpperCAmelCase__		)


  @require_torch
  def  _lowercase      (			self    : Dict		)					->Dict:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Dict       =   self.feature_extraction_class(**self.feat_extract_dict		)
        SCREAMING_SNAKE_CASE :      List[Any]       =   self.feat_extract_tester.prepare_inputs_for_common()
        SCREAMING_SNAKE_CASE :      Optional[int]       =   feat_extract.model_input_names[0]

        SCREAMING_SNAKE_CASE :      Optional[Any]       =   BatchFeature({input_name: speech_inputs}		)

        SCREAMING_SNAKE_CASE :      Optional[Any]       =   feat_extract.pad(UpperCAmelCase__				,				padding="""longest"""				,				return_tensors="""np"""		)[input_name]
        SCREAMING_SNAKE_CASE :      Any       =   feat_extract.pad(UpperCAmelCase__				,				padding="""longest"""				,				return_tensors="""pt"""		)[input_name]

        self.assertTrue(abs(input_np.astype(np.floataa		).sum() - input_pt.numpy().astype(np.floataa		).sum()		) < 1e-2		)


  @require_tf
  def  _lowercase      (			self    : Optional[int]		)					->str:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Any       =   self.feature_extraction_class(**self.feat_extract_dict		)
        SCREAMING_SNAKE_CASE :      Tuple       =   self.feat_extract_tester.prepare_inputs_for_common()
        SCREAMING_SNAKE_CASE :      List[str]       =   feat_extract.model_input_names[0]

        SCREAMING_SNAKE_CASE :      int       =   BatchFeature({input_name: speech_inputs}		)

        SCREAMING_SNAKE_CASE :      Dict       =   feat_extract.pad(UpperCAmelCase__				,				padding="""longest"""				,				return_tensors="""np"""		)[input_name]
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   feat_extract.pad(UpperCAmelCase__				,				padding="""longest"""				,				return_tensors="""tf"""		)[input_name]

        self.assertTrue(abs(input_np.astype(np.floataa		).sum() - input_tf.numpy().astype(np.floataa		).sum()		) < 1e-2		)


  def  _lowercase      (			self    : List[Any]		)					->Dict:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      Dict       =   self.feat_extract_dict
        SCREAMING_SNAKE_CASE :      Any       =   True
        SCREAMING_SNAKE_CASE :      str       =   self.feature_extraction_class(**UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      Optional[int]       =   self.feat_extract_tester.prepare_inputs_for_common()
        SCREAMING_SNAKE_CASE :      Optional[int]       =   [len(UpperCAmelCase__		) for x in speech_inputs]
        SCREAMING_SNAKE_CASE :      Tuple       =   feat_extract.model_input_names[0]

        SCREAMING_SNAKE_CASE :      str       =   BatchFeature({input_name: speech_inputs}		)

        SCREAMING_SNAKE_CASE :      List[str]       =   feat_extract.pad(UpperCAmelCase__				,				padding="""longest"""				,				return_tensors="""np"""		)
        self.assertIn("""attention_mask"""				,				UpperCAmelCase__		)
        self.assertListEqual(list(processed.attention_mask.shape		)				,				list(processed[input_name].shape[:2]		)		)
        self.assertListEqual(processed.attention_mask.sum(-1		).tolist()				,				UpperCAmelCase__		)




  def  _lowercase      (			self    : Dict		)					->Any:
        """simple docstring"""







        SCREAMING_SNAKE_CASE :      int       =   self.feat_extract_dict
        SCREAMING_SNAKE_CASE :      int       =   True
        SCREAMING_SNAKE_CASE :      Dict       =   self.feature_extraction_class(**UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE :      str       =   self.feat_extract_tester.prepare_inputs_for_common()
        SCREAMING_SNAKE_CASE :      List[str]       =   [len(UpperCAmelCase__		) for x in speech_inputs]
        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   feat_extract.model_input_names[0]

        SCREAMING_SNAKE_CASE :      Union[str, Any]       =   BatchFeature({input_name: speech_inputs}		)
        SCREAMING_SNAKE_CASE :      Optional[Any]       =   min(UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE :      List[str]       =   feat_extract.pad(
            UpperCAmelCase__				,				padding="""max_length"""				,				max_length=UpperCAmelCase__				,				truncation=UpperCAmelCase__				,				return_tensors="""np"""		)
        self.assertIn("""attention_mask"""				,				UpperCAmelCase__		)
        self.assertListEqual(
            list(processed_pad.attention_mask.shape		)				,				[processed_pad[input_name].shape[0], max_length]		)
        self.assertListEqual(
            processed_pad.attention_mask[:, :max_length].sum(-1		).tolist()				,				[max_length for x in speech_inputs]		)







import importlib
import inspect
import json
import os
import re
import shutil
import sys
from pathlib import Path
from typing import Dict, Optional, Union
from urllib import request

from huggingface_hub import HfFolder, cached_download, hf_hub_download, model_info
from packaging import version

from .. import __version__
from . import DIFFUSERS_DYNAMIC_MODULE_NAME, HF_MODULES_CACHE, logging


UpperCAmelCase__ :     List[str]				    =    (
    """https://raw.githubusercontent.com/huggingface/diffusers/{revision}/examples/community/{pipeline}.py"""
)


UpperCAmelCase__ :     Union[str, Any]				    =    logging.get_logger(__name__)  # pylint: disable=invalid-name






def 					__lowercase      (     )     ->						List[Any]:
      SCREAMING_SNAKE_CASE :      Optional[int]       =   """https://pypi.org/pypi/diffusers/json"""
      SCREAMING_SNAKE_CASE :      Dict       =   json.loads(request.urlopen(_A			).read()			)["""releases"""].keys()
      return sorted(_A						,    key=lambda _A			: version.Version(_A			)			)






def 					__lowercase      (     )     ->						Any:
      # This function has already been executed if HF_MODULES_CACHE already is in the Python path.
      if HF_MODULES_CACHE in sys.path:
            return

      sys.path.append(_A			)
      os.makedirs(_A						,    exist_ok=_A			)
      SCREAMING_SNAKE_CASE :      Dict       =   Path(_A			) / """__init__.py"""
      if not init_path.exists():
            init_path.touch()






def 					__lowercase      (     _A			)     ->						Any:
      init_hf_modules()
      SCREAMING_SNAKE_CASE :      Any       =   Path(_A			) / name
      # If the parent module does not exist yet, recursively create it.
      if not dynamic_module_path.parent.exists():
            create_dynamic_module(dynamic_module_path.parent			)
      os.makedirs(_A						,    exist_ok=_A			)
      SCREAMING_SNAKE_CASE :      str       =   dynamic_module_path / """__init__.py"""
      if not init_path.exists():
            init_path.touch()






def 					__lowercase      (     _A			)     ->						Optional[int]:
      with open(_A						,    """r"""						,    encoding="""utf-8"""			) as f:
            SCREAMING_SNAKE_CASE :      Any       =   f.read()

      # Imports of the form `import .xxx`
      SCREAMING_SNAKE_CASE :      List[str]       =   re.findall("""^\s*import\s+\.(\S+)\s*$"""						,    _A						,    flags=re.MULTILINE			)
      # Imports of the form `from .xxx import yyy`
      relative_imports += re.findall("""^\s*from\s+\.(\S+)\s+import"""						,    _A						,    flags=re.MULTILINE			)
      # Unique-ify
      return list(set(_A			)			)






def 					__lowercase      (     _A			)     ->						Optional[Any]:
      SCREAMING_SNAKE_CASE :      Union[str, Any]       =   False
      SCREAMING_SNAKE_CASE :      List[Any]       =   [module_file]
      SCREAMING_SNAKE_CASE :      Any       =   []

      # Let's recurse through all relative imports
      while not no_change:
            SCREAMING_SNAKE_CASE :      List[str]       =   []
            for f in files_to_check:
                  new_imports.extend(get_relative_imports(_A			)			)

            SCREAMING_SNAKE_CASE :      List[str]       =   Path(_A			).parent
            SCREAMING_SNAKE_CASE :      int       =   [str(module_path / m			) for m in new_imports]
            SCREAMING_SNAKE_CASE :      int       =   [f for f in new_import_files if f not in all_relative_imports]
            SCREAMING_SNAKE_CASE :      Optional[Any]       =   [F"{f}.py" for f in new_import_files]

            SCREAMING_SNAKE_CASE :      List[Any]       =   len(_A			) == 0
            all_relative_imports.extend(_A			)

      return all_relative_imports






def 					__lowercase      (     _A			)     ->						Union[str, Any]:
      with open(_A						,    """r"""						,    encoding="""utf-8"""			) as f:
            SCREAMING_SNAKE_CASE :      List[Any]       =   f.read()

      # Imports of the form `import xxx`
      SCREAMING_SNAKE_CASE :      Any       =   re.findall("""^\s*import\s+(\S+)\s*$"""						,    _A						,    flags=re.MULTILINE			)
      # Imports of the form `from xxx import yyy`
      imports += re.findall("""^\s*from\s+(\S+)\s+import"""						,    _A						,    flags=re.MULTILINE			)
      # Only keep the top-level module
      SCREAMING_SNAKE_CASE :      List[Any]       =   [imp.split("""."""			)[0] for imp in imports if not imp.startswith("""."""			)]

      # Unique-ify and test we got them all
      SCREAMING_SNAKE_CASE :      Dict       =   list(set(_A			)			)
      SCREAMING_SNAKE_CASE :      Dict       =   []
      for imp in imports:
            try:
                  importlib.import_module(_A			)
            except ImportError:
                  missing_packages.append(_A			)

      if len(_A			) > 0:
            raise ImportError(
                """This modeling file requires the following packages that were not found in your environment: """
                F"{', '.join(_A			)}. Run `pip install {' '.join(_A			)}`"			)

      return get_relative_imports(_A			)






def 					__lowercase      (     _A						,    _A			)     ->						List[str]:
      SCREAMING_SNAKE_CASE :      int       =   module_path.replace(os.path.sep						,    """."""			)
      SCREAMING_SNAKE_CASE :      List[str]       =   importlib.import_module(_A			)

      if class_name is None:
            return find_pipeline_class(_A			)
      return getattr(_A						,    _A			)






def 					__lowercase      (     _A			)     ->						Optional[int]:
      from ..pipelines import DiffusionPipeline

      SCREAMING_SNAKE_CASE :      Tuple       =   dict(inspect.getmembers(_A						,    inspect.isclass			)			)

      SCREAMING_SNAKE_CASE :      Dict       =   None
      for cls_name, cls in cls_members.items():
            if (
                cls_name != DiffusionPipeline.__name__
                and issubclass(cls						,    _A			)
                and cls.__module__.split("""."""			)[0] != "diffusers"
            ):
                  if pipeline_class is not None:
                        raise ValueError(
                            F"Multiple classes that inherit from {DiffusionPipeline.__name__} have been found:"
                            F" {pipeline_class.__name__}, and {cls_name}. Please make sure to define only one in"
                            F" {loaded_module}."			)
                  SCREAMING_SNAKE_CASE :      List[str]       =   cls

      return pipeline_class






def 					__lowercase      (     _A						,    _A						,    _A = None						,    _A = False						,    _A = False						,    _A = None						,    _A = None						,    _A = None						,    _A = False						,    )     ->						List[str]:
      SCREAMING_SNAKE_CASE :      List[str]       =   str(_A			)

      SCREAMING_SNAKE_CASE :      List[Any]       =   os.path.join(_A						,    _A			)

      if os.path.isfile(_A			):
            SCREAMING_SNAKE_CASE :      Union[str, Any]       =   module_file_or_url
            SCREAMING_SNAKE_CASE :      Dict       =   """local"""
      elif pretrained_model_name_or_path.count("""/"""			) == 0:
            SCREAMING_SNAKE_CASE :      Tuple       =   get_diffusers_versions()
            # cut ".dev0"
            SCREAMING_SNAKE_CASE :      str       =   """v""" + """.""".join(__version__.split("""."""			)[:3]			)

            # retrieve github version that matches
            if revision is None:
                  SCREAMING_SNAKE_CASE :      Any       =   latest_version if latest_version[1:] in available_versions else """main"""
                  logger.info(F"Defaulting to latest_version: {revision}."			)
            elif revision in available_versions:
                  SCREAMING_SNAKE_CASE :      List[Any]       =   F"v{revision}"
            elif revision == "main":
                  SCREAMING_SNAKE_CASE :      Union[str, Any]       =   revision
            else:
                  raise ValueError(
                      F"`custom_revision`: {revision} does not exist. Please make sure to choose one of"
                      F" {', '.join(available_versions + ['main']			)}."			)

            # community pipeline on GitHub
            SCREAMING_SNAKE_CASE :      Tuple       =   COMMUNITY_PIPELINES_URL.format(revision=_A						,    pipeline=_A			)
            try:
                  SCREAMING_SNAKE_CASE :      str       =   cached_download(
                      _A						,    cache_dir=_A						,    force_download=_A						,    proxies=_A						,    resume_download=_A						,    local_files_only=_A						,    use_auth_token=_A						,    )
                  SCREAMING_SNAKE_CASE :      Tuple       =   """git"""
                  SCREAMING_SNAKE_CASE :      Optional[int]       =   pretrained_model_name_or_path + """.py"""
            except EnvironmentError:
                  logger.error(F"Could not locate the {module_file} inside {pretrained_model_name_or_path}."			)
                  raise
      else:
            try:
                  # Load from URL or cache if already cached
                  SCREAMING_SNAKE_CASE :      int       =   hf_hub_download(
                      _A						,    _A						,    cache_dir=_A						,    force_download=_A						,    proxies=_A						,    resume_download=_A						,    local_files_only=_A						,    use_auth_token=_A						,    )
                  SCREAMING_SNAKE_CASE :      Union[str, Any]       =   os.path.join("""local"""						,    """--""".join(pretrained_model_name_or_path.split("""/"""			)			)			)
            except EnvironmentError:
                  logger.error(F"Could not locate the {module_file} inside {pretrained_model_name_or_path}."			)
                  raise

    # Check we have all the requirements in our environment
      SCREAMING_SNAKE_CASE :      int       =   check_imports(_A			)

      # Now we move the module inside our cached dynamic modules.
      SCREAMING_SNAKE_CASE :      List[Any]       =   DIFFUSERS_DYNAMIC_MODULE_NAME + os.path.sep + submodule
      create_dynamic_module(_A			)
      SCREAMING_SNAKE_CASE :      Tuple       =   Path(_A			) / full_submodule
      if submodule == "local" or submodule == "git":
            # We always copy local files (we could hash the file to see if there was a change, and give them the name of
            # that hash, to only copy when there is a modification but it seems overkill for now).
            # The only reason we do the copy is to avoid putting too many folders in sys.path.
            shutil.copy(_A						,    submodule_path / module_file			)
            for module_needed in modules_needed:
                  SCREAMING_SNAKE_CASE :      Optional[int]       =   F"{module_needed}.py"
                  shutil.copy(os.path.join(_A						,    _A			)						,    submodule_path / module_needed			)
      else:
            # Get the commit hash
            # TODO: we will get this info in the etag soon, so retrieve it from there and not here.
            if isinstance(_A						,    _A			):
                  SCREAMING_SNAKE_CASE :      Tuple       =   use_auth_token
            elif use_auth_token is True:
                  SCREAMING_SNAKE_CASE :      str       =   HfFolder.get_token()
            else:
                  SCREAMING_SNAKE_CASE :      Optional[int]       =   None

            SCREAMING_SNAKE_CASE :      List[Any]       =   model_info(_A						,    revision=_A						,    token=_A			).sha

            # The module file will end up being placed in a subfolder with the git hash of the repo. This way we get the
            # benefit of versioning.
            SCREAMING_SNAKE_CASE :      int       =   submodule_path / commit_hash
            SCREAMING_SNAKE_CASE :      Tuple       =   full_submodule + os.path.sep + commit_hash
            create_dynamic_module(_A			)

            if not (submodule_path / module_file).exists():
                  shutil.copy(_A						,    submodule_path / module_file			)
            # Make sure we also have every file with relative
            for module_needed in modules_needed:
                  if not (submodule_path / module_needed).exists():
                        get_cached_module_file(
                            _A						,    F"{module_needed}.py"						,    cache_dir=_A						,    force_download=_A						,    resume_download=_A						,    proxies=_A						,    use_auth_token=_A						,    revision=_A						,    local_files_only=_A						,    )
      return os.path.join(_A						,    _A			)







def 					__lowercase      (     _A						,    _A						,    _A = None						,    _A = None						,    _A = False						,    _A = False						,    _A = None						,    _A = None						,    _A = None						,    _A = False						,    **_A						,    )     ->						Tuple:
      SCREAMING_SNAKE_CASE :      Dict       =   get_cached_module_file(
          _A						,    _A						,    cache_dir=_A						,    force_download=_A						,    resume_download=_A						,    proxies=_A						,    use_auth_token=_A						,    revision=_A						,    local_files_only=_A						,    )
      return get_class_in_module(_A						,    final_module.replace(""".py"""						,    """"""			)			)




