




"""simple docstring"""


import argparse
import gc
import json
import os
import shutil
import warnings

import torch

from transformers import LlamaConfig, LlamaForCausalLM, LlamaTokenizer


try:
     from transformers import LlamaTokenizerFast
except ImportError as e:
     warnings.warn(e)
     warnings.warn(
         '''The converted tokenizer will be the `slow` tokenizer. To use the fast, update your `tokenizers` library and re-run the tokenizer conversion'''
     )
     A_      			=       None
A_      			=       {
    '''7B''': 1_10_08,
    '''13B''': 1_38_24,
    '''30B''': 1_79_20,
    '''65B''': 2_20_16,
    '''70B''': 2_86_72,
}
A_      			=       {
    '''7B''': 1,
    '''7Bf''': 1,
    '''13B''': 2,
    '''13Bf''': 2,
    '''30B''': 4,
    '''65B''': 8,
    '''70B''': 8,
    '''70Bf''': 8,
}


def       UpperCAmelCase__ (snake_case__							:      Dict	,     snake_case__							:      str=1	,     snake_case__							:      List[str]=2_56					):
  """simple docstring"""

  return multiple_of * ((int(ffn_dim_multiplier * int(8 * n / 3					)					) + multiple_of - 1) // multiple_of)


def       UpperCAmelCase__ (snake_case__							:      List[str]					):
  """simple docstring"""

  with open(a_	,     """r"""					) as f:
    return json.load(a_					)


def       UpperCAmelCase__ (snake_case__							:      List[Any]	,     snake_case__							:      List[Any]					):
  """simple docstring"""

  with open(a_	,     """w"""					) as f:
    json.dump(a_	,     a_					)


def       UpperCAmelCase__ (snake_case__							:      int	,     snake_case__							:      Dict	,     snake_case__							:      List[Any]	,     snake_case__							:      Any=True					):
  """simple docstring"""

  os.makedirs(a_	,     exist_ok=a_					)
  _snake_case							:	Any			=						os.path.join(a_	,     """tmp"""					)
  os.makedirs(a_	,     exist_ok=a_					)

  _snake_case							:	Union[str, Any]			=						read_json(os.path.join(a_	,     """params.json"""					)					)
  _snake_case							:	List[str]			=						NUM_SHARDS[model_size]
  _snake_case							:	Union[str, Any]			=						params['''n_layers''']
  _snake_case							:	List[Any]			=						params['''n_heads''']
  _snake_case							:	Any			=						n_heads // num_shards
  _snake_case							:	Tuple			=						params['''dim''']
  _snake_case							:	List[Any]			=						dim // n_heads
  _snake_case							:	List[Any]			=						1_00_00.0
  _snake_case							:	List[Any]			=						1.0 / (base ** (torch.arange(0	,     a_	,     2					).float() / dims_per_head))

  if "n_kv_heads" in params:
    _snake_case							:	int			=						params['''n_kv_heads''']  # for GQA / MQA
    _snake_case							:	Optional[Any]			=						n_heads_per_shard // num_key_value_heads
    _snake_case							:	Any			=						dim // num_key_value_heads
  else:  # compatibility with other checkpoints
    _snake_case							:	List[Any]			=						n_heads
    _snake_case							:	int			=						n_heads_per_shard
    _snake_case							:	Union[str, Any]			=						dim

  # permute for sliced rotary
  def permute(snake_case__							:      Any	,     snake_case__							:      Optional[int]=n_heads	,     snake_case__							:      Any=dim	,     snake_case__							:      int=dim					):
    return w.view(a_	,     dima // n_heads // 2	,     2	,     a_					).transpose(1	,     2					).reshape(a_	,     a_					)

  print(F"Fetching all parameters from the checkpoint at {input_base_path}."					)
  # Load weights
  if model_size == "7B":
    # Not sharded
    # (The sharded implementation would also work, but this is simpler.)
    _snake_case							:	Any			=						torch.load(os.path.join(a_	,     """consolidated.00.pth"""					)	,     map_location="""cpu"""					)
  else:
    # Sharded
    _snake_case							:	Dict			=						[
        torch.load(os.path.join(a_	,     F"consolidated.{i:02d}.pth"					)	,     map_location="""cpu"""					)
        for i in range(a_					)
    ]
  _snake_case							:	Tuple			=						0
  _snake_case							:	Any			=						{'''weight_map''': {}}
  for layer_i in range(a_					):
    _snake_case							:	Optional[int]			=						F"pytorch_model-{layer_i + 1}-of-{n_layers + 1}.bin"
    if model_size == "7B":
      # Unsharded
      _snake_case							:	Tuple			=						{
          F"model.layers.{layer_i}.self_attn.q_proj.weight": permute(
              loaded[F"layers.{layer_i}.attention.wq.weight"]					),
          F"model.layers.{layer_i}.self_attn.k_proj.weight": permute(
              loaded[F"layers.{layer_i}.attention.wk.weight"]					),
          F"model.layers.{layer_i}.self_attn.v_proj.weight": loaded[F"layers.{layer_i}.attention.wv.weight"],
          F"model.layers.{layer_i}.self_attn.o_proj.weight": loaded[F"layers.{layer_i}.attention.wo.weight"],
          F"model.layers.{layer_i}.mlp.gate_proj.weight": loaded[F"layers.{layer_i}.feed_forward.w1.weight"],
          F"model.layers.{layer_i}.mlp.down_proj.weight": loaded[F"layers.{layer_i}.feed_forward.w2.weight"],
          F"model.layers.{layer_i}.mlp.up_proj.weight": loaded[F"layers.{layer_i}.feed_forward.w3.weight"],
          F"model.layers.{layer_i}.input_layernorm.weight": loaded[F"layers.{layer_i}.attention_norm.weight"],
          F"model.layers.{layer_i}.post_attention_layernorm.weight": loaded[F"layers.{layer_i}.ffn_norm.weight"],
      }
    else:
      # Sharded
      # Note that attention.w{q,k,v,o}, feed_fordward.w[1,2,3], attention_norm.weight and ffn_norm.weight share
      # the same storage object, saving attention_norm and ffn_norm will save other weights too, which is
      # redundant as other weights will be stitched from multiple shards. To avoid that, they are cloned.

      _snake_case							:	int			=						{
          F"model.layers.{layer_i}.input_layernorm.weight": loaded[0][
              F"layers.{layer_i}.attention_norm.weight"
          ].clone(),
          F"model.layers.{layer_i}.post_attention_layernorm.weight": loaded[0][
              F"layers.{layer_i}.ffn_norm.weight"
          ].clone(),
      }
      _snake_case							:	Union[str, Any]			=						permute(
          torch.cat(
              [
                  loaded[i][F"layers.{layer_i}.attention.wq.weight"].view(a_	,     a_	,     a_					)
                  for i in range(a_					)
              ]	,     dim=0	,     ).reshape(a_	,     a_					)					)
      _snake_case							:	Optional[Any]			=						permute(
          torch.cat(
              [
                  loaded[i][F"layers.{layer_i}.attention.wk.weight"].view(
                      a_	,     a_	,     a_					)
                  for i in range(a_					)
              ]	,     dim=0	,     ).reshape(a_	,     a_					)	,     a_	,     a_	,     a_	,     )
      _snake_case							:	Dict			=						torch.cat(
          [
              loaded[i][F"layers.{layer_i}.attention.wv.weight"].view(
                  a_	,     a_	,     a_					)
              for i in range(a_					)
          ]	,     dim=0	,     ).reshape(a_	,     a_					)

      _snake_case							:	Dict			=						torch.cat(
          [loaded[i][F"layers.{layer_i}.attention.wo.weight"] for i in range(a_					)]	,     dim=1					)
      _snake_case							:	List[Any]			=						torch.cat(
          [loaded[i][F"layers.{layer_i}.feed_forward.w1.weight"] for i in range(a_					)]	,     dim=0					)
      _snake_case							:	List[Any]			=						torch.cat(
          [loaded[i][F"layers.{layer_i}.feed_forward.w2.weight"] for i in range(a_					)]	,     dim=1					)
      _snake_case							:	List[str]			=						torch.cat(
          [loaded[i][F"layers.{layer_i}.feed_forward.w3.weight"] for i in range(a_					)]	,     dim=0					)

    _snake_case							:	Union[str, Any]			=						inv_freq
    for k, v in state_dict.items():
      _snake_case							:	List[Any]			=						filename
      param_count += v.numel()
    torch.save(a_	,     os.path.join(a_	,     a_					)					)

  _snake_case							:	List[Any]			=						F"pytorch_model-{n_layers + 1}-of-{n_layers + 1}.bin"
  if model_size == "7B":
    # Unsharded
    _snake_case							:	Optional[int]			=						{
        '''model.embed_tokens.weight''': loaded['''tok_embeddings.weight'''],
        '''model.norm.weight''': loaded['''norm.weight'''],
        '''lm_head.weight''': loaded['''output.weight'''],
    }
  else:
    _snake_case							:	List[Any]			=						{
        '''model.norm.weight''': loaded[0]['''norm.weight'''],
        '''model.embed_tokens.weight''': torch.cat(
            [loaded[i]["""tok_embeddings.weight"""] for i in range(a_					)]	,     dim=1					),
        '''lm_head.weight''': torch.cat([loaded[i]["""output.weight"""] for i in range(a_					)]	,     dim=0					),
    }

  for k, v in state_dict.items():
    _snake_case							:	Any			=						filename
    param_count += v.numel()
  torch.save(a_	,     os.path.join(a_	,     a_					)					)

  # Write configs
  _snake_case							:	Optional[int]			=						{'''total_size''': param_count * 2}
  write_json(a_	,     os.path.join(a_	,     """pytorch_model.bin.index.json"""					)					)
  _snake_case							:	Union[str, Any]			=						params['''ffn_dim_multiplier'''] if '''ffn_dim_multiplier''' in params else 1
  _snake_case							:	Any			=						params['''multiple_of'''] if '''multiple_of''' in params else 2_56
  _snake_case							:	Optional[int]			=						LlamaConfig(
      hidden_size=a_	,     intermediate_size=compute_intermediate_size(a_	,     a_	,     a_					)	,     num_attention_heads=params["""n_heads"""]	,     num_hidden_layers=params["""n_layers"""]	,     rms_norm_eps=params["""norm_eps"""]	,     num_key_value_heads=a_	,     )
  config.save_pretrained(a_					)

  # Make space so we can load the model properly now.
  del state_dict
  del loaded
  gc.collect()

  print("""Loading the checkpoint in a Llama model."""					)
  _snake_case							:	str			=						LlamaForCausalLM.from_pretrained(a_	,     torch_dtype=torch.floataa	,     low_cpu_mem_usage=a_					)
  # Avoid saving this as part of the config.
  del model.config._name_or_path

  print("""Saving in the Transformers format."""					)
  model.save_pretrained(a_	,     safe_serialization=a_					)
  shutil.rmtree(a_					)


def       UpperCAmelCase__ (snake_case__							:      Any	,     snake_case__							:      str					):
  """simple docstring"""

  _snake_case							:	Optional[int]			=						LlamaTokenizer if LlamaTokenizerFast is None else LlamaTokenizerFast
  print(F"Saving a {tokenizer_class.__name__} to {tokenizer_path}."					)
  _snake_case							:	str			=						tokenizer_class(a_					)
  tokenizer.save_pretrained(a_					)


def       UpperCAmelCase__ ():
  """simple docstring"""

  _snake_case							:	Optional[int]			=						argparse.ArgumentParser()
  parser.add_argument(
      """--input_dir"""	,     help="""Location of LLaMA weights, which contains tokenizer.model and model folders"""	,     )
  parser.add_argument(
      """--model_size"""	,     choices=["""7B""", """7Bf""", """13B""", """13Bf""", """30B""", """65B""", """70B""", """70Bf""", """tokenizer_only"""]	,     )
  parser.add_argument(
      """--output_dir"""	,     help="""Location to write HF model and tokenizer"""	,     )
  parser.add_argument("""--safe_serialization"""	,     type=a_	,     help="""Whether or not to save using `safetensors`."""					)
  _snake_case							:	Optional[Any]			=						parser.parse_args()
  if args.model_size != "tokenizer_only":
    write_model(
        model_path=args.output_dir	,     input_base_path=os.path.join(args.input_dir	,     args.model_size					)	,     model_size=args.model_size	,     safe_serialization=args.safe_serialization	,     )
  _snake_case							:	Optional[Any]			=						os.path.join(args.input_dir	,     """tokenizer.model"""					)
  write_tokenizer(args.output_dir	,     a_					)


if __name__ == "__main__":
     main()








"""simple docstring"""


import os
from glob import glob

import imageio
import torch
import torchvision
import wandb
from img_processing import custom_to_pil, loop_post_process, preprocess, preprocess_vqgan
from loaders import load_vqgan
from PIL import Image
from torch import nn

from transformers import CLIPModel, CLIPTokenizerFast
from utils import get_device, get_timestamp, show_pil






class       lowercase:







 '''simple docstring'''





 def __init__(       self:	List[Any],   a_:	str = "cpu",   a_:	str = "openai/clip-vit-large-patch14"	):




   '''simple docstring'''



   _snake_case							:	Optional[int]			=						device
   _snake_case							:	str			=						CLIPTokenizerFast.from_pretrained(a_	)
   _snake_case							:	Union[str, Any]			=						[0.48_145_466, 0.4_578_275, 0.40_821_073]
   _snake_case							:	Optional[int]			=						[0.26_862_954, 0.26_130_258, 0.27_577_711]
   _snake_case							:	str			=						torchvision.transforms.Normalize(self.image_mean,   self.image_std	)
   _snake_case							:	Optional[int]			=						torchvision.transforms.Resize(224	)
   _snake_case							:	str			=						torchvision.transforms.CenterCrop(224	)


 def 				UpperCamelCase_     (       self:	List[str],   a_:	str	):




   '''simple docstring'''



   _snake_case							:	Optional[int]			=						self.resize(a_	)
   _snake_case							:	List[Any]			=						self.center_crop(a_	)
   _snake_case							:	Optional[Any]			=						self.normalize(a_	)
   return images


 def __call__(       self:	Any,   a_:	Optional[int]=None,   a_:	str=None,   **a_:	str	):




   '''simple docstring'''



   _snake_case							:	Optional[int]			=						self.tokenizer(text=a_,   **a_	)
   _snake_case							:	Any			=						self.preprocess_img(a_	)
   _snake_case							:	Union[str, Any]			=						{key: value.to(self.device	) for (key, value) in encoding.items()}
   return encoding







class       lowercase(  nn.Module				):







 '''simple docstring'''





 def __init__(       self:	List[Any],   a_:	List[Any]=10,   a_:	Optional[Any]=0.01,   a_:	List[str]=None,   a_:	str=None,   a_:	Any=None,   a_:	Tuple=None,   a_:	List[str]=None,   a_:	List[str]=None,   a_:	str=False,   a_:	List[str]=True,   a_:	Any="image",   a_:	Optional[Any]=True,   a_:	Dict=False,   a_:	List[str]=False,   a_:	Optional[int]=False,   ):




   '''simple docstring'''



   super().__init__()
   _snake_case							:	int			=						None
   _snake_case							:	List[str]			=						device if device else get_device()
   if vqgan:
     _snake_case							:	Any			=						vqgan
   else:
     _snake_case							:	Optional[Any]			=						load_vqgan(self.device,   conf_path=a_,   ckpt_path=a_	)
   self.vqgan.eval()
   if clip:
     _snake_case							:	Tuple			=						clip
   else:
     _snake_case							:	Optional[Any]			=						CLIPModel.from_pretrained("""openai/clip-vit-base-patch32"""	)
   self.clip.to(self.device	)
   _snake_case							:	List[str]			=						ProcessorGradientFlow(device=self.device	)

   _snake_case							:	Union[str, Any]			=						iterations
   _snake_case							:	Dict			=						lr
   _snake_case							:	Optional[int]			=						log
   _snake_case							:	List[str]			=						make_grid
   _snake_case							:	Union[str, Any]			=						return_val
   _snake_case							:	List[str]			=						quantize
   _snake_case							:	List[str]			=						self.vqgan.decoder.z_shape


 def 				UpperCamelCase_     (       self:	Tuple,   a_:	str=None,   a_:	Dict=None,   a_:	Dict=5,   a_:	Dict=True	):




   '''simple docstring'''



   _snake_case							:	Dict			=						[]
   if output_path is None:
     _snake_case							:	Tuple			=						"""./animation.gif"""
   if input_path is None:
     _snake_case							:	Any			=						self.save_path
   _snake_case							:	Optional[int]			=						sorted(glob(input_path + """/*"""	)	)
   if not len(a_	):
     raise ValueError(
         """No images found in save path, aborting (did you pass save_intermediate=True to the generate"""
         """ function?)"""	)
   if len(a_	) == 1:
     print("""Only one image found in save path, (did you pass save_intermediate=True to the generate function?)"""	)
   _snake_case							:	List[Any]			=						total_duration / len(a_	)
   _snake_case							:	Optional[Any]			=						[frame_duration] * len(a_	)
   if extend_frames:
     _snake_case							:	Optional[int]			=						1.5
     _snake_case							:	int			=						3
   for file_name in paths:
     if file_name.endswith(""".png"""	):
       images.append(imageio.imread(a_	)	)
   imageio.mimsave(a_,   a_,   duration=a_	)
   print(f"gif saved to {output_path}"	)


 def 				UpperCamelCase_     (       self:	str,   a_:	Tuple=None,   a_:	Optional[Any]=None	):




   '''simple docstring'''



   if not (path or img):
     raise ValueError("""Input either path or tensor"""	)
   if img is not None:
     raise NotImplementedError
   _snake_case							:	int			=						preprocess(Image.open(a_	),   target_image_size=256	).to(self.device	)
   _snake_case							:	int			=						preprocess_vqgan(a_	)
   _snake_case	,      *_snake_case							:	List[Any]			=						self.vqgan.encode(a_	)
   return z


 def 				UpperCamelCase_     (       self:	Union[str, Any],   a_:	Optional[Any]	):




   '''simple docstring'''



   _snake_case							:	Optional[int]			=						self.latent.detach().requires_grad_()
   _snake_case							:	Tuple			=						base_latent + transform_vector
   if self.quantize:
     _snake_case	,      *_snake_case							:	Any			=						self.vqgan.quantize(a_	)
   else:
     _snake_case							:	List[Any]			=						trans_latent
   return self.vqgan.decode(a_	)


 def 				UpperCamelCase_     (       self:	List[Any],   a_:	Any,   a_:	Union[str, Any],   a_:	Dict=None	):




   '''simple docstring'''



   _snake_case							:	Tuple			=						self.clip_preprocessor(text=a_,   images=a_,   return_tensors="""pt""",   padding=a_	)
   _snake_case							:	Any			=						self.clip(**a_	)
   _snake_case							:	str			=						clip_outputs.logits_per_image
   if weights is not None:
     _snake_case							:	Any			=						similarity_logits * weights
   return similarity_logits.sum()


 def 				UpperCamelCase_     (       self:	Any,   a_:	Any,   a_:	List[str],   a_:	Dict	):




   '''simple docstring'''



   _snake_case							:	List[Any]			=						self._get_clip_similarity(pos_prompts["""prompts"""],   a_,   weights=(1 / pos_prompts["""weights"""])	)
   if neg_prompts:
     _snake_case							:	List[str]			=						self._get_clip_similarity(neg_prompts["""prompts"""],   a_,   weights=neg_prompts["""weights"""]	)
   else:
     _snake_case							:	Tuple			=						torch.tensor([1],   device=self.device	)
   _snake_case							:	int			=						-torch.log(a_	) + torch.log(a_	)
   return loss


 def 				UpperCamelCase_     (       self:	Optional[Any],   a_:	Tuple,   a_:	Union[str, Any],   a_:	List[str]	):




   '''simple docstring'''



   _snake_case							:	Tuple			=						torch.randn_like(self.latent,   requires_grad=a_,   device=self.device	)
   _snake_case							:	Dict			=						torch.optim.Adam([vector],   lr=self.lr	)

   for i in range(self.iterations	):
     optim.zero_grad()
     _snake_case							:	str			=						self._add_vector(a_	)
     _snake_case							:	List[Any]			=						loop_post_process(a_	)
     _snake_case							:	List[Any]			=						self._get_CLIP_loss(a_,   a_,   a_	)
     print("""CLIP loss""",   a_	)
     if self.log:
       wandb.log({"""CLIP Loss""": clip_loss}	)
     clip_loss.backward(retain_graph=a_	)
     optim.step()
     if self.return_val == "image":
       yield custom_to_pil(transformed_img[0]	)
     else:
       yield vector


 def 				UpperCamelCase_     (       self:	int,   a_:	Any,   a_:	Union[str, Any],   a_:	Optional[int]	):




   '''simple docstring'''



   wandb.init(reinit=a_,   project="""face-editor"""	)
   wandb.config.update({"""Positive Prompts""": positive_prompts}	)
   wandb.config.update({"""Negative Prompts""": negative_prompts}	)
   wandb.config.update({"""lr""": self.lr, """iterations""": self.iterations}	)
   if image_path:
     _snake_case							:	Any			=						Image.open(a_	)
     _snake_case							:	str			=						image.resize((256, 256)	)
     wandb.log("""Original Image""",   wandb.Image(a_	)	)


 def 				UpperCamelCase_     (       self:	str,   a_:	List[Any]	):




   '''simple docstring'''



   if not prompts:
     return []
   _snake_case							:	List[str]			=						[]
   _snake_case							:	Tuple			=						[]
   if isinstance(a_,   a_	):
     _snake_case							:	Union[str, Any]			=						[prompt.strip() for prompt in prompts.split("""|"""	)]
   for prompt in prompts:
     if isinstance(a_,   (tuple, list)	):
       _snake_case							:	List[Any]			=						prompt[0]
       _snake_case							:	Optional[Any]			=						float(prompt[1]	)
     elif ":" in prompt:
       _snake_case	,      _snake_case							:	List[Any]			=						prompt.split(""":"""	)
       _snake_case							:	str			=						float(a_	)
     else:
       _snake_case							:	int			=						prompt
       _snake_case							:	Union[str, Any]			=						1.0
     processed_prompts.append(a_	)
     weights.append(a_	)
   return {
       "prompts": processed_prompts,
       "weights": torch.tensor(a_,   device=self.device	),
   }


 def 				UpperCamelCase_     (       self:	Dict,   a_:	List[Any],   a_:	List[Any]=None,   a_:	Optional[Any]=None,   a_:	Optional[Any]=True,   a_:	Dict=False,   a_:	Optional[Any]=True,   a_:	Optional[Any]=True,   a_:	Any=None,   ):




   '''simple docstring'''



   if image_path:
     _snake_case							:	Union[str, Any]			=						self._get_latent(a_	)
   else:
     _snake_case							:	Any			=						torch.randn(self.latent_dim,   device=self.device	)
   if self.log:
     self._init_logging(a_,   a_,   a_	)

   assert pos_prompts, "You must provide at least one positive prompt."
   _snake_case							:	str			=						self.process_prompts(a_	)
   _snake_case							:	Dict			=						self.process_prompts(a_	)
   if save_final and save_path is None:
     _snake_case							:	Any			=						os.path.join("""./outputs/""",   """_""".join(pos_prompts["""prompts"""]	)	)
   if not os.path.exists(a_	):
     os.makedirs(a_	)
   else:
     _snake_case							:	List[Any]			=						save_path + """_""" + get_timestamp()
     os.makedirs(a_	)
   _snake_case							:	Optional[Any]			=						save_path

   _snake_case							:	List[Any]			=						self.vqgan.decode(self.latent	)[0]
   if show_intermediate:
     print("""Original Image"""	)
     show_pil(custom_to_pil(a_	)	)

   _snake_case							:	List[Any]			=						loop_post_process(a_	)
   for iter, transformed_img in enumerate(self._optimize_CLIP(a_,   a_,   a_	)	):
     if show_intermediate:
       show_pil(a_	)
     if save_intermediate:
       transformed_img.save(os.path.join(self.save_path,   f"iter_{iter:03d}.png"	)	)
     if self.log:
       wandb.log({"""Image""": wandb.Image(a_	)}	)
   if show_final:
     show_pil(a_	)
   if save_final:
     transformed_img.save(os.path.join(self.save_path,   f"iter_{iter:03d}_final.png"	)	)


