





import math
from dataclasses import dataclass
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, logging, randn_tensor
from .scheduling_utils import KarrasDiffusionSchedulers, SchedulerMixin


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

@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->EulerAncestralDiscrete
class 						UpperCamelCase__					( lowerCAmelCase_  ):



     '''simple docstring'''

     __snake_case       :      torch.FloatTensor
     __snake_case       :      Optional[torch.FloatTensor]						      =     None



def   __lowercase  ( _SCREAMING_SNAKE_CASE					,		_SCREAMING_SNAKE_CASE=0.999					,		_SCREAMING_SNAKE_CASE="cosine"					,		)					->  List[str]:


      '''simple docstring'''


      if alpha_transform_type == "cosine":

            def alpha_bar_fn(_SCREAMING_SNAKE_CASE      ):
                  return math.cos((t + 0.008) / 1.008 * math.pi / 2      ) ** 2

      elif alpha_transform_type == "exp":

            def alpha_bar_fn(_SCREAMING_SNAKE_CASE      ):
                  return math.exp(t * -12.0      )

      else:
            raise ValueError(F"""Unsupported alpha_tranform_type: {alpha_transform_type}"""      )

      SCREAMING_SNAKE_CASE						=		[]
      for i in range(_SCREAMING_SNAKE_CASE      ):
            SCREAMING_SNAKE_CASE						=		i / num_diffusion_timesteps
            SCREAMING_SNAKE_CASE						=		(i + 1) / num_diffusion_timesteps
            betas.append(min(1 - alpha_bar_fn(_SCREAMING_SNAKE_CASE      ) / alpha_bar_fn(_SCREAMING_SNAKE_CASE      )					,		_SCREAMING_SNAKE_CASE      )      )
      return torch.tensor(_SCREAMING_SNAKE_CASE					,		dtype=torch.floataa      )



class 						UpperCamelCase__					( lowerCAmelCase_				, lowerCAmelCase_  ):



     '''simple docstring'''

     __snake_case       :      List[Any]						      =     [e.name for e in KarrasDiffusionSchedulers]
     __snake_case       :      List[Any]						      =     1






     @register_to_config
     def __init__(				self		:       Dict		,lowerCamelCase__		:       int = 1000		,lowerCamelCase__		:       float = 0.0001		,lowerCamelCase__		:       float = 0.02		,lowerCamelCase__		:       str = "linear"		,lowerCamelCase__		:       Optional[Union[np.ndarray, List[float]]] = None		,lowerCamelCase__		:       str = "epsilon"		,lowerCamelCase__		:       str = "linspace"		,lowerCamelCase__		:       int = 0		,)	->						str:




           '''simple docstring'''


           if trained_betas is not None:
                 SCREAMING_SNAKE_CASE						=		torch.tensor(lowerCamelCase__		,dtype=torch.floataa			)
           elif beta_schedule == "linear":
                 SCREAMING_SNAKE_CASE						=		torch.linspace(lowerCamelCase__		,lowerCamelCase__		,lowerCamelCase__		,dtype=torch.floataa			)
           elif beta_schedule == "scaled_linear":
                 # this schedule is very specific to the latent diffusion model.
                 SCREAMING_SNAKE_CASE						=		(
                     torch.linspace(beta_start**0.5		,beta_end**0.5		,lowerCamelCase__		,dtype=torch.floataa			) ** 2
                 )
           elif beta_schedule == "squaredcos_cap_v2":
                 # Glide cosine schedule
                 SCREAMING_SNAKE_CASE						=		betas_for_alpha_bar(lowerCamelCase__			)
           else:
                 raise NotImplementedError(F"""{beta_schedule} does is not implemented for {self.__class__}"""			)

           SCREAMING_SNAKE_CASE						=		1.0 - self.betas
           SCREAMING_SNAKE_CASE						=		torch.cumprod(self.alphas		,dim=0			)

           SCREAMING_SNAKE_CASE						=		np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5			)
           SCREAMING_SNAKE_CASE						=		np.concatenate([sigmas[::-1], [0.0]]			).astype(np.floataa			)
           SCREAMING_SNAKE_CASE						=		torch.from_numpy(lowerCamelCase__			)

           # setable values
           SCREAMING_SNAKE_CASE						=		None
           SCREAMING_SNAKE_CASE						=		np.linspace(0		,num_train_timesteps - 1		,lowerCamelCase__		,dtype=lowerCamelCase__			)[::-1].copy()
           SCREAMING_SNAKE_CASE						=		torch.from_numpy(lowerCamelCase__			)
           SCREAMING_SNAKE_CASE						=		False






     @property
     def 							SCREAMING_SNAKE_CASE__    (				self		:       List[str]			)	->						Union[str, Any]:




           '''simple docstring'''


           if self.config.timestep_spacing in ["linspace", "trailing"]:
                 return self.sigmas.max()

           return (self.sigmas.max() ** 2 + 1) ** 0.5






     def 							SCREAMING_SNAKE_CASE__    (				self		:       List[str]		,lowerCamelCase__		:       torch.FloatTensor		,lowerCamelCase__		:       Union[float, torch.FloatTensor]			)	->						torch.FloatTensor:




           '''simple docstring'''


           if isinstance(lowerCamelCase__		,torch.Tensor			):
                 SCREAMING_SNAKE_CASE						=		timestep.to(self.timesteps.device			)
           SCREAMING_SNAKE_CASE						=		(self.timesteps == timestep).nonzero().item()
           SCREAMING_SNAKE_CASE						=		self.sigmas[step_index]
           SCREAMING_SNAKE_CASE						=		sample / ((sigma**2 + 1) ** 0.5)
           SCREAMING_SNAKE_CASE						=		True
           return sample






     def 							SCREAMING_SNAKE_CASE__    (				self		:       Tuple		,lowerCamelCase__		:       int		,lowerCamelCase__		:       Union[str, torch.device] = None			)	->						Optional[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		num_inference_steps

           # "linspace", "leading", "trailing" corresponds to annotation of Table 2. of https://arxiv.org/abs/2305.08891
           if self.config.timestep_spacing == "linspace":
                 SCREAMING_SNAKE_CASE						=		np.linspace(0		,self.config.num_train_timesteps - 1		,lowerCamelCase__		,dtype=lowerCamelCase__			)[
                     ::-1
                 ].copy()
           elif self.config.timestep_spacing == "leading":
                 SCREAMING_SNAKE_CASE						=		self.config.num_train_timesteps // self.num_inference_steps
                 # creates integer timesteps by multiplying by ratio
                 # casting to int to avoid issues when num_inference_step is power of 3
                 SCREAMING_SNAKE_CASE						=		(np.arange(0		,lowerCamelCase__			) * step_ratio).round()[::-1].copy().astype(lowerCamelCase__			)
                 timesteps += self.config.steps_offset
           elif self.config.timestep_spacing == "trailing":
                 SCREAMING_SNAKE_CASE						=		self.config.num_train_timesteps / self.num_inference_steps
                 # creates integer timesteps by multiplying by ratio
                 # casting to int to avoid issues when num_inference_step is power of 3
                 SCREAMING_SNAKE_CASE						=		(np.arange(self.config.num_train_timesteps		,0		,-step_ratio			)).round().copy().astype(lowerCamelCase__			)
                 timesteps -= 1
           else:
                 raise ValueError(
                     F"""{self.config.timestep_spacing} is not supported. Please make sure to choose one of 'linspace', 'leading' or 'trailing'."""			)

           SCREAMING_SNAKE_CASE						=		np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5			)
           SCREAMING_SNAKE_CASE						=		np.interp(lowerCamelCase__		,np.arange(0		,len(lowerCamelCase__			)			)		,lowerCamelCase__			)
           SCREAMING_SNAKE_CASE						=		np.concatenate([sigmas, [0.0]]			).astype(np.floataa			)
           SCREAMING_SNAKE_CASE						=		torch.from_numpy(lowerCamelCase__			).to(device=lowerCamelCase__			)
           if str(lowerCamelCase__			).startswith("""mps"""			):
                 # mps does not support float64
                 SCREAMING_SNAKE_CASE						=		torch.from_numpy(lowerCamelCase__			).to(lowerCamelCase__		,dtype=torch.floataa			)
           else:
                 SCREAMING_SNAKE_CASE						=		torch.from_numpy(lowerCamelCase__			).to(device=lowerCamelCase__			)






     def 							SCREAMING_SNAKE_CASE__    (				self		:       Dict		,lowerCamelCase__		:       torch.FloatTensor		,lowerCamelCase__		:       Union[float, torch.FloatTensor]		,lowerCamelCase__		:       torch.FloatTensor		,lowerCamelCase__		:       Optional[torch.Generator] = None		,lowerCamelCase__		:       bool = True		,)	->						Union[EulerAncestralDiscreteSchedulerOutput, Tuple]:




           '''simple docstring'''


           if (
               isinstance(lowerCamelCase__		,lowerCamelCase__			)
               or isinstance(lowerCamelCase__		,torch.IntTensor			)
               or isinstance(lowerCamelCase__		,torch.LongTensor			)
           ):
                 raise ValueError(
                     (
                         """Passing integer indices (e.g. from `enumerate(timesteps)`) as timesteps to"""
                         """ `EulerDiscreteScheduler.step()` is not supported. Make sure to pass"""
                         """ one of the `scheduler.timesteps` as a timestep."""
                     )		,)

           if not self.is_scale_input_called:
                 logger.warning(
                     """The `scale_model_input` function should be called before `step` to ensure correct denoising. """
                     """See `StableDiffusionPipeline` for a usage example."""			)

           if isinstance(lowerCamelCase__		,torch.Tensor			):
                 SCREAMING_SNAKE_CASE						=		timestep.to(self.timesteps.device			)

           SCREAMING_SNAKE_CASE						=		(self.timesteps == timestep).nonzero().item()
           SCREAMING_SNAKE_CASE						=		self.sigmas[step_index]

           # 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
           if self.config.prediction_type == "epsilon":
                 SCREAMING_SNAKE_CASE						=		sample - sigma * model_output
           elif self.config.prediction_type == "v_prediction":
                 # * c_out + input * c_skip
                 SCREAMING_SNAKE_CASE						=		model_output * (-sigma / (sigma**2 + 1) ** 0.5) + (sample / (sigma**2 + 1))
           elif self.config.prediction_type == "sample":
                 raise NotImplementedError("""prediction_type not implemented yet: sample"""			)
           else:
                 raise ValueError(
                     F"""prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`"""			)

           SCREAMING_SNAKE_CASE						=		self.sigmas[step_index]
           SCREAMING_SNAKE_CASE						=		self.sigmas[step_index + 1]
           SCREAMING_SNAKE_CASE						=		(sigma_to**2 * (sigma_from**2 - sigma_to**2) / sigma_from**2) ** 0.5
           SCREAMING_SNAKE_CASE						=		(sigma_to**2 - sigma_up**2) ** 0.5

           # 2. Convert to an ODE derivative
           SCREAMING_SNAKE_CASE						=		(sample - pred_original_sample) / sigma

           SCREAMING_SNAKE_CASE						=		sigma_down - sigma

           SCREAMING_SNAKE_CASE						=		sample + derivative * dt

           SCREAMING_SNAKE_CASE						=		model_output.device
           SCREAMING_SNAKE_CASE						=		randn_tensor(model_output.shape		,dtype=model_output.dtype		,device=lowerCamelCase__		,generator=lowerCamelCase__			)

           SCREAMING_SNAKE_CASE						=		prev_sample + noise * sigma_up

           if not return_dict:
                 return (prev_sample,)

           return EulerAncestralDiscreteSchedulerOutput(
               prev_sample=lowerCamelCase__		,pred_original_sample=lowerCamelCase__			)






     def 							SCREAMING_SNAKE_CASE__    (				self		:       Any		,lowerCamelCase__		:       torch.FloatTensor		,lowerCamelCase__		:       torch.FloatTensor		,lowerCamelCase__		:       torch.FloatTensor		,)	->						torch.FloatTensor:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		self.sigmas.to(device=original_samples.device		,dtype=original_samples.dtype			)
           if original_samples.device.type == "mps" and torch.is_floating_point(lowerCamelCase__			):
                 # mps does not support float64
                 SCREAMING_SNAKE_CASE						=		self.timesteps.to(original_samples.device		,dtype=torch.floataa			)
                 SCREAMING_SNAKE_CASE						=		timesteps.to(original_samples.device		,dtype=torch.floataa			)
           else:
                 SCREAMING_SNAKE_CASE						=		self.timesteps.to(original_samples.device			)
                 SCREAMING_SNAKE_CASE						=		timesteps.to(original_samples.device			)

           SCREAMING_SNAKE_CASE						=		[(schedule_timesteps == t).nonzero().item() for t in timesteps]

           SCREAMING_SNAKE_CASE						=		sigmas[step_indices].flatten()
           while len(sigma.shape			) < len(original_samples.shape			):
                 SCREAMING_SNAKE_CASE						=		sigma.unsqueeze(-1			)

           SCREAMING_SNAKE_CASE						=		original_samples + noise * sigma
           return noisy_samples






     def __len__(				self		:       Tuple			)	->						str:




           '''simple docstring'''


           return self.config.num_train_timesteps









import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin


if is_torch_available():
      import torch

if is_vision_available():
      from PIL import Image

      from transformers import TvltImageProcessor



def   __lowercase  ( _SCREAMING_SNAKE_CASE					,		_SCREAMING_SNAKE_CASE=10					,		_SCREAMING_SNAKE_CASE=10					,		_SCREAMING_SNAKE_CASE=False					,		_SCREAMING_SNAKE_CASE=False      )					->  Optional[Any]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE						=		[]
      for i in range(image_processor_tester.num_frames      ):
            video.append(np.random.randint(2_55					,		size=(image_processor_tester.num_channels, width, height)					,		dtype=np.uinta      )      )

      if not numpify and not torchify:
            # PIL expects the channel dimension as last dimension
            SCREAMING_SNAKE_CASE						=		[Image.fromarray(np.moveaxis(_SCREAMING_SNAKE_CASE					,		0					,		-1      )      ) for frame in video]

      if torchify:
            SCREAMING_SNAKE_CASE						=		[torch.from_numpy(_SCREAMING_SNAKE_CASE      ) for frame in video]

      return video



def   __lowercase  ( _SCREAMING_SNAKE_CASE					,		_SCREAMING_SNAKE_CASE=False					,		_SCREAMING_SNAKE_CASE=False					,		_SCREAMING_SNAKE_CASE=False      )					->  Tuple:


      '''simple docstring'''


      assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

      SCREAMING_SNAKE_CASE						=		[]
      for i in range(image_processor_tester.batch_size      ):
            if equal_resolution:
                  SCREAMING_SNAKE_CASE						=		SCREAMING_SNAKE_CASE						=		image_processor_tester.max_resolution
            else:
                  SCREAMING_SNAKE_CASE,   SCREAMING_SNAKE_CASE						=		np.random.choice(
                      np.arange(image_processor_tester.min_resolution					,		image_processor_tester.max_resolution      )					,		2      )
                  SCREAMING_SNAKE_CASE						=		prepare_video(
                      image_processor_tester=_SCREAMING_SNAKE_CASE					,		width=_SCREAMING_SNAKE_CASE					,		height=_SCREAMING_SNAKE_CASE					,		numpify=_SCREAMING_SNAKE_CASE					,		torchify=_SCREAMING_SNAKE_CASE					,		)
            video_inputs.append(_SCREAMING_SNAKE_CASE      )

      return video_inputs

class 						UpperCamelCase__					( unittest.TestCase  ):



     '''simple docstring'''

     def __init__(				self		:       Union[str, Any]		,lowerCamelCase__		:       Tuple		,lowerCamelCase__		:       Any=7		,lowerCamelCase__		:       str=3		,lowerCamelCase__		:       Union[str, Any]=4		,lowerCamelCase__		:       Union[str, Any]=18		,lowerCamelCase__		:       Any=30		,lowerCamelCase__		:       Union[str, Any]=400		,lowerCamelCase__		:       Any=True		,lowerCamelCase__		:       Any=None		,lowerCamelCase__		:       Tuple=True		,lowerCamelCase__		:       int=[0.5, 0.5, 0.5]		,lowerCamelCase__		:       Any=[0.5, 0.5, 0.5]		,lowerCamelCase__		:       Dict=True		,lowerCamelCase__		:       Tuple=None		,)	->						Tuple:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		size if size is not None else {"""shortest_edge""": 18}
           SCREAMING_SNAKE_CASE						=		crop_size if crop_size is not None else {"""height""": 18, """width""": 18}

           SCREAMING_SNAKE_CASE						=		parent
           SCREAMING_SNAKE_CASE						=		batch_size
           SCREAMING_SNAKE_CASE						=		num_channels
           SCREAMING_SNAKE_CASE						=		num_frames
           SCREAMING_SNAKE_CASE						=		image_size
           SCREAMING_SNAKE_CASE						=		min_resolution
           SCREAMING_SNAKE_CASE						=		max_resolution
           SCREAMING_SNAKE_CASE						=		do_resize
           SCREAMING_SNAKE_CASE						=		size
           SCREAMING_SNAKE_CASE						=		do_normalize
           SCREAMING_SNAKE_CASE						=		image_mean
           SCREAMING_SNAKE_CASE						=		image_std
           SCREAMING_SNAKE_CASE						=		do_center_crop
           SCREAMING_SNAKE_CASE						=		crop_size






     def 							SCREAMING_SNAKE_CASE__    (				self		:       List[Any]			)	->						int:




           '''simple docstring'''


           return {
               "image_mean": self.image_mean,
               "image_std": self.image_std,
               "do_normalize": self.do_normalize,
               "do_resize": self.do_resize,
               "size": self.size,
               "do_center_crop": self.do_center_crop,
               "crop_size": self.crop_size,
           }

@require_torch
@require_vision
class 						UpperCamelCase__					( lowerCAmelCase_				, unittest.TestCase  ):



     '''simple docstring'''

     __snake_case       :      int						      =     TvltImageProcessor if is_vision_available() else None






     def 							SCREAMING_SNAKE_CASE__    (				self		:       Dict			)	->						Tuple:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		TvltImageProcessorTester(self			)






     @property
     def 							SCREAMING_SNAKE_CASE__    (				self		:       Optional[int]			)	->						Dict:




           '''simple docstring'''


           return self.image_processor_tester.prepare_image_processor_dict()






     def 							SCREAMING_SNAKE_CASE__    (				self		:       Any			)	->						List[str]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		self.image_processing_class(**self.image_processor_dict			)
           self.assertTrue(hasattr(lowerCamelCase__		,"""image_mean"""			)			)
           self.assertTrue(hasattr(lowerCamelCase__		,"""image_std"""			)			)
           self.assertTrue(hasattr(lowerCamelCase__		,"""do_normalize"""			)			)
           self.assertTrue(hasattr(lowerCamelCase__		,"""do_resize"""			)			)
           self.assertTrue(hasattr(lowerCamelCase__		,"""do_center_crop"""			)			)
           self.assertTrue(hasattr(lowerCamelCase__		,"""size"""			)			)






     def 							SCREAMING_SNAKE_CASE__    (				self		:       List[Any]			)	->						List[str]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		self.image_processing_class(**self.image_processor_dict			)
           # create random PIL videos
           SCREAMING_SNAKE_CASE						=		prepare_video_inputs(self.image_processor_tester		,equal_resolution=lowerCamelCase__			)
           for video in video_inputs:
                 self.assertIsInstance(lowerCamelCase__		,lowerCamelCase__			)
                 self.assertIsInstance(video[0]		,Image.Image			)

           # Test not batched input
           SCREAMING_SNAKE_CASE						=		image_processor(video_inputs[0]		,return_tensors="""pt"""			).pixel_values
           self.assertEqual(
               encoded_videos.shape		,(
                   1,
                   self.image_processor_tester.num_frames,
                   self.image_processor_tester.num_channels,
                   self.image_processor_tester.crop_size["""height"""],
                   self.image_processor_tester.crop_size["""width"""],
               )		,)

           # Test batched
           SCREAMING_SNAKE_CASE						=		image_processor(lowerCamelCase__		,return_tensors="""pt"""			).pixel_values
           self.assertEqual(
               encoded_videos.shape		,(
                   self.image_processor_tester.batch_size,
                   self.image_processor_tester.num_frames,
                   self.image_processor_tester.num_channels,
                   self.image_processor_tester.crop_size["""height"""],
                   self.image_processor_tester.crop_size["""width"""],
               )		,)






     def 							SCREAMING_SNAKE_CASE__    (				self		:       int			)	->						int:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		self.image_processing_class(**self.image_processor_dict			)
           # create random numpy tensors
           SCREAMING_SNAKE_CASE						=		prepare_video_inputs(self.image_processor_tester		,equal_resolution=lowerCamelCase__		,numpify=lowerCamelCase__			)
           for video in video_inputs:
                 self.assertIsInstance(lowerCamelCase__		,lowerCamelCase__			)
                 self.assertIsInstance(video[0]		,np.ndarray			)

           # Test not batched input
           SCREAMING_SNAKE_CASE						=		image_processor(video_inputs[0]		,return_tensors="""pt"""			).pixel_values
           self.assertEqual(
               encoded_videos.shape		,(
                   1,
                   self.image_processor_tester.num_frames,
                   self.image_processor_tester.num_channels,
                   self.image_processor_tester.crop_size["""height"""],
                   self.image_processor_tester.crop_size["""width"""],
               )		,)

           # Test batched
           SCREAMING_SNAKE_CASE						=		image_processor(lowerCamelCase__		,return_tensors="""pt"""			).pixel_values
           self.assertEqual(
               encoded_videos.shape		,(
                   self.image_processor_tester.batch_size,
                   self.image_processor_tester.num_frames,
                   self.image_processor_tester.num_channels,
                   self.image_processor_tester.crop_size["""height"""],
                   self.image_processor_tester.crop_size["""width"""],
               )		,)






     def 							SCREAMING_SNAKE_CASE__    (				self		:       str			)	->						Optional[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		self.image_processing_class(**self.image_processor_dict			)
           # create random PyTorch tensors
           SCREAMING_SNAKE_CASE						=		prepare_video_inputs(self.image_processor_tester		,equal_resolution=lowerCamelCase__		,torchify=lowerCamelCase__			)
           for video in video_inputs:
                 self.assertIsInstance(lowerCamelCase__		,lowerCamelCase__			)
                 self.assertIsInstance(video[0]		,torch.Tensor			)

           # Test not batched input
           SCREAMING_SNAKE_CASE						=		image_processor(video_inputs[0]		,return_tensors="""pt"""			).pixel_values
           self.assertEqual(
               encoded_videos.shape		,(
                   1,
                   self.image_processor_tester.num_frames,
                   self.image_processor_tester.num_channels,
                   self.image_processor_tester.crop_size["""height"""],
                   self.image_processor_tester.crop_size["""width"""],
               )		,)

           # Test batched
           SCREAMING_SNAKE_CASE						=		image_processor(lowerCamelCase__		,return_tensors="""pt"""			).pixel_values
           self.assertEqual(
               encoded_videos.shape		,(
                   self.image_processor_tester.batch_size,
                   self.image_processor_tester.num_frames,
                   self.image_processor_tester.num_channels,
                   self.image_processor_tester.crop_size["""height"""],
                   self.image_processor_tester.crop_size["""width"""],
               )		,)


