

'''simple docstring'''


import tempfile
import unittest

from transformers import SPIECE_UNDERLINE, BatchEncoding, PLBartTokenizer, is_torch_available
from transformers.testing_utils import (
    get_tests_dir,
    nested_simplify,
    require_sentencepiece,
    require_tokenizers,
    require_torch,
)

from ...test_tokenization_common import TokenizerTesterMixin


lowerCAmelCase__						:       int										=						get_tests_dir("""fixtures/test_sentencepiece.model""")


if is_torch_available():
		from transformers.models.plbart.modeling_plbart import shift_tokens_right

lowerCAmelCase__						:       Optional[Any]										=						50003
lowerCAmelCase__						:       Optional[Any]										=						50002




@require_sentencepiece
@require_tokenizers
class     a			(				SCREAMING_SNAKE_CASE			,			unittest.TestCase       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    PLBartTokenizer
						__UpperCAmelCase			     =    None
						__UpperCAmelCase			     =    False







						def       __magic_name__							(							self					:      Tuple  ):



								'''simple docstring'''



								super().setUp()

								# We have a SentencePiece fixture for testing
								snake_case__						:       int       						=  PLBartTokenizer(snake_case_			,			language_codes='''base'''			,			keep_accents=snake_case_  )
								tokenizer.save_pretrained(self.tmpdirname  )







						def       __magic_name__							(							self					:      Dict  ):



								'''simple docstring'''



								snake_case__						:       Optional[Any]       						=  PLBartTokenizer(snake_case_			,			language_codes='''base'''			,			keep_accents=snake_case_  )

								snake_case__						:       Tuple       						=  tokenizer.tokenize('''This is a test'''  )
								self.assertListEqual(snake_case_			,			['''▁This''', '''▁is''', '''▁a''', '''▁t''', '''est''']  )

								self.assertListEqual(
								    tokenizer.convert_tokens_to_ids(snake_case_  )			,			[value + tokenizer.fairseq_offset for value in [2_8_5, 4_6, 1_0, 1_7_0, 3_8_2]]			,			)

								snake_case__						:       Union[str, Any]       						=  tokenizer.tokenize('''I was born in 92000, and this is falsé.'''  )
								self.assertListEqual(
								    snake_case_			,			[
								        SPIECE_UNDERLINE + '''I''',
								        SPIECE_UNDERLINE + '''was''',
								        SPIECE_UNDERLINE + '''b''',
								        '''or''',
								        '''n''',
								        SPIECE_UNDERLINE + '''in''',
								        SPIECE_UNDERLINE + '''''',
								        '''9''',
								        '''2''',
								        '''0''',
								        '''0''',
								        '''0''',
								        ''',''',
								        SPIECE_UNDERLINE + '''and''',
								        SPIECE_UNDERLINE + '''this''',
								        SPIECE_UNDERLINE + '''is''',
								        SPIECE_UNDERLINE + '''f''',
								        '''al''',
								        '''s''',
								        '''é''',
								        '''.''',
								    ]			,			)
								snake_case__						:       Tuple       						=  tokenizer.convert_tokens_to_ids(snake_case_  )
								self.assertListEqual(
								    snake_case_			,			[
								        value + tokenizer.fairseq_offset
								        for value in [8, 2_1, 8_4, 5_5, 2_4, 1_9, 7, 2, 6_0_2, 3_4_7, 3_4_7, 3_4_7, 3, 1_2, 6_6, 4_6, 7_2, 8_0, 6, 2, 4]
								    ]			,			)

								snake_case__						:       Optional[int]       						=  tokenizer.convert_ids_to_tokens(snake_case_  )
								self.assertListEqual(
								    snake_case_			,			[
								        SPIECE_UNDERLINE + '''I''',
								        SPIECE_UNDERLINE + '''was''',
								        SPIECE_UNDERLINE + '''b''',
								        '''or''',
								        '''n''',
								        SPIECE_UNDERLINE + '''in''',
								        SPIECE_UNDERLINE + '''''',
								        '''<unk>''',
								        '''2''',
								        '''0''',
								        '''0''',
								        '''0''',
								        ''',''',
								        SPIECE_UNDERLINE + '''and''',
								        SPIECE_UNDERLINE + '''this''',
								        SPIECE_UNDERLINE + '''is''',
								        SPIECE_UNDERLINE + '''f''',
								        '''al''',
								        '''s''',
								        '''<unk>''',
								        '''.''',
								    ]			,			)

								snake_case__						:       int       						=  tokenizer.vocab_size
								snake_case__						:       Optional[int]       						=  [tokenizer.convert_ids_to_tokens(snake_case_  ) for x in range(end - 4			,			snake_case_  )]

								self.assertListEqual(snake_case_			,			['''__java__''', '''__python__''', '''__en_XX__''', '''<mask>''']  )

								snake_case__						:       int       						=  '''java.lang.Exception, python.lang.Exception, javascript, php, ruby, go'''
								snake_case__						:       Any       						=  tokenizer(snake_case_  ).input_ids
								self.assertEqual(
								    tokenizer.decode(snake_case_			,			skip_special_tokens=snake_case_			,			clean_up_tokenization_spaces=snake_case_  )			,			snake_case_			,			)







						def       __magic_name__							(							self					:      Tuple  ):



								'''simple docstring'''



								snake_case__						:       Optional[Any]       						=  PLBartTokenizer(snake_case_			,			language_codes='''multi'''			,			keep_accents=snake_case_  )

								snake_case__						:       Dict       						=  tokenizer.tokenize('''This is a test'''  )
								self.assertListEqual(snake_case_			,			['''▁This''', '''▁is''', '''▁a''', '''▁t''', '''est''']  )

								self.assertListEqual(
								    tokenizer.convert_tokens_to_ids(snake_case_  )			,			[value + tokenizer.fairseq_offset for value in [2_8_5, 4_6, 1_0, 1_7_0, 3_8_2]]			,			)

								snake_case__						:       Union[str, Any]       						=  tokenizer.tokenize('''I was born in 92000, and this is falsé.'''  )
								self.assertListEqual(
								    snake_case_			,			[
								        SPIECE_UNDERLINE + '''I''',
								        SPIECE_UNDERLINE + '''was''',
								        SPIECE_UNDERLINE + '''b''',
								        '''or''',
								        '''n''',
								        SPIECE_UNDERLINE + '''in''',
								        SPIECE_UNDERLINE + '''''',
								        '''9''',
								        '''2''',
								        '''0''',
								        '''0''',
								        '''0''',
								        ''',''',
								        SPIECE_UNDERLINE + '''and''',
								        SPIECE_UNDERLINE + '''this''',
								        SPIECE_UNDERLINE + '''is''',
								        SPIECE_UNDERLINE + '''f''',
								        '''al''',
								        '''s''',
								        '''é''',
								        '''.''',
								    ]			,			)
								snake_case__						:       Optional[int]       						=  tokenizer.convert_tokens_to_ids(snake_case_  )
								self.assertListEqual(
								    snake_case_			,			[
								        value + tokenizer.fairseq_offset
								        for value in [8, 2_1, 8_4, 5_5, 2_4, 1_9, 7, 2, 6_0_2, 3_4_7, 3_4_7, 3_4_7, 3, 1_2, 6_6, 4_6, 7_2, 8_0, 6, 2, 4]
								    ]			,			)

								snake_case__						:       Optional[Any]       						=  tokenizer.convert_ids_to_tokens(snake_case_  )
								self.assertListEqual(
								    snake_case_			,			[
								        SPIECE_UNDERLINE + '''I''',
								        SPIECE_UNDERLINE + '''was''',
								        SPIECE_UNDERLINE + '''b''',
								        '''or''',
								        '''n''',
								        SPIECE_UNDERLINE + '''in''',
								        SPIECE_UNDERLINE + '''''',
								        '''<unk>''',
								        '''2''',
								        '''0''',
								        '''0''',
								        '''0''',
								        ''',''',
								        SPIECE_UNDERLINE + '''and''',
								        SPIECE_UNDERLINE + '''this''',
								        SPIECE_UNDERLINE + '''is''',
								        SPIECE_UNDERLINE + '''f''',
								        '''al''',
								        '''s''',
								        '''<unk>''',
								        '''.''',
								    ]			,			)
								snake_case__						:       List[Any]       						=  tokenizer.vocab_size
								snake_case__						:       Dict       						=  [tokenizer.convert_ids_to_tokens(snake_case_  ) for x in range(end - 7			,			snake_case_  )]

								self.assertListEqual(
								    snake_case_			,			['''__java__''', '''__python__''', '''__en_XX__''', '''__javascript__''', '''__php__''', '''__ruby__''', '''__go__''']  )
								snake_case__						:       List[Any]       						=  '''java.lang.Exception, python.lang.Exception, javascript, php, ruby, go'''
								snake_case__						:       int       						=  tokenizer(snake_case_  ).input_ids
								self.assertEqual(
								    tokenizer.decode(snake_case_			,			skip_special_tokens=snake_case_			,			clean_up_tokenization_spaces=snake_case_  )			,			snake_case_			,			)





@require_torch
@require_sentencepiece
@require_tokenizers
class     a			(				unittest.TestCase       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    """uclanlp/plbart-python-en_XX"""
						__UpperCAmelCase			     =    [
						    """def maximum(a,b,c):NEW_LINE_INDENTreturn max([a,b,c])""",
						    """def sum(a,b,c):NEW_LINE_INDENTreturn sum([a,b,c])""",
						]
						__UpperCAmelCase			     =    [
						    """Returns the maximum value of a b c.""",
						    """Sums the values of a b c.""",
						]
						__UpperCAmelCase			     =    [
						    134,
						    5452,
						    3_3460,
						    3_3441,
						    3_3463,
						    3_3465,
						    3_3463,
						    3_3449,
						    988,
						    20,
						    3_3456,
						    19,
						    3_3456,
						    771,
						    39,
						    4258,
						    889,
						    3318,
						    3_3441,
						    3_3463,
						    3_3465,
						    3_3463,
						    3_3449,
						    2471,
						    2,
						    PYTHON_CODE,
						]







						@classmethod
						def       __magic_name__							(							cls					:      List[str]  ):



								'''simple docstring'''



								snake_case__						:       PLBartTokenizer       						=  PLBartTokenizer.from_pretrained(
								    cls.checkpoint_name			,			language_codes='''base'''			,			src_lang='''python'''			,			tgt_lang='''en_XX'''  )
								snake_case__						:       Tuple       						=  1
								return cls







						def       __magic_name__							(							self					:      Union[str, Any]  ):



								'''simple docstring'''



								self.assertEqual(self.tokenizer.fairseq_tokens_to_ids['''__java__''']			,			5_0_0_0_1  )
								self.assertEqual(self.tokenizer.fairseq_tokens_to_ids['''__python__''']			,			5_0_0_0_2  )
								self.assertEqual(self.tokenizer.fairseq_tokens_to_ids['''__en_XX__''']			,			5_0_0_0_3  )







						def       __magic_name__							(							self					:      Optional[int]  ):



								'''simple docstring'''



								snake_case__						:       Optional[int]       						=  self.tokenizer.batch_encode_plus(self.src_text  ).input_ids[0]
								self.assertListEqual(self.expected_src_tokens			,			snake_case_  )







						def       __magic_name__							(							self					:      List[str]  ):



								'''simple docstring'''



								self.assertIn(snake_case_			,			self.tokenizer.all_special_ids  )
								snake_case__						:       str       						=  [EN_CODE, 9_0_3_7, 3_3_4_4_2, 5_7, 7_5_2, 1_5_3, 1_4, 5_6, 1_8, 9, 2]
								snake_case__						:       Optional[int]       						=  self.tokenizer.decode(snake_case_			,			skip_special_tokens=snake_case_  )
								snake_case__						:       Any       						=  self.tokenizer.decode(generated_ids[1:]			,			skip_special_tokens=snake_case_  )
								self.assertEqual(snake_case_			,			snake_case_  )
								self.assertNotIn(self.tokenizer.eos_token			,			snake_case_  )







						def       __magic_name__							(							self					:      Optional[int]  ):



								'''simple docstring'''



								snake_case__						:       str       						=  ['''def sum(a,b,c):NEW_LINE_INDENTreturn sum([a,b,c])''' * 2_0]
								self.assertIsInstance(src_text[0]			,			snake_case_  )
								snake_case__						:       int       						=  1_0
								snake_case__						:       int       						=  self.tokenizer(snake_case_			,			max_length=snake_case_			,			truncation=snake_case_  ).input_ids[0]
								self.assertEqual(ids[-2]			,			2  )
								self.assertEqual(ids[-1]			,			snake_case_  )
								self.assertEqual(len(snake_case_  )			,			snake_case_  )







						def       __magic_name__							(							self					:      Union[str, Any]  ):



								'''simple docstring'''



								self.assertListEqual(self.tokenizer.convert_tokens_to_ids(['''<mask>''', '''__java__''']  )			,			[5_0_0_0_4, 5_0_0_0_1]  )







						def       __magic_name__							(							self					:      Optional[int]  ):



								'''simple docstring'''



								snake_case__						:       List[Any]       						=  tempfile.mkdtemp()
								snake_case__						:       List[Any]       						=  self.tokenizer.fairseq_tokens_to_ids
								self.tokenizer.save_pretrained(snake_case_  )
								snake_case__						:       Any       						=  PLBartTokenizer.from_pretrained(snake_case_  )
								self.assertDictEqual(new_tok.fairseq_tokens_to_ids			,			snake_case_  )







						@require_torch
						def       __magic_name__							(							self					:      Optional[int]  ):



								'''simple docstring'''



								snake_case__						:       List[str]       						=  self.tokenizer(self.src_text			,			text_target=self.tgt_text			,			padding=snake_case_			,			return_tensors='''pt'''  )
								snake_case__						:       Any       						=  shift_tokens_right(batch['''labels''']			,			self.tokenizer.pad_token_id  )

								# fairseq batch: https://gist.github.com/sshleifer/cba08bc2109361a74ac3760a7e30e4f4
								self.assertEqual(batch.input_ids[1][-2:].tolist()			,			[2, PYTHON_CODE]  )
								self.assertEqual(batch.decoder_input_ids[1][0]			,			snake_case_  )
								self.assertEqual(batch.decoder_input_ids[1][-1]			,			2  )
								self.assertEqual(batch.labels[1][-2:].tolist()			,			[2, EN_CODE]  )







						@require_torch
						def       __magic_name__							(							self					:      str  ):



								'''simple docstring'''



								snake_case__						:       List[Any]       						=  self.tokenizer(
								    self.src_text			,			text_target=self.tgt_text			,			padding=snake_case_			,			truncation=snake_case_			,			max_length=len(self.expected_src_tokens  )			,			return_tensors='''pt'''			,			)
								snake_case__						:       Optional[int]       						=  shift_tokens_right(batch['''labels''']			,			self.tokenizer.pad_token_id  )

								self.assertIsInstance(snake_case_			,			snake_case_  )

								self.assertEqual((2, 2_6)			,			batch.input_ids.shape  )
								self.assertEqual((2, 2_6)			,			batch.attention_mask.shape  )
								snake_case__						:       Any       						=  batch.input_ids.tolist()[0]
								self.assertListEqual(self.expected_src_tokens			,			snake_case_  )
								self.assertEqual(2			,			batch.decoder_input_ids[0, -1]  )  # EOS
								# Test that special tokens are reset
								self.assertEqual(self.tokenizer.prefix_tokens			,			[]  )
								self.assertEqual(self.tokenizer.suffix_tokens			,			[self.tokenizer.eos_token_id, PYTHON_CODE]  )







						def       __magic_name__							(							self					:      Any  ):



								'''simple docstring'''



								snake_case__						:       Any       						=  self.tokenizer(self.src_text			,			padding=snake_case_			,			truncation=snake_case_			,			max_length=3			,			return_tensors='''pt'''  )
								snake_case__						:       Tuple       						=  self.tokenizer(
								    text_target=self.tgt_text			,			padding=snake_case_			,			truncation=snake_case_			,			max_length=1_0			,			return_tensors='''pt'''  )
								snake_case__						:       Dict       						=  targets['''input_ids''']
								snake_case__						:       Tuple       						=  shift_tokens_right(snake_case_			,			self.tokenizer.pad_token_id  )

								self.assertEqual(batch.input_ids.shape[1]			,			3  )
								self.assertEqual(batch.decoder_input_ids.shape[1]			,			1_0  )







						@require_torch
						def       __magic_name__							(							self					:      List[str]  ):



								'''simple docstring'''



								snake_case__						:       str       						=  self.tokenizer._build_translation_inputs(
								    '''A test'''			,			return_tensors='''pt'''			,			src_lang='''en_XX'''			,			tgt_lang='''java'''  )

								self.assertEqual(
								    nested_simplify(snake_case_  )			,			{
								        # A, test, EOS, en_XX
								        '''input_ids''': [[1_5_0, 2_4_2, 2, 5_0_0_0_3]],
								        '''attention_mask''': [[1, 1, 1, 1]],
								        # java
								        '''forced_bos_token_id''': 5_0_0_0_1,
								    }			,			)






'''simple docstring'''


from argparse import ArgumentParser, Namespace
from typing import Any, List, Optional

from ..pipelines import Pipeline, get_supported_tasks, pipeline
from ..utils import logging
from . import BaseTransformersCLICommand


try:
					from fastapi import Body, FastAPI, HTTPException
					from fastapi.routing import APIRoute
					from pydantic import BaseModel
					from starlette.responses import JSONResponse
					from uvicorn import run

					lowerCAmelCase__						:       Any										=						True
except (ImportError, AttributeError):
					lowerCAmelCase__						:       Dict										=						object




					def 	_a       (				*__lowerCAmelCase		:      List[str]   ,							**__lowerCAmelCase		:      Tuple     ):

							"""simple docstring"""

							pass





					lowerCAmelCase__						:       str										=						False


lowerCAmelCase__						:       List[str]										=						logging.get_logger("""transformers-cli/serving""")




def 	_a       (				__lowerCAmelCase		:      Namespace     ):

		"""simple docstring"""

		snake_case__						:       Union[str, Any]       						=  pipeline(
		    task=args.task   ,							model=args.model if args.model else None   ,							config=args.config   ,							tokenizer=args.tokenizer   ,							device=args.device   ,							)
		return ServeCommand(__lowerCAmelCase   ,							args.host   ,							args.port   ,							args.workers     )




class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    42





class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    42
						__UpperCAmelCase			     =    42





class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    42





class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    42





class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						@staticmethod
						def       __magic_name__							(							snake_case_					:      ArgumentParser  ):



								'''simple docstring'''



								snake_case__						:       Optional[Any]       						=  parser.add_parser(
								    '''serve'''			,			help='''CLI tool to run inference requests through REST and GraphQL endpoints.'''  )
								serve_parser.add_argument(
								    '''--task'''			,			type=snake_case_			,			choices=get_supported_tasks()			,			help='''The task to run the pipeline on'''			,			)
								serve_parser.add_argument('''--host'''			,			type=snake_case_			,			default='''localhost'''			,			help='''Interface the server will listen on.'''  )
								serve_parser.add_argument('''--port'''			,			type=snake_case_			,			default=8_8_8_8			,			help='''Port the serving will listen to.'''  )
								serve_parser.add_argument('''--workers'''			,			type=snake_case_			,			default=1			,			help='''Number of http workers'''  )
								serve_parser.add_argument('''--model'''			,			type=snake_case_			,			help='''Model\'s name or path to stored model.'''  )
								serve_parser.add_argument('''--config'''			,			type=snake_case_			,			help='''Model\'s config name or path to stored model.'''  )
								serve_parser.add_argument('''--tokenizer'''			,			type=snake_case_			,			help='''Tokenizer name to use.'''  )
								serve_parser.add_argument(
								    '''--device'''			,			type=snake_case_			,			default=-1			,			help='''Indicate the device to run onto, -1 indicates CPU, >= 0 indicates GPU (default: -1)'''			,			)
								serve_parser.set_defaults(func=snake_case_  )







						def __init__(							self					:      Union[str, Any]			,			snake_case_					:      Pipeline			,			snake_case_					:      str			,			snake_case_					:      int			,			snake_case_					:      int  ):



								'''simple docstring'''



								snake_case__						:       Any       						=  pipeline

								snake_case__						:       Tuple       						=  host
								snake_case__						:       Optional[Any]       						=  port
								snake_case__						:       Tuple       						=  workers

								if not _serve_dependencies_installed:
										raise RuntimeError(
										    '''Using serve command requires FastAPI and uvicorn. '''
										    '''Please install transformers with [serving]: pip install "transformers[serving]".'''
										    '''Or install FastAPI and uvicorn separately.'''  )
								else:
										logger.info(F"""Serving model over {host}:{port}"""  )
										snake_case__						:       str       						=  FastAPI(
										    routes=[
										        APIRoute(
										            '''/'''			,			self.model_info			,			response_model=snake_case_			,			response_class=snake_case_			,			methods=['''GET''']			,			),
										        APIRoute(
										            '''/tokenize'''			,			self.tokenize			,			response_model=snake_case_			,			response_class=snake_case_			,			methods=['''POST''']			,			),
										        APIRoute(
										            '''/detokenize'''			,			self.detokenize			,			response_model=snake_case_			,			response_class=snake_case_			,			methods=['''POST''']			,			),
										        APIRoute(
										            '''/forward'''			,			self.forward			,			response_model=snake_case_			,			response_class=snake_case_			,			methods=['''POST''']			,			),
										    ]			,			timeout=6_0_0			,			)







						def       __magic_name__							(							self					:      str  ):



								'''simple docstring'''



								run(self._app			,			host=self.host			,			port=self.port			,			workers=self.workers  )







						def       __magic_name__							(							self					:      Optional[Any]  ):



								'''simple docstring'''



								return ServeModelInfoResult(infos=vars(self._pipeline.model.config  )  )







						def       __magic_name__							(							self					:      List[str]			,			snake_case_					:      str = Body(snake_case_			,			embed=snake_case_  )			,			snake_case_					:      bool = Body(snake_case_			,			embed=snake_case_  )  ):



								'''simple docstring'''



								try:
										snake_case__						:       Optional[Any]       						=  self._pipeline.tokenizer.tokenize(snake_case_  )

										if return_ids:
												snake_case__						:       Optional[int]       						=  self._pipeline.tokenizer.convert_tokens_to_ids(snake_case_  )
												return ServeTokenizeResult(tokens=snake_case_			,			tokens_ids=snake_case_  )
										else:
												return ServeTokenizeResult(tokens=snake_case_  )

								except Exception as e:
										raise HTTPException(status_code=5_0_0			,			detail={'''model''': '''''', '''error''': str(snake_case_  )}  )







						def       __magic_name__							(							self					:      List[Any]			,			snake_case_					:      List[int] = Body(snake_case_			,			embed=snake_case_  )			,			snake_case_					:      bool = Body(snake_case_			,			embed=snake_case_  )			,			snake_case_					:      bool = Body(snake_case_			,			embed=snake_case_  )			,			):



								'''simple docstring'''



								try:
										snake_case__						:       Optional[int]       						=  self._pipeline.tokenizer.decode(snake_case_			,			snake_case_			,			snake_case_  )
										return ServeDeTokenizeResult(model=''''''			,			text=snake_case_  )
								except Exception as e:
										raise HTTPException(status_code=5_0_0			,			detail={'''model''': '''''', '''error''': str(snake_case_  )}  )







						async def       __magic_name__							(							self					:      Tuple			,			snake_case_					:      List[str]=Body(snake_case_			,			embed=snake_case_  )  ):



								'''simple docstring'''



								if len(snake_case_  ) == 0:
										return ServeForwardResult(output=[]			,			attention=[]  )

								try:
										# Forward through the model
										snake_case__						:       Tuple       						=  self._pipeline(snake_case_  )
										return ServeForwardResult(output=snake_case_  )
								except Exception as e:
										raise HTTPException(5_0_0			,			{'''error''': str(snake_case_  )}  )



