



#
# Sort large text files in a minimum amount of memory
#
import argparse
import os





class    __lowerCAmelCase					:
				__lowerCamelCase					=							'''block_{0}.dat'''




				def __init__(				self		,							_snake_case						):



									"""simple docstring"""





									_lowerCAmelCase         =       filename
									_lowerCAmelCase         =       []




				def 				snake_case  (				self		,							_snake_case		,							_snake_case						):



									"""simple docstring"""





									_lowerCAmelCase         =       self.BLOCK_FILENAME_FORMAT.format(_snake_case						)
									with open(_snake_case		,							"""w"""						) as file:
														file.write(_snake_case						)
									self.block_filenames.append(_snake_case						)




				def 				snake_case  (				self						):



									"""simple docstring"""





									return self.block_filenames




				def 				snake_case  (				self		,							_snake_case		,							_snake_case=None						):



									"""simple docstring"""





									_lowerCAmelCase         =       0
									with open(self.filename						) as file:
														while True:
																			_lowerCAmelCase         =       file.readlines(_snake_case						)

																			if lines == []:
																								break

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

																			self.write_block("""""".join(_snake_case						)		,							_snake_case						)
																			i += 1




				def 				snake_case  (				self						):



									"""simple docstring"""





									map(os.remove		,							self.block_filenames						)





class    __lowerCAmelCase					:




				def 				snake_case  (				self		,							_snake_case						):



									"""simple docstring"""





									_lowerCAmelCase         =       -1
									_lowerCAmelCase         =       None

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

									return min_index





class    __lowerCAmelCase					:




				def __init__(				self		,							_snake_case						):



									"""simple docstring"""





									_lowerCAmelCase         =       files
									_lowerCAmelCase         =       set()
									_lowerCAmelCase         =       len(_snake_case						)
									_lowerCAmelCase         =       {i: None for i in range(self.num_buffers						)}




				def 				snake_case  (				self						):



									"""simple docstring"""





									return {
									    i: self.buffers[i] for i in range(self.num_buffers						) if i not in self.empty
									}




				def 				snake_case  (				self						):



									"""simple docstring"""





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

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

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

									return True




				def 				snake_case  (				self		,							_snake_case						):



									"""simple docstring"""





									_lowerCAmelCase         =       self.buffers[index]
									_lowerCAmelCase         =       None

									return value





class    __lowerCAmelCase					:




				def __init__(				self		,							_snake_case						):



									"""simple docstring"""





									_lowerCAmelCase         =       merge_strategy




				def 				snake_case  (				self		,							_snake_case		,							_snake_case		,							_snake_case						):



									"""simple docstring"""





									_lowerCAmelCase         =       FilesArray(self.get_file_handles(_snake_case		,							_snake_case						)						)
									with open(_snake_case		,							"""w"""		,							_snake_case						) as outfile:
														while buffers.refresh():
																			_lowerCAmelCase         =       self.merge_strategy.select(buffers.get_dict()						)
																			outfile.write(buffers.unshift(_snake_case						)						)




				def 				snake_case  (				self		,							_snake_case		,							_snake_case						):



									"""simple docstring"""





									_lowerCAmelCase         =       {}

									for i in range(len(_snake_case						)						):
														_lowerCAmelCase         =       open(filenames[i]		,							"""r"""		,							_snake_case						)  # noqa: UP015

									return files





class    __lowerCAmelCase					:




				def __init__(				self		,							_snake_case						):



									"""simple docstring"""





									_lowerCAmelCase         =       block_size




				def 				snake_case  (				self		,							_snake_case		,							_snake_case=None						):



									"""simple docstring"""





									_lowerCAmelCase         =       self.get_number_blocks(_snake_case		,							self.block_size						)
									_lowerCAmelCase         =       FileSplitter(_snake_case						)
									splitter.split(self.block_size		,							_snake_case						)

									_lowerCAmelCase         =       FileMerger(NWayMerge()						)
									_lowerCAmelCase         =       self.block_size / (num_blocks + 1)
									merger.merge(splitter.get_block_filenames()		,							filename + """.out"""		,							_snake_case						)

									splitter.cleanup()




				def 				snake_case  (				self		,							_snake_case		,							_snake_case						):



									"""simple docstring"""





									return (os.stat(_snake_case						).st_size / block_size) + 1


def 						_UpperCAmelCase							( snake_case     ):





					"""simple docstring"""

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


def 						_UpperCAmelCase							( ):





					"""simple docstring"""

					_lowerCAmelCase         =       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"""     )
					_lowerCAmelCase         =       parser.parse_args()

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


if __name__ == "__main__":
	main()





'''simple docstring'''


from typing import Any, Callable, Dict, List, Optional, Union

import PIL.Image
import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionImgaImgPipeline,
    StableDiffusionInpaintPipelineLegacy,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.configuration_utils import FrozenDict
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.utils import deprecate, logging


__SCREAMING_SNAKE_CASE						=      logging.get_logger(__name__)  # pylint: disable=invalid-name




class     __snake_case			(    _a	):
		a_          =					["safety_checker", "feature_extractor"]




		def __init__(  self					:str			,    _SCREAMING_SNAKE_CASE					:AutoencoderKL			,    _SCREAMING_SNAKE_CASE					:CLIPTextModel			,    _SCREAMING_SNAKE_CASE					:CLIPTokenizer			,    _SCREAMING_SNAKE_CASE					:UNetaDConditionModel			,    _SCREAMING_SNAKE_CASE					:Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]			,    _SCREAMING_SNAKE_CASE					:StableDiffusionSafetyChecker			,    _SCREAMING_SNAKE_CASE					:CLIPImageProcessor			,    _SCREAMING_SNAKE_CASE					:bool = True			,    )					->			int:






								'''simple docstring'''






								super().__init__()
								if hasattr(scheduler.config			,    """steps_offset""") and scheduler.config.steps_offset != 1:
														UpperCamelCase_     :	Optional[Any]					     =     (
														    F"""The configuration file of this scheduler: {scheduler} is outdated. `steps_offset`"""
														    F""" should be set to 1 instead of {scheduler.config.steps_offset}. Please make sure """
														    """to update the config accordingly as leaving `steps_offset` might led to incorrect results"""
														    """ in future versions. If you have downloaded this checkpoint from the Hugging Face Hub,"""
														    """ it would be very nice if you could open a Pull request for the `scheduler/scheduler_config.json`"""
														    """ file"""
														)
														deprecate("""steps_offset!=1"""			,    """1.0.0"""			,    _SCREAMING_SNAKE_CASE			,    standard_warn=_SCREAMING_SNAKE_CASE)
														UpperCamelCase_     :	Union[str, Any]					     =     dict(scheduler.config)
														UpperCamelCase_     :	Optional[Any]					     =     1
														UpperCamelCase_     :	int					     =     FrozenDict(_SCREAMING_SNAKE_CASE)

								self.register_modules(
								    vae=_SCREAMING_SNAKE_CASE			,    text_encoder=_SCREAMING_SNAKE_CASE			,    tokenizer=_SCREAMING_SNAKE_CASE			,    unet=_SCREAMING_SNAKE_CASE			,    scheduler=_SCREAMING_SNAKE_CASE			,    safety_checker=_SCREAMING_SNAKE_CASE			,    feature_extractor=_SCREAMING_SNAKE_CASE			,    )
								self.register_to_config(requires_safety_checker=_SCREAMING_SNAKE_CASE)




		@property
		def 	_snake_case	(  self					:Tuple)					->			Dict[str, Any]:






								'''simple docstring'''






								return {k: getattr(self			,    _SCREAMING_SNAKE_CASE) for k in self.config.keys() if not k.startswith("""_""")}




		def 	_snake_case	(  self					:Optional[int]			,    _SCREAMING_SNAKE_CASE					:Optional[Union[str, int]] = "auto")					->			Union[str, Any]:






								'''simple docstring'''






								if slice_size == "auto":
														# half the attention head size is usually a good trade-off between
														# speed and memory
														UpperCamelCase_     :	int					     =     self.unet.config.attention_head_dim // 2
								self.unet.set_attention_slice(_SCREAMING_SNAKE_CASE)




		def 	_snake_case	(  self					:Union[str, Any])					->			Dict:






								'''simple docstring'''






								self.enable_attention_slicing(_SCREAMING_SNAKE_CASE)




		@torch.no_grad()
		def 	_snake_case	(  self					:Union[str, Any]			,    _SCREAMING_SNAKE_CASE					:Union[str, List[str]]			,    _SCREAMING_SNAKE_CASE					:Union[torch.FloatTensor, PIL.Image.Image]			,    _SCREAMING_SNAKE_CASE					:Union[torch.FloatTensor, PIL.Image.Image]			,    _SCREAMING_SNAKE_CASE					:float = 0.8			,    _SCREAMING_SNAKE_CASE					:Optional[int] = 5_0			,    _SCREAMING_SNAKE_CASE					:Optional[float] = 7.5			,    _SCREAMING_SNAKE_CASE					:Optional[Union[str, List[str]]] = None			,    _SCREAMING_SNAKE_CASE					:Optional[int] = 1			,    _SCREAMING_SNAKE_CASE					:Optional[float] = 0.0			,    _SCREAMING_SNAKE_CASE					:Optional[torch.Generator] = None			,    _SCREAMING_SNAKE_CASE					:Optional[str] = "pil"			,    _SCREAMING_SNAKE_CASE					:bool = True			,    _SCREAMING_SNAKE_CASE					:Optional[Callable[[int, int, torch.FloatTensor], None]] = None			,    _SCREAMING_SNAKE_CASE					:int = 1			,    )					->			List[str]:






								'''simple docstring'''






								return StableDiffusionInpaintPipelineLegacy(**self.components)(
								    prompt=_SCREAMING_SNAKE_CASE			,    image=_SCREAMING_SNAKE_CASE			,    mask_image=_SCREAMING_SNAKE_CASE			,    strength=_SCREAMING_SNAKE_CASE			,    num_inference_steps=_SCREAMING_SNAKE_CASE			,    guidance_scale=_SCREAMING_SNAKE_CASE			,    negative_prompt=_SCREAMING_SNAKE_CASE			,    num_images_per_prompt=_SCREAMING_SNAKE_CASE			,    eta=_SCREAMING_SNAKE_CASE			,    generator=_SCREAMING_SNAKE_CASE			,    output_type=_SCREAMING_SNAKE_CASE			,    return_dict=_SCREAMING_SNAKE_CASE			,    callback=_SCREAMING_SNAKE_CASE			,    )




		@torch.no_grad()
		def 	_snake_case	(  self					:List[str]			,    _SCREAMING_SNAKE_CASE					:Union[str, List[str]]			,    _SCREAMING_SNAKE_CASE					:Union[torch.FloatTensor, PIL.Image.Image]			,    _SCREAMING_SNAKE_CASE					:float = 0.8			,    _SCREAMING_SNAKE_CASE					:Optional[int] = 5_0			,    _SCREAMING_SNAKE_CASE					:Optional[float] = 7.5			,    _SCREAMING_SNAKE_CASE					:Optional[Union[str, List[str]]] = None			,    _SCREAMING_SNAKE_CASE					:Optional[int] = 1			,    _SCREAMING_SNAKE_CASE					:Optional[float] = 0.0			,    _SCREAMING_SNAKE_CASE					:Optional[torch.Generator] = None			,    _SCREAMING_SNAKE_CASE					:Optional[str] = "pil"			,    _SCREAMING_SNAKE_CASE					:bool = True			,    _SCREAMING_SNAKE_CASE					:Optional[Callable[[int, int, torch.FloatTensor], None]] = None			,    _SCREAMING_SNAKE_CASE					:int = 1			,    **_SCREAMING_SNAKE_CASE					:int			,    )					->			Any:






								'''simple docstring'''






								return StableDiffusionImgaImgPipeline(**self.components)(
								    prompt=_SCREAMING_SNAKE_CASE			,    image=_SCREAMING_SNAKE_CASE			,    strength=_SCREAMING_SNAKE_CASE			,    num_inference_steps=_SCREAMING_SNAKE_CASE			,    guidance_scale=_SCREAMING_SNAKE_CASE			,    negative_prompt=_SCREAMING_SNAKE_CASE			,    num_images_per_prompt=_SCREAMING_SNAKE_CASE			,    eta=_SCREAMING_SNAKE_CASE			,    generator=_SCREAMING_SNAKE_CASE			,    output_type=_SCREAMING_SNAKE_CASE			,    return_dict=_SCREAMING_SNAKE_CASE			,    callback=_SCREAMING_SNAKE_CASE			,    callback_steps=_SCREAMING_SNAKE_CASE			,    )





		@torch.no_grad()
		def 	_snake_case	(  self					:Any			,    _SCREAMING_SNAKE_CASE					:Union[str, List[str]]			,    _SCREAMING_SNAKE_CASE					:int = 5_1_2			,    _SCREAMING_SNAKE_CASE					:int = 5_1_2			,    _SCREAMING_SNAKE_CASE					:int = 5_0			,    _SCREAMING_SNAKE_CASE					:float = 7.5			,    _SCREAMING_SNAKE_CASE					:Optional[Union[str, List[str]]] = None			,    _SCREAMING_SNAKE_CASE					:Optional[int] = 1			,    _SCREAMING_SNAKE_CASE					:float = 0.0			,    _SCREAMING_SNAKE_CASE					:Optional[torch.Generator] = None			,    _SCREAMING_SNAKE_CASE					:Optional[torch.FloatTensor] = None			,    _SCREAMING_SNAKE_CASE					:Optional[str] = "pil"			,    _SCREAMING_SNAKE_CASE					:bool = True			,    _SCREAMING_SNAKE_CASE					:Optional[Callable[[int, int, torch.FloatTensor], None]] = None			,    _SCREAMING_SNAKE_CASE					:int = 1			,    )					->			List[Any]:






								'''simple docstring'''






								return StableDiffusionPipeline(**self.components)(
								    prompt=_SCREAMING_SNAKE_CASE			,    height=_SCREAMING_SNAKE_CASE			,    width=_SCREAMING_SNAKE_CASE			,    num_inference_steps=_SCREAMING_SNAKE_CASE			,    guidance_scale=_SCREAMING_SNAKE_CASE			,    negative_prompt=_SCREAMING_SNAKE_CASE			,    num_images_per_prompt=_SCREAMING_SNAKE_CASE			,    eta=_SCREAMING_SNAKE_CASE			,    generator=_SCREAMING_SNAKE_CASE			,    latents=_SCREAMING_SNAKE_CASE			,    output_type=_SCREAMING_SNAKE_CASE			,    return_dict=_SCREAMING_SNAKE_CASE			,    callback=_SCREAMING_SNAKE_CASE			,    callback_steps=_SCREAMING_SNAKE_CASE			,    )




