



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        snake_case__    (unittest.TestCase		):







						"""simple docstring"""




						__lowerCAmelCase					:Tuple      =					MODEL_FOR_CAUSAL_LM_MAPPING
						__lowerCAmelCase					:Tuple      =					TF_MODEL_FOR_CAUSAL_LM_MAPPING

						@require_torch
						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				List[str]:
									"""simple docstring"""




									a__       :						Dict							=  pipeline(task="""text-generation""" ,       model="""sshleifer/tiny-ctrl""" ,       framework="""pt"""  )
									# Using `do_sample=False` to force deterministic output
									a__       :						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@@"""
									            )
									        }
									    ] ,       )

									a__       :						Any							=  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@@"""
									                )
									            }
									        ],
									    ] ,       )

									a__       :						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  )},
									    ] ,       )
									a__       :						List[str]							=  text_generator.model.config.eos_token_id
									a__       :						Tuple							=  """<pad>"""
									a__       :						Optional[int]							=  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 							SCREAMING_SNAKE_CASE__(       self  ) ->				Dict:
									"""simple docstring"""




									a__       :						Tuple							=  pipeline(task="""text-generation""" ,       model="""sshleifer/tiny-ctrl""" ,       framework="""tf"""  )

									# Using `do_sample=False` to force deterministic output
									a__       :						Dict							=  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,"""
									            )
									        }
									    ] ,       )

									a__       :						Optional[int]							=  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 							SCREAMING_SNAKE_CASE__(       self ,       __lowercase ,       __lowercase ,       __lowercase  ) ->				str:
									"""simple docstring"""




									a__       :						int							=  TextGenerationPipeline(model=__lowercase ,       tokenizer=__lowercase  )
									return text_generator, ["This is a test", "Another test"]

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Any:
									"""simple docstring"""




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

									a__       :						int							=  text_generator(__lowercase ,       stop_sequence=""" fe"""  )
									self.assertEqual(__lowercase ,       [{"""generated_text""": """Hello I believe in fe"""}]  )

						def 							SCREAMING_SNAKE_CASE__(       self ,       __lowercase ,       __lowercase  ) ->				Dict:
									"""simple docstring"""




									a__       :						Optional[Any]							=  text_generator.model
									a__       :						Tuple							=  text_generator.tokenizer

									a__       :						Optional[int]							=  text_generator("""This is a test"""  )
									self.assertEqual(__lowercase ,       [{"""generated_text""": ANY(__lowercase  )}]  )
									self.assertTrue(outputs[0]["""generated_text"""].startswith("""This is a test"""  )  )

									a__       :						List[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"""]  )

									a__       :						Union[str, Any]							=  pipeline(task="""text-generation""" ,       model=__lowercase ,       tokenizer=__lowercase ,       return_full_text=__lowercase  )
									a__       :						Optional[int]							=  text_generator("""This is a test"""  )
									self.assertEqual(__lowercase ,       [{"""generated_text""": ANY(__lowercase  )}]  )
									self.assertNotIn("""This is a test""" ,       outputs[0]["""generated_text"""]  )

									a__       :						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"""  )  )

									a__       :						List[str]							=  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:
												a__       :						int							=  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  ):
												a__       :						Union[str, Any]							=  text_generator("""test""" ,       return_full_text=__lowercase ,       return_text=__lowercase  )
									with self.assertRaises(__lowercase  ):
												a__       :						Union[str, Any]							=  text_generator("""test""" ,       return_full_text=__lowercase ,       return_tensors=__lowercase  )
									with self.assertRaises(__lowercase  ):
												a__       :						Optional[Any]							=  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__
									):
												a__       :						List[Any]							=  text_generator(""""""  )
												self.assertEqual(__lowercase ,       [{"""generated_text""": ANY(__lowercase  )}]  )
									else:
												with self.assertRaises((ValueError, AssertionError)  ):
															a__       :						List[str]							=  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.
									a__       :						str							=  ["""RwkvForCausalLM""", """XGLMForCausalLM""", """GPTNeoXForCausalLM"""]
									if (
									    tokenizer.model_max_length < 1_0_0_0_0
									    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""" * 5_0_0 ,       max_new_tokens=2_0  )

												a__       :						Optional[Any]							=  text_generator("""This is a test""" * 5_0_0 ,       handle_long_generation="""hole""" ,       max_new_tokens=2_0  )
												# Hole strategy cannot work
												with self.assertRaises(__lowercase  ):
															text_generator(
															    """This is a test""" * 5_0_0 ,       handle_long_generation="""hole""" ,       max_new_tokens=tokenizer.model_max_length + 1_0 ,       )

						@require_torch
						@require_accelerate
						@require_torch_gpu
						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				int:
									"""simple docstring"""




									import torch

									# Classic `model_kwargs`
									a__       :						Optional[int]							=  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  )
									a__       :						Optional[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"""
									            )
									        }
									    ] ,       )

									# Upgraded those two to real pipeline arguments (they just get sent for the model as they're unlikely to mean anything else.)
									a__       :						List[str]							=  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  )
									a__       :						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"""
									            )
									        }
									    ] ,       )

									# torch_dtype will be automatically set to float32 if not provided - check: https://github.com/huggingface/transformers/pull/20602
									a__       :						Union[str, Any]							=  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  )
									a__       :						Optional[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 							SCREAMING_SNAKE_CASE__(       self  ) ->				Optional[Any]:
									"""simple docstring"""




									import torch

									a__       :						str							=  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 							SCREAMING_SNAKE_CASE__(       self  ) ->				str:
									"""simple docstring"""




									import torch

									a__       :						Union[str, Any]							=  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 							SCREAMING_SNAKE_CASE__(       self  ) ->				int:
									"""simple docstring"""




									a__       :						Optional[int]							=  """Hello world"""
									a__       :						Optional[int]							=  pipeline("""text-generation""" ,       model="""hf-internal-testing/tiny-random-gpt2"""  )
									if text_generator.model.framework == "tf":
												a__       :						Optional[int]							=  logging.get_logger("""transformers.generation.tf_utils"""  )
									else:
												a__       :						Dict							=  logging.get_logger("""transformers.generation.utils"""  )
									a__       :						Union[str, 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:
												a__       :						List[str]							=  text_generator(__lowercase ,       max_length=1_0 ,       max_new_tokens=1  )
									self.assertIn(__lowercase ,       cl.out  )

									# The user only sets one -> no warning
									with CaptureLogger(__lowercase  ) as cl:
												a__       :						int							=  text_generator(__lowercase ,       max_new_tokens=1  )
									self.assertNotIn(__lowercase ,       cl.out  )

									with CaptureLogger(__lowercase  ) as cl:
												a__       :						List[Any]							=  text_generator(__lowercase ,       max_length=1_0  )
									self.assertNotIn(__lowercase ,       cl.out  )









import asyncio
import os
import shutil
import subprocess
import sys
import tempfile
import unittest
from distutils.util import strtobool
from functools import partial
from pathlib import Path
from typing import List, Union
from unittest import mock

import torch

from ..state import AcceleratorState, PartialState
from ..utils import (
    gather,
    is_bnb_available,
    is_comet_ml_available,
    is_datasets_available,
    is_deepspeed_available,
    is_mps_available,
    is_safetensors_available,
    is_tensorboard_available,
    is_torch_version,
    is_tpu_available,
    is_transformers_available,
    is_wandb_available,
    is_xpu_available,
)

def      lowerCAmelCase_					(  _lowercase    :       Union[str, Any]						,       _lowercase    :       Dict=False)							->	Any:







			"""simple docstring"""
			try:
						a__       :						str							=  os.environ[key]
			except KeyError:
						# KEY isn't set, default to `default`.
						a__       :						Optional[int]							=  default
			else:
						# KEY is set, convert it to True or False.
						try:
									a__       :						Optional[int]							=  strtobool(_lowercase)
						except ValueError:
									# More values are supported, but let's keep the message simple.
									raise ValueError(F'''If set, {key} must be yes or no.''')
			return _value


_lowercase		:				Dict      		=parse_flag_from_env("RUN_SLOW", default=False)

def      lowerCAmelCase_					(  _lowercase    :       Any)							->	str:







			"""simple docstring"""
			return unittest.skip("""Test was skipped""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       str)							->	List[str]:







			"""simple docstring"""
			return unittest.skipUnless(_run_slow_tests						,       """test is slow""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       List[Any])							->	Union[str, Any]:







			"""simple docstring"""
			return unittest.skipUnless(not torch.cuda.is_available()						,       """test requires only a CPU""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       List[Any])							->	List[Any]:







			"""simple docstring"""
			return unittest.skipUnless(torch.cuda.is_available()						,       """test requires a GPU""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       Optional[int])							->	Dict:







			"""simple docstring"""
			return unittest.skipUnless(is_xpu_available()						,       """test requires a XPU""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       Tuple)							->	List[str]:







			"""simple docstring"""
			return unittest.skipUnless(is_mps_available()						,       """test requires a `mps` backend support in `torch`""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       Dict)							->	Union[str, Any]:







			"""simple docstring"""
			return unittest.skipUnless(
			    is_transformers_available() and is_datasets_available()						,       """test requires the Hugging Face suite""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       Tuple)							->	Any:







			"""simple docstring"""
			return unittest.skipUnless(is_bnb_available()						,       """test requires the bitsandbytes library""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       Union[str, Any])							->	List[str]:







			"""simple docstring"""
			return unittest.skipUnless(is_tpu_available()						,       """test requires TPU""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       str)							->	int:







			"""simple docstring"""
			return unittest.skipUnless(torch.cuda.device_count() == 1						,       """test requires a GPU""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       Any)							->	List[str]:







			"""simple docstring"""
			return unittest.skipUnless(torch.xpu.device_count() == 1						,       """test requires a XPU""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       Union[str, Any])							->	Union[str, Any]:







			"""simple docstring"""
			return unittest.skipUnless(torch.cuda.device_count() > 1						,       """test requires multiple GPUs""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       int)							->	Tuple:







			"""simple docstring"""
			return unittest.skipUnless(torch.xpu.device_count() > 1						,       """test requires multiple XPUs""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       int)							->	Optional[Any]:







			"""simple docstring"""
			return unittest.skipUnless(is_safetensors_available()						,       """test requires safetensors""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       Optional[int])							->	List[str]:







			"""simple docstring"""
			return unittest.skipUnless(is_deepspeed_available()						,       """test requires DeepSpeed""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       Union[str, Any])							->	Optional[Any]:







			"""simple docstring"""
			return unittest.skipUnless(is_torch_version(""">="""						,       """1.12.0""")						,       """test requires torch version >= 1.12.0""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       Any=None						,       _lowercase    :       List[str]=None)							->	Dict:







			"""simple docstring"""
			if test_case is None:
						return partial(_lowercase						,       version=_lowercase)
			return unittest.skipUnless(is_torch_version(""">="""						,       _lowercase)						,       F'''test requires torch version >= {version}''')(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       Any)							->	int:







			"""simple docstring"""
			return unittest.skipUnless(is_tensorboard_available()						,       """test requires Tensorboard""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       str)							->	Union[str, Any]:







			"""simple docstring"""
			return unittest.skipUnless(is_wandb_available()						,       """test requires wandb""")(_lowercase)

def      lowerCAmelCase_					(  _lowercase    :       Optional[int])							->	Union[str, Any]:







			"""simple docstring"""
			return unittest.skipUnless(is_comet_ml_available()						,       """test requires comet_ml""")(_lowercase)


_lowercase		:				List[str]      		=(
    any([is_wandb_available(), is_tensorboard_available()]) and not is_comet_ml_available()
)

def      lowerCAmelCase_					(  _lowercase    :       Optional[int])							->	Union[str, Any]:







			"""simple docstring"""
			return unittest.skipUnless(
			    _atleast_one_tracker_available						,       """test requires at least one tracker to be available and for `comet_ml` to not be installed"""						,       )(_lowercase)

class        snake_case__    (unittest.TestCase		):







						"""simple docstring"""




						__lowerCAmelCase					:Optional[Any]      =					True

						@classmethod
						def 							SCREAMING_SNAKE_CASE__(       cls  ) ->				Optional[int]:
									"""simple docstring"""




									a__       :						Tuple							=  tempfile.mkdtemp()

						@classmethod
						def 							SCREAMING_SNAKE_CASE__(       cls  ) ->				Dict:
									"""simple docstring"""




									if os.path.exists(cls.tmpdir  ):
												shutil.rmtree(cls.tmpdir  )

						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				List[Any]:
									"""simple docstring"""




									if self.clear_on_setup:
												for path in Path(self.tmpdir  ).glob("""**/*"""  ):
															if path.is_file():
																		path.unlink()
															elif path.is_dir():
																		shutil.rmtree(__lowercase  )




class        snake_case__    (unittest.TestCase		):







						"""simple docstring"""




						def 							SCREAMING_SNAKE_CASE__(       self  ) ->				Union[str, Any]:
									"""simple docstring"""




									super().tearDown()
									# Reset the state of the AcceleratorState singleton.
									AcceleratorState._reset_state()
									PartialState._reset_state()




class        snake_case__    (unittest.TestCase		):







						"""simple docstring"""




						def 							SCREAMING_SNAKE_CASE__(       self ,       __lowercase  ) ->				Union[str, Any]:
									"""simple docstring"""




									a__       :						Tuple							=  mocks if isinstance(__lowercase ,       (tuple, list)  ) else [mocks]
									for m in self.mocks:
												m.start()
												self.addCleanup(m.stop  )




def      lowerCAmelCase_					(  _lowercase    :       Optional[int])							->	List[Any]:







			"""simple docstring"""
			a__       :						Tuple							=  AcceleratorState()
			a__       :						List[str]							=  tensor[None].clone().to(state.device)
			a__       :						Any							=  gather(_lowercase).cpu()
			a__       :						Optional[Any]							=  tensor[0].cpu()
			for i in range(tensors.shape[0]):
						if not torch.equal(tensors[i]						,       _lowercase):
									return False
			return True

class        snake_case__    :







						"""simple docstring"""




						def __init__(       self ,       __lowercase ,       __lowercase ,       __lowercase  ) ->				Any:
									"""simple docstring"""




									a__       :						Any							=  returncode
									a__       :						List[Any]							=  stdout
									a__       :						Any							=  stderr




async def      lowerCAmelCase_					(  _lowercase    :       str						,       _lowercase    :       str)							->	List[Any]:







			"""simple docstring"""
			while True:
						a__       :						str							=  await stream.readline()
						if line:
									callback(_lowercase)
						else:
									break

async def      lowerCAmelCase_					(  _lowercase    :       Any						,       _lowercase    :       Union[str, Any]=None						,       _lowercase    :       List[str]=None						,       _lowercase    :       Tuple=None						,       _lowercase    :       Optional[Any]=False						,       _lowercase    :       Dict=False)							->	_RunOutput:







			"""simple docstring"""
			if echo:
						print("""\nRunning: """						,       """ """.join(_lowercase))

			a__       :						int							=  await asyncio.create_subprocess_exec(
			    cmd[0]						,       *cmd[1:]						,       stdin=_lowercase						,       stdout=asyncio.subprocess.PIPE						,       stderr=asyncio.subprocess.PIPE						,       env=_lowercase						,       )

			# note: there is a warning for a possible deadlock when using `wait` with huge amounts of data in the pipe
			# https://docs.python.org/3/library/asyncio-subprocess.html#asyncio.asyncio.subprocess.Process.wait
			#
			# If it starts hanging, will need to switch to the following code. The problem is that no data
			# will be seen until it's done and if it hangs for example there will be no debug info.
			# out, err = await p.communicate()
			# return _RunOutput(p.returncode, out, err)

			a__       :						int							=  []
			a__       :						Optional[int]							=  []

			def tee(_lowercase    :       List[str]						,       _lowercase    :       Optional[int]						,       _lowercase    :       Any						,       _lowercase    :       Optional[Any]=""):
						a__       :						int							=  line.decode("""utf-8""").rstrip()
						sink.append(_lowercase)
						if not quiet:
									print(_lowercase						,       _lowercase						,       file=_lowercase)

    # XXX: the timeout doesn't seem to make any difference here
			await asyncio.wait(
			    [
			        asyncio.create_task(_read_stream(p.stdout						,       lambda _lowercase: tee(_lowercase						,       _lowercase						,       sys.stdout						,       label="""stdout:"""))),
			        asyncio.create_task(_read_stream(p.stderr						,       lambda _lowercase: tee(_lowercase						,       _lowercase						,       sys.stderr						,       label="""stderr:"""))),
			    ]						,       timeout=_lowercase						,       )
			return _RunOutput(await p.wait()						,       _lowercase						,       _lowercase)

def      lowerCAmelCase_					(  _lowercase    :       Optional[int]						,       _lowercase    :       Optional[int]=None						,       _lowercase    :       Tuple=None						,       _lowercase    :       Any=180						,       _lowercase    :       List[Any]=False						,       _lowercase    :       Dict=True)							->	_RunOutput:







			"""simple docstring"""
			a__       :						Any							=  asyncio.get_event_loop()
			a__       :						List[Any]							=  loop.run_until_complete(
			    _stream_subprocess(_lowercase						,       env=_lowercase						,       stdin=_lowercase						,       timeout=_lowercase						,       quiet=_lowercase						,       echo=_lowercase))

			a__       :						Optional[int]							=  """ """.join(_lowercase)
			if result.returncode > 0:
						a__       :						List[Any]							=  """\n""".join(result.stderr)
						raise RuntimeError(
						    F'''\'{cmd_str}\' failed with returncode {result.returncode}\n\n'''
						    F'''The combined stderr from workers follows:\n{stderr}''')

			return result

class        snake_case__    (A__		):







						"""simple docstring"""




						pass




def      lowerCAmelCase_					(  _lowercase    :       List[str]						,       _lowercase    :       Optional[int]=False)							->	Dict:







			"""simple docstring"""
			try:
						a__       :						List[Any]							=  subprocess.check_output(_lowercase						,       stderr=subprocess.STDOUT)
						if return_stdout:
									if hasattr(_lowercase						,       """decode"""):
												a__       :						Tuple							=  output.decode("""utf-8""")
									return output
			except subprocess.CalledProcessError as e:
						raise SubprocessCallException(
						    F'''Command `{' '.join(_lowercase)}` failed with the following error:\n\n{e.output.decode()}''') from e




