
"""simple docstring"""




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, prepare_video_inputs


if is_torch_available():
	import torch

if is_vision_available():
	from PIL import Image

	from transformers import VideoMAEImageProcessor







class 	A_       (      unittest.TestCase       ):




	'''simple docstring'''


	def __init__(	self		,   snake_case		,   snake_case=7		,   snake_case=3		,   snake_case=10		,   snake_case=18		,   snake_case=30		,   snake_case=400		,   snake_case=True		,   snake_case=None		,   snake_case=True		,   snake_case=[0.5, 0.5, 0.5]		,   snake_case=[0.5, 0.5, 0.5]		,   snake_case=None		,   ):
		lowercase					  =   size if size is not None else {'shortest_edge': 18}
		lowercase					  =   crop_size if crop_size is not None else {'height': 18, 'width': 18}

		lowercase					  =   parent
		lowercase					  =   batch_size
		lowercase					  =   num_channels
		lowercase					  =   num_frames
		lowercase					  =   image_size
		lowercase					  =   min_resolution
		lowercase					  =   max_resolution
		lowercase					  =   do_resize
		lowercase					  =   size
		lowercase					  =   do_normalize
		lowercase					  =   image_mean
		lowercase					  =   image_std
		lowercase					  =   crop_size





	def      SCREAMING_SNAKE_CASE__						(	self	):
		return {
		    "image_mean": self.image_mean,
		    "image_std": self.image_std,
		    "do_normalize": self.do_normalize,
		    "do_resize": self.do_resize,
		    "size": self.size,
		    "crop_size": self.crop_size,
		}







@require_torch
@require_vision
class 	A_       (      __lowerCamelCase				,     unittest.TestCase       ):




	'''simple docstring'''


	_UpperCamelCase					:					List[str]										=	VideoMAEImageProcessor if is_vision_available() else None





	def      SCREAMING_SNAKE_CASE__						(	self	):
		lowercase					  =   VideoMAEImageProcessingTester(self	)





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





	def      SCREAMING_SNAKE_CASE__						(	self	):
		lowercase					  =   self.image_processing_class(**self.image_processor_dict	)
		self.assertTrue(hasattr(snake_case		,   'image_mean'	)	)
		self.assertTrue(hasattr(snake_case		,   'image_std'	)	)
		self.assertTrue(hasattr(snake_case		,   'do_normalize'	)	)
		self.assertTrue(hasattr(snake_case		,   'do_resize'	)	)
		self.assertTrue(hasattr(snake_case		,   'do_center_crop'	)	)
		self.assertTrue(hasattr(snake_case		,   'size'	)	)





	def      SCREAMING_SNAKE_CASE__						(	self	):
		lowercase					  =   self.image_processing_class.from_dict(self.image_processor_dict	)
		self.assertEqual(image_processor.size		,   {'shortest_edge': 18}	)
		self.assertEqual(image_processor.crop_size		,   {'height': 18, 'width': 18}	)

		lowercase					  =   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	):
		pass





	def      SCREAMING_SNAKE_CASE__						(	self	):
		# Initialize image_processing
		lowercase					  =   self.image_processing_class(**self.image_processor_dict	)
		# create random PIL videos
		lowercase					  =   prepare_video_inputs(self.image_processor_tester		,   equal_resolution=snake_case	)
		for video in video_inputs:
			self.assertIsInstance(snake_case		,   snake_case	)
			self.assertIsInstance(video[0]		,   Image.Image	)

		# Test not batched input
		lowercase					  =   image_processing(video_inputs[0]		,   return_tensors='pt'	).pixel_values
		self.assertEqual(
		    encoded_videos.shape		,   (
		        1,
		        self.image_processor_tester.num_frames,
		        self.image_processor_tester.num_channels,
		        self.image_processor_tester.crop_size['height'],
		        self.image_processor_tester.crop_size['width'],
		    )		,   )

		# Test batched
		lowercase					  =   image_processing(snake_case		,   return_tensors='pt'	).pixel_values
		self.assertEqual(
		    encoded_videos.shape		,   (
		        self.image_processor_tester.batch_size,
		        self.image_processor_tester.num_frames,
		        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	):
		# Initialize image_processing
		lowercase					  =   self.image_processing_class(**self.image_processor_dict	)
		# create random numpy tensors
		lowercase					  =   prepare_video_inputs(self.image_processor_tester		,   equal_resolution=snake_case		,   numpify=snake_case	)
		for video in video_inputs:
			self.assertIsInstance(snake_case		,   snake_case	)
			self.assertIsInstance(video[0]		,   np.ndarray	)

		# Test not batched input
		lowercase					  =   image_processing(video_inputs[0]		,   return_tensors='pt'	).pixel_values
		self.assertEqual(
		    encoded_videos.shape		,   (
		        1,
		        self.image_processor_tester.num_frames,
		        self.image_processor_tester.num_channels,
		        self.image_processor_tester.crop_size['height'],
		        self.image_processor_tester.crop_size['width'],
		    )		,   )

		# Test batched
		lowercase					  =   image_processing(snake_case		,   return_tensors='pt'	).pixel_values
		self.assertEqual(
		    encoded_videos.shape		,   (
		        self.image_processor_tester.batch_size,
		        self.image_processor_tester.num_frames,
		        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	):
		# Initialize image_processing
		lowercase					  =   self.image_processing_class(**self.image_processor_dict	)
		# create random PyTorch tensors
		lowercase					  =   prepare_video_inputs(self.image_processor_tester		,   equal_resolution=snake_case		,   torchify=snake_case	)
		for video in video_inputs:
			self.assertIsInstance(snake_case		,   snake_case	)
			self.assertIsInstance(video[0]		,   torch.Tensor	)

		# Test not batched input
		lowercase					  =   image_processing(video_inputs[0]		,   return_tensors='pt'	).pixel_values
		self.assertEqual(
		    encoded_videos.shape		,   (
		        1,
		        self.image_processor_tester.num_frames,
		        self.image_processor_tester.num_channels,
		        self.image_processor_tester.crop_size['height'],
		        self.image_processor_tester.crop_size['width'],
		    )		,   )

		# Test batched
		lowercase					  =   image_processing(snake_case		,   return_tensors='pt'	).pixel_values
		self.assertEqual(
		    encoded_videos.shape		,   (
		        self.image_processor_tester.batch_size,
		        self.image_processor_tester.num_frames,
		        self.image_processor_tester.num_channels,
		        self.image_processor_tester.crop_size['height'],
		        self.image_processor_tester.crop_size['width'],
		    )		,   )



# This test is meant to be run in on an instance with TPUs like this:
#
#   python examples/pytorch/xla_spawn.py --num_cores=8 tests/test_trainer_tpu.py
#
# Replace 8 with the number of TPU cores you have.
#

import sys
from typing import Dict

from transformers import EvalPrediction, HfArgumentParser, TrainingArguments, is_torch_available
from transformers.utils import logging


UpperCAmelCase     				=			logging.get_logger(__name__)


if is_torch_available():
	import torch
	from torch import nn
	from torch.utils.data import Dataset

	from transformers import Trainer







	class 	A_       (      __lowerCamelCase       ):




		'''simple docstring'''


		def __init__(	self		,   snake_case = 101	):
			lowercase					  =   length





		def __len__(	self	):
			return self.length





		def __getitem__(	self		,   snake_case	):
			return i







	class 	A_       :




		'''simple docstring'''


		def __call__(	self		,   snake_case	):
			return {"input_ids": torch.tensor(snake_case	), "labels": torch.tensor(snake_case	)}







	class 	A_       (      nn.Module       ):




		'''simple docstring'''


		def __init__(	self	):
			super().__init__()
			# Add some (unused) params otherwise DDP will complain.
			lowercase					  =   nn.Linear(120		,   80	)





		def      SCREAMING_SNAKE_CASE__						(	self		,   snake_case		,   snake_case=None	):
			if labels is not None:
				return torch.tensor(0.0		,   device=input_ids.device	), input_ids
			else:
				return input_ids






def 							UpperCAmelCase_		(							):
	lowercase					  =   HfArgumentParser((TrainingArguments,)							)
	sys.argv += ["--output_dir", "./examples"]
	lowercase					  =   parser.parse_args_into_dataclasses()[0]

	logger.warning(
	    F'''Process rank: {training_args.local_rank}, device: {training_args.device}, '''
	    F'''tpu_num_cores: {training_args.tpu_num_cores}'''			,							)

	# Essentially, what we want to verify in the distributed case is
	# that we get all samples back, in the right order.
	# (this is crucial for prediction for instance)
	for dataset_length in [1001, 256, 15]:
		lowercase					  =   DummyDataset(__SCREAMING_SNAKE_CASE							)

		def compute_metrics(__SCREAMING_SNAKE_CASE							) -> Dict:
			lowercase					  =   list(range(len(__SCREAMING_SNAKE_CASE							)							)							)
			lowercase					  =   p.predictions.tolist() == sequential and p.label_ids.tolist() == sequential
			return {"success": success}

		lowercase					  =   Trainer(
		    model=DummyModel()			,							args=__SCREAMING_SNAKE_CASE			,							data_collator=DummyDataCollator()			,							eval_dataset=__SCREAMING_SNAKE_CASE			,							compute_metrics=__SCREAMING_SNAKE_CASE			,							)
		lowercase					  =   trainer.evaluate()
		logger.info(__SCREAMING_SNAKE_CASE							)
		if metrics["eval_success"] is not True:
			logger.error(__SCREAMING_SNAKE_CASE							)
			exit(1							)

		lowercase					  =   trainer.predict(__SCREAMING_SNAKE_CASE							)
		logger.info(p.metrics							)
		if p.metrics["test_success"] is not True:
			logger.error(p.metrics							)
			exit(1							)

		lowercase					  =   2

		lowercase					  =   trainer.evaluate()
		logger.info(__SCREAMING_SNAKE_CASE							)
		if metrics["eval_success"] is not True:
			logger.error(__SCREAMING_SNAKE_CASE							)
			exit(1							)

		lowercase					  =   trainer.predict(__SCREAMING_SNAKE_CASE							)
		logger.info(p.metrics							)
		if p.metrics["test_success"] is not True:
			logger.error(p.metrics							)
			exit(1							)

		lowercase					  =   None

	logger.info('🔥 All distributed tests successful'							)






def 							UpperCAmelCase_		(							__SCREAMING_SNAKE_CASE							):
	# For xla_spawn (TPUs)
	main()


if __name__ == "__main__":
	main()

