
"""simple docstring"""





import unittest

from transformers import MODEL_FOR_DOCUMENT_QUESTION_ANSWERING_MAPPING, AutoTokenizer, is_vision_available
from transformers.pipelines import pipeline
from transformers.pipelines.document_question_answering import apply_tesseract
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_detectrona,
    require_pytesseract,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
		from PIL import Image

		from transformers.image_utils import load_image
else:



		class __lowerCAmelCase				:




					"""simple docstring"""







					@staticmethod
					def      lowerCamelCase__     (				*_snake_case						:					List[str]       ,   **_snake_case						:					Dict	)      -> Tuple:







										"""simple docstring"""
										pass





		def 							A_     (__a				):

							'''simple docstring'''





							return None


# This is a pinned image from a specific revision of a document question answering space, hosted by HuggingFace,
# so we can expect it to be available.
UpperCamelCase_					:						int				= (
    '''https://huggingface.co/spaces/impira/docquery/resolve/2f6c96314dc84dfda62d40de9da55f2f5165d403/invoice.png'''
)



@is_pipeline_test
@require_torch
@require_vision
class __lowerCAmelCase				(					unittest.TestCase      ):




			"""simple docstring"""







			snake_case	      =				MODEL_FOR_DOCUMENT_QUESTION_ANSWERING_MAPPING




			@require_pytesseract
			@require_vision
			def      lowerCamelCase__     (				self						:					List[Any]       ,   _snake_case						:					int       ,   _snake_case						:					Union[str, Any]       ,   _snake_case						:					Tuple	)      -> Dict:







								"""simple docstring"""
								A_				 =						pipeline(
								    "document-question-answering"       ,   model=_snake_case       ,   tokenizer=_snake_case       ,   image_processor=_snake_case	)

								A_				 =						INVOICE_URL
								A_				 =						list(zip(*apply_tesseract(load_image(_snake_case	)       ,   _snake_case       ,   ""	)	)	)
								A_				 =						"What is the placebo?"
								A_				 =						[
								    {
								        "image": load_image(_snake_case	),
								        "question": question,
								    },
								    {
								        "image": image,
								        "question": question,
								    },
								    {
								        "image": image,
								        "question": question,
								        "word_boxes": word_boxes,
								    },
								]
								return dqa_pipeline, examples




			def      lowerCamelCase__     (				self						:					Any       ,   _snake_case						:					List[Any]       ,   _snake_case						:					Dict	)      -> Dict:







								"""simple docstring"""
								A_				 =						dqa_pipeline(_snake_case       ,   top_k=2	)
								self.assertEqual(
								    _snake_case       ,   [
								        [
								            {"score": ANY(_snake_case	), "answer": ANY(_snake_case	), "start": ANY(_snake_case	), "end": ANY(_snake_case	)},
								            {"score": ANY(_snake_case	), "answer": ANY(_snake_case	), "start": ANY(_snake_case	), "end": ANY(_snake_case	)},
								        ]
								    ]
								    * 3       ,   )




			@require_torch
			@require_detectrona
			@require_pytesseract
			def      lowerCamelCase__     (				self						:					Optional[Any]	)      -> Tuple:







								"""simple docstring"""
								A_				 =						pipeline("document-question-answering"       ,   model="hf-internal-testing/tiny-random-layoutlmv2"	)
								A_				 =						INVOICE_URL
								A_				 =						"How many cats are there?"

								A_				 =						[
								    {"score": 0.0_0_0_1, "answer": "oy 2312/2019", "start": 38, "end": 39},
								    {"score": 0.0_0_0_1, "answer": "oy 2312/2019 DUE", "start": 38, "end": 40},
								]
								A_				 =						dqa_pipeline(image=_snake_case       ,   question=_snake_case       ,   top_k=2	)
								self.assertEqual(nested_simplify(_snake_case       ,   decimals=4	)       ,   _snake_case	)

								A_				 =						dqa_pipeline({"image": image, "question": question}       ,   top_k=2	)
								self.assertEqual(nested_simplify(_snake_case       ,   decimals=4	)       ,   _snake_case	)

								# This image does not detect ANY text in it, meaning layoutlmv2 should fail.
								# Empty answer probably
								A_				 =						"./tests/fixtures/tests_samples/COCO/000000039769.png"
								A_				 =						dqa_pipeline(image=_snake_case       ,   question=_snake_case       ,   top_k=2	)
								self.assertEqual(_snake_case       ,   []	)

								# We can optionnally pass directly the words and bounding boxes
								A_				 =						"./tests/fixtures/tests_samples/COCO/000000039769.png"
								A_				 =						[]
								A_				 =						[]
								A_				 =						dqa_pipeline(image=_snake_case       ,   question=_snake_case       ,   words=_snake_case       ,   boxes=_snake_case       ,   top_k=2	)
								self.assertEqual(_snake_case       ,   []	)




			@slow
			@require_torch
			@require_detectrona
			@require_pytesseract
			def      lowerCamelCase__     (				self						:					Dict	)      -> Optional[Any]:







								"""simple docstring"""
								A_				 =						pipeline(
								    "document-question-answering"       ,   model="tiennvcs/layoutlmv2-base-uncased-finetuned-docvqa"       ,   revision="9977165"       ,   )
								A_				 =						INVOICE_URL
								A_				 =						"What is the invoice number?"

								A_				 =						dqa_pipeline(image=_snake_case       ,   question=_snake_case       ,   top_k=2	)
								self.assertEqual(
								    nested_simplify(_snake_case       ,   decimals=4	)       ,   [
								        {"score": 0.9_9_4_4, "answer": "us-001", "start": 16, "end": 16},
								        {"score": 0.0_0_0_9, "answer": "us-001", "start": 16, "end": 16},
								    ]       ,   )

								A_				 =						dqa_pipeline({"image": image, "question": question}       ,   top_k=2	)
								self.assertEqual(
								    nested_simplify(_snake_case       ,   decimals=4	)       ,   [
								        {"score": 0.9_9_4_4, "answer": "us-001", "start": 16, "end": 16},
								        {"score": 0.0_0_0_9, "answer": "us-001", "start": 16, "end": 16},
								    ]       ,   )

								A_				 =						dqa_pipeline(
								    [{"image": image, "question": question}, {"image": image, "question": question}]       ,   top_k=2	)
								self.assertEqual(
								    nested_simplify(_snake_case       ,   decimals=4	)       ,   [
								        [
								            {"score": 0.9_9_4_4, "answer": "us-001", "start": 16, "end": 16},
								            {"score": 0.0_0_0_9, "answer": "us-001", "start": 16, "end": 16},
								        ],
								    ]
								    * 2       ,   )




			@slow
			@require_torch
			@require_detectrona
			@require_pytesseract
			def      lowerCamelCase__     (				self						:					int	)      -> Any:







								"""simple docstring"""
								A_				 =						pipeline(
								    "document-question-answering"       ,   model="tiennvcs/layoutlmv2-base-uncased-finetuned-docvqa"       ,   revision="9977165"       ,   max_seq_len=50       ,   )
								A_				 =						INVOICE_URL
								A_				 =						"What is the invoice number?"

								A_				 =						dqa_pipeline(image=_snake_case       ,   question=_snake_case       ,   top_k=2	)
								self.assertEqual(
								    nested_simplify(_snake_case       ,   decimals=4	)       ,   [
								        {"score": 0.9_9_7_4, "answer": "1110212019", "start": 23, "end": 23},
								        {"score": 0.9_9_4_8, "answer": "us-001", "start": 16, "end": 16},
								    ]       ,   )

								A_				 =						dqa_pipeline({"image": image, "question": question}       ,   top_k=2	)
								self.assertEqual(
								    nested_simplify(_snake_case       ,   decimals=4	)       ,   [
								        {"score": 0.9_9_7_4, "answer": "1110212019", "start": 23, "end": 23},
								        {"score": 0.9_9_4_8, "answer": "us-001", "start": 16, "end": 16},
								    ]       ,   )

								A_				 =						dqa_pipeline(
								    [{"image": image, "question": question}, {"image": image, "question": question}]       ,   top_k=2	)
								self.assertEqual(
								    nested_simplify(_snake_case       ,   decimals=4	)       ,   [
								        [
								            {"score": 0.9_9_7_4, "answer": "1110212019", "start": 23, "end": 23},
								            {"score": 0.9_9_4_8, "answer": "us-001", "start": 16, "end": 16},
								        ]
								    ]
								    * 2       ,   )




			@slow
			@require_torch
			@require_pytesseract
			@require_vision
			def      lowerCamelCase__     (				self						:					Union[str, Any]	)      -> Optional[int]:







								"""simple docstring"""
								A_				 =						AutoTokenizer.from_pretrained(
								    "impira/layoutlm-document-qa"       ,   revision="3dc6de3"       ,   add_prefix_space=_snake_case	)
								A_				 =						pipeline(
								    "document-question-answering"       ,   model="impira/layoutlm-document-qa"       ,   tokenizer=_snake_case       ,   revision="3dc6de3"       ,   )
								A_				 =						INVOICE_URL
								A_				 =						"What is the invoice number?"

								A_				 =						dqa_pipeline(image=_snake_case       ,   question=_snake_case       ,   top_k=2	)
								self.assertEqual(
								    nested_simplify(_snake_case       ,   decimals=4	)       ,   [
								        {"score": 0.4_2_5_1, "answer": "us-001", "start": 16, "end": 16},
								        {"score": 0.0_8_1_9, "answer": "1110212019", "start": 23, "end": 23},
								    ]       ,   )

								A_				 =						dqa_pipeline({"image": image, "question": question}       ,   top_k=2	)
								self.assertEqual(
								    nested_simplify(_snake_case       ,   decimals=4	)       ,   [
								        {"score": 0.4_2_5_1, "answer": "us-001", "start": 16, "end": 16},
								        {"score": 0.0_8_1_9, "answer": "1110212019", "start": 23, "end": 23},
								    ]       ,   )

								A_				 =						dqa_pipeline(
								    [{"image": image, "question": question}, {"image": image, "question": question}]       ,   top_k=2	)
								self.assertEqual(
								    nested_simplify(_snake_case       ,   decimals=4	)       ,   [
								        [
								            {"score": 0.4_2_5_1, "answer": "us-001", "start": 16, "end": 16},
								            {"score": 0.0_8_1_9, "answer": "1110212019", "start": 23, "end": 23},
								        ]
								    ]
								    * 2       ,   )

								A_				 =						list(zip(*apply_tesseract(load_image(_snake_case	)       ,   _snake_case       ,   ""	)	)	)

								# This model should also work if `image` is set to None
								A_				 =						dqa_pipeline({"image": None, "word_boxes": word_boxes, "question": question}       ,   top_k=2	)
								self.assertEqual(
								    nested_simplify(_snake_case       ,   decimals=4	)       ,   [
								        {"score": 0.4_2_5_1, "answer": "us-001", "start": 16, "end": 16},
								        {"score": 0.0_8_1_9, "answer": "1110212019", "start": 23, "end": 23},
								    ]       ,   )




			@slow
			@require_torch
			@require_pytesseract
			@require_vision
			def      lowerCamelCase__     (				self						:					List[Any]	)      -> Dict:







								"""simple docstring"""
								A_				 =						AutoTokenizer.from_pretrained(
								    "impira/layoutlm-document-qa"       ,   revision="3dc6de3"       ,   add_prefix_space=_snake_case	)
								A_				 =						pipeline(
								    "document-question-answering"       ,   model="impira/layoutlm-document-qa"       ,   tokenizer=_snake_case       ,   revision="3dc6de3"       ,   max_seq_len=50       ,   )
								A_				 =						INVOICE_URL
								A_				 =						"What is the invoice number?"

								A_				 =						dqa_pipeline(image=_snake_case       ,   question=_snake_case       ,   top_k=2	)
								self.assertEqual(
								    nested_simplify(_snake_case       ,   decimals=4	)       ,   [
								        {"score": 0.9_9_9_9, "answer": "us-001", "start": 16, "end": 16},
								        {"score": 0.9_9_9_8, "answer": "us-001", "start": 16, "end": 16},
								    ]       ,   )

								A_				 =						dqa_pipeline(
								    [{"image": image, "question": question}, {"image": image, "question": question}]       ,   top_k=2	)
								self.assertEqual(
								    nested_simplify(_snake_case       ,   decimals=4	)       ,   [
								        [
								            {"score": 0.9_9_9_9, "answer": "us-001", "start": 16, "end": 16},
								            {"score": 0.9_9_9_8, "answer": "us-001", "start": 16, "end": 16},
								        ]
								    ]
								    * 2       ,   )

								A_				 =						list(zip(*apply_tesseract(load_image(_snake_case	)       ,   _snake_case       ,   ""	)	)	)

								# This model should also work if `image` is set to None
								A_				 =						dqa_pipeline({"image": None, "word_boxes": word_boxes, "question": question}       ,   top_k=2	)
								self.assertEqual(
								    nested_simplify(_snake_case       ,   decimals=4	)       ,   [
								        {"score": 0.9_9_9_9, "answer": "us-001", "start": 16, "end": 16},
								        {"score": 0.9_9_9_8, "answer": "us-001", "start": 16, "end": 16},
								    ]       ,   )




			@slow
			@require_torch
			def      lowerCamelCase__     (				self						:					Dict	)      -> int:







								"""simple docstring"""
								A_				 =						pipeline(
								    "document-question-answering"       ,   model="naver-clova-ix/donut-base-finetuned-docvqa"       ,   tokenizer=AutoTokenizer.from_pretrained("naver-clova-ix/donut-base-finetuned-docvqa"	)       ,   feature_extractor="naver-clova-ix/donut-base-finetuned-docvqa"       ,   )

								A_				 =						INVOICE_URL
								A_				 =						"What is the invoice number?"
								A_				 =						dqa_pipeline(image=_snake_case       ,   question=_snake_case       ,   top_k=2	)
								self.assertEqual(nested_simplify(_snake_case       ,   decimals=4	)       ,   [{"answer": "us-001"}]	)






			@require_tf
			@unittest.skip("Document question answering not implemented in TF"	)
			def      lowerCamelCase__     (				self						:					Optional[int]	)      -> Union[str, Any]:







								"""simple docstring"""
								pass






"""simple docstring"""





from collections import deque



class __lowerCAmelCase				:




			"""simple docstring"""







			def __init__(				self						:					Optional[Any]       ,   _snake_case						:					str       ,   _snake_case						:					int       ,   _snake_case						:					int	)      -> 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 __lowerCAmelCase				:




			"""simple docstring"""







			def __init__(				self						:					str       ,   _snake_case						:					int       ,   _snake_case						:					list[int]       ,   _snake_case						:					deque[Process]       ,   _snake_case						:					int       ,   )      -> None:







								"""simple docstring"""
								# total number of mlfq's queues
								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      lowerCamelCase__     (				self						:					List[str]	)      -> list[str]:







								"""simple docstring"""
								A_				 =						[]
								for i in range(len(self.finish_queue	)	):
													sequence.append(self.finish_queue[i].process_name	)
								return sequence




			def      lowerCamelCase__     (				self						:					str       ,   _snake_case						:					list[Process]	)      -> list[int]:







								"""simple docstring"""
								A_				 =						[]
								for i in range(len(_snake_case	)	):
													waiting_times.append(queue[i].waiting_time	)
								return waiting_times




			def      lowerCamelCase__     (				self						:					Tuple       ,   _snake_case						:					list[Process]	)      -> list[int]:







								"""simple docstring"""
								A_				 =						[]
								for i in range(len(_snake_case	)	):
													turnaround_times.append(queue[i].turnaround_time	)
								return turnaround_times




			def      lowerCamelCase__     (				self						:					Tuple       ,   _snake_case						:					list[Process]	)      -> list[int]:







								"""simple docstring"""
								A_				 =						[]
								for i in range(len(_snake_case	)	):
													completion_times.append(queue[i].stop_time	)
								return completion_times




			def      lowerCamelCase__     (				self						:					Optional[int]       ,   _snake_case						:					deque[Process]	)      -> list[int]:







								"""simple docstring"""
								return [q.burst_time for q in queue]




			def      lowerCamelCase__     (				self						:					int       ,   _snake_case						:					Process	)      -> int:







								"""simple docstring"""
								process.waiting_time += self.current_time - process.stop_time
								return process.waiting_time




			def      lowerCamelCase__     (				self						:					Dict       ,   _snake_case						:					deque[Process]	)      -> deque[Process]:







								"""simple docstring"""
								A_				 =						deque()  # sequence deque of finished process
								while len(_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(_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(_snake_case	)

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




			def      lowerCamelCase__     (				self						:					List[str]       ,   _snake_case						:					deque[Process]       ,   _snake_case						:					int	)      -> 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(_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(_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(_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(_snake_case	)

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






			def      lowerCamelCase__     (				self						:					Any	)      -> deque[Process]:







								"""simple docstring"""

								#  all queues except last one have round_robin algorithm
								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

		UpperCamelCase_					:						Tuple				= Process('''P1''', 0, 53)
		UpperCamelCase_					:						List[Any]				= Process('''P2''', 0, 17)
		UpperCamelCase_					:						Optional[int]				= Process('''P3''', 0, 68)
		UpperCamelCase_					:						Union[str, Any]				= Process('''P4''', 0, 24)
		UpperCamelCase_					:						Any				= 3
		UpperCamelCase_					:						List[str]				= [17, 25]
		UpperCamelCase_					:						List[str]				= 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])})

		UpperCamelCase_					:						str				= Process('''P1''', 0, 53)
		UpperCamelCase_					:						Dict				= Process('''P2''', 0, 17)
		UpperCamelCase_					:						List[Any]				= Process('''P3''', 0, 68)
		UpperCamelCase_					:						int				= Process('''P4''', 0, 24)
		UpperCamelCase_					:						Dict				= 3
		UpperCamelCase_					:						str				= [17, 25]
		UpperCamelCase_					:						Optional[int]				= deque([Pa, Pa, Pa, Pa])
		UpperCamelCase_					:						str				= MLFQ(number_of_queues, time_slices, queue, 0)
		UpperCamelCase_					:						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()}"""
		)




