

'''simple docstring'''






import math
from typing import Dict, Iterable, List, Optional, Tuple, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import normalize, rescale, resize, to_channel_dimension_format
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    get_image_size,
    is_torch_available,
    is_torch_tensor,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_torch_available():
 import torch

if is_vision_available():
 import PIL


lowerCAmelCase__   		=      logging.get_logger(__name__)


def 						_A	(       A__	,							A__	,							A__	,							A__ ):






 """simple docstring"""



 def constraint_to_multiple_of(A__	,							A__	,							A__=0	,							A__=None ):
  __lowercase				    =					round(val / multiple ) * multiple

  if max_val is not None and x > max_val:
   __lowercase				    =					math.floor(val / multiple ) * multiple

  if x < min_val:
   __lowercase				    =					math.ceil(val / multiple ) * multiple

  return x

 __lowercase				    =					(output_size, output_size) if isinstance(A__	,							A__ ) else output_size

 __lowercase    ,      __lowercase				    =					get_image_size(A__ )
 __lowercase    ,      __lowercase				    =					output_size

 # determine new height and width
 __lowercase				    =					output_height / input_height
 __lowercase				    =					output_width / input_width

 if keep_aspect_ratio:
  # scale as little as possible
  if abs(1 - scale_width ) < abs(1 - scale_height ):
   # fit width
   __lowercase				    =					scale_width
  else:
   # fit height
   __lowercase				    =					scale_height

 __lowercase				    =					constraint_to_multiple_of(scale_height * input_height	,							multiple=A__ )
 __lowercase				    =					constraint_to_multiple_of(scale_width * input_width	,							multiple=A__ )

 return (new_height, new_width)






class      lowercase_      (lowerCamelCase__ ):




  """simple docstring"""




  SCREAMING_SNAKE_CASE       :    List[Any]							     =     ['pixel_values']






  def __init__( self  :	Tuple   ,lowercase__  :	bool = True   ,lowercase__  :	Dict[str, int] = None   ,lowercase__  :	PILImageResampling = PILImageResampling.BILINEAR   ,lowercase__  :	bool = False   ,lowercase__  :	int = 1   ,lowercase__  :	bool = True   ,lowercase__  :	Union[int, float] = 1 / 2_5_5   ,lowercase__  :	bool = True   ,lowercase__  :	Optional[Union[float, List[float]]] = None   ,lowercase__  :	Optional[Union[float, List[float]]] = None   ,**lowercase__  :	Dict   ,):
   super().__init__(**lowercase__     )
   __lowercase				    =					size if size is not None else {'''height''': 3_8_4, '''width''': 3_8_4}
   __lowercase				    =					get_size_dict(lowercase__     )
   __lowercase				    =					do_resize
   __lowercase				    =					size
   __lowercase				    =					keep_aspect_ratio
   __lowercase				    =					ensure_multiple_of
   __lowercase				    =					resample
   __lowercase				    =					do_rescale
   __lowercase				    =					rescale_factor
   __lowercase				    =					do_normalize
   __lowercase				    =					image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
   __lowercase				    =					image_std if image_std is not None else IMAGENET_STANDARD_STD






  def    SCREAMING_SNAKE_CASE ( self  :	str   ,lowercase__  :	np.ndarray   ,lowercase__  :	Dict[str, int]   ,lowercase__  :	bool = False   ,lowercase__  :	int = 1   ,lowercase__  :	PILImageResampling = PILImageResampling.BICUBIC   ,lowercase__  :	Optional[Union[str, ChannelDimension]] = None   ,**lowercase__  :	Tuple   ,):
   __lowercase				    =					get_size_dict(lowercase__     )
   if "height" not in size or "width" not in size:
    raise ValueError(F"The size dictionary must contain the keys 'height' and 'width'. Got {size.keys()}"     )
   __lowercase				    =					get_resize_output_image_size(
       lowercase__   ,output_size=(size['''height'''], size['''width'''])   ,keep_aspect_ratio=lowercase__   ,multiple=lowercase__   ,)
   return resize(lowercase__   ,size=lowercase__   ,resample=lowercase__   ,data_format=lowercase__   ,**lowercase__     )






  def    SCREAMING_SNAKE_CASE ( self  :	Optional[Any]   ,lowercase__  :	np.ndarray   ,lowercase__  :	Union[int, float]   ,lowercase__  :	Optional[Union[str, ChannelDimension]] = None   ,**lowercase__  :	Optional[Any]   ,):
   return rescale(lowercase__   ,scale=lowercase__   ,data_format=lowercase__   ,**lowercase__     )






  def    SCREAMING_SNAKE_CASE ( self  :	Union[str, Any]   ,lowercase__  :	np.ndarray   ,lowercase__  :	Union[float, List[float]]   ,lowercase__  :	Union[float, List[float]]   ,lowercase__  :	Optional[Union[str, ChannelDimension]] = None   ,**lowercase__  :	Tuple   ,):
   return normalize(lowercase__   ,mean=lowercase__   ,std=lowercase__   ,data_format=lowercase__   ,**lowercase__     )






  def    SCREAMING_SNAKE_CASE ( self  :	Tuple   ,lowercase__  :	ImageInput   ,lowercase__  :	bool = None   ,lowercase__  :	int = None   ,lowercase__  :	bool = None   ,lowercase__  :	int = None   ,lowercase__  :	PILImageResampling = None   ,lowercase__  :	bool = None   ,lowercase__  :	float = None   ,lowercase__  :	bool = None   ,lowercase__  :	Optional[Union[float, List[float]]] = None   ,lowercase__  :	Optional[Union[float, List[float]]] = None   ,lowercase__  :	Optional[Union[str, TensorType]] = None   ,lowercase__  :	ChannelDimension = ChannelDimension.FIRST   ,**lowercase__  :	str   ,):
   __lowercase				    =					do_resize if do_resize is not None else self.do_resize
   __lowercase				    =					size if size is not None else self.size
   __lowercase				    =					get_size_dict(lowercase__     )
   __lowercase				    =					keep_aspect_ratio if keep_aspect_ratio is not None else self.keep_aspect_ratio
   __lowercase				    =					ensure_multiple_of if ensure_multiple_of is not None else self.ensure_multiple_of
   __lowercase				    =					resample if resample is not None else self.resample
   __lowercase				    =					do_rescale if do_rescale is not None else self.do_rescale
   __lowercase				    =					rescale_factor if rescale_factor is not None else self.rescale_factor
   __lowercase				    =					do_normalize if do_normalize is not None else self.do_normalize
   __lowercase				    =					image_mean if image_mean is not None else self.image_mean
   __lowercase				    =					image_std if image_std is not None else self.image_std

   __lowercase				    =					make_list_of_images(lowercase__     )

   if not valid_images(lowercase__     ):
    raise ValueError(
        '''Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '''
        '''torch.Tensor, tf.Tensor or jax.ndarray.'''     )

   if do_resize and size is None or resample is None:
    raise ValueError('''Size and resample must be specified if do_resize is True.'''     )

   if do_rescale and rescale_factor is None:
    raise ValueError('''Rescale factor must be specified if do_rescale is True.'''     )

   if do_normalize and (image_mean is None or image_std is None):
    raise ValueError('''Image mean and std must be specified if do_normalize is True.'''     )

   # All transformations expect numpy arrays.
   __lowercase				    =					[to_numpy_array(lowercase__     ) for image in images]

   if do_resize:
    __lowercase				    =					[self.resize(image=lowercase__   ,size=lowercase__   ,resample=lowercase__     ) for image in images]

   if do_rescale:
    __lowercase				    =					[self.rescale(image=lowercase__   ,scale=lowercase__     ) for image in images]

   if do_normalize:
    __lowercase				    =					[self.normalize(image=lowercase__   ,mean=lowercase__   ,std=lowercase__     ) for image in images]

   __lowercase				    =					[to_channel_dimension_format(lowercase__   ,lowercase__     ) for image in images]

   __lowercase				    =					{'''pixel_values''': images}
   return BatchFeature(data=lowercase__   ,tensor_type=lowercase__     )






  def    SCREAMING_SNAKE_CASE ( self  :	str   ,lowercase__  :	Tuple   ,lowercase__  :	List[Tuple] = None     ):
   __lowercase				    =					outputs.logits

   # Resize logits and compute semantic segmentation maps
   if target_sizes is not None:
    if len(lowercase__     ) != len(lowercase__     ):
     raise ValueError(
         '''Make sure that you pass in as many target sizes as the batch dimension of the logits'''     )

    if is_torch_tensor(lowercase__     ):
     __lowercase				    =					target_sizes.numpy()

    __lowercase				    =					[]

    for idx in range(len(lowercase__     )     ):
     __lowercase				    =					torch.nn.functional.interpolate(
         logits[idx].unsqueeze(dim=0     )   ,size=target_sizes[idx]   ,mode='''bilinear'''   ,align_corners=lowercase__     )
     __lowercase				    =					resized_logits[0].argmax(dim=0     )
     semantic_segmentation.append(lowercase__     )
   else:
    __lowercase				    =					logits.argmax(dim=1     )
    __lowercase				    =					[semantic_segmentation[i] for i in range(semantic_segmentation.shape[0]     )]

   return semantic_segmentation




"""simple docstring"""

import itertools
import os
import random
import tempfile
import unittest

import numpy as np

from transformers import TvltFeatureExtractor, is_datasets_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
    import torch

if is_datasets_available():
    from datasets import load_dataset

_UpperCamelCase		     =    random.Random()



def       SCREAMING_SNAKE_CASE							( lowercase__					,					lowercase__=1.0					,					lowercase__=None					,					lowercase__=None   )			->						Optional[Any]:
 if rng is None:
  lowerCAmelCase__   :				Union[str, Any] 			=				global_rng

 lowerCAmelCase__   :				Dict 			=				[]
 for batch_idx in range(shape[0]   ):
  values.append([]   )
  for _ in range(shape[1]   ):
   values[-1].append(rng.random() * scale   )

 return values






class 			__a					(						unittest.TestCase       ):





       """simple docstring"""


       def __init__(   self			,					snake_case			,					snake_case=7			,					snake_case=400			,					snake_case=2_000			,					snake_case=2_048			,					snake_case=128			,					snake_case=1			,					snake_case=512			,					snake_case=30			,					snake_case=44_100			,					):





        """simple docstring"""

        lowerCAmelCase__   :				Optional[int] 			=				parent
        lowerCAmelCase__   :				Optional[int] 			=				batch_size
        lowerCAmelCase__   :				Optional[Any] 			=				min_seq_length
        lowerCAmelCase__   :				Optional[int] 			=				max_seq_length
        lowerCAmelCase__   :				Dict 			=				(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
        lowerCAmelCase__   :				Dict 			=				spectrogram_length
        lowerCAmelCase__   :				Any 			=				feature_size
        lowerCAmelCase__   :				int 			=				num_audio_channels
        lowerCAmelCase__   :				Optional[int] 			=				hop_length
        lowerCAmelCase__   :				List[str] 			=				chunk_length
        lowerCAmelCase__   :				Optional[Any] 			=				sampling_rate






       def 			SCREAMING_SNAKE_CASE_	(   self   ):





        """simple docstring"""

        return {
            "spectrogram_length": self.spectrogram_length,
            "feature_size": self.feature_size,
            "num_audio_channels": self.num_audio_channels,
            "hop_length": self.hop_length,
            "chunk_length": self.chunk_length,
            "sampling_rate": self.sampling_rate,
        }






       def 			SCREAMING_SNAKE_CASE_	(   self			,					snake_case=False			,					snake_case=False   ):





        """simple docstring"""

        def _flatten(snake_case   ):
         return list(itertools.chain(*snake_case   )   )

        if equal_length:
         lowerCAmelCase__   :				Dict 			=				[floats_list((self.max_seq_length, self.feature_size)   ) for _ in range(self.batch_size   )]
        else:
         # make sure that inputs increase in size
         lowerCAmelCase__   :				Tuple 			=				[
             floats_list((x, self.feature_size)   )
             for x in range(self.min_seq_length			,					self.max_seq_length			,					self.seq_length_diff   )
         ]
        if numpify:
         lowerCAmelCase__   :				Dict 			=				[np.asarray(snake_case   ) for x in speech_inputs]
        return speech_inputs






@require_torch
@require_torchaudio
class 			__a					(						__magic_name__       ,	unittest.TestCase       ):





       """simple docstring"""


       __UpperCamelCase					:      Optional[Any]    =						TvltFeatureExtractor






       def 			SCREAMING_SNAKE_CASE_	(   self   ):





        """simple docstring"""

        lowerCAmelCase__   :				Optional[Any] 			=				TvltFeatureExtractionTester(self   )






       def 			SCREAMING_SNAKE_CASE_	(   self   ):





        """simple docstring"""

        lowerCAmelCase__   :				List[str] 			=				self.feature_extraction_class(**self.feat_extract_dict   )
        self.assertTrue(hasattr(snake_case			,					"spectrogram_length"   )   )
        self.assertTrue(hasattr(snake_case			,					"feature_size"   )   )
        self.assertTrue(hasattr(snake_case			,					"num_audio_channels"   )   )
        self.assertTrue(hasattr(snake_case			,					"hop_length"   )   )
        self.assertTrue(hasattr(snake_case			,					"chunk_length"   )   )
        self.assertTrue(hasattr(snake_case			,					"sampling_rate"   )   )






       def 			SCREAMING_SNAKE_CASE_	(   self   ):





        """simple docstring"""

        lowerCAmelCase__   :				int 			=				self.feature_extraction_class(**self.feat_extract_dict   )

        with tempfile.TemporaryDirectory() as tmpdirname:
         lowerCAmelCase__   :				Union[str, Any] 			=				feat_extract_first.save_pretrained(snake_case   )[0]
         check_json_file_has_correct_format(snake_case   )
         lowerCAmelCase__   :				Tuple 			=				self.feature_extraction_class.from_pretrained(snake_case   )

        lowerCAmelCase__   :				Union[str, Any] 			=				feat_extract_first.to_dict()
        lowerCAmelCase__   :				int 			=				feat_extract_second.to_dict()
        lowerCAmelCase__   :				Union[str, Any] 			=				dict_first.pop("mel_filters"   )
        lowerCAmelCase__   :				List[Any] 			=				dict_second.pop("mel_filters"   )
        self.assertTrue(np.allclose(snake_case			,					snake_case   )   )
        self.assertEqual(snake_case			,					snake_case   )






       def 			SCREAMING_SNAKE_CASE_	(   self   ):





        """simple docstring"""

        lowerCAmelCase__   :				Any 			=				self.feature_extraction_class(**self.feat_extract_dict   )

        with tempfile.TemporaryDirectory() as tmpdirname:
         lowerCAmelCase__   :				List[Any] 			=				os.path.join(snake_case			,					"feat_extract.json"   )
         feat_extract_first.to_json_file(snake_case   )
         lowerCAmelCase__   :				List[Any] 			=				self.feature_extraction_class.from_json_file(snake_case   )

        lowerCAmelCase__   :				Optional[Any] 			=				feat_extract_first.to_dict()
        lowerCAmelCase__   :				Dict 			=				feat_extract_second.to_dict()
        lowerCAmelCase__   :				Tuple 			=				dict_first.pop("mel_filters"   )
        lowerCAmelCase__   :				Dict 			=				dict_second.pop("mel_filters"   )
        self.assertTrue(np.allclose(snake_case			,					snake_case   )   )
        self.assertEqual(snake_case			,					snake_case   )






       def 			SCREAMING_SNAKE_CASE_	(   self   ):





        """simple docstring"""

        lowerCAmelCase__   :				Optional[int] 			=				self.feature_extraction_class(**self.feat_extract_dict   )

        # create three inputs of length 800, 1000, and 1200
        lowerCAmelCase__   :				Union[str, Any] 			=				[floats_list((1, x)   )[0] for x in range(800			,					1_400			,					200   )]
        lowerCAmelCase__   :				Optional[int] 			=				[np.asarray(snake_case   ) for speech_input in speech_inputs]

        # Test not batched input
        lowerCAmelCase__   :				Tuple 			=				feature_extractor(np_speech_inputs[0]			,					return_tensors="np"			,					sampling_rate=44_100   ).audio_values

        self.assertTrue(encoded_audios.ndim == 4   )
        self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size   )
        self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length   )
        self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels   )

        # Test batched
        lowerCAmelCase__   :				Optional[int] 			=				feature_extractor(snake_case			,					return_tensors="np"			,					sampling_rate=44_100   ).audio_values

        self.assertTrue(encoded_audios.ndim == 4   )
        self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size   )
        self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length   )
        self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels   )

        # Test audio masking
        lowerCAmelCase__   :				List[str] 			=				feature_extractor(
            snake_case			,					return_tensors="np"			,					sampling_rate=44_100			,					mask_audio=snake_case   ).audio_values

        self.assertTrue(encoded_audios.ndim == 4   )
        self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size   )
        self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length   )
        self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels   )

        # Test 2-D numpy arrays are batched.
        lowerCAmelCase__   :				List[str] 			=				[floats_list((1, x)   )[0] for x in (800, 800, 800)]
        lowerCAmelCase__   :				Tuple 			=				np.asarray(snake_case   )
        lowerCAmelCase__   :				List[Any] 			=				feature_extractor(snake_case			,					return_tensors="np"			,					sampling_rate=44_100   ).audio_values
        self.assertTrue(encoded_audios.ndim == 4   )
        self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size   )
        self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length   )
        self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels   )






       def 			SCREAMING_SNAKE_CASE_	(   self			,					snake_case   ):





        """simple docstring"""

        lowerCAmelCase__   :				str 			=				load_dataset("hf-internal-testing/librispeech_asr_dummy"			,					"clean"			,					split="validation"   )
        # automatic decoding with librispeech
        lowerCAmelCase__   :				int 			=				ds.sort("id"   ).select(range(snake_case   )   )[:num_samples]["audio"]

        return [x["array"] for x in speech_samples]






       def 			SCREAMING_SNAKE_CASE_	(   self   ):





        """simple docstring"""

        lowerCAmelCase__   :				str 			=				self._load_datasamples(1   )
        lowerCAmelCase__   :				Optional[Any] 			=				TvltFeatureExtractor()
        lowerCAmelCase__   :				Dict 			=				feature_extractor(snake_case			,					return_tensors="pt"   ).audio_values

        self.assertEquals(audio_values.shape			,					(1, 1, 192, 128)   )

        lowerCAmelCase__   :				Dict 			=				torch.tensor([[-0.3_032, -0.2_708], [-0.4_434, -0.4_007]]   )
        self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2]			,					snake_case			,					atol=1e-4   )   )






