

import os
from shutil import copyfile
from typing import TYPE_CHECKING, Optional, Tuple

from tokenizers import processors

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging
from ...utils.versions import require_version


if TYPE_CHECKING:
 from transformers.pipelines.conversational import Conversation

require_version('''tokenizers>=0.13.3''')

if is_sentencepiece_available():
 from .tokenization_llama import LlamaTokenizer
else:
 __A    	=			None

__A    	=			logging.get_logger(__name__)
__A    	=			{'''vocab_file''': '''tokenizer.model''', '''tokenizer_file''': '''tokenizer.json'''}

__A	,  __A    	=			'''[INST]''', '''[/INST]'''
__A	,  __A    	=			'''<<SYS>>\n''', '''\n<</SYS>>\n\n'''

# fmt: off
__A    	=			'''You are a helpful, respectful and honest assistant. Always answer as helpfully as possible, while being safe. Your\
answers should not include any harmful, unethical, racist, sexist, toxic, dangerous, or illegal content. Please ensure\
that your responses are socially unbiased and positive in nature.

If a question does not make any sense, or is not factually coherent, explain why instead of answering something not\
correct. If you don\'t know the answer to a question, please don\'t share false information.'''



class 						lowercase_					(						__lowercase  ):
      UpperCamelCase_    :	Union[str, Any]		   =       VOCAB_FILES_NAMES
      UpperCamelCase_    :	Any		   =       LlamaTokenizer
      UpperCamelCase_    :	Dict		   =       "left"
      UpperCamelCase_    :	Optional[int]		   =       ["input_ids", "attention_mask"]



      def __init__(    self   :       Dict    ,      A__   :       str=None    ,      A__   :       str=None    ,      A__   :       Any=False    ,      A__   :       Union[str, Any]="<unk>"    ,      A__   :       int="<s>"    ,      A__   :       Union[str, Any]="</s>"    ,      A__   :       Union[str, Any]=True    ,      A__   :       List[Any]=False    ,      **A__   :       Tuple    ,      )					->  List[str]:
        super().__init__(
            vocab_file=A__    ,      tokenizer_file=A__    ,      clean_up_tokenization_spaces=A__    ,      unk_token=A__    ,      bos_token=A__    ,      eos_token=A__    ,      **A__    ,      )
        _snake_case		=							add_bos_token
        _snake_case		=							add_eos_token
        self.update_post_processor()

        _snake_case		=							vocab_file
        _snake_case		=							False if not self.vocab_file else True



      def 		UpperCamelCase_	(    self   :       Optional[Any]	)					->  Tuple:
        _snake_case		=							self.bos_token
        _snake_case		=							self.bos_token_id

        _snake_case		=							self.eos_token
        _snake_case		=							self.eos_token_id

        _snake_case		=							f"""{(bos+':0 ') * self.add_bos_token}$A:0{(' '+eos+':0') * self.add_eos_token}"""
        _snake_case		=							f"""{single}{(' '+bos+':1') * self.add_bos_token} $B:1{(' '+eos+':1') * self.add_eos_token}"""

        _snake_case		=							[]
        if self.add_bos_token:
          special_tokens.append((bos, bos_token_id)	)
        if self.add_eos_token:
          special_tokens.append((eos, eos_token_id)	)
        _snake_case		=							processors.TemplateProcessing(
            single=A__    ,      pair=A__    ,      special_tokens=A__	)



      @property
      def 		UpperCamelCase_	(    self   :       List[Any]	)					->  Optional[Any]:
        return self._add_eos_token



      @property
      def 		UpperCamelCase_	(    self   :       List[Any]	)					->  str:
        return self._add_bos_token



      @add_eos_token.setter
      def 		UpperCamelCase_	(    self   :       str    ,      A__   :       List[str]	)					->  Tuple:
        _snake_case		=							value
        self.update_post_processor()



      @add_bos_token.setter
      def 		UpperCamelCase_	(    self   :       Optional[Any]    ,      A__   :       Dict	)					->  Union[str, Any]:
        _snake_case		=							value
        self.update_post_processor()



      def 		UpperCamelCase_	(    self   :       Tuple    ,      A__   :       str    ,      A__   :       Optional[str] = None	)					->  Tuple[str]:
        if not self.can_save_slow_tokenizer:
          raise ValueError(
              '''Your fast tokenizer does not have the necessary information to save the vocabulary for a slow '''
              '''tokenizer.'''	)

        if not os.path.isdir(A__	):
          logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""	)
          return
        _snake_case		=							os.path.join(
            A__    ,      (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']	)

        if os.path.abspath(self.vocab_file	) != os.path.abspath(A__	):
          copyfile(self.vocab_file    ,      A__	)

        return (out_vocab_file,)







      def 		UpperCamelCase_	(    self   :       int    ,      A__   :       "Conversation"	)					->  Union[str, Any]:
        _snake_case		=							list(conversation.iter_texts()	)
        if not all([is_user for is_user, msg in dialogue[::2]]	) or not all(
            [not is_user for is_user, msg in dialogue[1::2]]	):
          raise ValueError(
              '''The model only supports \'user\' and \'assistant\' roles, starting with user and alternating (u/a/u/a/u...)'''	)

        _snake_case		=							[]
        if len(conversation.past_user_inputs	) > 0:
          if not conversation.past_user_inputs[0].startswith(A__	) or E_SYS not in conversation.past_user_inputs[0]:
            _snake_case		=							(
                B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + conversation.past_user_inputs[0]
            )
        elif not dialogue[0][1].startswith(A__	) or E_SYS not in dialogue[0][1]:
          _snake_case		=							(dialogue[0][0], B_SYS + DEFAULT_SYSTEM_PROMPT + E_SYS + dialogue[0][1])

        dialog_tokens += sum(
            [
                [self.bos_token_id]
                + self.encode(
                    f"""{B_INST} {(prompt[1]).strip()} {E_INST} {(answer[1]).strip()} """    ,      add_special_tokens=A__	)
                + [self.eos_token_id]
                for prompt, answer in zip(dialogue[::2]    ,      dialogue[1::2]	)
            ]    ,      []    ,      )
        if not (dialogue[-1][0]):
          raise ValueError(f"""Last message must be from user, got {dialogue[-1]['role']}"""	)
        dialog_tokens += [self.bos_token_id] + self.encode(
            f"""{B_INST} {(dialogue[-1][1]).strip()} {E_INST}"""    ,      add_special_tokens=A__	)
        return dialog_tokens




from typing import Callable, List, Optional, Union

import torch

from ...models import UNetaDModel
from ...schedulers import CMStochasticIterativeScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


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


__A    	=			'''
    Examples:
        ```py
        >>> import torch

        >>> from diffusers import ConsistencyModelPipeline

        >>> device = "cuda"
        >>> # Load the cd_imagenet64_l2 checkpoint.
        >>> model_id_or_path = "openai/diffusers-cd_imagenet64_l2"
        >>> pipe = ConsistencyModelPipeline.from_pretrained(model_id_or_path, torch_dtype=torch.float16)
        >>> pipe.to(device)

        >>> # Onestep Sampling
        >>> image = pipe(num_inference_steps=1).images[0]
        >>> image.save("cd_imagenet64_l2_onestep_sample.png")

        >>> # Onestep sampling, class-conditional image generation
        >>> # ImageNet-64 class label 145 corresponds to king penguins
        >>> image = pipe(num_inference_steps=1, class_labels=145).images[0]
        >>> image.save("cd_imagenet64_l2_onestep_sample_penguin.png")

        >>> # Multistep sampling, class-conditional image generation
        >>> # Timesteps can be explicitly specified; the particular timesteps below are from the original Github repo:
        >>> # https://github.com/openai/consistency_models/blob/main/scripts/launch.sh#L77
        >>> image = pipe(num_inference_steps=None, timesteps=[22, 0], class_labels=145).images[0]
        >>> image.save("cd_imagenet64_l2_multistep_sample_penguin.png")
        ```
'''



class 						lowercase_					(						__lowercase  ):



      def __init__(    self   :       str    ,      A__   :       UNetaDModel    ,      A__   :       CMStochasticIterativeScheduler	)					->  None:
        super().__init__()

        self.register_modules(
            unet=A__    ,      scheduler=A__    ,      )

        _snake_case		=							None



      def 		UpperCamelCase_	(    self   :       Union[str, Any]    ,      A__   :       Optional[int]=0	)					->  List[str]:
        if is_accelerate_available() and is_accelerate_version('''>='''    ,      '''0.14.0'''	):
          from accelerate import cpu_offload
        else:
          raise ImportError('''`enable_sequential_cpu_offload` requires `accelerate v0.14.0` or higher'''	)

        _snake_case		=							torch.device(f"""cuda:{gpu_id}"""	)

        if self.device.type != "cpu":
          self.to('''cpu'''    ,      silence_dtype_warnings=A__	)
          torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

        for cpu_offloaded_model in [self.unet]:
          cpu_offload(A__    ,      A__	)

        if self.safety_checker is not None:
          cpu_offload(self.safety_checker    ,      execution_device=A__    ,      offload_buffers=A__	)



      def 		UpperCamelCase_	(    self   :       Optional[int]    ,      A__   :       Optional[Any]=0	)					->  Union[str, Any]:
        if is_accelerate_available() and is_accelerate_version('''>='''    ,      '''0.17.0.dev0'''	):
          from accelerate import cpu_offload_with_hook
        else:
          raise ImportError('''`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher.'''	)

        _snake_case		=							torch.device(f"""cuda:{gpu_id}"""	)

        if self.device.type != "cpu":
          self.to('''cpu'''    ,      silence_dtype_warnings=A__	)
          torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

        _snake_case		=							None
        for cpu_offloaded_model in [self.unet]:
          _snake_case,  _snake_case		=							cpu_offload_with_hook(A__    ,      A__    ,      prev_module_hook=A__	)

        if self.safety_checker is not None:
          _snake_case,  _snake_case		=							cpu_offload_with_hook(self.safety_checker    ,      A__    ,      prev_module_hook=A__	)

        # We'll offload the last model manually.
        _snake_case		=							hook



      @property
      # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
      def 		UpperCamelCase_	(    self   :       int	)					->  List[str]:
        if not hasattr(self.unet    ,      '''_hf_hook'''	):
          return self.device
        for module in self.unet.modules():
          if (
              hasattr(A__    ,      '''_hf_hook'''	)
              and hasattr(module._hf_hook    ,      '''execution_device'''	)
              and module._hf_hook.execution_device is not None
          ):
            return torch.device(module._hf_hook.execution_device	)
        return self.device



      def 		UpperCamelCase_	(    self   :       List[str]    ,      A__   :       Tuple    ,      A__   :       str    ,      A__   :       Tuple    ,      A__   :       Union[str, Any]    ,      A__   :       Union[str, Any]    ,      A__   :       Union[str, Any]    ,      A__   :       Tuple    ,      A__   :       Union[str, Any]=None	)					->  Optional[int]:
        _snake_case		=							(batch_size, num_channels, height, width)
        if isinstance(A__    ,      A__	) and len(A__	) != batch_size:
          raise ValueError(
              f"""You have passed a list of generators of length {len(A__	)}, but requested an effective batch"""
              f""" size of {batch_size}. Make sure the batch size matches the length of the generators."""	)

        if latents is None:
          _snake_case		=							randn_tensor(A__    ,      generator=A__    ,      device=A__    ,      dtype=A__	)
        else:
          _snake_case		=							latents.to(device=A__    ,      dtype=A__	)

        # scale the initial noise by the standard deviation required by the scheduler
        _snake_case		=							latents * self.scheduler.init_noise_sigma
        return latents



      def 		UpperCamelCase_	(    self   :       Optional[Any]    ,      A__   :       torch.FloatTensor    ,      A__   :       str = "pil"	)					->  Optional[int]:
        if output_type not in ["pt", "np", "pil"]:
          raise ValueError(
              f"""output_type={output_type} is not supported. Make sure to choose one of ['pt', 'np', or 'pil']"""	)

        # Equivalent to diffusers.VaeImageProcessor.denormalize
        _snake_case		=							(sample / 2 + 0.5).clamp(0    ,      1	)
        if output_type == "pt":
          return sample

        # Equivalent to diffusers.VaeImageProcessor.pt_to_numpy
        _snake_case		=							sample.cpu().permute(0    ,      2    ,      3    ,      1	).numpy()
        if output_type == "np":
          return sample

        # Output_type must be 'pil'
        _snake_case		=							self.numpy_to_pil(A__	)
        return sample



      def 		UpperCamelCase_	(    self   :       Any    ,      A__   :       Optional[Any]    ,      A__   :       Dict    ,      A__   :       int=None	)					->  Dict:
        if self.unet.config.num_class_embeds is not None:
          if isinstance(A__    ,      A__	):
            _snake_case		=							torch.tensor(A__    ,      dtype=torch.int	)
          elif isinstance(A__    ,      A__	):
            assert batch_size == 1, "Batch size must be 1 if classes is an int"
            _snake_case		=							torch.tensor([class_labels]    ,      dtype=torch.int	)
          elif class_labels is None:
            # Randomly generate batch_size class labels
            # TODO: should use generator here? int analogue of randn_tensor is not exposed in ...utils
            _snake_case		=							torch.randint(0    ,      self.unet.config.num_class_embeds    ,      size=(batch_size,)	)
          _snake_case		=							class_labels.to(A__	)
        else:
          _snake_case		=							None
        return class_labels



      def 		UpperCamelCase_	(    self   :       List[str]    ,      A__   :       List[str]    ,      A__   :       List[str]    ,      A__   :       str    ,      A__   :       List[Any]    ,      A__   :       Optional[Any]    ,      A__   :       Dict	)					->  Optional[Any]:
        if num_inference_steps is None and timesteps is None:
          raise ValueError('''Exactly one of `num_inference_steps` or `timesteps` must be supplied.'''	)

        if num_inference_steps is not None and timesteps is not None:
          logger.warning(
              f"""Both `num_inference_steps`: {num_inference_steps} and `timesteps`: {timesteps} are supplied;"""
              ''' `timesteps` will be used over `num_inference_steps`.'''	)

        if latents is not None:
          _snake_case		=							(batch_size, 3, img_size, img_size)
          if latents.shape != expected_shape:
            raise ValueError(f"""The shape of latents is {latents.shape} but is expected to be {expected_shape}."""	)

        if (callback_steps is None) or (
            callback_steps is not None and (not isinstance(A__    ,      A__	) or callback_steps <= 0)
        ):
          raise ValueError(
              f"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
              f""" {type(A__	)}."""	)







      @torch.no_grad()
      @replace_example_docstring(A__	)
      def __call__(    self   :       Any    ,      A__   :       int = 1    ,      A__   :       Optional[Union[torch.Tensor, List[int], int]] = None    ,      A__   :       int = 1    ,      A__   :       List[int] = None    ,      A__   :       Optional[Union[torch.Generator, List[torch.Generator]]] = None    ,      A__   :       Optional[torch.FloatTensor] = None    ,      A__   :       Optional[str] = "pil"    ,      A__   :       bool = True    ,      A__   :       Optional[Callable[[int, int, torch.FloatTensor], None]] = None    ,      A__   :       int = 1    ,      )					->  Dict:
        _snake_case		=							self.unet.config.sample_size
        _snake_case		=							self._execution_device

        # 1. Check inputs
        self.check_inputs(A__    ,      A__    ,      A__    ,      A__    ,      A__    ,      A__	)

        # 2. Prepare image latents
        # Sample image latents x_0 ~ N(0, sigma_0^2 * I)
        _snake_case		=							self.prepare_latents(
            batch_size=A__    ,      num_channels=self.unet.config.in_channels    ,      height=A__    ,      width=A__    ,      dtype=self.unet.dtype    ,      device=A__    ,      generator=A__    ,      latents=A__    ,      )

        # 3. Handle class_labels for class-conditional models
        _snake_case		=							self.prepare_class_labels(A__    ,      A__    ,      class_labels=A__	)

        # 4. Prepare timesteps
        if timesteps is not None:
          self.scheduler.set_timesteps(timesteps=A__    ,      device=A__	)
          _snake_case		=							self.scheduler.timesteps
          _snake_case		=							len(A__	)
        else:
          self.scheduler.set_timesteps(A__	)
          _snake_case		=							self.scheduler.timesteps

        # 5. Denoising loop
        # Multistep sampling: implements Algorithm 1 in the paper
        with self.progress_bar(total=A__	) as progress_bar:
          for i, t in enumerate(A__	):
            _snake_case		=							self.scheduler.scale_model_input(A__    ,      A__	)
            _snake_case		=							self.unet(A__    ,      A__    ,      class_labels=A__    ,      return_dict=A__	)[0]

            _snake_case		=							self.scheduler.step(A__    ,      A__    ,      A__    ,      generator=A__	)[0]

            # call the callback, if provided
            progress_bar.update()
            if callback is not None and i % callback_steps == 0:
              callback(A__    ,      A__    ,      A__	)

        # 6. Post-process image sample
        _snake_case		=							self.postprocess_image(A__    ,      output_type=A__	)

        # Offload last model to CPU
        if hasattr(self    ,      '''final_offload_hook'''	) and self.final_offload_hook is not None:
          self.final_offload_hook.offload()

        if not return_dict:
          return (image,)

        return ImagePipelineOutput(images=A__	)

