





import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np
import torch

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


_UpperCamelCase							=			logging.get_logger(__name__)




class 					_lowerCamelCase			(     UpperCamelCase__     ):



      """simple docstring"""






      UpperCAmelCase_    :   Any      =["input_features", "is_longer"]







      def __init__(			self					,    UpperCAmelCase=64					,    UpperCAmelCase=48000					,    UpperCAmelCase=480					,    UpperCAmelCase=10					,    UpperCAmelCase=1024					,    UpperCAmelCase=0.0					,    UpperCAmelCase=False					,    UpperCAmelCase = 0					,    UpperCAmelCase = 14000					,    UpperCAmelCase = None					,    UpperCAmelCase = "fusion"					,    UpperCAmelCase = "repeatpad"					,    **UpperCAmelCase					,    )							-> Dict:






        '''simple docstring'''






        super().__init__(
            feature_size=lowercase_					,    sampling_rate=lowercase_					,    padding_value=lowercase_					,    return_attention_mask=lowercase_					,    **lowercase_					,    )
        __snake_case :					List[str]          =							top_db
        __snake_case :					Tuple          =							truncation
        __snake_case :					Any          =							padding
        __snake_case :					List[str]          =							fft_window_size
        __snake_case :					Dict          =							(fft_window_size >> 1) + 1
        __snake_case :					Union[str, Any]          =							hop_length
        __snake_case :					Optional[Any]          =							max_length_s
        __snake_case :					Optional[Any]          =							max_length_s * sampling_rate
        __snake_case :					List[Any]          =							sampling_rate
        __snake_case :					List[Any]          =							frequency_min
        __snake_case :					int          =							frequency_max
        __snake_case :					Union[str, Any]          =							mel_filter_bank(
            num_frequency_bins=self.nb_frequency_bins					,    num_mel_filters=lowercase_					,    min_frequency=lowercase_					,    max_frequency=lowercase_					,    sampling_rate=lowercase_					,    norm=lowercase_					,    mel_scale="htk"					,    )
        __snake_case :					Any          =							mel_filter_bank(
            num_frequency_bins=self.nb_frequency_bins					,    num_mel_filters=lowercase_					,    min_frequency=lowercase_					,    max_frequency=lowercase_					,    sampling_rate=lowercase_					,    norm="slaney"					,    mel_scale="slaney"					,    )







      def     UpperCAmelCase		(			self )							-> Dict[str, Any]:






        '''simple docstring'''






        __snake_case :					Union[str, Any]          =							copy.deepcopy(self.__dict__ )
        __snake_case :					List[str]          =							self.__class__.__name__
        if "mel_filters" in output:
          del output["mel_filters"]
        if "mel_filters_slaney" in output:
          del output["mel_filters_slaney"]
        return output







      def     UpperCAmelCase		(			self					,    UpperCAmelCase					,    UpperCAmelCase = None )							-> np.ndarray:






        '''simple docstring'''






        __snake_case :					Tuple          =							spectrogram(
            lowercase_					,    window_function(self.fft_window_size					,    "hann" )					,    frame_length=self.fft_window_size					,    hop_length=self.hop_length					,    power=2.0					,    mel_filters=lowercase_					,    log_mel="dB"					,    )
        return log_mel_spectrogram.T







      def     UpperCAmelCase		(			self					,    UpperCAmelCase					,    UpperCAmelCase					,    UpperCAmelCase )							-> Any:






        '''simple docstring'''






        __snake_case :					List[Any]          =							np.array_split(list(range(0					,    total_frames - chunk_frames + 1 ) )					,    3 )
        if len(ranges[1] ) == 0:
          # if the audio is too short, we just use the first chunk
          __snake_case :					Dict          =							[0]
        if len(ranges[2] ) == 0:
          # if the audio is too short, we just use the first chunk
          __snake_case :					Optional[Any]          =							[0]
        # randomly choose index for each part
        __snake_case :					str          =							np.random.choice(ranges[0] )
        __snake_case :					Tuple          =							np.random.choice(ranges[1] )
        __snake_case :					Union[str, Any]          =							np.random.choice(ranges[2] )

        __snake_case :					List[str]          =							mel[idx_front : idx_front + chunk_frames, :]
        __snake_case :					Dict          =							mel[idx_middle : idx_middle + chunk_frames, :]
        __snake_case :					str          =							mel[idx_back : idx_back + chunk_frames, :]

        __snake_case :					Optional[Any]          =							torch.tensor(mel[None, None, :] )
        __snake_case :					str          =							torch.nn.functional.interpolate(
            lowercase_					,    size=[chunk_frames, 64]					,    mode="bilinear"					,    align_corners=lowercase_ )
        __snake_case :					str          =							mel_shrink[0][0].numpy()
        __snake_case :					List[str]          =							np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back]					,    axis=0 )
        return mel_fusion







      def     UpperCAmelCase		(			self					,    UpperCAmelCase					,    UpperCAmelCase					,    UpperCAmelCase					,    UpperCAmelCase )							-> np.array:






        '''simple docstring'''






        if waveform.shape[0] > max_length:
          if truncation == "rand_trunc":
            __snake_case :					Dict          =							True
            # random crop to max_length (for compatibility) -> this should be handled by self.pad
            __snake_case :					Dict          =							len(lowercase_ ) - max_length
            __snake_case :					str          =							np.random.randint(0					,    overflow + 1 )
            __snake_case :					Dict          =							waveform[idx : idx + max_length]
            __snake_case :					Dict          =							self._np_extract_fbank_features(lowercase_					,    self.mel_filters_slaney )[None, :]
          elif truncation == "fusion":
            __snake_case :					List[str]          =							self._np_extract_fbank_features(lowercase_					,    self.mel_filters )
            __snake_case :					Tuple          =							max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
            __snake_case :					Dict          =							mel.shape[0]
            if chunk_frames == total_frames:
              # there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
              # In this case, we just use the whole audio.
              __snake_case :					Optional[Any]          =							np.stack([mel, mel, mel, mel]					,    axis=0 )
              __snake_case :					str          =							False
            else:
              __snake_case :					List[Any]          =							self._random_mel_fusion(lowercase_					,    lowercase_					,    lowercase_ )
              __snake_case :					Union[str, Any]          =							True
          else:
            raise NotImplementedError(F"""data_truncating {truncation} not implemented""" )

        else:
          __snake_case :					List[Any]          =							False
          # only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
          if waveform.shape[0] < max_length:
            if padding == "repeat":
              __snake_case :					Optional[Any]          =							int(max_length / len(lowercase_ ) )
              __snake_case :					str          =							np.stack(np.tile(lowercase_					,    n_repeat + 1 ) )[:max_length]
            if padding == "repeatpad":
              __snake_case :					Dict          =							int(max_length / len(lowercase_ ) )
              __snake_case :					List[str]          =							np.stack(np.tile(lowercase_					,    lowercase_ ) )
            __snake_case :					Optional[Any]          =							np.pad(lowercase_					,    (0, max_length - waveform.shape[0])					,    mode="constant"					,    constant_values=0 )

          if truncation == "fusion":
            __snake_case :					Optional[Any]          =							self._np_extract_fbank_features(lowercase_					,    self.mel_filters )
            __snake_case :					List[Any]          =							np.stack([input_mel, input_mel, input_mel, input_mel]					,    axis=0 )
          else:
            __snake_case :					Union[str, Any]          =							self._np_extract_fbank_features(lowercase_					,    self.mel_filters_slaney )[None, :]

        return input_mel, longer







      def __call__(			self					,    UpperCAmelCase					,    UpperCAmelCase = None					,    UpperCAmelCase = None					,    UpperCAmelCase = None					,    UpperCAmelCase = None					,    UpperCAmelCase = None					,    **UpperCAmelCase					,    )							-> BatchFeature:






        '''simple docstring'''






        __snake_case :					Union[str, Any]          =							truncation if truncation is not None else self.truncation
        __snake_case :					Any          =							padding if padding else self.padding

        if sampling_rate is not None:
          if sampling_rate != self.sampling_rate:
            raise ValueError(
                F"""The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a"""
                F""" sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input"""
                F""" was sampled with {self.sampling_rate} and not {sampling_rate}.""" )
        else:
          logger.warning(
              "It is strongly recommended to pass the `sampling_rate` argument to this function. "
              "Failing to do so can result in silent errors that might be hard to debug." )

        __snake_case :					List[Any]          =							isinstance(lowercase_					,    np.ndarray ) and len(raw_speech.shape ) > 1
        if is_batched_numpy and len(raw_speech.shape ) > 2:
          raise ValueError(F"""Only mono-channel audio is supported for input to {self}""" )
        __snake_case :					str          =							is_batched_numpy or (
            isinstance(lowercase_					,    (list, tuple) ) and (isinstance(raw_speech[0]					,    (np.ndarray, tuple, list) ))
        )

        if is_batched:
          __snake_case :					Tuple          =							[np.asarray(lowercase_					,    dtype=np.floataa ) for speech in raw_speech]
        elif not is_batched and not isinstance(lowercase_					,    np.ndarray ):
          __snake_case :					Union[str, Any]          =							np.asarray(lowercase_					,    dtype=np.floataa )
        elif isinstance(lowercase_					,    np.ndarray ) and raw_speech.dtype is np.dtype(np.floataa ):
          __snake_case :					Any          =							raw_speech.astype(np.floataa )

        # always return batch
        if not is_batched:
          __snake_case :					Dict          =							[np.asarray(lowercase_ )]

        # convert to mel spectrogram, truncate and pad if needed.
        __snake_case :					int          =							[
            self._get_input_mel(lowercase_					,    max_length if max_length else self.nb_max_samples					,    lowercase_					,    lowercase_ )
            for waveform in raw_speech
        ]

        __snake_case :					Dict          =							[]
        __snake_case :					Optional[int]          =							[]
        for mel, longer in padded_inputs:
          input_mel.append(lowercase_ )
          is_longer.append(lowercase_ )

        if truncation == "fusion" and sum(lowercase_ ) == 0:
          # if no audio is longer than 10s, then randomly select one audio to be longer
          __snake_case :					List[Any]          =							np.random.randint(0					,    len(lowercase_ ) )
          __snake_case :					List[Any]          =							True

        if isinstance(input_mel[0]					,    lowercase_ ):
          __snake_case :					Any          =							[np.asarray(lowercase_					,    dtype=np.floataa ) for feature in input_mel]

        # is_longer is a list of bool
        __snake_case :					Any          =							[[longer] for longer in is_longer]

        __snake_case :					str          =							{"input_features": input_mel, "is_longer": is_longer}
        __snake_case :					Optional[Any]          =							BatchFeature(lowercase_ )

        if return_tensors is not None:
          __snake_case :					int          =							input_features.convert_to_tensors(lowercase_ )

        return input_features








'''simple docstring'''

import unittest

from transformers import MODEL_FOR_DOCUMENT_QUESTION_ANSWERING_MAPPING, AutoTokenizer, is_vision_available
from transformers.pipelines import pipeline
from transformers.pipelines.document_question_answering import apply_tesseract
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_detectrona,
    require_pytesseract,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
  from PIL import Image

  from transformers.image_utils import load_image
else:






  class   UpperCAmelCase      :
   @staticmethod
   def 	UpperCAmelCase_				(							*lowercase_ :Optional[Any]		,			**lowercase_ :List[Any]				)->				Optional[int]:
       pass





  def 		UpperCamelCase							(				_lowerCamelCase   :   List[Any]			):
      return None


# This is a pinned image from a specific revision of a document question answering space, hosted by HuggingFace,
# so we can expect it to be available.
__lowerCAmelCase				:      int						      =(
    "https://huggingface.co/spaces/impira/docquery/resolve/2f6c96314dc84dfda62d40de9da55f2f5165d403/invoice.png"
)






@is_pipeline_test
@require_torch
@require_vision
class   UpperCAmelCase      (			unittest.TestCase							):
 __lowercase  						=							MODEL_FOR_DOCUMENT_QUESTION_ANSWERING_MAPPING
 @require_pytesseract
 @require_vision
 def 	UpperCAmelCase_				(							self :int		,			lowercase_ :Optional[Any]		,			lowercase_ :Optional[Any]		,			lowercase_ :Dict				)->				int:
     A__							=     pipeline(
         "document-question-answering"		,			model=lowercase_		,			tokenizer=lowercase_		,			image_processor=lowercase_				)

     A__							=     INVOICE_URL
     A__							=     list(zip(*apply_tesseract(load_image(lowercase_				)		,			lowercase_		,			""				)				)				)
     A__							=     "What is the placebo?"
     A__							=     [
         {
             "image": load_image(lowercase_				),
             "question": question,
         },
         {
             "image": image,
             "question": question,
         },
         {
             "image": image,
             "question": question,
             "word_boxes": word_boxes,
         },
     ]
     return dqa_pipeline, examples
 def 	UpperCAmelCase_				(							self :str		,			lowercase_ :List[Any]		,			lowercase_ :List[str]				)->				List[str]:
     A__							=     dqa_pipeline(lowercase_		,			top_k=2				)
     self.assertEqual(
         lowercase_		,			[
             [
                 {"score": ANY(lowercase_				), "answer": ANY(lowercase_				), "start": ANY(lowercase_				), "end": ANY(lowercase_				)},
                 {"score": ANY(lowercase_				), "answer": ANY(lowercase_				), "start": ANY(lowercase_				), "end": ANY(lowercase_				)},
             ]
         ]
         * 3		,			)
 @require_torch
 @require_detectrona
 @require_pytesseract
 def 	UpperCAmelCase_				(							self :Dict				)->				Optional[Any]:
     A__							=     pipeline("document-question-answering"		,			model="hf-internal-testing/tiny-random-layoutlmv2"				)
     A__							=     INVOICE_URL
     A__							=     "How many cats are there?"

     A__							=     [
         {"score": 0.0_0_0_1, "answer": "oy 2312/2019", "start": 38, "end": 39},
         {"score": 0.0_0_0_1, "answer": "oy 2312/2019 DUE", "start": 38, "end": 40},
     ]
     A__							=     dqa_pipeline(image=lowercase_		,			question=lowercase_		,			top_k=2				)
     self.assertEqual(nested_simplify(lowercase_		,			decimals=4				)		,			lowercase_				)

     A__							=     dqa_pipeline({"image": image, "question": question}		,			top_k=2				)
     self.assertEqual(nested_simplify(lowercase_		,			decimals=4				)		,			lowercase_				)

     # This image does not detect ANY text in it, meaning layoutlmv2 should fail.
     # Empty answer probably
     A__							=     "./tests/fixtures/tests_samples/COCO/000000039769.png"
     A__							=     dqa_pipeline(image=lowercase_		,			question=lowercase_		,			top_k=2				)
     self.assertEqual(lowercase_		,			[]				)

     # We can optionnally pass directly the words and bounding boxes
     A__							=     "./tests/fixtures/tests_samples/COCO/000000039769.png"
     A__							=     []
     A__							=     []
     A__							=     dqa_pipeline(image=lowercase_		,			question=lowercase_		,			words=lowercase_		,			boxes=lowercase_		,			top_k=2				)
     self.assertEqual(lowercase_		,			[]				)
 @slow
 @require_torch
 @require_detectrona
 @require_pytesseract
 def 	UpperCAmelCase_				(							self :Optional[Any]				)->				Tuple:
     A__							=     pipeline(
         "document-question-answering"		,			model="tiennvcs/layoutlmv2-base-uncased-finetuned-docvqa"		,			revision="9977165"		,			)
     A__							=     INVOICE_URL
     A__							=     "What is the invoice number?"

     A__							=     dqa_pipeline(image=lowercase_		,			question=lowercase_		,			top_k=2				)
     self.assertEqual(
         nested_simplify(lowercase_		,			decimals=4				)		,			[
             {"score": 0.9_9_4_4, "answer": "us-001", "start": 16, "end": 16},
             {"score": 0.0_0_0_9, "answer": "us-001", "start": 16, "end": 16},
         ]		,			)

     A__							=     dqa_pipeline({"image": image, "question": question}		,			top_k=2				)
     self.assertEqual(
         nested_simplify(lowercase_		,			decimals=4				)		,			[
             {"score": 0.9_9_4_4, "answer": "us-001", "start": 16, "end": 16},
             {"score": 0.0_0_0_9, "answer": "us-001", "start": 16, "end": 16},
         ]		,			)

     A__							=     dqa_pipeline(
         [{"image": image, "question": question}, {"image": image, "question": question}]		,			top_k=2				)
     self.assertEqual(
         nested_simplify(lowercase_		,			decimals=4				)		,			[
             [
                 {"score": 0.9_9_4_4, "answer": "us-001", "start": 16, "end": 16},
                 {"score": 0.0_0_0_9, "answer": "us-001", "start": 16, "end": 16},
             ],
         ]
         * 2		,			)
 @slow
 @require_torch
 @require_detectrona
 @require_pytesseract
 def 	UpperCAmelCase_				(							self :Tuple				)->				Optional[Any]:
     A__							=     pipeline(
         "document-question-answering"		,			model="tiennvcs/layoutlmv2-base-uncased-finetuned-docvqa"		,			revision="9977165"		,			max_seq_len=50		,			)
     A__							=     INVOICE_URL
     A__							=     "What is the invoice number?"

     A__							=     dqa_pipeline(image=lowercase_		,			question=lowercase_		,			top_k=2				)
     self.assertEqual(
         nested_simplify(lowercase_		,			decimals=4				)		,			[
             {"score": 0.9_9_7_4, "answer": "1110212019", "start": 23, "end": 23},
             {"score": 0.9_9_4_8, "answer": "us-001", "start": 16, "end": 16},
         ]		,			)

     A__							=     dqa_pipeline({"image": image, "question": question}		,			top_k=2				)
     self.assertEqual(
         nested_simplify(lowercase_		,			decimals=4				)		,			[
             {"score": 0.9_9_7_4, "answer": "1110212019", "start": 23, "end": 23},
             {"score": 0.9_9_4_8, "answer": "us-001", "start": 16, "end": 16},
         ]		,			)

     A__							=     dqa_pipeline(
         [{"image": image, "question": question}, {"image": image, "question": question}]		,			top_k=2				)
     self.assertEqual(
         nested_simplify(lowercase_		,			decimals=4				)		,			[
             [
                 {"score": 0.9_9_7_4, "answer": "1110212019", "start": 23, "end": 23},
                 {"score": 0.9_9_4_8, "answer": "us-001", "start": 16, "end": 16},
             ]
         ]
         * 2		,			)
 @slow
 @require_torch
 @require_pytesseract
 @require_vision
 def 	UpperCAmelCase_				(							self :List[Any]				)->				Any:
     A__							=     AutoTokenizer.from_pretrained(
         "impira/layoutlm-document-qa"		,			revision="3dc6de3"		,			add_prefix_space=lowercase_				)
     A__							=     pipeline(
         "document-question-answering"		,			model="impira/layoutlm-document-qa"		,			tokenizer=lowercase_		,			revision="3dc6de3"		,			)
     A__							=     INVOICE_URL
     A__							=     "What is the invoice number?"

     A__							=     dqa_pipeline(image=lowercase_		,			question=lowercase_		,			top_k=2				)
     self.assertEqual(
         nested_simplify(lowercase_		,			decimals=4				)		,			[
             {"score": 0.4_2_5_1, "answer": "us-001", "start": 16, "end": 16},
             {"score": 0.0_8_1_9, "answer": "1110212019", "start": 23, "end": 23},
         ]		,			)

     A__							=     dqa_pipeline({"image": image, "question": question}		,			top_k=2				)
     self.assertEqual(
         nested_simplify(lowercase_		,			decimals=4				)		,			[
             {"score": 0.4_2_5_1, "answer": "us-001", "start": 16, "end": 16},
             {"score": 0.0_8_1_9, "answer": "1110212019", "start": 23, "end": 23},
         ]		,			)

     A__							=     dqa_pipeline(
         [{"image": image, "question": question}, {"image": image, "question": question}]		,			top_k=2				)
     self.assertEqual(
         nested_simplify(lowercase_		,			decimals=4				)		,			[
             [
                 {"score": 0.4_2_5_1, "answer": "us-001", "start": 16, "end": 16},
                 {"score": 0.0_8_1_9, "answer": "1110212019", "start": 23, "end": 23},
             ]
         ]
         * 2		,			)

     A__							=     list(zip(*apply_tesseract(load_image(lowercase_				)		,			lowercase_		,			""				)				)				)

     # This model should also work if `image` is set to None
     A__							=     dqa_pipeline({"image": None, "word_boxes": word_boxes, "question": question}		,			top_k=2				)
     self.assertEqual(
         nested_simplify(lowercase_		,			decimals=4				)		,			[
             {"score": 0.4_2_5_1, "answer": "us-001", "start": 16, "end": 16},
             {"score": 0.0_8_1_9, "answer": "1110212019", "start": 23, "end": 23},
         ]		,			)
 @slow
 @require_torch
 @require_pytesseract
 @require_vision
 def 	UpperCAmelCase_				(							self :Optional[int]				)->				Union[str, Any]:
     A__							=     AutoTokenizer.from_pretrained(
         "impira/layoutlm-document-qa"		,			revision="3dc6de3"		,			add_prefix_space=lowercase_				)
     A__							=     pipeline(
         "document-question-answering"		,			model="impira/layoutlm-document-qa"		,			tokenizer=lowercase_		,			revision="3dc6de3"		,			max_seq_len=50		,			)
     A__							=     INVOICE_URL
     A__							=     "What is the invoice number?"

     A__							=     dqa_pipeline(image=lowercase_		,			question=lowercase_		,			top_k=2				)
     self.assertEqual(
         nested_simplify(lowercase_		,			decimals=4				)		,			[
             {"score": 0.9_9_9_9, "answer": "us-001", "start": 16, "end": 16},
             {"score": 0.9_9_9_8, "answer": "us-001", "start": 16, "end": 16},
         ]		,			)

     A__							=     dqa_pipeline(
         [{"image": image, "question": question}, {"image": image, "question": question}]		,			top_k=2				)
     self.assertEqual(
         nested_simplify(lowercase_		,			decimals=4				)		,			[
             [
                 {"score": 0.9_9_9_9, "answer": "us-001", "start": 16, "end": 16},
                 {"score": 0.9_9_9_8, "answer": "us-001", "start": 16, "end": 16},
             ]
         ]
         * 2		,			)

     A__							=     list(zip(*apply_tesseract(load_image(lowercase_				)		,			lowercase_		,			""				)				)				)

     # This model should also work if `image` is set to None
     A__							=     dqa_pipeline({"image": None, "word_boxes": word_boxes, "question": question}		,			top_k=2				)
     self.assertEqual(
         nested_simplify(lowercase_		,			decimals=4				)		,			[
             {"score": 0.9_9_9_9, "answer": "us-001", "start": 16, "end": 16},
             {"score": 0.9_9_9_8, "answer": "us-001", "start": 16, "end": 16},
         ]		,			)
 @slow
 @require_torch
 def 	UpperCAmelCase_				(							self :Optional[int]				)->				Tuple:
     A__							=     pipeline(
         "document-question-answering"		,			model="naver-clova-ix/donut-base-finetuned-docvqa"		,			tokenizer=AutoTokenizer.from_pretrained("naver-clova-ix/donut-base-finetuned-docvqa"				)		,			feature_extractor="naver-clova-ix/donut-base-finetuned-docvqa"		,			)

     A__							=     INVOICE_URL
     A__							=     "What is the invoice number?"
     A__							=     dqa_pipeline(image=lowercase_		,			question=lowercase_		,			top_k=2				)
     self.assertEqual(nested_simplify(lowercase_		,			decimals=4				)		,			[{"answer": "us-001"}]				)



 @require_tf
 @unittest.skip("Document question answering not implemented in TF"				)
 def 	UpperCAmelCase_				(							self :Optional[Any]				)->				List[str]:
     pass



