

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__										=      "\n    Examples:\n        ```py\n        >>> import torch\n\n        >>> from diffusers import ConsistencyModelPipeline\n\n        >>> device = \"cuda\"\n        >>> # Load the cd_imagenet64_l2 checkpoint.\n        >>> model_id_or_path = \"openai/diffusers-cd_imagenet64_l2\"\n        >>> pipe = ConsistencyModelPipeline.from_pretrained(model_id_or_path, torch_dtype=torch.float16)\n        >>> pipe.to(device)\n\n        >>> # Onestep Sampling\n        >>> image = pipe(num_inference_steps=1).images[0]\n        >>> image.save(\"cd_imagenet64_l2_onestep_sample.png\")\n\n        >>> # Onestep sampling, class-conditional image generation\n        >>> # ImageNet-64 class label 145 corresponds to king penguins\n        >>> image = pipe(num_inference_steps=1, class_labels=145).images[0]\n        >>> image.save(\"cd_imagenet64_l2_onestep_sample_penguin.png\")\n\n        >>> # Multistep sampling, class-conditional image generation\n        >>> # Timesteps can be explicitly specified; the particular timesteps below are from the original Github repo:\n        >>> # https://github.com/openai/consistency_models/blob/main/scripts/launch.sh#L77\n        >>> image = pipe(num_inference_steps=None, timesteps=[22, 0], class_labels=145).images[0]\n        >>> image.save(\"cd_imagenet64_l2_multistep_sample_penguin.png\")\n        ```\n"







class  UpperCamelCase_      (	_snake_case      ):


  """simple docstring"""



  def __init__(						self     :					Optional[Any]  ,	A_     :					UNetaDModel  ,	A_     :					CMStochasticIterativeScheduler						)       ->			None:
      super().__init__()

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

      SCREAMING_SNAKE_CASE__         =    None



  def   _UpperCamelCase     (						self     :					Any  ,	A_     :					Dict=0						)       ->			Any:
      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"						)

      SCREAMING_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     :					List[str]  ,	A_     :					Optional[Any]=0						)       ->			int:
      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."						)

      SCREAMING_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)

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

      if self.safety_checker is not None:
          SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__         =    cpu_offload_with_hook(self.safety_checker  ,	A_  ,	prev_module_hook=A_						)

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



  @property
  # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
  def   _UpperCamelCase     (						self     :					Tuple						)       ->			List[Any]:
      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     :					str  ,	A_     :					List[Any]  ,	A_     :					List[Any]  ,	A_     :					Dict  ,	A_     :					Optional[int]  ,	A_     :					List[str]  ,	A_     :					List[str]  ,	A_     :					Any  ,	A_     :					Tuple=None						)       ->			Optional[Any]:
      SCREAMING_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:
          SCREAMING_SNAKE_CASE__         =    randn_tensor(A_  ,	generator=A_  ,	device=A_  ,	dtype=A_						)
      else:
          SCREAMING_SNAKE_CASE__         =    latents.to(device=A_  ,	dtype=A_						)

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



  def   _UpperCamelCase     (						self     :					int  ,	A_     :					torch.FloatTensor  ,	A_     :					str = "pil"						)       ->			Tuple:
      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
      SCREAMING_SNAKE_CASE__         =    (sample / 2 + 0.5).clamp(0  ,	1						)
      if output_type == "pt":
          return sample

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

      # Output_type must be 'pil'
      SCREAMING_SNAKE_CASE__         =    self.numpy_to_pil(A_						)
      return sample



  def   _UpperCamelCase     (						self     :					List[str]  ,	A_     :					List[Any]  ,	A_     :					Optional[Any]  ,	A_     :					Optional[Any]=None						)       ->			List[Any]:
      if self.unet.config.num_class_embeds is not None:
          if isinstance(A_  ,	A_						):
              SCREAMING_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"
              SCREAMING_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
              SCREAMING_SNAKE_CASE__         =    torch.randint(0  ,	self.unet.config.num_class_embeds  ,	size=(batch_size,)						)
          SCREAMING_SNAKE_CASE__         =    class_labels.to(A_						)
      else:
          SCREAMING_SNAKE_CASE__         =    None
      return class_labels



  def   _UpperCamelCase     (						self     :					List[Any]  ,	A_     :					List[str]  ,	A_     :					Dict  ,	A_     :					List[str]  ,	A_     :					Dict  ,	A_     :					Optional[int]  ,	A_     :					Tuple						)       ->			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:
          SCREAMING_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     :					Optional[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  ,	)       ->			Optional[Any]:
      SCREAMING_SNAKE_CASE__         =    self.unet.config.sample_size
      SCREAMING_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)
      SCREAMING_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
      SCREAMING_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_						)
          SCREAMING_SNAKE_CASE__         =    self.scheduler.timesteps
          SCREAMING_SNAKE_CASE__         =    len(A_						)
      else:
          self.scheduler.set_timesteps(A_						)
          SCREAMING_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_						):
              SCREAMING_SNAKE_CASE__         =    self.scheduler.scale_model_input(A_  ,	A_						)
              SCREAMING_SNAKE_CASE__         =    self.unet(A_  ,	A_  ,	class_labels=A_  ,	return_dict=A_						)[0]

              SCREAMING_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
      SCREAMING_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_						)





def 				__a							(       __A			,							__A			,							__A			,							__A						):
    return len_board * len_board_column * row + column





def 				__a							(       __A			,							__A			,							__A			,							__A			,							__A			,							__A			,							):

    if board[row][column] != word[word_index]:
        return False

    if word_index == len(__A						) - 1:
        return True

    SCREAMING_SNAKE_CASE__         =    [(0, 1), (0, -1), (-1, 0), (1, 0)]
    SCREAMING_SNAKE_CASE__         =    len(__A						)
    SCREAMING_SNAKE_CASE__         =    len(board[0]						)
    for direction in traverts_directions:
        SCREAMING_SNAKE_CASE__         =    row + direction[0]
        SCREAMING_SNAKE_CASE__         =    column + direction[1]
        if not (0 <= next_i < len_board and 0 <= next_j < len_board_column):
            continue

        SCREAMING_SNAKE_CASE__         =    get_point_key(__A			,							__A			,							__A			,							__A						)
        if key in visited_points_set:
            continue

        visited_points_set.add(__A						)
        if exits_word(__A			,							__A			,							__A			,							__A			,							word_index + 1			,							__A						):
            return True

        visited_points_set.remove(__A						)

    return False





def 				__a							(       __A			,							__A						):
    SCREAMING_SNAKE_CASE__         =    (
        "The board should be a non empty matrix of single chars strings."
    )

    SCREAMING_SNAKE_CASE__         =    len(__A						)
    if not isinstance(__A			,							__A						) or len(__A						) == 0:
        raise ValueError(__A						)

    for row in board:
        if not isinstance(__A			,							__A						) or len(__A						) == 0:
            raise ValueError(__A						)

        for item in row:
            if not isinstance(__A			,							__A						) or len(__A						) != 1:
                raise ValueError(__A						)

    # Validate word
    if not isinstance(__A			,							__A						) or len(__A						) == 0:
        raise ValueError(
            "The word parameter should be a string of length greater than 0."						)

    SCREAMING_SNAKE_CASE__         =    len(board[0]						)
    for i in range(__A						):
        for j in range(__A						):
            if exits_word(
                __A			,							__A			,							__A			,							__A			,							0			,							{get_point_key(__A			,							__A			,							__A			,							__A						)}						):
                return True

    return False


if __name__ == "__main__":
  import doctest

  doctest.testmod()


