"""simple docstring"""




from __future__ import annotations

import collections
import tempfile
import unittest

import numpy as np

from transformers.testing_utils import require_tf, require_vision, slow
from transformers.utils import is_tf_available, is_vision_available

from ...test_modeling_tf_common import floats_tensor, ids_tensor, random_attention_mask
from ..bert.test_modeling_tf_bert import TFBertModelTester
from ..clip.test_modeling_tf_clip import TFCLIPVisionModelTester
from ..deit.test_modeling_tf_deit import TFDeiTModelTester
from ..roberta.test_modeling_tf_roberta import TFRobertaModelTester
from ..vit.test_modeling_tf_vit import TFViTModelTester


if is_tf_available():
			from transformers import (
			    TFBertModel,
			    TFCLIPVisionModel,
			    TFDeiTModel,
			    TFRobertaModel,
			    TFVisionTextDualEncoderModel,
			    TFViTModel,
			    VisionTextDualEncoderConfig,
			)

if is_vision_available():
			from PIL import Image

			from transformers import VisionTextDualEncoderProcessor







def 				_UpperCAmelCase		(							lowerCamelCase__  ):





		"""simple docstring"""




		if isinstance(lowerCamelCase__      ,	collections.abc.Iterable  ):
				return x
		return (x, x)

@require_tf
class    a_   :




					def     _SCREAMING_SNAKE_CASE    (						self    :  Any	,    snake_case__    :  str	,    snake_case__    :  List[str]  ):
							pass




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[Any]  ):
							pass




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[int]  ):
							pass




					def     _SCREAMING_SNAKE_CASE    (						self    :  str	,    snake_case__    :  Optional[Any]	,    snake_case__    :  Optional[int]	,    snake_case__    :  int	,    snake_case__    :  Optional[Any]	,    snake_case__    :  int=None	,    **snake_case__    :  Optional[int]  ):
							lowerCAmelCase__   = VisionTextDualEncoderConfig.from_vision_text_configs(snake_case__	,    snake_case__  )

							lowerCAmelCase__   = TFVisionTextDualEncoderModel(snake_case__  )

							lowerCAmelCase__   = model(input_ids=snake_case__	,    pixel_values=snake_case__	,    attention_mask=snake_case__  )

							self.assertEqual(output["""text_embeds"""].shape	,    (input_ids.shape[0], config.projection_dim)  )
							self.assertEqual(output["""image_embeds"""].shape	,    (pixel_values.shape[0], config.projection_dim)  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  int	,    snake_case__    :  Union[str, Any]	,    snake_case__    :  List[str]	,    snake_case__    :  str	,    snake_case__    :  Optional[Any]	,    snake_case__    :  Dict=None	,    **snake_case__    :  Optional[int]  ):
							lowerCAmelCase__ ,       lowerCAmelCase__   = self.get_vision_text_model(snake_case__	,    snake_case__  )
							lowerCAmelCase__   = TFVisionTextDualEncoderModel(vision_model=snake_case__	,    text_model=snake_case__  )

							lowerCAmelCase__   = model(input_ids=snake_case__	,    pixel_values=snake_case__	,    attention_mask=snake_case__  )

							self.assertEqual(output["""text_embeds"""].shape	,    (input_ids.shape[0], model.config.projection_dim)  )
							self.assertEqual(output["""image_embeds"""].shape	,    (pixel_values.shape[0], model.config.projection_dim)  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[int]	,    snake_case__    :  int	,    snake_case__    :  Tuple	,    snake_case__    :  Union[str, Any]	,    snake_case__    :  Optional[Any]	,    snake_case__    :  int=None	,    **snake_case__    :  Optional[Any]  ):
							lowerCAmelCase__ ,       lowerCAmelCase__   = self.get_vision_text_model(snake_case__	,    snake_case__  )
							lowerCAmelCase__   = {"""vision_model""": vision_model, """text_model""": text_model}
							lowerCAmelCase__   = TFVisionTextDualEncoderModel.from_vision_text_pretrained(**snake_case__  )

							lowerCAmelCase__   = model(input_ids=snake_case__	,    pixel_values=snake_case__	,    attention_mask=snake_case__  )

							self.assertEqual(output["""text_embeds"""].shape	,    (input_ids.shape[0], model.config.projection_dim)  )
							self.assertEqual(output["""image_embeds"""].shape	,    (pixel_values.shape[0], model.config.projection_dim)  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Union[str, Any]	,    snake_case__    :  Optional[Any]	,    snake_case__    :  int	,    snake_case__    :  str	,    snake_case__    :  Optional[int]	,    snake_case__    :  Union[str, Any]=None	,    **snake_case__    :  Optional[int]  ):
							lowerCAmelCase__ ,       lowerCAmelCase__   = self.get_vision_text_model(snake_case__	,    snake_case__  )
							lowerCAmelCase__   = TFVisionTextDualEncoderModel(vision_model=snake_case__	,    text_model=snake_case__  )

							lowerCAmelCase__   = model(input_ids=snake_case__	,    pixel_values=snake_case__	,    attention_mask=snake_case__  )
							lowerCAmelCase__   = output[0].numpy()

							with tempfile.TemporaryDirectory() as tmpdirname:
									model.save_pretrained(snake_case__  )
									lowerCAmelCase__   = TFVisionTextDualEncoderModel.from_pretrained(snake_case__  )

									lowerCAmelCase__   = model(input_ids=snake_case__	,    pixel_values=snake_case__	,    attention_mask=snake_case__  )
									lowerCAmelCase__   = after_output[0].numpy()
									lowerCAmelCase__   = np.amax(np.abs(out_a - out_a  )  )
									self.assertLessEqual(snake_case__	,    1E-5  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Any	,    snake_case__    :  Optional[Any]	,    snake_case__    :  List[Any]	,    snake_case__    :  int	,    snake_case__    :  Tuple	,    snake_case__    :  Any=None	,    **snake_case__    :  Tuple  ):
							lowerCAmelCase__ ,       lowerCAmelCase__   = self.get_vision_text_model(snake_case__	,    snake_case__  )
							lowerCAmelCase__   = TFVisionTextDualEncoderModel(vision_model=snake_case__	,    text_model=snake_case__  )

							lowerCAmelCase__   = model(
							    input_ids=snake_case__	,    pixel_values=snake_case__	,    attention_mask=snake_case__	,    output_attentions=snake_case__  )

							lowerCAmelCase__   = output.vision_model_output.attentions
							self.assertEqual(len(snake_case__  )	,    vision_config.num_hidden_layers  )

							# in ViT, the seq_len equals the number of patches + 1 (we add 1 for the [CLS] token)
							lowerCAmelCase__   = to_atuple(vision_model.config.image_size  )
							lowerCAmelCase__   = to_atuple(vision_model.config.patch_size  )
							lowerCAmelCase__   = (image_size[1] // patch_size[1]) * (image_size[0] // patch_size[0])
							lowerCAmelCase__   = num_patches + 1
							self.assertEqual(vision_attentions[0].shape[-3:]	,    (vision_config.num_attention_heads, seq_len, seq_len)  )

							lowerCAmelCase__   = output.text_model_output.attentions
							self.assertEqual(len(snake_case__  )	,    text_config.num_hidden_layers  )

							self.assertEqual(
							    text_attentions[0].shape[-3:]	,    (text_config.num_attention_heads, input_ids.shape[-1], input_ids.shape[-1])	,    )




					def     _SCREAMING_SNAKE_CASE    (						self    :  str	,    snake_case__    :  np.ndarray	,    snake_case__    :  np.ndarray	,    snake_case__    :  float  ):
							lowerCAmelCase__   = np.abs((a - b)  ).max()
							self.assertLessEqual(snake_case__	,    snake_case__	,    F"""Difference between torch and flax is {diff} (>= {tol})."""  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  List[str]  ):
							lowerCAmelCase__   = self.prepare_config_and_inputs()
							self.check_vision_text_dual_encoder_model(**snake_case__  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Union[str, Any]  ):
							lowerCAmelCase__   = self.prepare_config_and_inputs()
							self.check_model_from_pretrained_configs(**snake_case__  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[int]  ):
							lowerCAmelCase__   = self.prepare_config_and_inputs()
							self.check_vision_text_dual_encoder_from_pretrained(**snake_case__  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  str  ):
							lowerCAmelCase__   = self.prepare_config_and_inputs()
							self.check_save_load(**snake_case__  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Dict  ):
							lowerCAmelCase__   = self.prepare_config_and_inputs()
							self.check_vision_text_output_attention(**snake_case__  )




					@slow
					def     _SCREAMING_SNAKE_CASE    (						self    :  int  ):
							lowerCAmelCase__ ,       lowerCAmelCase__   = self.get_pretrained_model_and_inputs()

							lowerCAmelCase__   = model_a(**snake_case__  )
							lowerCAmelCase__   = outputs[0].numpy()

							with tempfile.TemporaryDirectory() as tmp_dirname:
									model_a.save_pretrained(snake_case__  )
									lowerCAmelCase__   = TFVisionTextDualEncoderModel.from_pretrained(snake_case__  )

									lowerCAmelCase__   = model_a(**snake_case__  )
									lowerCAmelCase__   = after_outputs[0].numpy()
									lowerCAmelCase__   = np.amax(np.abs(out_a - out_a  )  )
									self.assertLessEqual(snake_case__	,    1E-5  )

@require_tf
class    a_   (  __UpperCamelCase			,		unittest.TestCase      ):




					def     _SCREAMING_SNAKE_CASE    (						self    :  List[str]  ):
							lowerCAmelCase__   = TFVisionTextDualEncoderModel.from_vision_text_pretrained(
							    """hf-internal-testing/tiny-random-vit"""	,    """hf-internal-testing/tiny-random-bert"""  )
							lowerCAmelCase__   = 13
							lowerCAmelCase__   = floats_tensor(
							    [
							        batch_size,
							        model.vision_model.config.num_channels,
							        model.vision_model.config.image_size,
							        model.vision_model.config.image_size,
							    ]  )
							lowerCAmelCase__   = ids_tensor([batch_size, 4]	,    model.text_model.config.vocab_size  )
							lowerCAmelCase__   = random_attention_mask([batch_size, 4]  )
							lowerCAmelCase__   = {"""pixel_values""": pixel_values, """input_ids""": input_ids, """attention_mask""": attention_mask}

							return model, inputs




					def     _SCREAMING_SNAKE_CASE    (						self    :  List[Any]	,    snake_case__    :  Tuple	,    snake_case__    :  Any  ):
							lowerCAmelCase__   = TFViTModel(snake_case__	,    name="""vision_model"""  )
							lowerCAmelCase__   = TFBertModel(snake_case__	,    name="""text_model"""  )
							return vision_model, text_model




					def     _SCREAMING_SNAKE_CASE    (						self    :  Dict  ):
							lowerCAmelCase__   = TFViTModelTester(self  )
							lowerCAmelCase__   = TFBertModelTester(self  )
							lowerCAmelCase__   = vit_model_tester.prepare_config_and_inputs()
							lowerCAmelCase__   = bert_model_tester.prepare_config_and_inputs()

							lowerCAmelCase__ ,       lowerCAmelCase__ ,       lowerCAmelCase__   = vision_config_and_inputs

							(
							    (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       
							)   = text_config_and_inputs

							return {
							    "text_config": text_config,
							    "vision_config": vision_config,
							    "pixel_values": pixel_values,
							    "attention_mask": input_mask,
							    "input_ids": input_ids,
							    "text_token_type_ids": token_type_ids,
							    "text_sequence_labels": sequence_labels,
							    "text_token_labels": token_labels,
							    "text_choice_labels": choice_labels,
							}

@require_tf
class    a_   (  __UpperCamelCase			,		unittest.TestCase      ):




					def     _SCREAMING_SNAKE_CASE    (						self    :  List[Any]  ):
							# DeiT repo doesn't have TF weights, but we don't actually use the weights at all so let's
							# just reinitialize it.
							lowerCAmelCase__   = TFVisionTextDualEncoderModel.from_vision_text_pretrained(
							    """Rocketknight1/tiny-random-deit-tf"""	,    """hf-internal-testing/tiny-random-roberta"""  )
							lowerCAmelCase__   = 13
							lowerCAmelCase__   = floats_tensor(
							    [
							        batch_size,
							        model.vision_model.config.num_channels,
							        model.vision_model.config.image_size,
							        model.vision_model.config.image_size,
							    ]  )
							lowerCAmelCase__   = ids_tensor([batch_size, 4]	,    model.text_model.config.vocab_size  )
							lowerCAmelCase__   = random_attention_mask([batch_size, 4]  )
							lowerCAmelCase__   = {"""pixel_values""": pixel_values, """input_ids""": input_ids, """attention_mask""": attention_mask}

							return model, inputs




					def     _SCREAMING_SNAKE_CASE    (						self    :  str	,    snake_case__    :  List[Any]	,    snake_case__    :  Optional[int]	,    snake_case__    :  Dict	,    snake_case__    :  Optional[int]	,    snake_case__    :  Tuple=None	,    **snake_case__    :  Dict  ):
							lowerCAmelCase__ ,       lowerCAmelCase__   = self.get_vision_text_model(snake_case__	,    snake_case__  )
							lowerCAmelCase__   = TFVisionTextDualEncoderModel(vision_model=snake_case__	,    text_model=snake_case__  )

							lowerCAmelCase__   = model(
							    input_ids=snake_case__	,    pixel_values=snake_case__	,    attention_mask=snake_case__	,    output_attentions=snake_case__  )

							lowerCAmelCase__   = output.vision_model_output.attentions
							self.assertEqual(len(snake_case__  )	,    vision_config.num_hidden_layers  )

							# in DEiT, the seq_len equals the number of patches + 2 (we add 2 for the [CLS] and distillation tokens)
							lowerCAmelCase__   = to_atuple(vision_model.config.image_size  )
							lowerCAmelCase__   = to_atuple(vision_model.config.patch_size  )
							lowerCAmelCase__   = (image_size[1] // patch_size[1]) * (image_size[0] // patch_size[0])
							lowerCAmelCase__   = num_patches + 2
							self.assertEqual(vision_attentions[0].shape[-3:]	,    (vision_config.num_attention_heads, seq_len, seq_len)  )

							lowerCAmelCase__   = output.text_model_output.attentions
							self.assertEqual(len(snake_case__  )	,    text_config.num_hidden_layers  )

							self.assertEqual(
							    text_attentions[0].shape[-3:]	,    (text_config.num_attention_heads, input_ids.shape[-1], input_ids.shape[-1])	,    )




					def     _SCREAMING_SNAKE_CASE    (						self    :  List[Any]	,    snake_case__    :  int	,    snake_case__    :  Optional[Any]  ):
							lowerCAmelCase__   = TFDeiTModel(snake_case__	,    name="""vision_model"""  )
							lowerCAmelCase__   = TFRobertaModel(snake_case__	,    name="""text_model"""  )
							return vision_model, text_model




					def     _SCREAMING_SNAKE_CASE    (						self    :  Dict  ):
							lowerCAmelCase__   = TFDeiTModelTester(self  )
							lowerCAmelCase__   = TFRobertaModelTester(self  )
							lowerCAmelCase__   = vit_model_tester.prepare_config_and_inputs()
							lowerCAmelCase__   = bert_model_tester.prepare_config_and_inputs()

							lowerCAmelCase__ ,       lowerCAmelCase__ ,       lowerCAmelCase__   = vision_config_and_inputs

							(
							    (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       
							)   = text_config_and_inputs

							return {
							    "text_config": text_config,
							    "vision_config": vision_config,
							    "pixel_values": pixel_values,
							    "attention_mask": input_mask,
							    "input_ids": input_ids,
							    "text_token_type_ids": token_type_ids,
							    "text_sequence_labels": sequence_labels,
							    "text_token_labels": token_labels,
							    "text_choice_labels": choice_labels,
							}

@require_tf
class    a_   (  __UpperCamelCase			,		unittest.TestCase      ):




					def     _SCREAMING_SNAKE_CASE    (						self    :  str  ):
							lowerCAmelCase__   = TFVisionTextDualEncoderModel.from_vision_text_pretrained(
							    """Rocketknight1/tiny-random-clip-tf"""	,    """hf-internal-testing/tiny-random-bert"""  )
							lowerCAmelCase__   = 13
							lowerCAmelCase__   = floats_tensor(
							    [
							        batch_size,
							        model.vision_model.config.num_channels,
							        model.vision_model.config.image_size,
							        model.vision_model.config.image_size,
							    ]  )
							lowerCAmelCase__   = ids_tensor([batch_size, 4]	,    model.text_model.config.vocab_size  )
							lowerCAmelCase__   = random_attention_mask([batch_size, 4]  )
							lowerCAmelCase__   = {"""pixel_values""": pixel_values, """input_ids""": input_ids, """attention_mask""": attention_mask}

							return model, inputs




					def     _SCREAMING_SNAKE_CASE    (						self    :  Dict	,    snake_case__    :  Optional[Any]	,    snake_case__    :  Tuple  ):
							lowerCAmelCase__   = TFCLIPVisionModel(snake_case__	,    name="""vision_model"""  )
							lowerCAmelCase__   = TFBertModel(snake_case__	,    name="""text_model"""  )
							return vision_model, text_model




					def     _SCREAMING_SNAKE_CASE    (						self    :  Union[str, Any]  ):
							lowerCAmelCase__   = TFCLIPVisionModelTester(self  )
							lowerCAmelCase__   = TFBertModelTester(self  )
							lowerCAmelCase__   = clip_model_tester.prepare_config_and_inputs()
							lowerCAmelCase__   = bert_model_tester.prepare_config_and_inputs()

							lowerCAmelCase__ ,       lowerCAmelCase__   = vision_config_and_inputs

							(
							    (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       (
							    lowerCAmelCase__
							) ,       
							)   = text_config_and_inputs

							return {
							    "text_config": text_config,
							    "vision_config": vision_config,
							    "pixel_values": pixel_values,
							    "attention_mask": input_mask,
							    "input_ids": input_ids,
							    "text_token_type_ids": token_type_ids,
							    "text_sequence_labels": sequence_labels,
							    "text_token_labels": token_labels,
							    "text_choice_labels": choice_labels,
							}

@require_vision
@require_tf
class    a_   (  unittest.TestCase      ):




					@slow
					def     _SCREAMING_SNAKE_CASE    (						self    :  Dict  ):
							lowerCAmelCase__   = TFVisionTextDualEncoderModel.from_pretrained(
							    """clip-italian/clip-italian"""	,    logit_scale_init_value=1.0	,    from_pt=snake_case__  )
							lowerCAmelCase__   = VisionTextDualEncoderProcessor.from_pretrained("""clip-italian/clip-italian"""  )

							lowerCAmelCase__   = Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""  )
							lowerCAmelCase__   = processor(
							    text=["""una foto di un gatto""", """una foto di un cane"""]	,    images=snake_case__	,    padding=snake_case__	,    return_tensors="""np"""  )

							lowerCAmelCase__   = model(**snake_case__  )

							# verify the logits
							self.assertEqual(outputs.logits_per_image.shape	,    (inputs.pixel_values.shape[0], inputs.input_ids.shape[0])  )
							self.assertEqual(
							    outputs.logits_per_text.shape	,    (inputs.input_ids.shape[0], inputs.pixel_values.shape[0])	,    )

							lowerCAmelCase__   = np.array([[1.228_4727, 0.310_4122]]  )

							self.assertTrue(np.allclose(outputs.logits_per_image.numpy()	,    snake_case__	,    atol=1E-3  )  )





"""simple docstring"""




from __future__ import annotations

from statistics import mean







def 				_UpperCAmelCase		(							lowerCamelCase__      ,	lowerCamelCase__      ,	lowerCamelCase__  ):





		"""simple docstring"""




		lowerCAmelCase__   = [0] * no_of_processes
		lowerCAmelCase__   = [0] * no_of_processes

		# Initialize remaining_time to waiting_time.

		for i in range(lowerCamelCase__  ):
				lowerCAmelCase__   = burst_time[i]
		lowerCAmelCase__   = []

		lowerCAmelCase__   = 0
		lowerCAmelCase__   = 0

		# When processes are not completed,
		# A process whose arrival time has passed \
		# and has remaining execution time is put into the ready_process.
		# The shortest process in the ready_process, target_process is executed.

		while completed != no_of_processes:
				lowerCAmelCase__   = []
				lowerCAmelCase__   = -1

				for i in range(lowerCamelCase__  ):
						if (arrival_time[i] <= total_time) and (remaining_time[i] > 0):
								ready_process.append(lowerCamelCase__  )

				if len(lowerCamelCase__  ) > 0:
						lowerCAmelCase__   = ready_process[0]
						for i in ready_process:
								if remaining_time[i] < remaining_time[target_process]:
										lowerCAmelCase__   = i
						total_time += burst_time[target_process]
						completed += 1
						lowerCAmelCase__   = 0
						lowerCAmelCase__   = (
						    total_time - arrival_time[target_process] - burst_time[target_process]
						)
				else:
						total_time += 1

		return waiting_time







def 				_UpperCAmelCase		(							lowerCamelCase__      ,	lowerCamelCase__      ,	lowerCamelCase__  ):





		"""simple docstring"""




		lowerCAmelCase__   = [0] * no_of_processes
		for i in range(lowerCamelCase__  ):
				lowerCAmelCase__   = burst_time[i] + waiting_time[i]
		return turn_around_time


if __name__ == "__main__":
			print("[TEST CASE 01]")

			__lowerCAmelCase     :				str    				=    4
			__lowerCAmelCase     :				Dict    				=    [2, 5, 3, 7]
			__lowerCAmelCase     :				List[str]    				=    [0, 0, 0, 0]
			__lowerCAmelCase     :				Union[str, Any]    				=    calculate_waitingtime(arrival_time, burst_time, no_of_processes)
			__lowerCAmelCase     :				Union[str, Any]    				=    calculate_turnaroundtime(
			    burst_time, no_of_processes, waiting_time
			)

			# Printing the Result
			print("PID\tBurst Time\tArrival Time\tWaiting Time\tTurnaround Time")
			for i, process_id in enumerate(list(range(1, 5))):
						print(
						    F"{process_id}\t{burst_time[i]}\t\t\t{arrival_time[i]}\t\t\t\t"
						    F"{waiting_time[i]}\t\t\t\t{turn_around_time[i]}"
						)
			print(F"\nAverage waiting time = {mean(waiting_time):.5f}")
			print(F"Average turnaround time = {mean(turn_around_time):.5f}")




