



"""simple docstring"""





import argparse

import torch

from diffusers.pipelines.stable_diffusion.convert_from_ckpt import download_from_original_stable_diffusion_ckpt


if __name__ == "__main__":
					__lowerCAmelCase							:       List[Any]          =       argparse.ArgumentParser()

					parser.add_argument(
					    '''--checkpoint_path''', default=None, type=str, required=True, help='''Path to the checkpoint to convert.'''
					)
					# !wget https://raw.githubusercontent.com/CompVis/stable-diffusion/main/configs/stable-diffusion/v1-inference.yaml
					parser.add_argument(
					    '''--original_config_file''',
					    default=None,
					    type=str,
					    help='''The YAML config file corresponding to the original architecture.''',
					)
					parser.add_argument(
					    '''--num_in_channels''',
					    default=None,
					    type=int,
					    help='''The number of input channels. If `None` number of input channels will be automatically inferred.''',
					)
					parser.add_argument(
					    '''--scheduler_type''',
					    default='''pndm''',
					    type=str,
					    help='''Type of scheduler to use. Should be one of [\'pndm\', \'lms\', \'ddim\', \'euler\', \'euler-ancestral\', \'dpm\']''',
					)
					parser.add_argument(
					    '''--pipeline_type''',
					    default=None,
					    type=str,
					    help=(
					        '''The pipeline type. One of \'FrozenOpenCLIPEmbedder\', \'FrozenCLIPEmbedder\', \'PaintByExample\''''
					        '''. If `None` pipeline will be automatically inferred.'''
					    ),
					)
					parser.add_argument(
					    '''--image_size''',
					    default=None,
					    type=int,
					    help=(
					        '''The image size that the model was trained on. Use 512 for Stable Diffusion v1.X and Stable Siffusion v2'''
					        ''' Base. Use 768 for Stable Diffusion v2.'''
					    ),
					)
					parser.add_argument(
					    '''--prediction_type''',
					    default=None,
					    type=str,
					    help=(
					        '''The prediction type that the model was trained on. Use \'epsilon\' for Stable Diffusion v1.X and Stable'''
					        ''' Diffusion v2 Base. Use \'v_prediction\' for Stable Diffusion v2.'''
					    ),
					)
					parser.add_argument(
					    '''--extract_ema''',
					    action='''store_true''',
					    help=(
					        '''Only relevant for checkpoints that have both EMA and non-EMA weights. Whether to extract the EMA weights'''
					        ''' or not. Defaults to `False`. Add `--extract_ema` to extract the EMA weights. EMA weights usually yield'''
					        ''' higher quality images for inference. Non-EMA weights are usually better to continue fine-tuning.'''
					    ),
					)
					parser.add_argument(
					    '''--upcast_attention''',
					    action='''store_true''',
					    help=(
					        '''Whether the attention computation should always be upcasted. This is necessary when running stable'''
					        ''' diffusion 2.1.'''
					    ),
					)
					parser.add_argument(
					    '''--from_safetensors''',
					    action='''store_true''',
					    help='''If `--checkpoint_path` is in `safetensors` format, load checkpoint with safetensors instead of PyTorch.''',
					)
					parser.add_argument(
					    '''--to_safetensors''',
					    action='''store_true''',
					    help='''Whether to store pipeline in safetensors format or not.''',
					)
					parser.add_argument('''--dump_path''', default=None, type=str, required=True, help='''Path to the output model.''')
					parser.add_argument('''--device''', type=str, help='''Device to use (e.g. cpu, cuda:0, cuda:1, etc.)''')
					parser.add_argument(
					    '''--stable_unclip''',
					    type=str,
					    default=None,
					    required=False,
					    help='''Set if this is a stable unCLIP model. One of \'txt2img\' or \'img2img\'.''',
					)
					parser.add_argument(
					    '''--stable_unclip_prior''',
					    type=str,
					    default=None,
					    required=False,
					    help='''Set if this is a stable unCLIP txt2img model. Selects which prior to use. If `--stable_unclip` is set to `txt2img`, the karlo prior (https://huggingface.co/kakaobrain/karlo-v1-alpha/tree/main/prior) is selected by default.''',
					)
					parser.add_argument(
					    '''--clip_stats_path''',
					    type=str,
					    help='''Path to the clip stats file. Only required if the stable unclip model\'s config specifies `model.params.noise_aug_config.params.clip_stats_path`.''',
					    required=False,
					)
					parser.add_argument(
					    '''--controlnet''', action='''store_true''', default=None, help='''Set flag if this is a controlnet checkpoint.'''
					)
					parser.add_argument('''--half''', action='''store_true''', help='''Save weights in half precision.''')
					parser.add_argument(
					    '''--vae_path''',
					    type=str,
					    default=None,
					    required=False,
					    help='''Set to a path, hub id to an already converted vae to not convert it again.''',
					)
					__lowerCAmelCase							:       List[Any]          =       parser.parse_args()

					__lowerCAmelCase							:       Dict          =       download_from_original_stable_diffusion_ckpt(
					    checkpoint_path=args.checkpoint_path,
					    original_config_file=args.original_config_file,
					    image_size=args.image_size,
					    prediction_type=args.prediction_type,
					    model_type=args.pipeline_type,
					    extract_ema=args.extract_ema,
					    scheduler_type=args.scheduler_type,
					    num_in_channels=args.num_in_channels,
					    upcast_attention=args.upcast_attention,
					    from_safetensors=args.from_safetensors,
					    device=args.device,
					    stable_unclip=args.stable_unclip,
					    stable_unclip_prior=args.stable_unclip_prior,
					    clip_stats_path=args.clip_stats_path,
					    controlnet=args.controlnet,
					    vae_path=args.vae_path,
					)

					if args.half:
										pipe.to(torch_dtype=torch.floataa)

					if args.controlnet:
										# only save the controlnet model
										pipe.controlnet.save_pretrained(args.dump_path, safe_serialization=args.to_safetensors)
					else:
										pipe.save_pretrained(args.dump_path, safe_serialization=args.to_safetensors)









from __future__ import annotations

from collections.abc import Callable
from typing import Generic, TypeVar

lowercase    				=   TypeVar("T")
lowercase    				=   TypeVar("U")




class 					UpperCamelCase_ (     Generic[T, U]							):



	'''simple docstring'''




	def __init__(			self			,     a			,     a   )			->       List[Any]:
								snake_case_           =				key
								snake_case_           =				val
								snake_case_           =				None
								snake_case_           =				None






	def __repr__(			self   )			->       str:
								return (
								    F'''Node: key: {self.key}, val: {self.val}, '''
								    F'''has next: {bool(self.next   )}, has prev: {bool(self.prev   )}'''
								)






class 					UpperCamelCase_ (     Generic[T, U]							):



	'''simple docstring'''




	def __init__(			self   )			->       None:
								snake_case_           =				DoubleLinkedListNode(a			,     a   )
								snake_case_           =				DoubleLinkedListNode(a			,     a   )
								snake_case_     ,						snake_case_           =				self.rear, self.head




	def __repr__(			self   )			->       str:
								snake_case_           =				['DoubleLinkedList']
								snake_case_           =				self.head
								while node.next is not None:
															rep.append(str(a   )   )
															snake_case_           =				node.next
								rep.append(str(self.rear   )   )
								return ",\n    ".join(a   )




	def 				_UpperCamelCase     (			self			,     a   )			->       None:
								snake_case_           =				self.rear.prev

								# All nodes other than self.head are guaranteed to have non-None previous
								assert previous is not None

								snake_case_           =				node
								snake_case_           =				previous
								snake_case_           =				node
								snake_case_           =				self.rear






	def 				_UpperCamelCase     (			self			,     a   )			->       DoubleLinkedListNode[T, U] | None:

								if node.prev is None or node.next is None:
															return None

								snake_case_           =				node.next
								snake_case_           =				node.prev
								snake_case_           =				None
								snake_case_           =				None
								return node






class 					UpperCamelCase_ (     Generic[T, U]							):



	'''simple docstring'''




	lowerCAmelCase						=					{}




	def __init__(			self			,     a   )			->       str:
								snake_case_           =				DoubleLinkedList()
								snake_case_           =				capacity
								snake_case_           =				0
								snake_case_           =				0
								snake_case_           =				0
								snake_case_           =				{}




	def __repr__(			self   )			->       str:
								return (
								    F'''CacheInfo(hits={self.hits}, misses={self.miss}, '''
								    F'''capacity={self.capacity}, current size={self.num_keys})'''
								)




	def __contains__(			self			,     a   )			->       bool:
								return key in self.cache




	def 				_UpperCamelCase     (			self			,     a   )			->       U | None:
								# Note: pythonic interface would throw KeyError rather than return None

								if key in self.cache:
															self.hits += 1
															snake_case_           =				self.cache[key]
															snake_case_           =				self.list.remove(self.cache[key]   )
															assert node == value_node

															# node is guaranteed not None because it is in self.cache
															assert node is not None
															self.list.add(a   )
															return node.val
								self.miss += 1
								return None




	def 				_UpperCamelCase     (			self			,     a			,     a   )			->       None:

								if key not in self.cache:
															if self.num_keys >= self.capacity:
																						# delete first node (oldest) when over capacity
																						snake_case_           =				self.list.head.next

																						# guaranteed to have a non-None first node when num_keys > 0
																						# explain to type checker via assertions
																						assert first_node is not None
																						assert first_node.key is not None
																						assert (
																						    self.list.remove(a   ) is not None
																						)  # node guaranteed to be in list assert node.key is not None

																						del self.cache[first_node.key]
																						self.num_keys -= 1
															snake_case_           =				DoubleLinkedListNode(a			,     a   )
															self.list.add(self.cache[key]   )
															self.num_keys += 1

								else:
															# bump node to the end of the list, update value
															snake_case_           =				self.list.remove(self.cache[key]   )
															assert node is not None  # node guaranteed to be in list
															snake_case_           =				value
															self.list.add(a   )






	@classmethod
	def 				_UpperCamelCase     (			cls			,     a = 1_28   )			->       Callable[[Callable[[T], U]], Callable[..., U]]:

								def cache_decorator_inner(a   ) -> Callable[..., U]:
															def cache_decorator_wrapper(*a   ) -> U:
																						if func not in cls.decorator_function_to_instance_map:
																													snake_case_           =				LRUCache(a   )

																						snake_case_           =				cls.decorator_function_to_instance_map[func].get(args[0]   )
																						if result is None:
																													snake_case_           =				func(*a   )
																													cls.decorator_function_to_instance_map[func].put(args[0]			,     a   )
																						return result

															def cache_info() -> LRUCache[T, U]:
																						return cls.decorator_function_to_instance_map[func]

															setattr(a			,     'cache_info'			,     a   )  # noqa: B010

															return cache_decorator_wrapper

								return cache_decorator_inner


if __name__ == "__main__":
	import doctest

	doctest.testmod()





