



"""simple docstring"""






import doctest
import logging
import os
import unittest
from pathlib import Path
from typing import List, Union

import transformers
from transformers.testing_utils import require_tf, require_torch, slow


__snake_case							:   Tuple      =    logging.getLogger()



@unittest.skip('Temporarily disable the doc tests.'							)
@require_torch
@require_tf
@slow
class 			A__   (  unittest.TestCase							):




	'''simple docstring'''





	def 							_SCREAMING_SNAKE_CASE   (     self:		Optional[Any]       ,			_SCREAMING_SNAKE_CASE:		Path       ,			_SCREAMING_SNAKE_CASE:		Union[str, None] = None       ,			_SCREAMING_SNAKE_CASE:		Union[List[str], None] = None       ,			_SCREAMING_SNAKE_CASE:		Union[str, List[str], None] = None       ,			_SCREAMING_SNAKE_CASE:		bool = True       ,			)				->       int:

				"""simple docstring"""

				__lowerCAmelCase      :  str												=       [file for file in os.listdir(_SCREAMING_SNAKE_CASE) if os.path.isfile(os.path.join(_SCREAMING_SNAKE_CASE       ,			_SCREAMING_SNAKE_CASE))]

				if identifier is not None:
							__lowerCAmelCase      :  Tuple												=       [file for file in files if identifier in file]

				if n_identifier is not None:
							if isinstance(_SCREAMING_SNAKE_CASE       ,			_SCREAMING_SNAKE_CASE):
										for n_ in n_identifier:
													__lowerCAmelCase      :  Dict												=       [file for file in files if n_ not in file]
							else:
										__lowerCAmelCase      :  List[str]												=       [file for file in files if n_identifier not in file]

				__lowerCAmelCase      :  List[Any]												=       ignore_files or []
				ignore_files.append("__init__.py")
				__lowerCAmelCase      :  List[str]												=       [file for file in files if file not in ignore_files]

				for file in files:
							# Open all files
							print("Testing"       ,			_SCREAMING_SNAKE_CASE)

							if only_modules:
										__lowerCAmelCase      :  Any												=       file.split(".")[0]
										try:
													__lowerCAmelCase      :  Any												=       getattr(_SCREAMING_SNAKE_CASE       ,			_SCREAMING_SNAKE_CASE)
													__lowerCAmelCase      :  Any												=       doctest.DocTestSuite(_SCREAMING_SNAKE_CASE)
													__lowerCAmelCase      :  str												=       unittest.TextTestRunner().run(_SCREAMING_SNAKE_CASE)
													self.assertIs(len(result.failures)       ,			0)
										except AttributeError:
													logger.info(F"""{module_identifier} is not a module.""")
							else:
										__lowerCAmelCase      :  List[Any]												=       doctest.testfile(str(".." / directory / file)       ,			optionflags=doctest.ELLIPSIS)
										self.assertIs(result.failed       ,			0)

	def 							_SCREAMING_SNAKE_CASE   (     self:		Optional[Any])				->       str:

				"""simple docstring"""

				__lowerCAmelCase      :  Union[str, Any]												=       Path("src/transformers")
				__lowerCAmelCase      :  Optional[Any]												=       "modeling"
				__lowerCAmelCase      :  List[str]												=       [
				    "modeling_ctrl.py",
				    "modeling_tf_ctrl.py",
				]
				self.analyze_directory(_SCREAMING_SNAKE_CASE       ,			identifier=_SCREAMING_SNAKE_CASE       ,			ignore_files=_SCREAMING_SNAKE_CASE)

	def 							_SCREAMING_SNAKE_CASE   (     self:		List[Any])				->       Any:

				"""simple docstring"""

				__lowerCAmelCase      :  Any												=       Path("src/transformers")
				__lowerCAmelCase      :  Optional[Any]												=       "tokenization"
				self.analyze_directory(_SCREAMING_SNAKE_CASE       ,			identifier=_SCREAMING_SNAKE_CASE)

	def 							_SCREAMING_SNAKE_CASE   (     self:		Dict)				->       Optional[Any]:

				"""simple docstring"""

				__lowerCAmelCase      :  Union[str, Any]												=       Path("src/transformers")
				__lowerCAmelCase      :  Union[str, Any]												=       "configuration"
				self.analyze_directory(_SCREAMING_SNAKE_CASE       ,			identifier=_SCREAMING_SNAKE_CASE)

	def 							_SCREAMING_SNAKE_CASE   (     self:		List[str])				->       List[str]:

				"""simple docstring"""

				__lowerCAmelCase      :  Union[str, Any]												=       Path("src/transformers")
				__lowerCAmelCase      :  Any												=       ["configuration", "modeling", "tokenization"]
				self.analyze_directory(_SCREAMING_SNAKE_CASE       ,			n_identifier=_SCREAMING_SNAKE_CASE)







	def 							_SCREAMING_SNAKE_CASE   (     self:		Optional[int])				->       List[Any]:

				"""simple docstring"""

				__lowerCAmelCase      :  Union[str, Any]												=       Path("docs/source")
				__lowerCAmelCase      :  Dict												=       ["favicon.ico"]
				self.analyze_directory(_SCREAMING_SNAKE_CASE       ,			ignore_files=_SCREAMING_SNAKE_CASE       ,			only_modules=_SCREAMING_SNAKE_CASE)




"""simple docstring"""






import gc
import threading
import time

import psutil
import torch



class 			A__   :




	'''simple docstring'''





	def __init__(     self:		str)				->       Optional[Any]:

				"""simple docstring"""

				__lowerCAmelCase      :  Optional[Any]												=       psutil.Process()
				__lowerCAmelCase      :  str												=       False

	def 							_SCREAMING_SNAKE_CASE   (     self:		int)				->       Optional[int]:

				"""simple docstring"""

				__lowerCAmelCase      :  Optional[Any]												=       -1

				while True:
							__lowerCAmelCase      :  str												=       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 							_SCREAMING_SNAKE_CASE   (     self:		Tuple)				->       int:

				"""simple docstring"""

				__lowerCAmelCase      :  List[str]												=       True
				__lowerCAmelCase      :  str												=       threading.Thread(target=self.peak_monitor)
				__lowerCAmelCase      :  Tuple												=       True
				self.thread.start()







	def 							_SCREAMING_SNAKE_CASE   (     self:		Optional[Any])				->       List[str]:

				"""simple docstring"""

				__lowerCAmelCase      :  Tuple												=       False
				self.thread.join()
				return self.cpu_memory_peak






__snake_case							:   Tuple      =    PeakCPUMemory()


def 			_lowercase				(  )						->   str:
			# Time
			__lowerCAmelCase      :  str												=       {"time": time.time()}

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

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

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

			return measures


def 			_lowercase				(  __snake_case	)						->   Optional[Any]:
			# Time
			__lowerCAmelCase      :  str												=       {"time": time.time() - start_measures["time"]}

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

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

			# GPU mem
			for i in range(torch.cuda.device_count()	):
						__lowerCAmelCase      :  Union[str, Any]												=       (torch.cuda.memory_allocated(__snake_case	) - start_measures[str(__snake_case	)]) / 2**20
						__lowerCAmelCase      :  Any												=       (torch.cuda.max_memory_allocated(__snake_case	) - start_measures[str(__snake_case	)]) / 2**20

			return measures





def 			_lowercase				(  __snake_case					,__snake_case	)						->   Dict:
			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(__snake_case	)]:.2f}MiB"""	)
						__lowerCAmelCase      :  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"""	)