import json
import os
import shutil
import tempfile
import unittest
from multiprocessing import get_context
from pathlib import Path

import datasets
import numpy as np
from datasets import load_dataset
from parameterized import parameterized

from transformers import AutoProcessor
from transformers.models.wavaveca import WavaVecaCTCTokenizer, WavaVecaFeatureExtractor
from transformers.models.wavaveca.tokenization_wavaveca import VOCAB_FILES_NAMES
from transformers.testing_utils import require_pyctcdecode, require_torch, require_torchaudio, slow
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_pyctcdecode_available, is_torch_available

from ..wavaveca.test_feature_extraction_wavaveca import floats_list


if is_pyctcdecode_available():
				from huggingface_hub import snapshot_download
				from pyctcdecode import BeamSearchDecoderCTC

				from transformers.models.wavaveca_with_lm import WavaVecaProcessorWithLM
				from transformers.models.wavaveca_with_lm.processing_wavaveca_with_lm import WavaVecaDecoderWithLMOutput

if is_torch_available():
				from transformers import WavaVecaForCTC



@require_pyctcdecode
class        lowerCamelCase__				(   unittest.TestCase				):





			def   SCREAMING_SNAKE_CASE__     (       self	:	str						):
										_lowerCAmelCase							=							'| <pad> <unk> <s> </s> a b c d e f g h i j k'.split()
										_lowerCAmelCase							=							dict(zip(lowercase__  ,  range(len(lowercase__						)						)						)						)

										_lowerCAmelCase							=							{
										    'unk_token': '<unk>',
										    'bos_token': '<s>',
										    'eos_token': '</s>',
										}
										_lowerCAmelCase							=							{
										    'feature_size': 1,
										    'padding_value': 0.0,
										    'sampling_rate': 1_60_00,
										    'return_attention_mask': False,
										    'do_normalize': True,
										}

										_lowerCAmelCase							=							tempfile.mkdtemp()
										_lowerCAmelCase							=							os.path.join(self.tmpdirname  ,  VOCAB_FILES_NAMES['vocab_file']						)
										_lowerCAmelCase							=							os.path.join(self.tmpdirname  ,  lowercase__						)
										with open(self.vocab_file  ,  'w'  ,  encoding='utf-8'						) as fp:
																	fp.write(json.dumps(lowercase__						) + '\n'						)

										with open(self.feature_extraction_file  ,  'w'  ,  encoding='utf-8'						) as fp:
																	fp.write(json.dumps(lowercase__						) + '\n'						)

										# load decoder from hub
										_lowerCAmelCase							=							'hf-internal-testing/ngram-beam-search-decoder'





			def   SCREAMING_SNAKE_CASE__     (       self	:	Optional[int]  ,  **lowercase__	:	Any						):
										_lowerCAmelCase							=							self.add_kwargs_tokens_map.copy()
										kwargs.update(lowercase__						)
										return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname  ,  **lowercase__						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	Optional[int]  ,  **lowercase__	:	str						):
										return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname  ,  **lowercase__						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	Any  ,  **lowercase__	:	Optional[int]						):
										return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name  ,  **lowercase__						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	Optional[Any]						):
										shutil.rmtree(self.tmpdirname						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	Union[str, Any]						):
										_lowerCAmelCase							=							self.get_tokenizer()
										_lowerCAmelCase							=							self.get_feature_extractor()
										_lowerCAmelCase							=							self.get_decoder()

										_lowerCAmelCase							=							WavaVecaProcessorWithLM(tokenizer=lowercase__  ,  feature_extractor=lowercase__  ,  decoder=lowercase__						)

										processor.save_pretrained(self.tmpdirname						)
										_lowerCAmelCase							=							WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname						)

										# tokenizer
										self.assertEqual(processor.tokenizer.get_vocab()  ,  tokenizer.get_vocab()						)
										self.assertIsInstance(processor.tokenizer  ,  lowercase__						)

										# feature extractor
										self.assertEqual(processor.feature_extractor.to_json_string()  ,  feature_extractor.to_json_string()						)
										self.assertIsInstance(processor.feature_extractor  ,  lowercase__						)

										# decoder
										self.assertEqual(processor.decoder._alphabet.labels  ,  decoder._alphabet.labels						)
										self.assertEqual(
										    processor.decoder.model_container[decoder._model_key]._unigram_set  ,  decoder.model_container[decoder._model_key]._unigram_set  ,  )
										self.assertIsInstance(processor.decoder  ,  lowercase__						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	Dict						):
										_lowerCAmelCase							=							WavaVecaProcessorWithLM(
										    tokenizer=self.get_tokenizer()  ,  feature_extractor=self.get_feature_extractor()  ,  decoder=self.get_decoder()						)
										processor.save_pretrained(self.tmpdirname						)

										# make sure that error is thrown when decoder alphabet doesn't match
										_lowerCAmelCase							=							WavaVecaProcessorWithLM.from_pretrained(
										    self.tmpdirname  ,  alpha=5.0  ,  beta=3.0  ,  score_boundary=-7.0  ,  unk_score_offset=3						)

										# decoder
										self.assertEqual(processor.language_model.alpha  ,  5.0						)
										self.assertEqual(processor.language_model.beta  ,  3.0						)
										self.assertEqual(processor.language_model.score_boundary  ,  -7.0						)
										self.assertEqual(processor.language_model.unk_score_offset  ,  3						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	Tuple						):
										_lowerCAmelCase							=							self.get_tokenizer()
										# add token to trigger raise
										tokenizer.add_tokens(['xx']						)
										with self.assertRaisesRegex(lowercase__  ,  'include'						):
																	WavaVecaProcessorWithLM(
																	    tokenizer=lowercase__  ,  feature_extractor=self.get_feature_extractor()  ,  decoder=self.get_decoder()						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	List[str]						):
										_lowerCAmelCase							=							self.get_feature_extractor()
										_lowerCAmelCase							=							self.get_tokenizer()
										_lowerCAmelCase							=							self.get_decoder()

										_lowerCAmelCase							=							WavaVecaProcessorWithLM(tokenizer=lowercase__  ,  feature_extractor=lowercase__  ,  decoder=lowercase__						)

										_lowerCAmelCase							=							floats_list((3, 10_00)						)

										_lowerCAmelCase							=							feature_extractor(lowercase__  ,  return_tensors='np'						)
										_lowerCAmelCase							=							processor(lowercase__  ,  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   SCREAMING_SNAKE_CASE__     (       self	:	int						):
										_lowerCAmelCase							=							self.get_feature_extractor()
										_lowerCAmelCase							=							self.get_tokenizer()
										_lowerCAmelCase							=							self.get_decoder()

										_lowerCAmelCase							=							WavaVecaProcessorWithLM(tokenizer=lowercase__  ,  feature_extractor=lowercase__  ,  decoder=lowercase__						)

										_lowerCAmelCase							=							'This is a test string'

										_lowerCAmelCase							=							processor(text=lowercase__						)

										_lowerCAmelCase							=							tokenizer(lowercase__						)

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





			def   SCREAMING_SNAKE_CASE__     (       self	:	Any  ,  lowercase__	:	Union[str, Any]=(2, 10, 16)  ,  lowercase__	:	Optional[Any]=77						):
										np.random.seed(lowercase__						)
										return np.random.rand(*lowercase__						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	str						):
										_lowerCAmelCase							=							self.get_feature_extractor()
										_lowerCAmelCase							=							self.get_tokenizer()
										_lowerCAmelCase							=							self.get_decoder()

										_lowerCAmelCase							=							WavaVecaProcessorWithLM(tokenizer=lowercase__  ,  feature_extractor=lowercase__  ,  decoder=lowercase__						)

										_lowerCAmelCase							=							self._get_dummy_logits(shape=(10, 16)  ,  seed=13						)

										_lowerCAmelCase							=							processor.decode(lowercase__						)

										_lowerCAmelCase							=							decoder.decode_beams(lowercase__						)[0]

										self.assertEqual(decoded_decoder[0]  ,  decoded_processor.text						)
										self.assertEqual('</s> <s> </s>'  ,  decoded_processor.text						)
										self.assertEqual(decoded_decoder[-2]  ,  decoded_processor.logit_score						)
										self.assertEqual(decoded_decoder[-1]  ,  decoded_processor.lm_score						)





			@parameterized.expand([[None], ['fork'], ['spawn']]						)
			def   SCREAMING_SNAKE_CASE__     (       self	:	int  ,  lowercase__	:	Optional[int]						):
										_lowerCAmelCase							=							self.get_feature_extractor()
										_lowerCAmelCase							=							self.get_tokenizer()
										_lowerCAmelCase							=							self.get_decoder()

										_lowerCAmelCase							=							WavaVecaProcessorWithLM(tokenizer=lowercase__  ,  feature_extractor=lowercase__  ,  decoder=lowercase__						)

										_lowerCAmelCase							=							self._get_dummy_logits()

										# note: pool should be instantiated *after* Wav2Vec2ProcessorWithLM.
										#       otherwise, the LM won't be available to the pool's sub-processes.
										# manual logic used to allow parameterized test for both pool=None and pool=Pool(...)
										if pool_context is None:
																	_lowerCAmelCase							=							processor.batch_decode(lowercase__						)
										else:
																	with get_context(lowercase__						).Pool() as pool:
																								_lowerCAmelCase							=							processor.batch_decode(lowercase__  ,  lowercase__						)

										_lowerCAmelCase							=							list(lowercase__						)

										with get_context('fork'						).Pool() as p:
																	_lowerCAmelCase							=							decoder.decode_beams_batch(lowercase__  ,  lowercase__						)

										_lowerCAmelCase			,		_lowerCAmelCase			,		_lowerCAmelCase							=							[], [], []
										for beams in decoded_beams:
																	texts_decoder.append(beams[0][0]						)
																	logit_scores_decoder.append(beams[0][-2]						)
																	lm_scores_decoder.append(beams[0][-1]						)

										self.assertListEqual(lowercase__  ,  decoded_processor.text						)
										self.assertListEqual(['<s> <s> </s>', '<s> <s> <s>']  ,  decoded_processor.text						)
										self.assertListEqual(lowercase__  ,  decoded_processor.logit_score						)
										self.assertListEqual(lowercase__  ,  decoded_processor.lm_score						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	Union[str, Any]						):
										_lowerCAmelCase							=							self.get_feature_extractor()
										_lowerCAmelCase							=							self.get_tokenizer()
										_lowerCAmelCase							=							self.get_decoder()

										_lowerCAmelCase							=							WavaVecaProcessorWithLM(tokenizer=lowercase__  ,  feature_extractor=lowercase__  ,  decoder=lowercase__						)

										_lowerCAmelCase							=							self._get_dummy_logits()

										_lowerCAmelCase							=							15
										_lowerCAmelCase							=							-2_0.0
										_lowerCAmelCase							=							-4.0

										_lowerCAmelCase							=							processor.batch_decode(
										    lowercase__  ,  beam_width=lowercase__  ,  beam_prune_logp=lowercase__  ,  token_min_logp=lowercase__  ,  )
										_lowerCAmelCase							=							decoded_processor_out.text

										_lowerCAmelCase							=							list(lowercase__						)

										with get_context('fork'						).Pool() as pool:
																	_lowerCAmelCase							=							decoder.decode_beams_batch(
																	    lowercase__  ,  lowercase__  ,  beam_width=lowercase__  ,  beam_prune_logp=lowercase__  ,  token_min_logp=lowercase__  ,  )

										_lowerCAmelCase							=							[d[0][0] for d in decoded_decoder_out]
										_lowerCAmelCase							=							[d[0][2] for d in decoded_decoder_out]
										_lowerCAmelCase							=							[d[0][3] for d in decoded_decoder_out]

										self.assertListEqual(lowercase__  ,  lowercase__						)
										self.assertListEqual(['</s> <s> <s>', '<s> <s> <s>']  ,  lowercase__						)

										self.assertTrue(np.array_equal(lowercase__  ,  decoded_processor_out.logit_score						)						)
										self.assertTrue(np.allclose([-2_0.0_5_4, -1_8.4_4_7]  ,  lowercase__  ,  atol=1e-3						)						)

										self.assertTrue(np.array_equal(lowercase__  ,  decoded_processor_out.lm_score						)						)
										self.assertTrue(np.allclose([-1_5.5_5_4, -1_3.9_4_7_4]  ,  lowercase__  ,  atol=1e-3						)						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	int						):
										_lowerCAmelCase							=							self.get_feature_extractor()
										_lowerCAmelCase							=							self.get_tokenizer()
										_lowerCAmelCase							=							self.get_decoder()

										_lowerCAmelCase							=							WavaVecaProcessorWithLM(tokenizer=lowercase__  ,  feature_extractor=lowercase__  ,  decoder=lowercase__						)

										_lowerCAmelCase							=							self._get_dummy_logits()

										_lowerCAmelCase							=							2.0
										_lowerCAmelCase							=							5.0
										_lowerCAmelCase							=							-2_0.0
										_lowerCAmelCase							=							True

										_lowerCAmelCase							=							processor.batch_decode(
										    lowercase__  ,  alpha=lowercase__  ,  beta=lowercase__  ,  unk_score_offset=lowercase__  ,  lm_score_boundary=lowercase__  ,  )
										_lowerCAmelCase							=							decoded_processor_out.text

										_lowerCAmelCase							=							list(lowercase__						)
										decoder.reset_params(
										    alpha=lowercase__  ,  beta=lowercase__  ,  unk_score_offset=lowercase__  ,  lm_score_boundary=lowercase__  ,  )

										with get_context('fork'						).Pool() as pool:
																	_lowerCAmelCase							=							decoder.decode_beams_batch(
																	    lowercase__  ,  lowercase__  ,  )

										_lowerCAmelCase							=							[d[0][0] for d in decoded_decoder_out]

										self.assertListEqual(lowercase__  ,  lowercase__						)
										self.assertListEqual(['<s> </s> <s> </s> </s>', '</s> </s> <s> </s> </s>']  ,  lowercase__						)
										_lowerCAmelCase							=							processor.decoder.model_container[processor.decoder._model_key]
										self.assertEqual(lm_model.alpha  ,  2.0						)
										self.assertEqual(lm_model.beta  ,  5.0						)
										self.assertEqual(lm_model.unk_score_offset  ,  -2_0.0						)
										self.assertEqual(lm_model.score_boundary  ,  lowercase__						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	Dict						):
										_lowerCAmelCase							=							WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'						)

										_lowerCAmelCase							=							processor.decoder.model_container[processor.decoder._model_key]
										_lowerCAmelCase							=							Path(language_model._kenlm_model.path.decode('utf-8'						)						).parent.parent.absolute()

										_lowerCAmelCase							=							os.listdir(lowercase__						)
										_lowerCAmelCase							=							['alphabet.json', 'language_model']

										downloaded_decoder_files.sort()
										expected_decoder_files.sort()

										# test that only decoder relevant files from
										# https://huggingface.co/hf-internal-testing/processor_with_lm/tree/main
										# are downloaded and none of the rest (e.g. README.md, ...)
										self.assertListEqual(lowercase__  ,  lowercase__						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	Any						):
										_lowerCAmelCase							=							snapshot_download('hf-internal-testing/processor_with_lm'						)

										_lowerCAmelCase							=							WavaVecaProcessorWithLM.from_pretrained(lowercase__						)

										_lowerCAmelCase							=							processor.decoder.model_container[processor.decoder._model_key]
										_lowerCAmelCase							=							Path(language_model._kenlm_model.path.decode('utf-8'						)						).parent.parent.absolute()

										_lowerCAmelCase							=							os.listdir(lowercase__						)
										_lowerCAmelCase							=							os.listdir(lowercase__						)

										local_decoder_files.sort()
										expected_decoder_files.sort()

										# test that both decoder form hub and local files in cache are the same
										self.assertListEqual(lowercase__  ,  lowercase__						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	Optional[Any]						):
										_lowerCAmelCase							=							WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'						)
										_lowerCAmelCase							=							AutoProcessor.from_pretrained('hf-internal-testing/processor_with_lm'						)

										_lowerCAmelCase							=							floats_list((3, 10_00)						)

										_lowerCAmelCase							=							processor_wavaveca(lowercase__  ,  return_tensors='np'						)
										_lowerCAmelCase							=							processor_auto(lowercase__  ,  return_tensors='np'						)

										for key in input_wavaveca.keys():
																	self.assertAlmostEqual(input_wavaveca[key].sum()  ,  input_auto[key].sum()  ,  delta=1e-2						)

										_lowerCAmelCase							=							self._get_dummy_logits()

										_lowerCAmelCase							=							processor_wavaveca.batch_decode(lowercase__						)
										_lowerCAmelCase							=							processor_auto.batch_decode(lowercase__						)

										self.assertListEqual(decoded_wavaveca.text  ,  decoded_auto.text						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	List[Any]						):
										_lowerCAmelCase							=							self.get_feature_extractor()
										_lowerCAmelCase							=							self.get_tokenizer()
										_lowerCAmelCase							=							self.get_decoder()

										_lowerCAmelCase							=							WavaVecaProcessorWithLM(tokenizer=lowercase__  ,  feature_extractor=lowercase__  ,  decoder=lowercase__						)

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





			@staticmethod
			def   SCREAMING_SNAKE_CASE__     (       lowercase__	:	Union[str, Any]  ,  lowercase__	:	str						):
										_lowerCAmelCase							=							[d[key] for d in offsets]
										return retrieved_list





			def   SCREAMING_SNAKE_CASE__     (       self	:	Optional[Any]						):
										_lowerCAmelCase							=							WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'						)
										_lowerCAmelCase							=							self._get_dummy_logits()[0]

										_lowerCAmelCase							=							processor.decode(lowercase__  ,  output_word_offsets=lowercase__						)
										# check Wav2Vec2CTCTokenizerOutput keys for word
										self.assertEqual(len(outputs.keys()						)  ,  4						)
										self.assertTrue('text' in outputs						)
										self.assertTrue('word_offsets' in outputs						)
										self.assertTrue(isinstance(lowercase__  ,  lowercase__						)						)

										self.assertEqual(' '.join(self.get_from_offsets(outputs['word_offsets']  ,  'word'						)						)  ,  outputs.text						)
										self.assertListEqual(self.get_from_offsets(outputs['word_offsets']  ,  'word'						)  ,  ['<s>', '<s>', '</s>']						)
										self.assertListEqual(self.get_from_offsets(outputs['word_offsets']  ,  'start_offset'						)  ,  [0, 2, 4]						)
										self.assertListEqual(self.get_from_offsets(outputs['word_offsets']  ,  'end_offset'						)  ,  [1, 3, 5]						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	List[str]						):
										_lowerCAmelCase							=							WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'						)
										_lowerCAmelCase							=							self._get_dummy_logits()

										_lowerCAmelCase							=							processor.batch_decode(lowercase__  ,  output_word_offsets=lowercase__						)

										# check Wav2Vec2CTCTokenizerOutput keys for word
										self.assertEqual(len(outputs.keys()						)  ,  4						)
										self.assertTrue('text' in outputs						)
										self.assertTrue('word_offsets' in outputs						)
										self.assertTrue(isinstance(lowercase__  ,  lowercase__						)						)

										self.assertListEqual(
										    [' '.join(self.get_from_offsets(lowercase__  ,  'word'						)						) for o in outputs['word_offsets']]  ,  outputs.text						)
										self.assertListEqual(self.get_from_offsets(outputs['word_offsets'][0]  ,  'word'						)  ,  ['<s>', '<s>', '</s>']						)
										self.assertListEqual(self.get_from_offsets(outputs['word_offsets'][0]  ,  'start_offset'						)  ,  [0, 2, 4]						)
										self.assertListEqual(self.get_from_offsets(outputs['word_offsets'][0]  ,  'end_offset'						)  ,  [1, 3, 5]						)





			@slow
			@require_torch
			@require_torchaudio
			def   SCREAMING_SNAKE_CASE__     (       self	:	Dict						):
										import torch

										_lowerCAmelCase							=							load_dataset('common_voice'  ,  'en'  ,  split='train'  ,  streaming=lowercase__						)
										_lowerCAmelCase							=							ds.cast_column('audio'  ,  datasets.Audio(sampling_rate=1_60_00						)						)
										_lowerCAmelCase							=							iter(lowercase__						)
										_lowerCAmelCase							=							next(lowercase__						)

										_lowerCAmelCase							=							AutoProcessor.from_pretrained('patrickvonplaten/wav2vec2-base-100h-with-lm'						)
										_lowerCAmelCase							=							WavaVecaForCTC.from_pretrained('patrickvonplaten/wav2vec2-base-100h-with-lm'						)

										# compare to filename `common_voice_en_100038.mp3` of dataset viewer on https://huggingface.co/datasets/common_voice/viewer/en/train
										_lowerCAmelCase							=							processor(sample['audio']['array']  ,  return_tensors='pt'						).input_values

										with torch.no_grad():
																	_lowerCAmelCase							=							model(lowercase__						).logits.cpu().numpy()

										_lowerCAmelCase							=							processor.decode(logits[0]  ,  output_word_offsets=lowercase__						)

										_lowerCAmelCase							=							model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
										_lowerCAmelCase							=							[
										    {
										        'start_time': d['start_offset'] * time_offset,
										        'end_time': d['end_offset'] * time_offset,
										        'word': d['word'],
										    }
										    for d in output['word_offsets']
										]

										_lowerCAmelCase							=							'WHY DOES MILISANDRA LOOK LIKE SHE WANTS TO CONSUME JOHN SNOW ON THE RIVER AT THE WALL'

										# output words
										self.assertEqual(' '.join(self.get_from_offsets(lowercase__  ,  'word'						)						)  ,  lowercase__						)
										self.assertEqual(' '.join(self.get_from_offsets(lowercase__  ,  'word'						)						)  ,  output.text						)

										# output times
										_lowerCAmelCase							=							torch.tensor(self.get_from_offsets(lowercase__  ,  'start_time'						)						)
										_lowerCAmelCase							=							torch.tensor(self.get_from_offsets(lowercase__  ,  'end_time'						)						)

										# fmt: off
										_lowerCAmelCase							=							torch.tensor([1.4_1_9_9, 1.6_5_9_9, 2.2_5_9_9, 3.0, 3.2_4, 3.5_9_9_9, 3.7_9_9_9, 4.0_9_9_9, 4.2_6, 4.9_4, 5.2_8, 5.6_5_9_9, 5.7_8, 5.9_4, 6.3_2, 6.5_3_9_9, 6.6_5_9_9]						)
										_lowerCAmelCase							=							torch.tensor([1.5_3_9_9, 1.8_9_9_9, 2.9, 3.1_6, 3.5_3_9_9, 3.7_2, 4.0_1_9_9, 4.1_7_9_9, 4.7_6, 5.1_5_9_9, 5.5_5_9_9, 5.6_9_9_9, 5.8_6, 6.1_9_9_9, 6.3_8, 6.6_1_9_9, 6.9_4]						)
										# fmt: on

										self.assertTrue(torch.allclose(lowercase__  ,  lowercase__  ,  atol=0.0_1						)						)
										self.assertTrue(torch.allclose(lowercase__  ,  lowercase__  ,  atol=0.0_1						)						)







import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin


if is_torch_available():
				import torch

if is_vision_available():
				from PIL import Image

				from transformers import ChineseCLIPImageProcessor



class        lowerCamelCase__				(   unittest.TestCase				):





			def __init__(       self	:	List[str]  ,  lowercase__	:	Any  ,  lowercase__	:	List[Any]=7  ,  lowercase__	:	List[str]=3  ,  lowercase__	:	str=18  ,  lowercase__	:	List[Any]=30  ,  lowercase__	:	Optional[int]=4_00  ,  lowercase__	:	Dict=True  ,  lowercase__	:	List[str]=None  ,  lowercase__	:	int=True  ,  lowercase__	:	Tuple=None  ,  lowercase__	:	int=True  ,  lowercase__	:	Tuple=[0.4_8_1_4_5_4_6_6, 0.4_5_7_8_2_7_5, 0.4_0_8_2_1_0_7_3]  ,  lowercase__	:	Optional[int]=[0.2_6_8_6_2_9_5_4, 0.2_6_1_3_0_2_5_8, 0.2_7_5_7_7_7_1_1]  ,  lowercase__	:	Any=True  ,  ):
										_lowerCAmelCase							=							size if size is not None else {'height': 2_24, 'width': 2_24}
										_lowerCAmelCase							=							crop_size if crop_size is not None else {'height': 18, 'width': 18}
										_lowerCAmelCase							=							parent
										_lowerCAmelCase							=							batch_size
										_lowerCAmelCase							=							num_channels
										_lowerCAmelCase							=							image_size
										_lowerCAmelCase							=							min_resolution
										_lowerCAmelCase							=							max_resolution
										_lowerCAmelCase							=							do_resize
										_lowerCAmelCase							=							size
										_lowerCAmelCase							=							do_center_crop
										_lowerCAmelCase							=							crop_size
										_lowerCAmelCase							=							do_normalize
										_lowerCAmelCase							=							image_mean
										_lowerCAmelCase							=							image_std
										_lowerCAmelCase							=							do_convert_rgb





			def   SCREAMING_SNAKE_CASE__     (       self	:	Optional[Any]						):
										return {
										    "do_resize": self.do_resize,
										    "size": self.size,
										    "do_center_crop": self.do_center_crop,
										    "crop_size": self.crop_size,
										    "do_normalize": self.do_normalize,
										    "image_mean": self.image_mean,
										    "image_std": self.image_std,
										    "do_convert_rgb": self.do_convert_rgb,
										}





			def   SCREAMING_SNAKE_CASE__     (       self	:	List[str]  ,  lowercase__	:	Tuple=False  ,  lowercase__	:	List[Any]=False  ,  lowercase__	:	str=False						):
										assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

										if equal_resolution:
																	_lowerCAmelCase							=							[]
																	for i in range(self.batch_size						):
																								image_inputs.append(
																								    np.random.randint(
																								        2_55  ,  size=(self.num_channels, self.max_resolution, self.max_resolution)  ,  dtype=np.uinta						)						)
										else:
																	_lowerCAmelCase							=							[]
																	for i in range(self.batch_size						):
																								_lowerCAmelCase			,		_lowerCAmelCase							=							np.random.choice(np.arange(self.min_resolution  ,  self.max_resolution						)  ,  2						)
																								image_inputs.append(np.random.randint(2_55  ,  size=(self.num_channels, width, height)  ,  dtype=np.uinta						)						)

										if not numpify and not torchify:
																	# PIL expects the channel dimension as last dimension
																	_lowerCAmelCase							=							[Image.fromarray(np.moveaxis(lowercase__  ,  0  ,  -1						)						) for x in image_inputs]

										if torchify:
																	_lowerCAmelCase							=							[torch.from_numpy(lowercase__						) for x in image_inputs]

										return image_inputs







@require_torch
@require_vision
class        lowerCamelCase__				(   UpperCAmelCase       ,unittest.TestCase				):
			UpperCamelCase__    							=ChineseCLIPImageProcessor if is_vision_available() else None





			def   SCREAMING_SNAKE_CASE__     (       self	:	Tuple						):
										_lowerCAmelCase							=							ChineseCLIPImageProcessingTester(self  ,  do_center_crop=lowercase__						)





			@property
			def   SCREAMING_SNAKE_CASE__     (       self	:	Any						):
										return self.image_processor_tester.prepare_image_processor_dict()





			def   SCREAMING_SNAKE_CASE__     (       self	:	Tuple						):
										_lowerCAmelCase							=							self.image_processing_class(**self.image_processor_dict						)
										self.assertTrue(hasattr(lowercase__  ,  'do_resize'						)						)
										self.assertTrue(hasattr(lowercase__  ,  'size'						)						)
										self.assertTrue(hasattr(lowercase__  ,  'do_center_crop'						)						)
										self.assertTrue(hasattr(lowercase__  ,  'center_crop'						)						)
										self.assertTrue(hasattr(lowercase__  ,  'do_normalize'						)						)
										self.assertTrue(hasattr(lowercase__  ,  'image_mean'						)						)
										self.assertTrue(hasattr(lowercase__  ,  'image_std'						)						)
										self.assertTrue(hasattr(lowercase__  ,  'do_convert_rgb'						)						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	Union[str, Any]						):
										_lowerCAmelCase							=							self.image_processing_class.from_dict(self.image_processor_dict						)
										self.assertEqual(image_processor.size  ,  {'height': 2_24, 'width': 2_24}						)
										self.assertEqual(image_processor.crop_size  ,  {'height': 18, 'width': 18}						)

										_lowerCAmelCase							=							self.image_processing_class.from_dict(self.image_processor_dict  ,  size=42  ,  crop_size=84						)
										self.assertEqual(image_processor.size  ,  {'shortest_edge': 42}						)
										self.assertEqual(image_processor.crop_size  ,  {'height': 84, 'width': 84}						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	List[Any]						):
										pass





			def   SCREAMING_SNAKE_CASE__     (       self	:	str						):
										# Initialize image_processing
										_lowerCAmelCase							=							self.image_processing_class(**self.image_processor_dict						)
										# create random PIL images
										_lowerCAmelCase							=							self.image_processor_tester.prepare_inputs(equal_resolution=lowercase__						)
										for image in image_inputs:
																	self.assertIsInstance(lowercase__  ,  Image.Image						)

										# Test not batched input
										_lowerCAmelCase							=							image_processing(image_inputs[0]  ,  return_tensors='pt'						).pixel_values
										self.assertEqual(
										    encoded_images.shape  ,  (
										        1,
										        self.image_processor_tester.num_channels,
										        self.image_processor_tester.crop_size['height'],
										        self.image_processor_tester.crop_size['width'],
										    )  ,  )

										# Test batched
										_lowerCAmelCase							=							image_processing(lowercase__  ,  return_tensors='pt'						).pixel_values
										self.assertEqual(
										    encoded_images.shape  ,  (
										        self.image_processor_tester.batch_size,
										        self.image_processor_tester.num_channels,
										        self.image_processor_tester.crop_size['height'],
										        self.image_processor_tester.crop_size['width'],
										    )  ,  )





			def   SCREAMING_SNAKE_CASE__     (       self	:	Any						):
										# Initialize image_processing
										_lowerCAmelCase							=							self.image_processing_class(**self.image_processor_dict						)
										# create random numpy tensors
										_lowerCAmelCase							=							self.image_processor_tester.prepare_inputs(equal_resolution=lowercase__  ,  numpify=lowercase__						)
										for image in image_inputs:
																	self.assertIsInstance(lowercase__  ,  np.ndarray						)

										# Test not batched input
										_lowerCAmelCase							=							image_processing(image_inputs[0]  ,  return_tensors='pt'						).pixel_values
										self.assertEqual(
										    encoded_images.shape  ,  (
										        1,
										        self.image_processor_tester.num_channels,
										        self.image_processor_tester.crop_size['height'],
										        self.image_processor_tester.crop_size['width'],
										    )  ,  )

										# Test batched
										_lowerCAmelCase							=							image_processing(lowercase__  ,  return_tensors='pt'						).pixel_values
										self.assertEqual(
										    encoded_images.shape  ,  (
										        self.image_processor_tester.batch_size,
										        self.image_processor_tester.num_channels,
										        self.image_processor_tester.crop_size['height'],
										        self.image_processor_tester.crop_size['width'],
										    )  ,  )





			def   SCREAMING_SNAKE_CASE__     (       self	:	int						):
										# Initialize image_processing
										_lowerCAmelCase							=							self.image_processing_class(**self.image_processor_dict						)
										# create random PyTorch tensors
										_lowerCAmelCase							=							self.image_processor_tester.prepare_inputs(equal_resolution=lowercase__  ,  torchify=lowercase__						)
										for image in image_inputs:
																	self.assertIsInstance(lowercase__  ,  torch.Tensor						)

										# Test not batched input
										_lowerCAmelCase							=							image_processing(image_inputs[0]  ,  return_tensors='pt'						).pixel_values
										self.assertEqual(
										    encoded_images.shape  ,  (
										        1,
										        self.image_processor_tester.num_channels,
										        self.image_processor_tester.crop_size['height'],
										        self.image_processor_tester.crop_size['width'],
										    )  ,  )

										# Test batched
										_lowerCAmelCase							=							image_processing(lowercase__  ,  return_tensors='pt'						).pixel_values
										self.assertEqual(
										    encoded_images.shape  ,  (
										        self.image_processor_tester.batch_size,
										        self.image_processor_tester.num_channels,
										        self.image_processor_tester.crop_size['height'],
										        self.image_processor_tester.crop_size['width'],
										    )  ,  )







@require_torch
@require_vision
class        lowerCamelCase__				(   UpperCAmelCase       ,unittest.TestCase				):
			UpperCamelCase__    							=ChineseCLIPImageProcessor if is_vision_available() else None





			def   SCREAMING_SNAKE_CASE__     (       self	:	Optional[Any]						):
										_lowerCAmelCase							=							ChineseCLIPImageProcessingTester(self  ,  num_channels=4  ,  do_center_crop=lowercase__						)
										_lowerCAmelCase							=							3





			@property
			def   SCREAMING_SNAKE_CASE__     (       self	:	Tuple						):
										return self.image_processor_tester.prepare_image_processor_dict()





			def   SCREAMING_SNAKE_CASE__     (       self	:	List[Any]						):
										_lowerCAmelCase							=							self.image_processing_class(**self.image_processor_dict						)
										self.assertTrue(hasattr(lowercase__  ,  'do_resize'						)						)
										self.assertTrue(hasattr(lowercase__  ,  'size'						)						)
										self.assertTrue(hasattr(lowercase__  ,  'do_center_crop'						)						)
										self.assertTrue(hasattr(lowercase__  ,  'center_crop'						)						)
										self.assertTrue(hasattr(lowercase__  ,  'do_normalize'						)						)
										self.assertTrue(hasattr(lowercase__  ,  'image_mean'						)						)
										self.assertTrue(hasattr(lowercase__  ,  'image_std'						)						)
										self.assertTrue(hasattr(lowercase__  ,  'do_convert_rgb'						)						)





			def   SCREAMING_SNAKE_CASE__     (       self	:	Union[str, Any]						):
										pass





			def   SCREAMING_SNAKE_CASE__     (       self	:	Dict						):
										# Initialize image_processing
										_lowerCAmelCase							=							self.image_processing_class(**self.image_processor_dict						)
										# create random PIL images
										_lowerCAmelCase							=							self.image_processor_tester.prepare_inputs(equal_resolution=lowercase__						)
										for image in image_inputs:
																	self.assertIsInstance(lowercase__  ,  Image.Image						)

										# Test not batched input
										_lowerCAmelCase							=							image_processing(image_inputs[0]  ,  return_tensors='pt'						).pixel_values
										self.assertEqual(
										    encoded_images.shape  ,  (
										        1,
										        self.expected_encoded_image_num_channels,
										        self.image_processor_tester.crop_size['height'],
										        self.image_processor_tester.crop_size['width'],
										    )  ,  )

										# Test batched
										_lowerCAmelCase							=							image_processing(lowercase__  ,  return_tensors='pt'						).pixel_values
										self.assertEqual(
										    encoded_images.shape  ,  (
										        self.image_processor_tester.batch_size,
										        self.expected_encoded_image_num_channels,
										        self.image_processor_tester.crop_size['height'],
										        self.image_processor_tester.crop_size['width'],
										    )  ,  )






