


import shutil
import tempfile
import unittest

import pytest

from transformers import WhisperTokenizer, is_speech_available
from transformers.testing_utils import require_sentencepiece, require_torch, require_torchaudio

from .test_feature_extraction_whisper import floats_list


if is_speech_available():
		from transformers import WhisperFeatureExtractor, WhisperProcessor


lowerCamelCase		=	5_0358
lowerCamelCase		=	5_0362





@require_torch
@require_torchaudio
@require_sentencepiece
class       _a							(       unittest.TestCase):




	def 							UpperCAmelCase__(      self		:	int	)-> Dict:
			lowerCAmelCase__     :    Optional[Any]					  =      '''openai/whisper-small.en'''
			lowerCAmelCase__     :    List[Any]					  =      tempfile.mkdtemp()




	def 							UpperCAmelCase__(      self		:	List[Any]	,      **_SCREAMING_SNAKE_CASE		:	Tuple	)-> Optional[Any]:
			return WhisperTokenizer.from_pretrained(self.checkpoint	,      **_SCREAMING_SNAKE_CASE	)




	def 							UpperCAmelCase__(      self		:	List[str]	,      **_SCREAMING_SNAKE_CASE		:	Any	)-> Optional[int]:
			return WhisperFeatureExtractor.from_pretrained(self.checkpoint	,      **_SCREAMING_SNAKE_CASE	)




	def 							UpperCAmelCase__(      self		:	Any	)-> str:
			shutil.rmtree(self.tmpdirname	)




	def 							UpperCAmelCase__(      self		:	int	)-> List[str]:
			lowerCAmelCase__     :    Union[str, Any]					  =      self.get_tokenizer()
			lowerCAmelCase__     :    Tuple					  =      self.get_feature_extractor()

			lowerCAmelCase__     :    Optional[Any]					  =      WhisperProcessor(tokenizer=_SCREAMING_SNAKE_CASE	,      feature_extractor=_SCREAMING_SNAKE_CASE	)

			processor.save_pretrained(self.tmpdirname	)
			lowerCAmelCase__     :    Union[str, Any]					  =      WhisperProcessor.from_pretrained(self.tmpdirname	)

			self.assertEqual(processor.tokenizer.get_vocab()	,      tokenizer.get_vocab()	)
			self.assertIsInstance(processor.tokenizer	,      _SCREAMING_SNAKE_CASE	)

			self.assertEqual(processor.feature_extractor.to_json_string()	,      feature_extractor.to_json_string()	)
			self.assertIsInstance(processor.feature_extractor	,      _SCREAMING_SNAKE_CASE	)




	def 							UpperCAmelCase__(      self		:	Union[str, Any]	)-> Dict:
			lowerCAmelCase__     :    List[Any]					  =      WhisperProcessor(tokenizer=self.get_tokenizer()	,      feature_extractor=self.get_feature_extractor()	)
			processor.save_pretrained(self.tmpdirname	)

			lowerCAmelCase__     :    int					  =      self.get_tokenizer(bos_token='''(BOS)'''	,      eos_token='''(EOS)'''	)
			lowerCAmelCase__     :    Dict					  =      self.get_feature_extractor(do_normalize=_SCREAMING_SNAKE_CASE	,      padding_value=1.0	)

			lowerCAmelCase__     :    int					  =      WhisperProcessor.from_pretrained(
			    self.tmpdirname	,      bos_token='''(BOS)'''	,      eos_token='''(EOS)'''	,      do_normalize=_SCREAMING_SNAKE_CASE	,      padding_value=1.0	)

			self.assertEqual(processor.tokenizer.get_vocab()	,      tokenizer_add_kwargs.get_vocab()	)
			self.assertIsInstance(processor.tokenizer	,      _SCREAMING_SNAKE_CASE	)

			self.assertEqual(processor.feature_extractor.to_json_string()	,      feature_extractor_add_kwargs.to_json_string()	)
			self.assertIsInstance(processor.feature_extractor	,      _SCREAMING_SNAKE_CASE	)




	def 							UpperCAmelCase__(      self		:	Dict	)-> Dict:
			lowerCAmelCase__     :    List[Any]					  =      self.get_feature_extractor()
			lowerCAmelCase__     :    Tuple					  =      self.get_tokenizer()

			lowerCAmelCase__     :    List[Any]					  =      WhisperProcessor(tokenizer=_SCREAMING_SNAKE_CASE	,      feature_extractor=_SCREAMING_SNAKE_CASE	)

			lowerCAmelCase__     :    List[str]					  =      floats_list((3, 1000)	)

			lowerCAmelCase__     :    List[str]					  =      feature_extractor(_SCREAMING_SNAKE_CASE	,      return_tensors='''np'''	)
			lowerCAmelCase__     :    Tuple					  =      processor(_SCREAMING_SNAKE_CASE	,      return_tensors='''np'''	)

			for key in input_feat_extract.keys():
					self.assertAlmostEqual(input_feat_extract[key].sum()	,      input_processor[key].sum()	,      delta=1E-2	)




	def 							UpperCAmelCase__(      self		:	Union[str, Any]	)-> int:
			lowerCAmelCase__     :    Tuple					  =      self.get_feature_extractor()
			lowerCAmelCase__     :    str					  =      self.get_tokenizer()

			lowerCAmelCase__     :    List[str]					  =      WhisperProcessor(tokenizer=_SCREAMING_SNAKE_CASE	,      feature_extractor=_SCREAMING_SNAKE_CASE	)

			lowerCAmelCase__     :    str					  =      '''This is a test string'''

			lowerCAmelCase__     :    List[Any]					  =      processor(text=_SCREAMING_SNAKE_CASE	)

			lowerCAmelCase__     :    Optional[int]					  =      tokenizer(_SCREAMING_SNAKE_CASE	)

			for key in encoded_tok.keys():
					self.assertListEqual(encoded_tok[key]	,      encoded_processor[key]	)




	def 							UpperCAmelCase__(      self		:	List[Any]	)-> Tuple:
			lowerCAmelCase__     :    Optional[Any]					  =      self.get_feature_extractor()
			lowerCAmelCase__     :    str					  =      self.get_tokenizer()

			lowerCAmelCase__     :    List[str]					  =      WhisperProcessor(tokenizer=_SCREAMING_SNAKE_CASE	,      feature_extractor=_SCREAMING_SNAKE_CASE	)

			lowerCAmelCase__     :    Dict					  =      [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

			lowerCAmelCase__     :    Union[str, Any]					  =      processor.batch_decode(_SCREAMING_SNAKE_CASE	)
			lowerCAmelCase__     :    Optional[int]					  =      tokenizer.batch_decode(_SCREAMING_SNAKE_CASE	)

			self.assertListEqual(_SCREAMING_SNAKE_CASE	,      _SCREAMING_SNAKE_CASE	)




	def 							UpperCAmelCase__(      self		:	Union[str, Any]	)-> Optional[int]:
			lowerCAmelCase__     :    List[str]					  =      self.get_feature_extractor()
			lowerCAmelCase__     :    Any					  =      self.get_tokenizer()

			lowerCAmelCase__     :    Optional[Any]					  =      WhisperProcessor(tokenizer=_SCREAMING_SNAKE_CASE	,      feature_extractor=_SCREAMING_SNAKE_CASE	)

			self.assertListEqual(
			    processor.model_input_names	,      feature_extractor.model_input_names	,      msg='''`processor` and `feature_extractor` model input names do not match'''	,      )




	def 							UpperCAmelCase__(      self		:	Any	)-> Union[str, Any]:
			lowerCAmelCase__     :    int					  =      self.get_feature_extractor()
			lowerCAmelCase__     :    int					  =      self.get_tokenizer()

			lowerCAmelCase__     :    int					  =      WhisperProcessor(tokenizer=_SCREAMING_SNAKE_CASE	,      feature_extractor=_SCREAMING_SNAKE_CASE	)
			lowerCAmelCase__     :    List[str]					  =      processor.get_decoder_prompt_ids(task='''transcribe'''	,      no_timestamps=_SCREAMING_SNAKE_CASE	)

			self.assertIsInstance(_SCREAMING_SNAKE_CASE	,      _SCREAMING_SNAKE_CASE	)
			for ids in forced_decoder_ids:
					self.assertIsInstance(_SCREAMING_SNAKE_CASE	,      (list, tuple)	)

			lowerCAmelCase__     :    int					  =      [TRANSCRIBE, NOTIMESTAMPS]
			self.assertListEqual([ids[-1] for ids in forced_decoder_ids]	,      _SCREAMING_SNAKE_CASE	)




	def 							UpperCAmelCase__(      self		:	Optional[int]	)-> List[Any]:
			lowerCAmelCase__     :    str					  =      WhisperProcessor(tokenizer=self.get_tokenizer()	,      feature_extractor=self.get_feature_extractor()	)
			lowerCAmelCase__     :    List[str]					  =      processor.get_prompt_ids('''Mr. Quilter'''	)
			lowerCAmelCase__     :    Optional[Any]					  =      processor.tokenizer.decode(_SCREAMING_SNAKE_CASE	)

			self.assertListEqual(prompt_ids.tolist()	,      [5_0360, 1770, 13, 2264, 346, 353]	)
			self.assertEqual(_SCREAMING_SNAKE_CASE	,      '''<|startofprev|> Mr. Quilter'''	)




	def 							UpperCAmelCase__(      self		:	Optional[int]	)-> List[Any]:
			lowerCAmelCase__     :    Dict					  =      WhisperProcessor(tokenizer=self.get_tokenizer()	,      feature_extractor=self.get_feature_extractor()	)
			lowerCAmelCase__     :    Dict					  =      processor.get_prompt_ids(''''''	)
			lowerCAmelCase__     :    Union[str, Any]					  =      processor.tokenizer.decode(_SCREAMING_SNAKE_CASE	)

			self.assertListEqual(prompt_ids.tolist()	,      [5_0360, 220]	)
			self.assertEqual(_SCREAMING_SNAKE_CASE	,      '''<|startofprev|> '''	)




	def 							UpperCAmelCase__(      self		:	Optional[Any]	)-> Dict:
			lowerCAmelCase__     :    Optional[int]					  =      WhisperProcessor(tokenizer=self.get_tokenizer()	,      feature_extractor=self.get_feature_extractor()	)

			def _test_prompt_error_raised_helper(_SCREAMING_SNAKE_CASE		:	int	,      _SCREAMING_SNAKE_CASE		:	int	):
					with pytest.raises(_SCREAMING_SNAKE_CASE	) as excinfo:
							processor.get_prompt_ids(_SCREAMING_SNAKE_CASE	)
					lowerCAmelCase__     :    List[Any]					  =      F'Encountered text in the prompt corresponding to disallowed special token: {special_token}.'
					self.assertEqual(_SCREAMING_SNAKE_CASE	,      str(excinfo.value	)	)

			_test_prompt_error_raised_helper('''<|startofprev|> test'''	,      '''<|startofprev|>'''	)
			_test_prompt_error_raised_helper('''test <|notimestamps|>'''	,      '''<|notimestamps|>'''	)
			_test_prompt_error_raised_helper('''test <|zh|> test <|transcribe|>'''	,      '''<|zh|>'''	)








import getopt
import json
import os

# import numpy as np
import sys
from collections import OrderedDict

import datasets
import numpy as np
import torch
from modeling_frcnn import GeneralizedRCNN
from processing_image import Preprocess

from utils import Config
lowerCamelCase		=	False
lowerCamelCase		=	Config.from_pretrained('''unc-nlp/frcnn-vg-finetuned''')
lowerCamelCase		=	datasets.Features(
    OrderedDict(
        {
            '''attr_ids''': datasets.Sequence(length=CONFIG.MAX_DETECTIONS, feature=datasets.Value('''float32''')),
            '''attr_probs''': datasets.Sequence(length=CONFIG.MAX_DETECTIONS, feature=datasets.Value('''float32''')),
            '''boxes''': datasets.ArrayaD((CONFIG.MAX_DETECTIONS, 4), dtype='''float32'''),
            '''img_id''': datasets.Value('''int32'''),
            '''obj_ids''': datasets.Sequence(length=CONFIG.MAX_DETECTIONS, feature=datasets.Value('''float32''')),
            '''obj_probs''': datasets.Sequence(length=CONFIG.MAX_DETECTIONS, feature=datasets.Value('''float32''')),
            '''roi_features''': datasets.ArrayaD((CONFIG.MAX_DETECTIONS, 2048), dtype='''float32'''),
            '''sizes''': datasets.Sequence(length=2, feature=datasets.Value('''float32''')),
            '''preds_per_image''': datasets.Value(dtype='''int32'''),
        }
    )
)





class       _a							:




	def __init__(      self		:	str	,      _SCREAMING_SNAKE_CASE		:	Dict=sys.argv[1:]	)-> str:
			lowerCAmelCase__     :    int					  =      None
			lowerCAmelCase__     :    Optional[int]					  =      None
			lowerCAmelCase__     :    Union[str, Any]					  =      None
			lowerCAmelCase__     :    Union[str, Any]					  =      1
			lowerCAmelCase__  ,			lowerCAmelCase__     :    List[str]					  =      getopt.getopt(_SCREAMING_SNAKE_CASE	,      '''i:o:b:s'''	,      ['''inputdir=''', '''outfile=''', '''batch_size=''', '''subset_list=''']	)
			for opt, arg in opts:
					if opt in ("-i", "--inputdir"):
							lowerCAmelCase__     :    List[Any]					  =      arg
					elif opt in ("-o", "--outfile"):
							lowerCAmelCase__     :    Optional[Any]					  =      arg
					elif opt in ("-b", "--batch_size"):
							lowerCAmelCase__     :    Tuple					  =      int(_SCREAMING_SNAKE_CASE	)
					elif opt in ("-s", "--subset_list"):
							lowerCAmelCase__     :    str					  =      arg

			assert inputdir is not None  # and os.path.isdir(inputdir), f"{inputdir}"
			assert outputfile is not None and not os.path.isfile(_SCREAMING_SNAKE_CASE	), F'{outputfile}'
			if subset_list is not None:
					with open(os.path.realpath(_SCREAMING_SNAKE_CASE	)	) as f:
							lowerCAmelCase__     :    Union[str, Any]					  =      {self._vqa_file_split()[0] for x in tryload(_SCREAMING_SNAKE_CASE	)}
			else:
					lowerCAmelCase__     :    int					  =      None

			lowerCAmelCase__     :    List[Any]					  =      CONFIG
			if torch.cuda.is_available():
					lowerCAmelCase__     :    List[Any]					  =      '''cuda'''
			lowerCAmelCase__     :    Optional[int]					  =      os.path.realpath(_SCREAMING_SNAKE_CASE	)
			lowerCAmelCase__     :    str					  =      os.path.realpath(_SCREAMING_SNAKE_CASE	)
			lowerCAmelCase__     :    str					  =      Preprocess(self.config	)
			lowerCAmelCase__     :    str					  =      GeneralizedRCNN.from_pretrained('''unc-nlp/frcnn-vg-finetuned'''	,      config=self.config	)
			lowerCAmelCase__     :    Tuple					  =      batch_size if batch_size != 0 else 1
			lowerCAmelCase__     :    Dict					  =      DEFAULT_SCHEMA




	def 							UpperCAmelCase__(      self		:	List[str]	,      _SCREAMING_SNAKE_CASE		:	str	)-> Optional[Any]:
			lowerCAmelCase__     :    Optional[Any]					  =      int(file.split('''.'''	)[0].split('''_'''	)[-1]	)
			lowerCAmelCase__     :    List[Any]					  =      os.path.join(self.inputdir	,      _SCREAMING_SNAKE_CASE	)
			return (img_id, filepath)




	@property
	def 							UpperCAmelCase__(      self		:	Dict	)-> Tuple:
			lowerCAmelCase__     :    Any					  =      []
			for i, file in enumerate(os.listdir(self.inputdir	)	):
					if self.subset_list is not None and i not in self.subset_list:
							continue
					batch.append(self._vqa_file_split(_SCREAMING_SNAKE_CASE	)	)
					if len(_SCREAMING_SNAKE_CASE	) == self.batch:
							lowerCAmelCase__     :    Optional[int]					  =      batch
							lowerCAmelCase__     :    Union[str, Any]					  =      []
							yield list(map(_SCREAMING_SNAKE_CASE	,      zip(*_SCREAMING_SNAKE_CASE	)	)	)

			for i in range(1	):
					yield list(map(_SCREAMING_SNAKE_CASE	,      zip(*_SCREAMING_SNAKE_CASE	)	)	)




	def __call__(      self		:	int	)-> Dict:
			# make writer
			if not TEST:
					lowerCAmelCase__     :    Union[str, Any]					  =      datasets.ArrowWriter(features=self.schema	,      path=self.outputfile	)
			# do file generator
			for i, (img_ids, filepaths) in enumerate(self.file_generator	):
					lowerCAmelCase__  ,			lowerCAmelCase__  ,			lowerCAmelCase__     :    Optional[Any]					  =      self.preprocess(_SCREAMING_SNAKE_CASE	)
					lowerCAmelCase__     :    List[Any]					  =      self.model(
					    _SCREAMING_SNAKE_CASE	,      _SCREAMING_SNAKE_CASE	,      scales_yx=_SCREAMING_SNAKE_CASE	,      padding='''max_detections'''	,      max_detections=self.config.MAX_DETECTIONS	,      pad_value=0	,      return_tensors='''np'''	,      location='''cpu'''	,      )
					lowerCAmelCase__     :    Any					  =      output_dict.pop('''normalized_boxes'''	)
					if not TEST:
							lowerCAmelCase__     :    Optional[int]					  =      np.array(_SCREAMING_SNAKE_CASE	)
							lowerCAmelCase__     :    str					  =      self.schema.encode_batch(_SCREAMING_SNAKE_CASE	)
							writer.write_batch(_SCREAMING_SNAKE_CASE	)
					if TEST:
							break
					# finalizer the writer
			if not TEST:
					lowerCAmelCase__  ,			lowerCAmelCase__     :    Optional[int]					  =      writer.finalize()
					print(F'Success! You wrote {num_examples} entry(s) and {num_bytes >> 20} mb'	)






def   lowerCamelCase_					( _a ):




		"""simple docstring"""






		try:
				lowerCAmelCase__     :    Any					  =      json.load(_a )
				try:
						lowerCAmelCase__     :    Tuple					  =      list(data.keys() )
				except Exception:
						lowerCAmelCase__     :    List[str]					  =      [d['''img_id'''] for d in data]
		except Exception:
				try:
						lowerCAmelCase__     :    int					  =      eval(stream.read() )
				except Exception:
						lowerCAmelCase__     :    Tuple					  =      stream.read().split('''\n''' )
		return data


if __name__ == "__main__":
		lowerCamelCase		=	Extract(sys.argv[1:])
		extract()
		if not TEST:
				lowerCamelCase		=	datasets.Dataset.from_file(extract.outputfile)
				# wala!
				# print(np.array(dataset[0:2]["roi_features"]).shape)




