
"""simple docstring"""
import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np

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


lowercase_       :						Tuple     					=	logging.get_logger(__name__)
class 			UpperCamelCase      (							__SCREAMING_SNAKE_CASE  ):
     A__      =			["""input_features"""]
     def __init__(     self  ,     snake_case__=80  ,     snake_case__=16000  ,     snake_case__=160  ,     snake_case__=30  ,     snake_case__=400  ,     snake_case__=0.0  ,     snake_case__=False  ,     **snake_case__  ,     ):




       """simple docstring"""


       super().__init__(
           feature_size=snake_case__  ,     sampling_rate=snake_case__  ,     padding_value=snake_case__  ,     return_attention_mask=snake_case__  ,     **snake_case__  ,     )
       _SCREAMING_SNAKE_CASE   :				str							    = n_fft
       _SCREAMING_SNAKE_CASE   :				int							    = hop_length
       _SCREAMING_SNAKE_CASE   :				Tuple							    = chunk_length
       _SCREAMING_SNAKE_CASE   :				Optional[int]							    = chunk_length * sampling_rate
       _SCREAMING_SNAKE_CASE   :				Tuple							    = self.n_samples // hop_length
       _SCREAMING_SNAKE_CASE   :				Any							    = sampling_rate
       _SCREAMING_SNAKE_CASE   :				int							    = mel_filter_bank(
           num_frequency_bins=1 + n_fft // 2  ,     num_mel_filters=snake_case__  ,     min_frequency=0.0  ,     max_frequency=8_000.0  ,     sampling_rate=snake_case__  ,     norm="slaney"  ,     mel_scale="slaney"  ,     )
     def 				__SCREAMING_SNAKE_CASE							(     self  ,     snake_case__				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				Optional[int]							    = spectrogram(
           snake_case__  ,     window_function(self.n_fft  ,     "hann"				)  ,     frame_length=self.n_fft  ,     hop_length=self.hop_length  ,     power=2.0  ,     mel_filters=self.mel_filters  ,     log_mel="log10"  ,     )
       _SCREAMING_SNAKE_CASE   :				str							    = log_spec[:, :-1]
       _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = np.maximum(snake_case__  ,     log_spec.max() - 8.0				)
       _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = (log_spec + 4.0) / 4.0
       return log_spec
     @staticmethod
     # Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
     def 				__SCREAMING_SNAKE_CASE							(     snake_case__  ,     snake_case__  ,     snake_case__ = 0.0				):




       """simple docstring"""


       if attention_mask is not None:
         _SCREAMING_SNAKE_CASE   :				int							    = np.array(snake_case__  ,     np.intaa				)
         _SCREAMING_SNAKE_CASE   :				str							    = []

         for vector, length in zip(snake_case__  ,     attention_mask.sum(-1				)				):
           _SCREAMING_SNAKE_CASE   :				Optional[int]							    = (vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7				)
           if length < normed_slice.shape[0]:
             _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = padding_value

           normed_input_values.append(snake_case__				)
       else:
         _SCREAMING_SNAKE_CASE   :				Optional[int]							    = [(x - x.mean()) / np.sqrt(x.var() + 1E-7				) for x in input_values]

       return normed_input_values
     def __call__(     self  ,     snake_case__  ,     snake_case__ = True  ,     snake_case__ = None  ,     snake_case__ = None  ,     snake_case__ = None  ,     snake_case__ = "max_length"  ,     snake_case__ = None  ,     snake_case__ = None  ,     snake_case__ = None  ,     **snake_case__  ,     ):




       """simple docstring"""


       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."				)

       _SCREAMING_SNAKE_CASE   :				int							    = isinstance(snake_case__  ,     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}'''				)
       _SCREAMING_SNAKE_CASE   :				List[Any]							    = is_batched_numpy or (
           isinstance(snake_case__  ,     (list, tuple)				) and (isinstance(raw_speech[0]  ,     (np.ndarray, tuple, list)				))
       )

       if is_batched:
         _SCREAMING_SNAKE_CASE   :				Any							    = [np.asarray([speech]  ,     dtype=np.floataa				).T for speech in raw_speech]
       elif not is_batched and not isinstance(snake_case__  ,     np.ndarray				):
         _SCREAMING_SNAKE_CASE   :				Dict							    = np.asarray(snake_case__  ,     dtype=np.floataa				)
       elif isinstance(snake_case__  ,     np.ndarray				) and raw_speech.dtype is np.dtype(np.floataa				):
         _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = raw_speech.astype(np.floataa				)

       # always return batch
       if not is_batched:
         _SCREAMING_SNAKE_CASE   :				Any							    = [np.asarray([raw_speech]				).T]

       _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = BatchFeature({"input_features": raw_speech}				)

       # convert into correct format for padding

       _SCREAMING_SNAKE_CASE   :				List[str]							    = self.pad(
           snake_case__  ,     padding=snake_case__  ,     max_length=max_length if max_length else self.n_samples  ,     truncation=snake_case__  ,     pad_to_multiple_of=snake_case__  ,     return_attention_mask=return_attention_mask or do_normalize  ,     )

       # zero-mean and unit-variance normalization
       if do_normalize:
         _SCREAMING_SNAKE_CASE   :				Tuple							    = self.zero_mean_unit_var_norm(
             padded_inputs["input_features"]  ,     attention_mask=padded_inputs["attention_mask"]  ,     padding_value=self.padding_value  ,     )
         _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = np.stack(padded_inputs["input_features"]  ,     axis=0				)

       # make sure list is in array format
       _SCREAMING_SNAKE_CASE   :				str							    = padded_inputs.get("input_features"				).transpose(2  ,     0  ,     1				)

       _SCREAMING_SNAKE_CASE   :				List[Any]							    = [self._np_extract_fbank_features(snake_case__				) for waveform in input_features[0]]

       if isinstance(input_features[0]  ,     snake_case__				):
         _SCREAMING_SNAKE_CASE   :				Any							    = [np.asarray(snake_case__  ,     dtype=np.floataa				) for feature in input_features]
       else:
         _SCREAMING_SNAKE_CASE   :				int							    = input_features

       if return_attention_mask:
         # rescale from sample (48000) to feature (3000)
         _SCREAMING_SNAKE_CASE   :				Dict							    = padded_inputs["attention_mask"][:, :: self.hop_length]

       if return_tensors is not None:
         _SCREAMING_SNAKE_CASE   :				Optional[int]							    = padded_inputs.convert_to_tensors(snake_case__				)

       return padded_inputs




     def 				__SCREAMING_SNAKE_CASE							(     self				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				int							    = copy.deepcopy(self.__dict__				)
       _SCREAMING_SNAKE_CASE   :				int							    = self.__class__.__name__
       if "mel_filters" in output:
         del output["mel_filters"]
       return output





"""simple docstring"""
# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import re

from ..models.auto import AutoProcessor
from ..models.vision_encoder_decoder import VisionEncoderDecoderModel
from ..utils import is_vision_available
from .base import PipelineTool


if is_vision_available():
      from PIL import Image
class 			UpperCamelCase      (							__SCREAMING_SNAKE_CASE  ):
     A__      =			"""naver-clova-ix/donut-base-finetuned-docvqa"""
     A__      =			(
         """This is a tool that answers a question about an document (pdf). It takes an input named `document` which """
         """should be the document containing the information, as well as a `question` that is the question about the """
         """document. It returns a text that contains the answer to the question."""
     )
     A__      =			"""document_qa"""
     A__      =			AutoProcessor
     A__      =			VisionEncoderDecoderModel

     A__      =			["""image""", """text"""]
     A__      =			["""text"""]
     def __init__(     self  ,     *snake_case__  ,     **snake_case__				):




       """simple docstring"""


       if not is_vision_available():
         raise ValueError("Pillow must be installed to use the DocumentQuestionAnsweringTool."				)

       super().__init__(*snake_case__  ,     **snake_case__				)
     def 				__SCREAMING_SNAKE_CASE							(     self  ,     snake_case__  ,     snake_case__				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				List[str]							    = "<s_docvqa><s_question>{user_input}</s_question><s_answer>"
       _SCREAMING_SNAKE_CASE   :				Optional[int]							    = task_prompt.replace("{user_input}"  ,     snake_case__				)
       _SCREAMING_SNAKE_CASE   :				Tuple							    = self.pre_processor.tokenizer(
           snake_case__  ,     add_special_tokens=snake_case__  ,     return_tensors="pt"				).input_ids
       _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = self.pre_processor(snake_case__  ,     return_tensors="pt"				).pixel_values

       return {"decoder_input_ids": decoder_input_ids, "pixel_values": pixel_values}
     def 				__SCREAMING_SNAKE_CASE							(     self  ,     snake_case__				):




       """simple docstring"""


       return self.model.generate(
           inputs["pixel_values"].to(self.device				)  ,     decoder_input_ids=inputs["decoder_input_ids"].to(self.device				)  ,     max_length=self.model.decoder.config.max_position_embeddings  ,     early_stopping=snake_case__  ,     pad_token_id=self.pre_processor.tokenizer.pad_token_id  ,     eos_token_id=self.pre_processor.tokenizer.eos_token_id  ,     use_cache=snake_case__  ,     num_beams=1  ,     bad_words_ids=[[self.pre_processor.tokenizer.unk_token_id]]  ,     return_dict_in_generate=snake_case__  ,     ).sequences




     def 				__SCREAMING_SNAKE_CASE							(     self  ,     snake_case__				):




       """simple docstring"""


       _SCREAMING_SNAKE_CASE   :				int							    = self.pre_processor.batch_decode(snake_case__				)[0]
       _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = sequence.replace(self.pre_processor.tokenizer.eos_token  ,     ""				)
       _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = sequence.replace(self.pre_processor.tokenizer.pad_token  ,     ""				)
       _SCREAMING_SNAKE_CASE   :				Dict							    = re.sub(r"<.*?>"  ,     ""  ,     snake_case__  ,     count=1				).strip()  # remove first task start token
       _SCREAMING_SNAKE_CASE   :				Dict							    = self.pre_processor.tokenajson(snake_case__				)

       return sequence["answer"]



