






import inspect
import unittest

from huggingface_hub import hf_hub_download

from transformers import ASTConfig
from transformers.testing_utils import require_torch, require_torchaudio, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_torchaudio_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
							import torch
							from torch import nn

							from transformers import ASTForAudioClassification, ASTModel
							from transformers.models.audio_spectrogram_transformer.modeling_audio_spectrogram_transformer import (
							    AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
							)


if is_torchaudio_available():
							import torchaudio

							from transformers import ASTFeatureExtractor

class 				lowerCAmelCase_	:


				def __init__(					self ,						_lowerCAmelCase ,						_lowerCAmelCase=1_3 ,						_lowerCAmelCase=2 ,						_lowerCAmelCase=2_4 ,						_lowerCAmelCase=1_6 ,						_lowerCAmelCase=True ,						_lowerCAmelCase=True ,						_lowerCAmelCase=3_2 ,						_lowerCAmelCase=5 ,						_lowerCAmelCase=4 ,						_lowerCAmelCase=3_7 ,						_lowerCAmelCase="gelu" ,						_lowerCAmelCase=0.1 ,						_lowerCAmelCase=0.1 ,						_lowerCAmelCase=1_0 ,						_lowerCAmelCase=0.02 ,						_lowerCAmelCase=None ,						_lowerCAmelCase=2 ,						_lowerCAmelCase=2 ,						):
											_lowercase							:   List[Any] 							=   parent
											_lowercase							:   Dict 							=   batch_size
											_lowercase							:   int 							=   patch_size
											_lowercase							:   int 							=   max_length
											_lowercase							:   Optional[Any] 							=   num_mel_bins
											_lowercase							:   Any 							=   is_training
											_lowercase							:   int 							=   use_labels
											_lowercase							:   Dict 							=   hidden_size
											_lowercase							:   int 							=   num_hidden_layers
											_lowercase							:   Any 							=   num_attention_heads
											_lowercase							:   str 							=   intermediate_size
											_lowercase							:   Tuple 							=   hidden_act
											_lowercase							:   Optional[int] 							=   hidden_dropout_prob
											_lowercase							:   List[Any] 							=   attention_probs_dropout_prob
											_lowercase							:   Dict 							=   type_sequence_label_size
											_lowercase							:   List[str] 							=   initializer_range
											_lowercase							:   List[Any] 							=   scope
											_lowercase							:   str 							=   frequency_stride
											_lowercase							:   Optional[Any] 							=   time_stride

											# in AST, the seq length equals the number of patches + 2 (we add 2 for the [CLS] and distillation tokens)
											_lowercase							:   Optional[Any] 							=   (self.num_mel_bins - self.patch_size) // self.frequency_stride + 1
											_lowercase							:   int 							=   (self.max_length - self.patch_size) // self.time_stride + 1
											_lowercase							:   List[Any] 							=   frequency_out_dimension * time_out_dimension
											_lowercase							:   Tuple 							=   num_patches + 2


				def 						__a					(					self      ):
											_lowercase							:   List[str] 							=   floats_tensor([self.batch_size, self.max_length, self.num_mel_bins]      )

											_lowercase							:   Dict 							=   None
											if self.use_labels:
																		_lowercase							:   int 							=   ids_tensor([self.batch_size] ,						self.type_sequence_label_size      )

											_lowercase							:   Union[str, Any] 							=   self.get_config()

											return config, input_values, labels


				def 						__a					(					self      ):
											return ASTConfig(
											    patch_size=self.patch_size ,						max_length=self.max_length ,						num_mel_bins=self.num_mel_bins ,						hidden_size=self.hidden_size ,						num_hidden_layers=self.num_hidden_layers ,						num_attention_heads=self.num_attention_heads ,						intermediate_size=self.intermediate_size ,						hidden_act=self.hidden_act ,						hidden_dropout_prob=self.hidden_dropout_prob ,						attention_probs_dropout_prob=self.attention_probs_dropout_prob ,						is_decoder=_lowerCAmelCase ,						initializer_range=self.initializer_range ,						frequency_stride=self.frequency_stride ,						time_stride=self.time_stride ,						)


				def 						__a					(					self ,						_lowerCAmelCase ,						_lowerCAmelCase ,						_lowerCAmelCase      ):
											_lowercase							:   Dict 							=   ASTModel(config=_lowerCAmelCase      )
											model.to(_lowerCAmelCase      )
											model.eval()
											_lowercase							:   List[str] 							=   model(_lowerCAmelCase      )
											self.parent.assertEqual(result.last_hidden_state.shape ,						(self.batch_size, self.seq_length, self.hidden_size)      )


				def 						__a					(					self      ):
											_lowercase							:   Optional[int] 							=   self.prepare_config_and_inputs()
											(
											    (
											    _lowercase
											)							,							(
											    _lowercase
											)							,							(
											    _lowercase
											)							,							
											)							:   Union[str, Any] 							=   config_and_inputs
											_lowercase							:   Optional[int] 							=   {'input_values': input_values}
											return config, inputs_dict




@require_torch
class 				lowerCAmelCase_	(		__snake_case   ,		__snake_case   ,		unittest.TestCase			):
				_UpperCamelCase					:       List[Any]							       = (
				    (
				        ASTModel,
				        ASTForAudioClassification,
				    )
				    if is_torch_available()
				    else ()
				)
				_UpperCamelCase					:       Tuple							       = (
				    {"audio-classification": ASTForAudioClassification, "feature-extraction": ASTModel}
				    if is_torch_available()
				    else {}
				)
				_UpperCamelCase					:       int							       = False
				_UpperCamelCase					:       Optional[Any]							       = False
				_UpperCamelCase					:       Any							       = False
				_UpperCamelCase					:       str							       = False


				def 						__a					(					self ,						_lowerCAmelCase ,						_lowerCAmelCase ,						_lowerCAmelCase ,						_lowerCAmelCase ,						_lowerCAmelCase      ):
											if pipeline_test_casse_name == "AudioClassificationPipelineTests":
																		return True

											return False


				def 						__a					(					self      ):
											_lowercase							:   Optional[Any] 							=   ASTModelTester(self      )
											_lowercase							:   List[Any] 							=   ConfigTester(self ,						config_class=_lowerCAmelCase ,						has_text_modality=_lowerCAmelCase ,						hidden_size=3_7      )


				def 						__a					(					self      ):
											self.config_tester.run_common_tests()


				@unittest.skip(reason='AST does not use inputs_embeds'      )
				def 						__a					(					self      ):
											pass


				def 						__a					(					self      ):
											_lowercase							,							_lowercase							:   List[str] 							=   self.model_tester.prepare_config_and_inputs_for_common()

											for model_class in self.all_model_classes:
																		_lowercase							:   List[Any] 							=   model_class(_lowerCAmelCase      )
																		self.assertIsInstance(model.get_input_embeddings() ,						(nn.Module)      )
																		_lowercase							:   Tuple 							=   model.get_output_embeddings()
																		self.assertTrue(x is None or isinstance(_lowerCAmelCase ,						nn.Linear      )      )


				def 						__a					(					self      ):
											_lowercase							,							_lowercase							:   Union[str, Any] 							=   self.model_tester.prepare_config_and_inputs_for_common()

											for model_class in self.all_model_classes:
																		_lowercase							:   List[Any] 							=   model_class(_lowerCAmelCase      )
																		_lowercase							:   Tuple 							=   inspect.signature(model.forward      )
																		# signature.parameters is an OrderedDict => so arg_names order is deterministic
																		_lowercase							:   List[str] 							=   [*signature.parameters.keys()]

																		_lowercase							:   Tuple 							=   ['input_values']
																		self.assertListEqual(arg_names[:1] ,						_lowerCAmelCase      )


				def 						__a					(					self      ):
											_lowercase							:   Optional[int] 							=   self.model_tester.prepare_config_and_inputs()
											self.model_tester.create_and_check_model(*_lowerCAmelCase      )


				@slow
				def 						__a					(					self      ):
											for model_name in AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																		_lowercase							:   Any 							=   ASTModel.from_pretrained(_lowerCAmelCase      )
																		self.assertIsNotNone(_lowerCAmelCase      )






def    __magic_name__    (	)     ->    Any:
							_lowercase							:   List[Any] 							=   hf_hub_download(
							    repo_id='nielsr/audio-spectogram-transformer-checkpoint'     ,     filename='sample_audio.flac'     ,     repo_type='dataset'				)

							_lowercase							,							_lowercase							:   Tuple 							=   torchaudio.load(SCREAMING_SNAKE_CASE				)

							return audio, sampling_rate






@require_torch
@require_torchaudio
class 				lowerCAmelCase_	(		unittest.TestCase			):


				@cached_property
				def 						__a					(					self      ):
											return (
											    ASTFeatureExtractor.from_pretrained('MIT/ast-finetuned-audioset-10-10-0.4593'      )
											    if is_torchaudio_available()
											    else None
											)


				@slow
				def 						__a					(					self      ):
											_lowercase							:   List[Any] 							=   self.default_feature_extractor
											_lowercase							:   List[Any] 							=   ASTForAudioClassification.from_pretrained('MIT/ast-finetuned-audioset-10-10-0.4593'      ).to(_lowerCAmelCase      )

											_lowercase							:   List[Any] 							=   self.default_feature_extractor
											_lowercase							,							_lowercase							:   Any 							=   prepare_audio()
											_lowercase							:   int 							=   audio.squeeze().numpy()
											_lowercase							:   List[str] 							=   feature_extractor(_lowerCAmelCase ,						sampling_rate=_lowerCAmelCase ,						return_tensors='pt'      ).to(_lowerCAmelCase      )

											# forward pass
											with torch.no_grad():
																		_lowercase							:   List[str] 							=   model(**_lowerCAmelCase      )

											# verify the logits
											_lowercase							:   Any 							=   torch.Size((1, 5_2_7)      )
											self.assertEqual(outputs.logits.shape ,						_lowerCAmelCase      )

											_lowercase							:   Dict 							=   torch.tensor([-0.87_60, -7.00_42, -8.66_02]      ).to(_lowerCAmelCase      )

											self.assertTrue(torch.allclose(outputs.logits[0, :3] ,						_lowerCAmelCase ,						atol=1E-4      )      )









from __future__ import annotations

import time

import numpy as np

lowercase_													=      [8, 5, 9, 7]
lowercase_													=      [
    [2, 0, 1, 1],
    [0, 1, 2, 1],
    [4, 0, 0, 3],
    [0, 2, 1, 0],
    [1, 0, 3, 0],
]
lowercase_													=      [
    [3, 2, 1, 4],
    [0, 2, 5, 2],
    [5, 1, 0, 5],
    [1, 5, 3, 0],
    [3, 0, 3, 3],
]






class     __a			:



	def __init__(    self				:     Tuple					,						snake_case_				:     list[int]					,						snake_case_				:     list[list[int]]					,						snake_case_				:     list[list[int]]					,						)->				None:
						__lowerCAmelCase						 =claim_vector
						__lowerCAmelCase						 =allocated_resources_table
						__lowerCAmelCase						 =maximum_claim_table



	def 		UpperCamelCase				(    self				:     Optional[int])->				list[int]:
						return [
						    sum(p_item[i] for p_item in self.__allocated_resources_table)
						    for i in range(len(self.__allocated_resources_table[0]))
						]



	def 		UpperCamelCase				(    self				:     Dict)->				list[int]:
						return np.array(self.__claim_vector) - np.array(
						    self.__processes_resource_summation())



	def 		UpperCamelCase				(    self				:     Optional[Any])->				list[list[int]]:
						return [
						    list(np.array(self.__maximum_claim_table[i]) - np.array(snake_case_))
						    for i, allocated_resource in enumerate(self.__allocated_resources_table)
						]



	def 		UpperCamelCase				(    self				:     Dict)->				dict[int, list[int]]:
						return {self.__need().index(snake_case_): i for i in self.__need()}



	def 		UpperCamelCase				(    self				:     Optional[Any]					,						**snake_case_				:     Optional[Any])->				None:
						__lowerCAmelCase						 =self.__need()
						__lowerCAmelCase						 =self.__allocated_resources_table
						__lowerCAmelCase						 =self.__available_resources()
						__lowerCAmelCase						 =self.__need_index_manager()
						for kw, val in kwargs.items():
											if kw and val is True:
																self.__pretty_data()
						print("""_""" * 50 + """\n""")
						while need_list:
											__lowerCAmelCase						 =False
											for each_need in need_list:
																__lowerCAmelCase						 =True
																for index, need in enumerate(snake_case_):
																					if need > available_resources[index]:
																										__lowerCAmelCase						 =False
																										break
																if execution:
																					__lowerCAmelCase						 =True
																					# get the original index of the process from ind_ctrl db
																					for original_need_index, need_clone in need_index_manager.items():
																										if each_need == need_clone:
																															__lowerCAmelCase						 =original_need_index
																					print(F"""Process {process_number + 1} is executing.""")
																					# remove the process run from stack
																					need_list.remove(snake_case_)
																					# update available/freed resources stack
																					__lowerCAmelCase						 =np.array(snake_case_) + np.array(
																					    alloc_resources_table[process_number])
																					print(
																					    """Updated available resource stack for processes: """
																					    + """ """.join([str(snake_case_) for x in available_resources]))
																					break
											if safe:
																print("""The process is in a safe state.\n""")
											else:
																print("""System in unsafe state. Aborting...\n""")
																break



	def 		UpperCamelCase				(    self				:     List[str])->				Optional[Any]:
						print(""" """ * 9 + """Allocated Resource Table""")
						for item in self.__allocated_resources_table:
											print(
											    F"""P{self.__allocated_resources_table.index(snake_case_) + 1}"""
											    + """ """.join(F"""{it:>8}""" for it in item)
											    + """\n""")
						print(""" """ * 9 + """System Resource Table""")
						for item in self.__maximum_claim_table:
											print(
											    F"""P{self.__maximum_claim_table.index(snake_case_) + 1}"""
											    + """ """.join(F"""{it:>8}""" for it in item)
											    + """\n""")
						print(
						    """Current Usage by Active Processes: """
						    + """ """.join(str(snake_case_) for x in self.__claim_vector))
						print(
						    """Initial Available Resources:       """
						    + """ """.join(str(snake_case_) for x in self.__available_resources()))
						time.sleep(1)


if __name__ == "__main__":
				import doctest

				doctest.testmod()




