
"""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     lowercase(unittest.TestCase       ):
     __snake_case:	Union[str, Any]						=     MODEL_FOR_CAUSAL_LM_MAPPING
     __snake_case:	Union[str, Any]						=     TF_MODEL_FOR_CAUSAL_LM_MAPPING



     @require_torch
     def 			lowercase__  (							self   )		->     Union[str, Any]:







         """simple docstring"""



         a__		=			pipeline(task='text-generation'     ,				model='sshleifer/tiny-ctrl'     ,				framework='pt'   )
         # Using `do_sample=False` to force deterministic output
         a__		=			text_generator('This is a test'     ,				do_sample=__SCREAMING_SNAKE_CASE   )
         self.assertEqual(
             __SCREAMING_SNAKE_CASE     ,				[
                 {
                     'generated_text': (
                         'This is a test ☃ ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy oscope.'
                         ' oscope. FiliFili@@'
                     )
                 }
             ]     ,				)

         a__		=			text_generator(['This is a test', 'This is a second test']   )
         self.assertEqual(
             __SCREAMING_SNAKE_CASE     ,				[
                 [
                     {
                         '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__		=			text_generator('This is a test'     ,				do_sample=__SCREAMING_SNAKE_CASE     ,				num_return_sequences=2     ,				return_tensors=__SCREAMING_SNAKE_CASE   )
         self.assertEqual(
             __SCREAMING_SNAKE_CASE     ,				[
                 {'generated_token_ids': ANY(__SCREAMING_SNAKE_CASE   )},
                 {'generated_token_ids': ANY(__SCREAMING_SNAKE_CASE   )},
             ]     ,				)
         a__		=			text_generator.model.config.eos_token_id
         a__		=			'<pad>'
         a__		=			text_generator(
             ['This is a test', 'This is a second test']     ,				do_sample=__SCREAMING_SNAKE_CASE     ,				num_return_sequences=2     ,				batch_size=2     ,				return_tensors=__SCREAMING_SNAKE_CASE     ,				)
         self.assertEqual(
             __SCREAMING_SNAKE_CASE     ,				[
                 [
                     {'generated_token_ids': ANY(__SCREAMING_SNAKE_CASE   )},
                     {'generated_token_ids': ANY(__SCREAMING_SNAKE_CASE   )},
                 ],
                 [
                     {'generated_token_ids': ANY(__SCREAMING_SNAKE_CASE   )},
                     {'generated_token_ids': ANY(__SCREAMING_SNAKE_CASE   )},
                 ],
             ]     ,				)



     @require_tf
     def 			lowercase__  (							self   )		->     Optional[int]:







         """simple docstring"""



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

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

         a__		=			text_generator(['This is a test', 'This is a second test']     ,				do_sample=__SCREAMING_SNAKE_CASE   )
         self.assertEqual(
             __SCREAMING_SNAKE_CASE     ,				[
                 [
                     {
                         '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     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )		->     Optional[int]:







         """simple docstring"""



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



     def 			lowercase__  (							self   )		->     int:







         """simple docstring"""



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

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



     def 			lowercase__  (							self     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )		->     Optional[int]:







         """simple docstring"""



         a__		=			text_generator.model
         a__		=			text_generator.tokenizer

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

         a__		=			text_generator('This is a test'     ,				return_full_text=__SCREAMING_SNAKE_CASE   )
         self.assertEqual(__SCREAMING_SNAKE_CASE     ,				[{'generated_text': ANY(__SCREAMING_SNAKE_CASE   )}]   )
         self.assertNotIn('This is a test'     ,				outputs[0]['generated_text']   )

         a__		=			pipeline(task='text-generation'     ,				model=__SCREAMING_SNAKE_CASE     ,				tokenizer=__SCREAMING_SNAKE_CASE     ,				return_full_text=__SCREAMING_SNAKE_CASE   )
         a__		=			text_generator('This is a test'   )
         self.assertEqual(__SCREAMING_SNAKE_CASE     ,				[{'generated_text': ANY(__SCREAMING_SNAKE_CASE   )}]   )
         self.assertNotIn('This is a test'     ,				outputs[0]['generated_text']   )

         a__		=			text_generator('This is a test'     ,				return_full_text=__SCREAMING_SNAKE_CASE   )
         self.assertEqual(__SCREAMING_SNAKE_CASE     ,				[{'generated_text': ANY(__SCREAMING_SNAKE_CASE   )}]   )
         self.assertTrue(outputs[0]['generated_text'].startswith('This is a test'   )   )

         a__		=			text_generator(['This is great !', 'Something else']     ,				num_return_sequences=2     ,				do_sample=__SCREAMING_SNAKE_CASE   )
         self.assertEqual(
             __SCREAMING_SNAKE_CASE     ,				[
                 [{'generated_text': ANY(__SCREAMING_SNAKE_CASE   )}, {'generated_text': ANY(__SCREAMING_SNAKE_CASE   )}],
                 [{'generated_text': ANY(__SCREAMING_SNAKE_CASE   )}, {'generated_text': ANY(__SCREAMING_SNAKE_CASE   )}],
             ]     ,				)

         if text_generator.tokenizer.pad_token is not None:
             a__		=			text_generator(
                 ['This is great !', 'Something else']     ,				num_return_sequences=2     ,				batch_size=2     ,				do_sample=__SCREAMING_SNAKE_CASE   )
             self.assertEqual(
                 __SCREAMING_SNAKE_CASE     ,				[
                     [{'generated_text': ANY(__SCREAMING_SNAKE_CASE   )}, {'generated_text': ANY(__SCREAMING_SNAKE_CASE   )}],
                     [{'generated_text': ANY(__SCREAMING_SNAKE_CASE   )}, {'generated_text': ANY(__SCREAMING_SNAKE_CASE   )}],
                 ]     ,				)

         with self.assertRaises(__SCREAMING_SNAKE_CASE   ):
             a__		=			text_generator('test'     ,				return_full_text=__SCREAMING_SNAKE_CASE     ,				return_text=__SCREAMING_SNAKE_CASE   )
         with self.assertRaises(__SCREAMING_SNAKE_CASE   ):
             a__		=			text_generator('test'     ,				return_full_text=__SCREAMING_SNAKE_CASE     ,				return_tensors=__SCREAMING_SNAKE_CASE   )
         with self.assertRaises(__SCREAMING_SNAKE_CASE   ):
             a__		=			text_generator('test'     ,				return_text=__SCREAMING_SNAKE_CASE     ,				return_tensors=__SCREAMING_SNAKE_CASE   )

         # 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__		=			text_generator(''   )
             self.assertEqual(__SCREAMING_SNAKE_CASE     ,				[{'generated_text': ANY(__SCREAMING_SNAKE_CASE   )}]   )
         else:
             with self.assertRaises((ValueError, AssertionError)   ):
                 a__		=			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__		=			['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__		=			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(__SCREAMING_SNAKE_CASE   ):
                 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 			lowercase__  (							self   )		->     List[Any]:







         """simple docstring"""



         import torch

         # Classic `model_kwargs`
         a__		=			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__		=			pipe('This is a test'   )
         self.assertEqual(
             __SCREAMING_SNAKE_CASE     ,				[
                 {
                     '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__		=			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__		=			pipe('This is a test'   )
         self.assertEqual(
             __SCREAMING_SNAKE_CASE     ,				[
                 {
                     '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__		=			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__		=			pipe('This is a test'   )
         self.assertEqual(
             __SCREAMING_SNAKE_CASE     ,				[
                 {
                     '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   )		->     Any:







         """simple docstring"""



         import torch

         a__		=			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   )		->     Dict:







         """simple docstring"""



         import torch

         a__		=			pipeline(model='hf-internal-testing/tiny-random-bloom'     ,				device_map='auto'     ,				torch_dtype=torch.floataa   )
         pipe('This is a test'     ,				do_sample=__SCREAMING_SNAKE_CASE     ,				top_p=0.5   )




     def 			lowercase__  (							self   )		->     int:







         """simple docstring"""



         a__		=			'Hello world'
         a__		=			pipeline('text-generation'     ,				model='hf-internal-testing/tiny-random-gpt2'   )
         if text_generator.model.framework == "tf":
             a__		=			logging.get_logger('transformers.generation.tf_utils'   )
         else:
             a__		=			logging.get_logger('transformers.generation.utils'   )
         a__		=			'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(__SCREAMING_SNAKE_CASE   ) as cl:
             a__		=			text_generator(__SCREAMING_SNAKE_CASE     ,				max_length=1_0     ,				max_new_tokens=1   )
         self.assertIn(__SCREAMING_SNAKE_CASE     ,				cl.out   )

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

         with CaptureLogger(__SCREAMING_SNAKE_CASE   ) as cl:
             a__		=			text_generator(__SCREAMING_SNAKE_CASE     ,				max_length=1_0   )
         self.assertNotIn(__SCREAMING_SNAKE_CASE     ,				cl.out   )




"""simple docstring"""




from collections import deque







class     lowercase:



     def __init__(							self     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )		->     None:







         """simple docstring"""



         a__		=			process_name  # process name
         a__		=			arrival_time  # arrival time of the process
         # completion time of finished process or last interrupted time
         a__		=			arrival_time
         a__		=			burst_time  # remaining burst time
         a__		=			0  # total time of the process wait in ready queue
         a__		=			0  # time from arrival time to completion time







class     lowercase:



     def __init__(							self     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE     ,				)		->     None:







         """simple docstring"""



         a__		=			number_of_queues
         # time slice of queues that round robin algorithm applied
         a__		=			time_slices
         # unfinished process is in this ready_queue
         a__		=			queue
         # current time
         a__		=			current_time
         # finished process is in this sequence queue
         a__		=			deque()



     def 			lowercase__  (							self   )		->     list[str]:







         """simple docstring"""



         a__		=			[]
         for i in range(len(self.finish_queue   )   ):
             sequence.append(self.finish_queue[i].process_name   )
         return sequence



     def 			lowercase__  (							self     ,				__SCREAMING_SNAKE_CASE   )		->     list[int]:







         """simple docstring"""



         a__		=			[]
         for i in range(len(__SCREAMING_SNAKE_CASE   )   ):
             waiting_times.append(queue[i].waiting_time   )
         return waiting_times



     def 			lowercase__  (							self     ,				__SCREAMING_SNAKE_CASE   )		->     list[int]:







         """simple docstring"""



         a__		=			[]
         for i in range(len(__SCREAMING_SNAKE_CASE   )   ):
             turnaround_times.append(queue[i].turnaround_time   )
         return turnaround_times



     def 			lowercase__  (							self     ,				__SCREAMING_SNAKE_CASE   )		->     list[int]:







         """simple docstring"""



         a__		=			[]
         for i in range(len(__SCREAMING_SNAKE_CASE   )   ):
             completion_times.append(queue[i].stop_time   )
         return completion_times



     def 			lowercase__  (							self     ,				__SCREAMING_SNAKE_CASE   )		->     list[int]:







         """simple docstring"""



         return [q.burst_time for q in queue]



     def 			lowercase__  (							self     ,				__SCREAMING_SNAKE_CASE   )		->     int:







         """simple docstring"""



         process.waiting_time += self.current_time - process.stop_time
         return process.waiting_time



     def 			lowercase__  (							self     ,				__SCREAMING_SNAKE_CASE   )		->     deque[Process]:







         """simple docstring"""



         a__		=			deque()  # sequence deque of finished process
         while len(__SCREAMING_SNAKE_CASE   ) != 0:
             a__		=			ready_queue.popleft()  # current process

             # if process's arrival time is later than current time, update current time
             if self.current_time < cp.arrival_time:
                 self.current_time += cp.arrival_time

             # update waiting time of current process
             self.update_waiting_time(__SCREAMING_SNAKE_CASE   )
             # update current time
             self.current_time += cp.burst_time
             # finish the process and set the process's burst-time 0
             a__		=			0
             # set the process's turnaround time because it is finished
             a__		=			self.current_time - cp.arrival_time
             # set the completion time
             a__		=			self.current_time
             # add the process to queue that has finished queue
             finished.append(__SCREAMING_SNAKE_CASE   )

         self.finish_queue.extend(__SCREAMING_SNAKE_CASE   )  # add finished process to finish queue
         # FCFS will finish all remaining processes
         return finished



     def 			lowercase__  (							self     ,				__SCREAMING_SNAKE_CASE     ,				__SCREAMING_SNAKE_CASE   )		->     tuple[deque[Process], deque[Process]]:







         """simple docstring"""



         a__		=			deque()  # sequence deque of terminated process
         # just for 1 cycle and unfinished processes will go back to queue
         for _ in range(len(__SCREAMING_SNAKE_CASE   )   ):
             a__		=			ready_queue.popleft()  # current process

             # if process's arrival time is later than current time, update current time
             if self.current_time < cp.arrival_time:
                 self.current_time += cp.arrival_time

             # update waiting time of unfinished processes
             self.update_waiting_time(__SCREAMING_SNAKE_CASE   )
             # if the burst time of process is bigger than time-slice
             if cp.burst_time > time_slice:
                 # use CPU for only time-slice
                 self.current_time += time_slice
                 # update remaining burst time
                 cp.burst_time -= time_slice
                 # update end point time
                 a__		=			self.current_time
                 # locate the process behind the queue because it is not finished
                 ready_queue.append(__SCREAMING_SNAKE_CASE   )
             else:
                 # use CPU for remaining burst time
                 self.current_time += cp.burst_time
                 # set burst time 0 because the process is finished
                 a__		=			0
                 # set the finish time
                 a__		=			self.current_time
                 # update the process' turnaround time because it is finished
                 a__		=			self.current_time - cp.arrival_time
                 # add the process to queue that has finished queue
                 finished.append(__SCREAMING_SNAKE_CASE   )

         self.finish_queue.extend(__SCREAMING_SNAKE_CASE   )  # add finished process to finish queue
         # return finished processes queue and remaining processes queue
         return finished, ready_queue




     def 			lowercase__  (							self   )		->     deque[Process]:







         """simple docstring"""



         for i in range(self.number_of_queues - 1   ):
             a__	,     a__		=			self.round_robin(
                 self.ready_queue     ,				self.time_slices[i]   )
         #  the last queue has first_come_first_served algorithm
         self.first_come_first_served(self.ready_queue   )

         return self.finish_queue


if __name__ == "__main__":
    import doctest

    a		:       List[str]     						=       Process('P1', 0, 53)
    a		:       Optional[int]     						=       Process('P2', 0, 17)
    a		:       Union[str, Any]     						=       Process('P3', 0, 68)
    a		:       Optional[int]     						=       Process('P4', 0, 24)
    a		:       Optional[int]     						=       3
    a		:       Optional[Any]     						=       [17, 25]
    a		:       Union[str, Any]     						=       deque([Pa, Pa, Pa, Pa])

    if len(time_slices) != number_of_queues - 1:
        raise SystemExit(0)

    doctest.testmod(extraglobs={'queue': deque([Pa, Pa, Pa, Pa])})

    a		:       Union[str, Any]     						=       Process('P1', 0, 53)
    a		:       Optional[int]     						=       Process('P2', 0, 17)
    a		:       Optional[Any]     						=       Process('P3', 0, 68)
    a		:       str     						=       Process('P4', 0, 24)
    a		:       Optional[Any]     						=       3
    a		:       Tuple     						=       [17, 25]
    a		:       int     						=       deque([Pa, Pa, Pa, Pa])
    a		:       Tuple     						=       MLFQ(number_of_queues, time_slices, queue, 0)
    a		:       Optional[int]     						=       mlfq.multi_level_feedback_queue()

    # print total waiting times of processes(P1, P2, P3, P4)
    print(
        F'''waiting time:\
        \t\t\t{MLFQ.calculate_waiting_time(mlfq, [Pa, Pa, Pa, Pa])}'''
    )
    # print completion times of processes(P1, P2, P3, P4)
    print(
        F'''completion time:\
        \t\t{MLFQ.calculate_completion_time(mlfq, [Pa, Pa, Pa, Pa])}'''
    )
    # print total turnaround times of processes(P1, P2, P3, P4)
    print(
        F'''turnaround time:\
        \t\t{MLFQ.calculate_turnaround_time(mlfq, [Pa, Pa, Pa, Pa])}'''
    )
    # print sequence of finished processes
    print(
        F'''sequence of finished processes:\
        {mlfq.calculate_sequence_of_finish_queue()}'''
    )


