




"""simple docstring"""




import gc
import threading
import time

import psutil
import torch

class    UpperCAmelCase_  :
		def __init__(		self    )		->    str:
					__lowercase       :   List[str]        =    psutil.Process()
					__lowercase       :   Optional[Any]        =    False
		def      _lowerCamelCase				(		self    )		->    List[str]:
					__lowercase       :   List[str]        =    -1

					while True:
								__lowercase       :   Union[str, Any]        =    max(self.process.memory_info().rss							,		self.cpu_memory_peak    )

								# can't sleep or will not catch the peak right (this comment is here on purpose)
								if not self.peak_monitoring:
											break
		def      _lowerCamelCase				(		self    )		->    Dict:
					__lowercase       :   Dict        =    True
					__lowercase       :   List[str]        =    threading.Thread(target=self.peak_monitor    )
					__lowercase       :   Optional[Any]        =    True
					self.thread.start()
		def      _lowerCamelCase				(		self    )		->    Union[str, Any]:
					__lowercase       :   Any        =    False
					self.thread.join()
					return self.cpu_memory_peak


a_	   =    PeakCPUMemory()







def 			__UpperCAmelCase			(      ):
			__lowercase       :   Dict        =    {"""time""": time.time()}

			gc.collect()
			torch.cuda.empty_cache()

			# CPU mem
			__lowercase       :   Optional[Any]        =    psutil.Process().memory_info().rss
			cpu_peak_tracker.start()

			# GPU mem
			for i in range(torch.cuda.device_count()     ):
						__lowercase       :   Union[str, Any]        =    torch.cuda.memory_allocated(__UpperCamelCase     )
			torch.cuda.reset_peak_memory_stats()

			return measures







def 			__UpperCAmelCase			(      __UpperCamelCase     ):
			__lowercase       :   List[Any]        =    {"""time""": time.time() - start_measures["""time"""]}

			gc.collect()
			torch.cuda.empty_cache()

			# CPU mem
			__lowercase       :   str        =    (psutil.Process().memory_info().rss - start_measures["""cpu"""]) / 2**20
			__lowercase       :   Optional[Any]        =    (cpu_peak_tracker.stop() - start_measures["""cpu"""]) / 2**20

			# GPU mem
			for i in range(torch.cuda.device_count()     ):
						__lowercase       :   Optional[int]        =    (torch.cuda.memory_allocated(__UpperCamelCase     ) - start_measures[str(__UpperCamelCase     )]) / 2**20
						__lowercase       :   Any        =    (torch.cuda.max_memory_allocated(__UpperCamelCase     ) - start_measures[str(__UpperCamelCase     )]) / 2**20

			return measures







def 			__UpperCAmelCase			(      __UpperCamelCase     ,							__UpperCamelCase     ):
			print(f"""{description}:"""     )
			print(f"""- Time: {measures["time"]:.2f}s"""     )
			for i in range(torch.cuda.device_count()     ):
						print(f"""- GPU {i} allocated: {measures[str(__UpperCamelCase     )]:.2f}MiB"""     )
						__lowercase       :   Optional[Any]        =    measures[f"""{i}-peak"""]
						print(f"""- GPU {i} peak: {peak:.2f}MiB"""     )
			print(f"""- CPU RAM allocated: {measures["cpu"]:.2f}MiB"""     )
			print(f"""- CPU RAM peak: {measures["cpu-peak"]:.2f}MiB"""     )











# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from ..models.whisper import WhisperForConditionalGeneration, WhisperProcessor
from .base import PipelineTool

class     _SCREAMING_SNAKE_CASE       (			_a      ):
	snake_case__							:     Any     	=  """openai/whisper-base"""
	snake_case__							:     Optional[int]     	=  (
	    """This is a tool that transcribes an audio into text. It takes an input named `audio` and returns the """
	    """transcribed text."""
	)
	snake_case__							:     Any     	=  """transcriber"""
	snake_case__							:     Optional[int]     	=  WhisperProcessor
	snake_case__							:     str     	=  WhisperForConditionalGeneration

	snake_case__							:     Optional[Any]     	=  ["""audio"""]
	snake_case__							:     Any     	=  ["""text"""]





	def 						_A (		self				:					str ,  __lowerCamelCase				:					Dict	):
		return self.pre_processor(__lowerCamelCase ,  return_tensors="""pt"""	).input_features





	def 						_A (		self				:					Dict ,  __lowerCamelCase				:					List[Any]	):
		return self.model.generate(inputs=__lowerCamelCase	)





	def 						_A (		self				:					Any ,  __lowerCamelCase				:					Optional[Any]	):
		return self.pre_processor.batch_decode(__lowerCamelCase ,  skip_special_tokens=__lowerCamelCase	)[0]
