



"""simple docstring"""







import unittest

from transformers import (
    MODEL_FOR_CAUSAL_LM_MAPPING,
    TF_MODEL_FOR_CAUSAL_LM_MAPPING,
    TextGenerationPipeline,
    logging,
    pipeline,
)
from transformers.testing_utils import (
    CaptureLogger,
    is_pipeline_test,
    require_accelerate,
    require_tf,
    require_torch,
    require_torch_gpu,
    require_torch_or_tf,
)

from .test_pipelines_common import ANY

@is_pipeline_test
@require_torch_or_tf
class       _a  ( unittest.TestCase):
			__magic_name__     					=     MODEL_FOR_CAUSAL_LM_MAPPING
			__magic_name__     					=     TF_MODEL_FOR_CAUSAL_LM_MAPPING



			@require_torch
			def 				__lowercase					(			self							:				Tuple		)							->					Optional[int]:
									snake_case   :			int						    = pipeline(task="text-generation"   ,	model="sshleifer/tiny-ctrl"   ,	framework="pt"		)
									# Using `do_sample=False` to force deterministic output
									snake_case   :			List[str]						    = text_generator("This is a test"   ,	do_sample=_lowercase		)
									self.assertEqual(
									    _lowercase   ,	[
									        {
									            "generated_text": (
									                "This is a test ☃ ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy oscope."
									                " oscope. FiliFili@@"
									            )
									        }
									    ]   ,	)

									snake_case   :			int						    = text_generator(["This is a test", "This is a second test"]		)
									self.assertEqual(
									    _lowercase   ,	[
									        [
									            {
									                "generated_text": (
									                    "This is a test ☃ ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy oscope."
									                    " oscope. FiliFili@@"
									                )
									            }
									        ],
									        [
									            {
									                "generated_text": (
									                    "This is a second test ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy"
									                    " oscope. oscope. FiliFili@@"
									                )
									            }
									        ],
									    ]   ,	)

									snake_case   :			List[Any]						    = text_generator("This is a test"   ,	do_sample=_lowercase   ,	num_return_sequences=2   ,	return_tensors=_lowercase		)
									self.assertEqual(
									    _lowercase   ,	[
									        {"generated_token_ids": ANY(_lowercase		)},
									        {"generated_token_ids": ANY(_lowercase		)},
									    ]   ,	)
									snake_case   :			Union[str, Any]						    = text_generator.model.config.eos_token_id
									snake_case   :			Tuple						    = "<pad>"
									snake_case   :			Tuple						    = text_generator(
									    ["This is a test", "This is a second test"]   ,	do_sample=_lowercase   ,	num_return_sequences=2   ,	batch_size=2   ,	return_tensors=_lowercase   ,	)
									self.assertEqual(
									    _lowercase   ,	[
									        [
									            {"generated_token_ids": ANY(_lowercase		)},
									            {"generated_token_ids": ANY(_lowercase		)},
									        ],
									        [
									            {"generated_token_ids": ANY(_lowercase		)},
									            {"generated_token_ids": ANY(_lowercase		)},
									        ],
									    ]   ,	)



			@require_tf
			def 				__lowercase					(			self							:				List[str]		)							->					int:
									snake_case   :			Any						    = pipeline(task="text-generation"   ,	model="sshleifer/tiny-ctrl"   ,	framework="tf"		)

									# Using `do_sample=False` to force deterministic output
									snake_case   :			Optional[int]						    = text_generator("This is a test"   ,	do_sample=_lowercase		)
									self.assertEqual(
									    _lowercase   ,	[
									        {
									            "generated_text": (
									                "This is a test FeyFeyFey(Croatis.), s.), Cannes Cannes Cannes 閲閲Cannes Cannes Cannes 攵"
									                " please,"
									            )
									        }
									    ]   ,	)

									snake_case   :			List[Any]						    = text_generator(["This is a test", "This is a second test"]   ,	do_sample=_lowercase		)
									self.assertEqual(
									    _lowercase   ,	[
									        [
									            {
									                "generated_text": (
									                    "This is a test FeyFeyFey(Croatis.), s.), Cannes Cannes Cannes 閲閲Cannes Cannes Cannes 攵"
									                    " please,"
									                )
									            }
									        ],
									        [
									            {
									                "generated_text": (
									                    "This is a second test Chieftain Chieftain prefecture prefecture prefecture Cannes Cannes"
									                    " Cannes 閲閲Cannes Cannes Cannes 攵 please,"
									                )
									            }
									        ],
									    ]   ,	)



			def 				__lowercase					(			self							:				int   ,	_lowercase							:				Tuple   ,	_lowercase							:				int   ,	_lowercase							:				int		)							->					Dict:
									snake_case   :			int						    = TextGenerationPipeline(model=_lowercase   ,	tokenizer=_lowercase		)
									return text_generator, ["This is a test", "Another test"]



			def 				__lowercase					(			self							:				Any		)							->					Tuple:
									snake_case   :			Optional[Any]						    = "Hello I believe in"
									snake_case   :			Tuple						    = pipeline("text-generation"   ,	model="hf-internal-testing/tiny-random-gpt2"		)
									snake_case   :			str						    = text_generator(_lowercase		)
									self.assertEqual(
									    _lowercase   ,	[{"generated_text": "Hello I believe in fe fe fe fe fe fe fe fe fe fe fe fe"}]   ,	)

									snake_case   :			str						    = text_generator(_lowercase   ,	stop_sequence=" fe"		)
									self.assertEqual(_lowercase   ,	[{"generated_text": "Hello I believe in fe"}]		)



			def 				__lowercase					(			self							:				Optional[Any]   ,	_lowercase							:				Union[str, Any]   ,	_lowercase							:				int		)							->					List[Any]:
									snake_case   :			Tuple						    = text_generator.model
									snake_case   :			int						    = text_generator.tokenizer

									snake_case   :			Union[str, Any]						    = text_generator("This is a test"		)
									self.assertEqual(_lowercase   ,	[{"generated_text": ANY(_lowercase		)}]		)
									self.assertTrue(outputs[0]["generated_text"].startswith("This is a test"		)		)

									snake_case   :			str						    = text_generator("This is a test"   ,	return_full_text=_lowercase		)
									self.assertEqual(_lowercase   ,	[{"generated_text": ANY(_lowercase		)}]		)
									self.assertNotIn("This is a test"   ,	outputs[0]["generated_text"]		)

									snake_case   :			Any						    = pipeline(task="text-generation"   ,	model=_lowercase   ,	tokenizer=_lowercase   ,	return_full_text=_lowercase		)
									snake_case   :			Dict						    = text_generator("This is a test"		)
									self.assertEqual(_lowercase   ,	[{"generated_text": ANY(_lowercase		)}]		)
									self.assertNotIn("This is a test"   ,	outputs[0]["generated_text"]		)

									snake_case   :			Dict						    = text_generator("This is a test"   ,	return_full_text=_lowercase		)
									self.assertEqual(_lowercase   ,	[{"generated_text": ANY(_lowercase		)}]		)
									self.assertTrue(outputs[0]["generated_text"].startswith("This is a test"		)		)

									snake_case   :			Optional[Any]						    = text_generator(["This is great !", "Something else"]   ,	num_return_sequences=2   ,	do_sample=_lowercase		)
									self.assertEqual(
									    _lowercase   ,	[
									        [{"generated_text": ANY(_lowercase		)}, {"generated_text": ANY(_lowercase		)}],
									        [{"generated_text": ANY(_lowercase		)}, {"generated_text": ANY(_lowercase		)}],
									    ]   ,	)

									if text_generator.tokenizer.pad_token is not None:
															snake_case   :			List[Any]						    = text_generator(
															    ["This is great !", "Something else"]   ,	num_return_sequences=2   ,	batch_size=2   ,	do_sample=_lowercase		)
															self.assertEqual(
															    _lowercase   ,	[
															        [{"generated_text": ANY(_lowercase		)}, {"generated_text": ANY(_lowercase		)}],
															        [{"generated_text": ANY(_lowercase		)}, {"generated_text": ANY(_lowercase		)}],
															    ]   ,	)

									with self.assertRaises(_lowercase		):
															snake_case   :			Tuple						    = text_generator("test"   ,	return_full_text=_lowercase   ,	return_text=_lowercase		)
									with self.assertRaises(_lowercase		):
															snake_case   :			int						    = text_generator("test"   ,	return_full_text=_lowercase   ,	return_tensors=_lowercase		)
									with self.assertRaises(_lowercase		):
															snake_case   :			Tuple						    = text_generator("test"   ,	return_text=_lowercase   ,	return_tensors=_lowercase		)

									# Empty prompt is slighly special
									# it requires BOS token to exist.
									# Special case for Pegasus which will always append EOS so will
									# work even without BOS.
									if (
									    text_generator.tokenizer.bos_token_id is not None
									    or "Pegasus" in tokenizer.__class__.__name__
									    or "Git" in model.__class__.__name__
									):
															snake_case   :			Any						    = text_generator(""		)
															self.assertEqual(_lowercase   ,	[{"generated_text": ANY(_lowercase		)}]		)
									else:
															with self.assertRaises((ValueError, AssertionError)		):
																					snake_case   :			Optional[Any]						    = text_generator(""		)

									if text_generator.framework == "tf":
															# TF generation does not support max_new_tokens, and it's impossible
															# to control long generation with only max_length without
															# fancy calculation, dismissing tests for now.
															return
									# We don't care about infinite range models.
									# They already work.
									# Skip this test for XGLM, since it uses sinusoidal positional embeddings which are resized on-the-fly.
									snake_case   :			str						    = ["RwkvForCausalLM", "XGLMForCausalLM", "GPTNeoXForCausalLM"]
									if (
									    tokenizer.model_max_length < 10000
									    and text_generator.model.__class__.__name__ not in EXTRA_MODELS_CAN_HANDLE_LONG_INPUTS
									):
															# Handling of large generations
															with self.assertRaises((RuntimeError, IndexError, ValueError, AssertionError)		):
																					text_generator("This is a test" * 500   ,	max_new_tokens=20		)

															snake_case   :			Tuple						    = text_generator("This is a test" * 500   ,	handle_long_generation="hole"   ,	max_new_tokens=20		)
															# Hole strategy cannot work
															with self.assertRaises(_lowercase		):
																					text_generator(
																					    "This is a test" * 500   ,	handle_long_generation="hole"   ,	max_new_tokens=tokenizer.model_max_length + 10   ,	)



			@require_torch
			@require_accelerate
			@require_torch_gpu
			def 				__lowercase					(			self							:				Any		)							->					str:
									import torch

									# Classic `model_kwargs`
									snake_case   :			List[str]						    = pipeline(
									    model="hf-internal-testing/tiny-random-bloom"   ,	model_kwargs={"device_map": "auto", "torch_dtype": torch.bfloataa}   ,	)
									self.assertEqual(pipe.model.device   ,	torch.device(0		)		)
									self.assertEqual(pipe.model.lm_head.weight.dtype   ,	torch.bfloataa		)
									snake_case   :			Tuple						    = pipe("This is a test"		)
									self.assertEqual(
									    _lowercase   ,	[
									        {
									            "generated_text": (
									                "This is a test test test test test test test test test test test test test test test test"
									                " test"
									            )
									        }
									    ]   ,	)

									# Upgraded those two to real pipeline arguments (they just get sent for the model as they're unlikely to mean anything else.)
									snake_case   :			int						    = pipeline(model="hf-internal-testing/tiny-random-bloom"   ,	device_map="auto"   ,	torch_dtype=torch.bfloataa		)
									self.assertEqual(pipe.model.device   ,	torch.device(0		)		)
									self.assertEqual(pipe.model.lm_head.weight.dtype   ,	torch.bfloataa		)
									snake_case   :			Union[str, Any]						    = pipe("This is a test"		)
									self.assertEqual(
									    _lowercase   ,	[
									        {
									            "generated_text": (
									                "This is a test test test test test test test test test test test test test test test test"
									                " test"
									            )
									        }
									    ]   ,	)

									# torch_dtype will be automatically set to float32 if not provided - check: https://github.com/huggingface/transformers/pull/20602
									snake_case   :			Optional[int]						    = pipeline(model="hf-internal-testing/tiny-random-bloom"   ,	device_map="auto"		)
									self.assertEqual(pipe.model.device   ,	torch.device(0		)		)
									self.assertEqual(pipe.model.lm_head.weight.dtype   ,	torch.floataa		)
									snake_case   :			List[Any]						    = pipe("This is a test"		)
									self.assertEqual(
									    _lowercase   ,	[
									        {
									            "generated_text": (
									                "This is a test test test test test test test test test test test test test test test test"
									                " test"
									            )
									        }
									    ]   ,	)



			@require_torch
			@require_torch_gpu
			def 				__lowercase					(			self							:				List[str]		)							->					Union[str, Any]:
									import torch

									snake_case   :			Union[str, Any]						    = pipeline(model="hf-internal-testing/tiny-random-bloom"   ,	device=0   ,	torch_dtype=torch.floataa		)
									pipe("This is a test"		)



			@require_torch
			@require_accelerate
			@require_torch_gpu
			def 				__lowercase					(			self							:				Optional[Any]		)							->					Any:
									import torch

									snake_case   :			Optional[int]						    = pipeline(model="hf-internal-testing/tiny-random-bloom"   ,	device_map="auto"   ,	torch_dtype=torch.floataa		)
									pipe("This is a test"   ,	do_sample=_lowercase   ,	top_p=0.5		)



			def 				__lowercase					(			self							:				Optional[Any]		)							->					Union[str, Any]:
									snake_case   :			Optional[int]						    = "Hello world"
									snake_case   :			List[Any]						    = pipeline("text-generation"   ,	model="hf-internal-testing/tiny-random-gpt2"		)
									if text_generator.model.framework == "tf":
															snake_case   :			List[str]						    = logging.get_logger("transformers.generation.tf_utils"		)
									else:
															snake_case   :			Tuple						    = logging.get_logger("transformers.generation.utils"		)
									snake_case   :			Optional[Any]						    = "Both `max_new_tokens`"  # The beggining of the message to be checked in this test

									# Both are set by the user -> log warning
									with CaptureLogger(_lowercase		) as cl:
															snake_case   :			List[str]						    = text_generator(_lowercase   ,	max_length=10   ,	max_new_tokens=1		)
									self.assertIn(_lowercase   ,	cl.out		)

									# The user only sets one -> no warning
									with CaptureLogger(_lowercase		) as cl:
															snake_case   :			Optional[int]						    = text_generator(_lowercase   ,	max_new_tokens=1		)
									self.assertNotIn(_lowercase   ,	cl.out		)

									with CaptureLogger(_lowercase		) as cl:
															snake_case   :			Union[str, Any]						    = text_generator(_lowercase   ,	max_length=10		)
									self.assertNotIn(_lowercase   ,	cl.out		)








"""simple docstring"""







import importlib
import shutil
import threading
import warnings
from typing import List

import fsspec
import fsspec.asyn

from . import compression
from .hffilesystem import HfFileSystem


A		=						importlib.util.find_spec('s3fs') is not None

if _has_safs:
			from .safilesystem import SaFileSystem  # noqa: F401

A		=						[
    compression.BzaFileSystem,
    compression.GzipFileSystem,
    compression.LzaFileSystem,
    compression.XzFileSystem,
    compression.ZstdFileSystem,
]

# Register custom filesystems
for fs_class in COMPRESSION_FILESYSTEMS + [HfFileSystem]:
			if fs_class.protocol in fsspec.registry and fsspec.registry[fs_class.protocol] is not fs_class:
						warnings.warn(F'''A filesystem protocol was already set for {fs_class.protocol} and will be overwritten.''')
			fsspec.register_implementation(fs_class.protocol, fs_class, clobber=True)

def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							str ):
						"""simple docstring"""






						if "://" in dataset_path:
												snake_case   :			List[str]						    = dataset_path.split("://" )[1]
						return dataset_path

def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							fsspec.AbstractFileSystem ):
						"""simple docstring"""






						if fs is not None and fs.protocol != "file":
												return True
						else:
												return False

def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							fsspec.AbstractFileSystem     ,		lowerCamelCase_:							str     ,		lowerCamelCase_:							str ):
						"""simple docstring"""






						snake_case   :			int						    = not is_remote_filesystem(lowerCamelCase_ )
						if is_local:
												# LocalFileSystem.mv does copy + rm, it is more efficient to simply move a local directory
												shutil.move(fs._strip_protocol(lowerCamelCase_ )     ,		fs._strip_protocol(lowerCamelCase_ ) )
						else:
												fs.mv(lowerCamelCase_     ,		lowerCamelCase_     ,		recursive=lowerCamelCase_ )

def  __SCREAMING_SNAKE_CASE		(			):
						"""simple docstring"""






						if hasattr(fsspec.asyn     ,		"reset_lock" ):
												# for future fsspec>2022.05.0
												fsspec.asyn.reset_lock()
						else:
												snake_case   :			Union[str, Any]						    = None
												snake_case   :			Optional[int]						    = None
												snake_case   :			List[str]						    = threading.Lock()



