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

lowercase_       :						Union[str, Any]     					=	random.Random()





def    _lowerCAmelCase    (     lowerCamelCase__							:							Optional[Any],						lowerCamelCase__							:							Optional[Any]=1.0,						lowerCamelCase__							:							List[str]=None,						lowerCamelCase__							:							List[Any]=None				)  ->    Dict:
  if rng is None:
    _SCREAMING_SNAKE_CASE   :				Any							    = global_rng

  _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = []
  for batch_idx in range(shape[0]				):
    values.append([]				)
    for _ in range(shape[1]				):
      values[-1].append(rng.random() * scale				)

  return values
class 			UpperCamelCase      (							unittest.TestCase  ):
   def __init__(     self  ,     snake_case__  ,     snake_case__=7  ,     snake_case__=400  ,     snake_case__=2000  ,     snake_case__=2048  ,     snake_case__=128  ,     snake_case__=1  ,     snake_case__=512  ,     snake_case__=30  ,     snake_case__=44100  ,     ):




     """simple docstring"""


     _SCREAMING_SNAKE_CASE   :				List[Any]							    = parent
     _SCREAMING_SNAKE_CASE   :				str							    = batch_size
     _SCREAMING_SNAKE_CASE   :				Dict							    = min_seq_length
     _SCREAMING_SNAKE_CASE   :				Any							    = max_seq_length
     _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
     _SCREAMING_SNAKE_CASE   :				Tuple							    = spectrogram_length
     _SCREAMING_SNAKE_CASE   :				List[Any]							    = feature_size
     _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = num_audio_channels
     _SCREAMING_SNAKE_CASE   :				str							    = hop_length
     _SCREAMING_SNAKE_CASE   :				Tuple							    = chunk_length
     _SCREAMING_SNAKE_CASE   :				int							    = 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:
       _SCREAMING_SNAKE_CASE   :				Dict							    = [floats_list((self.max_seq_length, self.feature_size)				) for _ in range(self.batch_size				)]
     else:
       # make sure that inputs increase in size
       _SCREAMING_SNAKE_CASE   :				str							    = [
           floats_list((x, self.feature_size)				)
           for x in range(self.min_seq_length  ,     self.max_seq_length  ,     self.seq_length_diff				)
       ]
     if numpify:
       _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = [np.asarray(snake_case__				) for x in speech_inputs]
     return speech_inputs
@require_torch
@require_torchaudio
class 			UpperCamelCase      (							__SCREAMING_SNAKE_CASE      ,       unittest.TestCase  ):
   A__      =			TvltFeatureExtractor
   def 				__SCREAMING_SNAKE_CASE							(     self				):




     """simple docstring"""


     _SCREAMING_SNAKE_CASE   :				str							    = TvltFeatureExtractionTester(self				)
   def 				__SCREAMING_SNAKE_CASE							(     self				):




     """simple docstring"""


     _SCREAMING_SNAKE_CASE   :				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"""


     _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = self.feature_extraction_class(**self.feat_extract_dict				)

     with tempfile.TemporaryDirectory() as tmpdirname:
       _SCREAMING_SNAKE_CASE   :				Dict							    = feat_extract_first.save_pretrained(snake_case__				)[0]
       check_json_file_has_correct_format(snake_case__				)
       _SCREAMING_SNAKE_CASE   :				Tuple							    = self.feature_extraction_class.from_pretrained(snake_case__				)

     _SCREAMING_SNAKE_CASE   :				Dict							    = feat_extract_first.to_dict()
     _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = feat_extract_second.to_dict()
     _SCREAMING_SNAKE_CASE   :				Optional[int]							    = dict_first.pop("mel_filters"				)
     _SCREAMING_SNAKE_CASE   :				int							    = 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"""


     _SCREAMING_SNAKE_CASE   :				List[str]							    = self.feature_extraction_class(**self.feat_extract_dict				)

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

     _SCREAMING_SNAKE_CASE   :				Union[str, Any]							    = feat_extract_first.to_dict()
     _SCREAMING_SNAKE_CASE   :				Dict							    = feat_extract_second.to_dict()
     _SCREAMING_SNAKE_CASE   :				int							    = dict_first.pop("mel_filters"				)
     _SCREAMING_SNAKE_CASE   :				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"""


     _SCREAMING_SNAKE_CASE   :				List[str]							    = self.feature_extraction_class(**self.feat_extract_dict				)

     # create three inputs of length 800, 1000, and 1200
     _SCREAMING_SNAKE_CASE   :				Dict							    = [floats_list((1, x)				)[0] for x in range(800  ,     1400  ,     200				)]
     _SCREAMING_SNAKE_CASE   :				Optional[Any]							    = [np.asarray(snake_case__				) for speech_input in speech_inputs]

     # Test not batched input
     _SCREAMING_SNAKE_CASE   :				int							    = feature_extractor(np_speech_inputs[0]  ,     return_tensors="np"  ,     sampling_rate=44100				).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
     _SCREAMING_SNAKE_CASE   :				Optional[int]							    = feature_extractor(snake_case__  ,     return_tensors="np"  ,     sampling_rate=44100				).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
     _SCREAMING_SNAKE_CASE   :				Any							    = feature_extractor(
         snake_case__  ,     return_tensors="np"  ,     sampling_rate=44100  ,     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.
     _SCREAMING_SNAKE_CASE   :				Dict							    = [floats_list((1, x)				)[0] for x in (800, 800, 800)]
     _SCREAMING_SNAKE_CASE   :				List[Any]							    = np.asarray(snake_case__				)
     _SCREAMING_SNAKE_CASE   :				Dict							    = feature_extractor(snake_case__  ,     return_tensors="np"  ,     sampling_rate=44100				).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"""


     _SCREAMING_SNAKE_CASE   :				List[Any]							    = load_dataset("hf-internal-testing/librispeech_asr_dummy"  ,     "clean"  ,     split="validation"				)
     # automatic decoding with librispeech
     _SCREAMING_SNAKE_CASE   :				Tuple							    = 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"""


     _SCREAMING_SNAKE_CASE   :				Optional[int]							    = self._load_datasamples(1				)
     _SCREAMING_SNAKE_CASE   :				int							    = TvltFeatureExtractor()
     _SCREAMING_SNAKE_CASE   :				Optional[int]							    = feature_extractor(snake_case__  ,     return_tensors="pt"				).audio_values

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

     _SCREAMING_SNAKE_CASE   :				Optional[int]							    = 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				)				)





"""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"]



