

"""simple docstring"""





# This code is adapted from OpenAI's release
# https://github.com/openai/human-eval/blob/master/human_eval/execution.py

import contextlib
import faulthandler
import io
import multiprocessing
import os
import platform
import signal
import tempfile

def   __SCREAMING_SNAKE_CASE  (      A_						,     A_						,     A_						,     A_ ):
	lowerCAmelCase__	:						Tuple  				=	multiprocessing.Manager()
	lowerCAmelCase__	:						List[Any]  				=	manager.list()

	lowerCAmelCase__	:						int  				=	multiprocessing.Process(target=A_						,     args=(check_program, result, timeout) )
	p.start()
	p.join(timeout=timeout + 1 )
	if p.is_alive():
		p.kill()

	if not result:
		result.append('''timed out''' )

	return {
	    "task_id": task_id,
	    "passed": result[0] == "passed",
	    "result": result[0],
	    "completion_id": completion_id,
	}

def   __SCREAMING_SNAKE_CASE  (      A_						,     A_						,     A_ ):
	with create_tempdir():
		# These system calls are needed when cleaning up tempdir.
		import os
		import shutil

		lowerCAmelCase__	:						List[str]  				=	shutil.rmtree
		lowerCAmelCase__	:						List[Any]  				=	os.rmdir
		lowerCAmelCase__	:						Dict  				=	os.chdir

		# Disable functionalities that can make destructive changes to the test.
		reliability_guard()

		# Run program.
		try:
			lowerCAmelCase__	:						Optional[int]  				=	{}
			with swallow_io():
				with time_limit(A_ ):
					exec(A_						,     A_ )
			result.append('''passed''' )
		except TimeoutException:
			result.append('''timed out''' )
		except BaseException as e:
			result.append(f'failed: {e}' )

		# Needed for cleaning up.
		lowerCAmelCase__	:						Tuple  				=	rmtree
		lowerCAmelCase__	:						Optional[Any]  				=	rmdir
		lowerCAmelCase__	:						int  				=	chdir

@contextlib.contextmanager
def   __SCREAMING_SNAKE_CASE  (      A_ ):
	def signal_handler(A_						,     A_ ):
		raise TimeoutException('''Timed out!''' )

	signal.setitimer(signal.ITIMER_REAL						,     A_ )
	signal.signal(signal.SIGALRM						,     A_ )
	try:
		yield
	finally:
		signal.setitimer(signal.ITIMER_REAL						,     0 )

@contextlib.contextmanager
def   __SCREAMING_SNAKE_CASE  (      ):
	lowerCAmelCase__	:						Optional[Any]  				=	WriteOnlyStringIO()
	with contextlib.redirect_stdout(A_ ):
		with contextlib.redirect_stderr(A_ ):
			with redirect_stdin(A_ ):
				yield

@contextlib.contextmanager
def   __SCREAMING_SNAKE_CASE  (      ):
	with tempfile.TemporaryDirectory() as dirname:
		with chdir(A_ ):
			yield dirname







class 				SCREAMING_SNAKE_CASE					(		a_       ):






					"""simple docstring"""



					pass







class 				SCREAMING_SNAKE_CASE					(		io.StringIO       ):






					"""simple docstring"""



					def        __lowerCAmelCase (       self			:				int   ,*lowercase_			:				Optional[Any]   ,**lowercase_			:				Optional[int]		):
						raise OSError







					def        __lowerCAmelCase (       self			:				List[str]   ,*lowercase_			:				Optional[int]   ,**lowercase_			:				List[Any]		):
						raise OSError







					def        __lowerCAmelCase (       self			:				Any   ,*lowercase_			:				List[str]   ,**lowercase_			:				Dict		):
						raise OSError







					def        __lowerCAmelCase (       self			:				Union[str, Any]   ,*lowercase_			:				List[Any]   ,**lowercase_			:				int		):
						return False







class 				SCREAMING_SNAKE_CASE					(		contextlib._RedirectStream       ):  # type: ignore






					"""simple docstring"""



					lowercase__				 =       "stdin"

@contextlib.contextmanager
def   __SCREAMING_SNAKE_CASE  (      A_ ):
	if root == ".":
		yield
		return
	lowerCAmelCase__	:						Any  				=	os.getcwd()
	os.chdir(A_ )
	try:
		yield
	except BaseException as exc:
		raise exc
	finally:
		os.chdir(A_ )





def   __SCREAMING_SNAKE_CASE  (      A_=None ):

	if maximum_memory_bytes is not None:
		import resource

		resource.setrlimit(resource.RLIMIT_AS						,     (maximum_memory_bytes, maximum_memory_bytes) )
		resource.setrlimit(resource.RLIMIT_DATA						,     (maximum_memory_bytes, maximum_memory_bytes) )
		if not platform.uname().system == "Darwin":
			resource.setrlimit(resource.RLIMIT_STACK						,     (maximum_memory_bytes, maximum_memory_bytes) )

	faulthandler.disable()

	import builtins

	lowerCAmelCase__	:						Dict  				=	None
	lowerCAmelCase__	:						Optional[int]  				=	None

	import os

	lowerCAmelCase__	:						Union[str, Any]  				=	'''1'''

	lowerCAmelCase__	:						Tuple  				=	None
	lowerCAmelCase__	:						Optional[int]  				=	None
	lowerCAmelCase__	:						List[str]  				=	None
	lowerCAmelCase__	:						str  				=	None
	lowerCAmelCase__	:						Dict  				=	None
	lowerCAmelCase__	:						Union[str, Any]  				=	None
	lowerCAmelCase__	:						Union[str, Any]  				=	None
	lowerCAmelCase__	:						List[Any]  				=	None
	lowerCAmelCase__	:						List[Any]  				=	None
	lowerCAmelCase__	:						str  				=	None
	lowerCAmelCase__	:						List[str]  				=	None
	lowerCAmelCase__	:						List[str]  				=	None
	lowerCAmelCase__	:						Dict  				=	None
	lowerCAmelCase__	:						Tuple  				=	None
	lowerCAmelCase__	:						Any  				=	None
	lowerCAmelCase__	:						List[Any]  				=	None
	lowerCAmelCase__	:						Any  				=	None
	lowerCAmelCase__	:						List[Any]  				=	None
	lowerCAmelCase__	:						Optional[int]  				=	None
	lowerCAmelCase__	:						str  				=	None
	lowerCAmelCase__	:						int  				=	None
	lowerCAmelCase__	:						Any  				=	None
	lowerCAmelCase__	:						Tuple  				=	None
	lowerCAmelCase__	:						Dict  				=	None
	lowerCAmelCase__	:						Union[str, Any]  				=	None
	lowerCAmelCase__	:						Any  				=	None
	lowerCAmelCase__	:						List[Any]  				=	None

	import shutil

	lowerCAmelCase__	:						str  				=	None
	lowerCAmelCase__	:						Union[str, Any]  				=	None
	lowerCAmelCase__	:						Optional[Any]  				=	None

	import subprocess

	lowerCAmelCase__	:						Any  				=	None  # type: ignore

	lowerCAmelCase__	:						Any  				=	None

	import sys

	lowerCAmelCase__	:						Any  				=	None
	lowerCAmelCase__	:						Any  				=	None
	lowerCAmelCase__	:						Tuple  				=	None
	lowerCAmelCase__	:						str  				=	None
	lowerCAmelCase__	:						Union[str, Any]  				=	None







"""simple docstring"""





import unittest

from huggingface_hub import hf_hub_download

from transformers import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING, VideoMAEFeatureExtractor
from transformers.pipelines import VideoClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_decord,
    require_tf,
    require_torch,
    require_torch_or_tf,
    require_vision,
)

from .test_pipelines_common import ANY







@is_pipeline_test
@require_torch_or_tf
@require_vision
@require_decord
class 				SCREAMING_SNAKE_CASE					(		unittest.TestCase       ):






					"""simple docstring"""



					lowercase__				 =       MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING







					def        __lowerCAmelCase (       self			:				Optional[Any]   ,lowercase_			:				Optional[Any]   ,lowercase_			:				int   ,lowercase_			:				str		):
						lowerCAmelCase__	:						Optional[int]  				=	hf_hub_download(
						    repo_id='''nateraw/video-demo'''   ,filename='''archery.mp4'''   ,repo_type='''dataset'''		)
						lowerCAmelCase__	:						Tuple  				=	VideoClassificationPipeline(model=lowercase_   ,image_processor=lowercase_   ,top_k=2		)
						lowerCAmelCase__	:						Optional[int]  				=	[
						    example_video_filepath,
						    '''https://huggingface.co/datasets/nateraw/video-demo/resolve/main/archery.mp4''',
						]
						return video_classifier, examples







					def        __lowerCAmelCase (       self			:				str   ,lowercase_			:				int   ,lowercase_			:				Dict		):
						for example in examples:
							lowerCAmelCase__	:						Dict  				=	video_classifier(lowercase_		)

							self.assertEqual(
							    lowercase_   ,[
							        {'''score''': ANY(lowercase_		), '''label''': ANY(lowercase_		)},
							        {'''score''': ANY(lowercase_		), '''label''': ANY(lowercase_		)},
							    ]   ,)







					@require_torch
					def        __lowerCAmelCase (       self			:				Optional[int]		):
						lowerCAmelCase__	:						Optional[int]  				=	'''hf-internal-testing/tiny-random-VideoMAEForVideoClassification'''
						lowerCAmelCase__	:						List[str]  				=	VideoMAEFeatureExtractor(
						    size={'''shortest_edge''': 1_0}   ,crop_size={'''height''': 1_0, '''width''': 1_0}		)
						lowerCAmelCase__	:						Optional[Any]  				=	pipeline(
						    '''video-classification'''   ,model=lowercase_   ,feature_extractor=lowercase_   ,frame_sampling_rate=4		)

						lowerCAmelCase__	:						Optional[int]  				=	hf_hub_download(repo_id='''nateraw/video-demo'''   ,filename='''archery.mp4'''   ,repo_type='''dataset'''		)
						lowerCAmelCase__	:						Optional[int]  				=	video_classifier(lowercase_   ,top_k=2		)
						self.assertEqual(
						    nested_simplify(lowercase_   ,decimals=4		)   ,[{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}]   ,)

						lowerCAmelCase__	:						Dict  				=	video_classifier(
						    [
						        video_file_path,
						        video_file_path,
						    ]   ,top_k=2   ,)
						self.assertEqual(
						    nested_simplify(lowercase_   ,decimals=4		)   ,[
						        [{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}],
						        [{'''score''': 0.5199, '''label''': '''LABEL_0'''}, {'''score''': 0.4801, '''label''': '''LABEL_1'''}],
						    ]   ,)







					@require_tf
					def        __lowerCAmelCase (       self			:				int		):
						pass




