



'''simple docstring'''
#
# Sort large text files in a minimum amount of memory
#
import argparse
import os
class      __UpperCAmelCase							:



					'''simple docstring'''





					_UpperCamelCase  	=		"""block_{0}.dat"""



					def __init__(  self    :	Dict			,					_lowercase    :	Optional[Any])						->       Optional[int]:
								A_       		=   filename
								A_       		=   []



					def 						__snake_case	(  self    :	Optional[Any]			,					_lowercase    :	Optional[Any]			,					_lowercase    :	Optional[Any])						->       List[str]:
								A_       		=   self.BLOCK_FILENAME_FORMAT.format(lowerCAmelCase_)
								with open(lowerCAmelCase_			,					'w') as file:
											file.write(lowerCAmelCase_)
								self.block_filenames.append(lowerCAmelCase_)



					def 						__snake_case	(  self    :	Tuple)						->       str:
								return self.block_filenames



					def 						__snake_case	(  self    :	List[Any]			,					_lowercase    :	Tuple			,					_lowercase    :	Optional[int]=None)						->       Optional[Any]:
								A_       		=   0
								with open(self.filename) as file:
											while True:
														A_       		=   file.readlines(lowerCAmelCase_)

														if lines == []:
																	break

														if sort_key is None:
																	lines.sort()
														else:
																	lines.sort(key=lowerCAmelCase_)

														self.write_block(''.join(lowerCAmelCase_)			,					lowerCAmelCase_)
														i += 1



					def 						__snake_case	(  self    :	Any)						->       str:
								map(os.remove			,					self.block_filenames)





class      __UpperCAmelCase							:



					'''simple docstring'''





					def 						__snake_case	(  self    :	Any			,					_lowercase    :	str)						->       Optional[Any]:
								A_       		=   -1
								A_       		=   None

								for i in range(len(lowerCAmelCase_)):
											if min_str is None or choices[i] < min_str:
														A_       		=   i

								return min_index





class      __UpperCAmelCase							:



					'''simple docstring'''





					def __init__(  self    :	Tuple			,					_lowercase    :	Tuple)						->       Union[str, Any]:
								A_       		=   files
								A_       		=   set()
								A_       		=   len(lowerCAmelCase_)
								A_       		=   {i: None for i in range(self.num_buffers)}



					def 						__snake_case	(  self    :	int)						->       Any:
								return {
								    i: self.buffers[i] for i in range(self.num_buffers) if i not in self.empty
								}



					def 						__snake_case	(  self    :	Dict)						->       int:






								for i in range(self.num_buffers):
											if self.buffers[i] is None and i not in self.empty:
														A_       		=   self.files[i].readline()

														if self.buffers[i] == "":
																	self.empty.add(lowerCAmelCase_)
																	self.files[i].close()

								if len(self.empty) == self.num_buffers:
											return False

								return True



					def 						__snake_case	(  self    :	int			,					_lowercase    :	Union[str, Any])						->       int:
								A_       		=   self.buffers[index]
								A_       		=   None

								return value





class      __UpperCAmelCase							:



					'''simple docstring'''





					def __init__(  self    :	Any			,					_lowercase    :	int)						->       Union[str, Any]:
								A_       		=   merge_strategy



					def 						__snake_case	(  self    :	List[str]			,					_lowercase    :	Dict			,					_lowercase    :	Union[str, Any]			,					_lowercase    :	Union[str, Any])						->       str:
								A_       		=   FilesArray(self.get_file_handles(lowerCAmelCase_			,					lowerCAmelCase_))
								with open(lowerCAmelCase_			,					'w'			,					lowerCAmelCase_) as outfile:
											while buffers.refresh():
														A_       		=   self.merge_strategy.select(buffers.get_dict())
														outfile.write(buffers.unshift(lowerCAmelCase_))



					def 						__snake_case	(  self    :	List[str]			,					_lowercase    :	Union[str, Any]			,					_lowercase    :	str)						->       Tuple:
								A_       		=   {}

								for i in range(len(lowerCAmelCase_)):
											A_       		=   open(filenames[i]			,					'r'			,					lowerCAmelCase_)  # noqa: UP015

								return files





class      __UpperCAmelCase							:



					'''simple docstring'''





					def __init__(  self    :	int			,					_lowercase    :	List[Any])						->       str:
								A_       		=   block_size



					def 						__snake_case	(  self    :	Optional[Any]			,					_lowercase    :	Optional[int]			,					_lowercase    :	int=None)						->       int:
								A_       		=   self.get_number_blocks(lowerCAmelCase_			,					self.block_size)
								A_       		=   FileSplitter(lowerCAmelCase_)
								splitter.split(self.block_size			,					lowerCAmelCase_)

								A_       		=   FileMerger(NWayMerge())
								A_       		=   self.block_size / (num_blocks + 1)
								merger.merge(splitter.get_block_filenames()			,					filename + '.out'			,					lowerCAmelCase_)

								splitter.cleanup()



					def 						__snake_case	(  self    :	int			,					_lowercase    :	Dict			,					_lowercase    :	Tuple)						->       List[Any]:
								return (os.stat(lowerCAmelCase_).st_size / block_size) + 1





def  lowerCamelCase(       SCREAMING_SNAKE_CASE_    )		->		List[Any]:



			if string[-1].lower() == "k":
						return int(string[:-1]    ) * 1024
			elif string[-1].lower() == "m":
						return int(string[:-1]    ) * 1024 * 1024
			elif string[-1].lower() == "g":
						return int(string[:-1]    ) * 1024 * 1024 * 1024
			else:
						return int(SCREAMING_SNAKE_CASE_    )





def  lowerCamelCase(       )		->		Tuple:
			A_       		=   argparse.ArgumentParser()
			parser.add_argument(
			    '-m'					,'--mem'					,help='amount of memory to use for sorting'					,default='100M'    )
			parser.add_argument(
			    'filename'					,metavar='<filename>'					,nargs=1					,help='name of file to sort'    )
			A_       		=   parser.parse_args()

			A_       		=   ExternalSort(parse_memory(args.mem    )    )
			sorter.sort(args.filename[0]    )


if __name__ == "__main__":
							main()










'''simple docstring'''
import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


_SCREAMING_SNAKE_CASE =	namedtuple("ExampleNamedTuple", "a b c")
class      __UpperCAmelCase							(    unittest.TestCase							):



					'''simple docstring'''





					def 						__snake_case	(  self    :	List[Any])						->       Tuple:
								A_       		=   torch.randn(5			,					2)
								A_       		=   torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')

								A_       		=   send_to_device(_lowercase			,					_lowercase)
								self.assertTrue(torch.equal(resulta.cpu()			,					_lowercase))

								A_       		=   send_to_device((tensor, [tensor, tensor], 1)			,					_lowercase)
								self.assertIsInstance(_lowercase			,					_lowercase)
								self.assertTrue(torch.equal(resulta[0].cpu()			,					_lowercase))
								self.assertIsInstance(resulta[1]			,					_lowercase)
								self.assertTrue(torch.equal(resulta[1][0].cpu()			,					_lowercase))
								self.assertTrue(torch.equal(resulta[1][1].cpu()			,					_lowercase))
								self.assertEqual(resulta[2]			,					1)

								A_       		=   send_to_device({'a': tensor, 'b': [tensor, tensor], 'c': 1}			,					_lowercase)
								self.assertIsInstance(_lowercase			,					_lowercase)
								self.assertTrue(torch.equal(resulta['a'].cpu()			,					_lowercase))
								self.assertIsInstance(resulta['b']			,					_lowercase)
								self.assertTrue(torch.equal(resulta['b'][0].cpu()			,					_lowercase))
								self.assertTrue(torch.equal(resulta['b'][1].cpu()			,					_lowercase))
								self.assertEqual(resulta['c']			,					1)

								A_       		=   send_to_device(ExampleNamedTuple(a=_lowercase			,					b=[tensor, tensor]			,					c=1)			,					_lowercase)
								self.assertIsInstance(_lowercase			,					_lowercase)
								self.assertTrue(torch.equal(resulta.a.cpu()			,					_lowercase))
								self.assertIsInstance(resulta.b			,					_lowercase)
								self.assertTrue(torch.equal(resulta.b[0].cpu()			,					_lowercase))
								self.assertTrue(torch.equal(resulta.b[1].cpu()			,					_lowercase))
								self.assertEqual(resulta.c			,					1)

								A_       		=   send_to_device(UserDict({'a': tensor, 'b': [tensor, tensor], 'c': 1})			,					_lowercase)
								self.assertIsInstance(_lowercase			,					_lowercase)
								self.assertTrue(torch.equal(resulta['a'].cpu()			,					_lowercase))
								self.assertIsInstance(resulta['b']			,					_lowercase)
								self.assertTrue(torch.equal(resulta['b'][0].cpu()			,					_lowercase))
								self.assertTrue(torch.equal(resulta['b'][1].cpu()			,					_lowercase))
								self.assertEqual(resulta['c']			,					1)



					def 						__snake_case	(  self    :	Dict)						->       str:
								with self.assertRaises(_lowercase) as cm:
											A_       		=   recursively_apply(torch.tensor			,					(torch.tensor(1), 1)			,					error_on_other_type=_lowercase)
								self.assertEqual(
								    str(cm.exception)			,					'Unsupported types (<class \'int\'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed.'			,					)



					def 						__snake_case	(  self    :	Union[str, Any])						->       List[str]:
								A_       		=   torch.tensor([1, 2, 3, 4, 5])
								self.assertEqual(listify(_lowercase)			,					[1, 2, 3, 4, 5])

								A_       		=   torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]])
								self.assertEqual(listify(_lowercase)			,					[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]])

								A_       		=   torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]])
								self.assertEqual(
								    listify(_lowercase)			,					[[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]])



					def 						__snake_case	(  self    :	List[Any])						->       Union[str, Any]:
								with patch_environment(aa=1			,					BB=2):
											self.assertEqual(os.environ.get('AA')			,					'1')
											self.assertEqual(os.environ.get('BB')			,					'2')

								self.assertNotIn('AA'			,					os.environ)
								self.assertNotIn('BB'			,					os.environ)



					def 						__snake_case	(  self    :	Any)						->       str:
								A_       		=   RegressionModel()
								A_       		=   model.forward
								A_       		=   convert_outputs_to_fpaa(model.forward)
								A_       		=   extract_model_from_parallel(_lowercase			,					keep_fpaa_wrapper=_lowercase)
								A_       		=   pickle.dumps(_lowercase)



					@require_cuda
					def 						__snake_case	(  self    :	Union[str, Any])						->       Union[str, Any]:
								A_       		=   RegressionModel()
								A_       		=   model.forward
								A_       		=   torch.cuda.amp.autocast(dtype=torch.floataa)(model.forward)
								A_       		=   convert_outputs_to_fpaa(model.forward)
								A_       		=   extract_model_from_parallel(_lowercase			,					keep_fpaa_wrapper=_lowercase)
								A_       		=   pickle.dumps(_lowercase)



					@require_cuda
					@require_torch_min_version(version='2.0')
					def 						__snake_case	(  self    :	Optional[Any])						->       Dict:
								A_       		=   RegressionModel()
								A_       		=   model.forward
								A_       		=   torch.cuda.amp.autocast(dtype=torch.floataa)(model.forward)
								A_       		=   convert_outputs_to_fpaa(model.forward)
								A_       		=   torch.compile(model.forward			,					backend='inductor')
								A_       		=   torch.randn(4			,					10).cuda()
								A_       		=   model(_lowercase)



					def 						__snake_case	(  self    :	Union[str, Any])						->       Optional[Any]:
								A_       		=   RegressionModel()
								# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
								A_       		=   torch.nn.parallel.DataParallel(_lowercase)
								A_       		=   extract_model_from_parallel(_lowercase)

								self.assertEqual(_lowercase			,					_lowercase)



					@require_torch_min_version(version='2.0')
					def 						__snake_case	(  self    :	str)						->       int:
								A_       		=   RegressionModel()
								A_       		=   torch.compile(_lowercase)

								# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
								A_       		=   torch.nn.parallel.DataParallel(_lowercase)
								A_       		=   torch.compile(_lowercase)
								A_       		=   extract_model_from_parallel(_lowercase)

								self.assertEqual(compiled_model._orig_mod			,					compiled_model_unwrapped._orig_mod)



					def 						__snake_case	(  self    :	List[str])						->       Optional[int]:
								self.assertEqual(find_device([1, 'a', torch.tensor([1, 2, 3])])			,					torch.device('cpu'))
								self.assertEqual(find_device({'a': 1, 'b': torch.tensor([1, 2, 3])})			,					torch.device('cpu'))
								self.assertIsNone(find_device([1, 'a']))




