


'''simple docstring'''






import tempfile

import torch

from diffusers import PNDMScheduler

from .test_schedulers import SchedulerCommonTest






class  __magic_name__							(		lowerCAmelCase    ):
       UpperCAmelCase										=(PNDMScheduler,)
       UpperCAmelCase										=(("num_inference_steps", 5_0),)






       def       lowerCAmelCase		(      self  ,							**snake_case)						->  Optional[Any]:




          '''simple docstring'''
          _UpperCAmelCase       :       List[Any]							   ={
              'num_train_timesteps': 1_0_0_0,
              'beta_start': 0.00_01,
              'beta_end': 0.02,
              'beta_schedule': 'linear',
          }

          config.update(**snake_case)
          return config






       def       lowerCAmelCase		(      self  ,							snake_case=0  ,							**snake_case)						->  Tuple:




          '''simple docstring'''
          _UpperCAmelCase       :       Dict							   =dict(self.forward_default_kwargs)
          _UpperCAmelCase       :       List[Any]							   =kwargs.pop('num_inference_steps'  ,							snake_case)
          _UpperCAmelCase       :       List[str]							   =self.dummy_sample
          _UpperCAmelCase       :       Any							   =0.1 * sample
          _UpperCAmelCase       :       int							   =[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

          for scheduler_class in self.scheduler_classes:
             _UpperCAmelCase       :       Tuple							   =self.get_scheduler_config(**snake_case)
             _UpperCAmelCase       :       Tuple							   =scheduler_class(**snake_case)
             scheduler.set_timesteps(snake_case)
             # copy over dummy past residuals
             _UpperCAmelCase       :       Any							   =dummy_past_residuals[:]

             with tempfile.TemporaryDirectory() as tmpdirname:
                scheduler.save_config(snake_case)
                _UpperCAmelCase       :       int							   =scheduler_class.from_pretrained(snake_case)
                new_scheduler.set_timesteps(snake_case)
                # copy over dummy past residuals
                _UpperCAmelCase       :       Tuple							   =dummy_past_residuals[:]

             _UpperCAmelCase       :       Optional[int]							   =scheduler.step_prk(snake_case  ,							snake_case  ,							snake_case  ,							**snake_case).prev_sample
             _UpperCAmelCase       :       List[Any]							   =new_scheduler.step_prk(snake_case  ,							snake_case  ,							snake_case  ,							**snake_case).prev_sample

             assert torch.sum(torch.abs(output - new_output)) < 1E-5, "Scheduler outputs are not identical"

             _UpperCAmelCase       :       List[Any]							   =scheduler.step_plms(snake_case  ,							snake_case  ,							snake_case  ,							**snake_case).prev_sample
             _UpperCAmelCase       :       int							   =new_scheduler.step_plms(snake_case  ,							snake_case  ,							snake_case  ,							**snake_case).prev_sample

             assert torch.sum(torch.abs(output - new_output)) < 1E-5, "Scheduler outputs are not identical"






       def       lowerCAmelCase		(      self)						->  List[str]:




          '''simple docstring'''
          pass






       def       lowerCAmelCase		(      self  ,							snake_case=0  ,							**snake_case)						->  List[Any]:




          '''simple docstring'''
          _UpperCAmelCase       :       str							   =dict(self.forward_default_kwargs)
          _UpperCAmelCase       :       str							   =kwargs.pop('num_inference_steps'  ,							snake_case)
          _UpperCAmelCase       :       Any							   =self.dummy_sample
          _UpperCAmelCase       :       Optional[int]							   =0.1 * sample
          _UpperCAmelCase       :       Optional[int]							   =[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]

          for scheduler_class in self.scheduler_classes:
             _UpperCAmelCase       :       Tuple							   =self.get_scheduler_config()
             _UpperCAmelCase       :       Optional[int]							   =scheduler_class(**snake_case)
             scheduler.set_timesteps(snake_case)

             # copy over dummy past residuals (must be after setting timesteps)
             _UpperCAmelCase       :       Union[str, Any]							   =dummy_past_residuals[:]

             with tempfile.TemporaryDirectory() as tmpdirname:
                scheduler.save_config(snake_case)
                _UpperCAmelCase       :       Union[str, Any]							   =scheduler_class.from_pretrained(snake_case)
                # copy over dummy past residuals
                new_scheduler.set_timesteps(snake_case)

                # copy over dummy past residual (must be after setting timesteps)
                _UpperCAmelCase       :       Any							   =dummy_past_residuals[:]

             _UpperCAmelCase       :       Optional[int]							   =scheduler.step_prk(snake_case  ,							snake_case  ,							snake_case  ,							**snake_case).prev_sample
             _UpperCAmelCase       :       Dict							   =new_scheduler.step_prk(snake_case  ,							snake_case  ,							snake_case  ,							**snake_case).prev_sample

             assert torch.sum(torch.abs(output - new_output)) < 1E-5, "Scheduler outputs are not identical"

             _UpperCAmelCase       :       Union[str, Any]							   =scheduler.step_plms(snake_case  ,							snake_case  ,							snake_case  ,							**snake_case).prev_sample
             _UpperCAmelCase       :       int							   =new_scheduler.step_plms(snake_case  ,							snake_case  ,							snake_case  ,							**snake_case).prev_sample

             assert torch.sum(torch.abs(output - new_output)) < 1E-5, "Scheduler outputs are not identical"






       def       lowerCAmelCase		(      self  ,							**snake_case)						->  Any:




          '''simple docstring'''
          _UpperCAmelCase       :       List[Any]							   =self.scheduler_classes[0]
          _UpperCAmelCase       :       Union[str, Any]							   =self.get_scheduler_config(**snake_case)
          _UpperCAmelCase       :       Optional[Any]							   =scheduler_class(**snake_case)

          _UpperCAmelCase       :       Optional[Any]							   =1_0
          _UpperCAmelCase       :       Any							   =self.dummy_model()
          _UpperCAmelCase       :       List[str]							   =self.dummy_sample_deter
          scheduler.set_timesteps(snake_case)

          for i, t in enumerate(scheduler.prk_timesteps):
             _UpperCAmelCase       :       int							   =model(snake_case  ,							snake_case)
             _UpperCAmelCase       :       List[Any]							   =scheduler.step_prk(snake_case  ,							snake_case  ,							snake_case).prev_sample

          for i, t in enumerate(scheduler.plms_timesteps):
             _UpperCAmelCase       :       Any							   =model(snake_case  ,							snake_case)
             _UpperCAmelCase       :       List[str]							   =scheduler.step_plms(snake_case  ,							snake_case  ,							snake_case).prev_sample

          return sample






       def       lowerCAmelCase		(      self)						->  Union[str, Any]:




          '''simple docstring'''
          _UpperCAmelCase       :       str							   =dict(self.forward_default_kwargs)

          _UpperCAmelCase       :       int							   =kwargs.pop('num_inference_steps'  ,							snake_case)

          for scheduler_class in self.scheduler_classes:
             _UpperCAmelCase       :       List[str]							   =self.get_scheduler_config()
             _UpperCAmelCase       :       Dict							   =scheduler_class(**snake_case)

             _UpperCAmelCase       :       Optional[int]							   =self.dummy_sample
             _UpperCAmelCase       :       Dict							   =0.1 * sample

             if num_inference_steps is not None and hasattr(snake_case  ,							'set_timesteps'):
                scheduler.set_timesteps(snake_case)
             elif num_inference_steps is not None and not hasattr(snake_case  ,							'set_timesteps'):
                _UpperCAmelCase       :       Dict							   =num_inference_steps

             # copy over dummy past residuals (must be done after set_timesteps)
             _UpperCAmelCase       :       int							   =[residual + 0.2, residual + 0.15, residual + 0.1, residual + 0.05]
             _UpperCAmelCase       :       Optional[Any]							   =dummy_past_residuals[:]

             _UpperCAmelCase       :       str							   =scheduler.step_prk(snake_case  ,							0  ,							snake_case  ,							**snake_case).prev_sample
             _UpperCAmelCase       :       Dict							   =scheduler.step_prk(snake_case  ,							1  ,							snake_case  ,							**snake_case).prev_sample

             self.assertEqual(output_a.shape  ,							sample.shape)
             self.assertEqual(output_a.shape  ,							output_a.shape)

             _UpperCAmelCase       :       str							   =scheduler.step_plms(snake_case  ,							0  ,							snake_case  ,							**snake_case).prev_sample
             _UpperCAmelCase       :       str							   =scheduler.step_plms(snake_case  ,							1  ,							snake_case  ,							**snake_case).prev_sample

             self.assertEqual(output_a.shape  ,							sample.shape)
             self.assertEqual(output_a.shape  ,							output_a.shape)






       def       lowerCAmelCase		(      self)						->  Optional[Any]:




          '''simple docstring'''
          for timesteps in [1_0_0, 1_0_0_0]:
             self.check_over_configs(num_train_timesteps=snake_case)






       def       lowerCAmelCase		(      self)						->  Any:




          '''simple docstring'''
          for steps_offset in [0, 1]:
             self.check_over_configs(steps_offset=snake_case)

          _UpperCAmelCase       :       Dict							   =self.scheduler_classes[0]
          _UpperCAmelCase       :       Optional[Any]							   =self.get_scheduler_config(steps_offset=1)
          _UpperCAmelCase       :       int							   =scheduler_class(**snake_case)
          scheduler.set_timesteps(1_0)
          assert torch.equal(
              scheduler.timesteps  ,							torch.LongTensor(
                  [9_0_1, 8_5_1, 8_5_1, 8_0_1, 8_0_1, 7_5_1, 7_5_1, 7_0_1, 7_0_1, 6_5_1, 6_5_1, 6_0_1, 6_0_1, 5_0_1, 4_0_1, 3_0_1, 2_0_1, 1_0_1, 1])  ,							)






       def       lowerCAmelCase		(      self)						->  Optional[Any]:




          '''simple docstring'''
          for beta_start, beta_end in zip([0.00_01, 0.0_01]  ,							[0.0_02, 0.02]):
             self.check_over_configs(beta_start=snake_case  ,							beta_end=snake_case)






       def       lowerCAmelCase		(      self)						->  Optional[int]:




          '''simple docstring'''
          for schedule in ["linear", "squaredcos_cap_v2"]:
             self.check_over_configs(beta_schedule=snake_case)






       def       lowerCAmelCase		(      self)						->  Any:




          '''simple docstring'''
          for prediction_type in ["epsilon", "v_prediction"]:
             self.check_over_configs(prediction_type=snake_case)






       def       lowerCAmelCase		(      self)						->  List[Any]:




          '''simple docstring'''
          for t in [1, 5, 1_0]:
             self.check_over_forward(time_step=snake_case)






       def       lowerCAmelCase		(      self)						->  Optional[Any]:




          '''simple docstring'''
          for t, num_inference_steps in zip([1, 5, 1_0]  ,							[1_0, 5_0, 1_0_0]):
             self.check_over_forward(num_inference_steps=snake_case)






       def       lowerCAmelCase		(      self)						->  int:




          '''simple docstring'''
          # earlier version of set_timesteps() caused an error indexing alpha's with inference steps as power of 3
          _UpperCAmelCase       :       Dict							   =2_7

          for scheduler_class in self.scheduler_classes:
             _UpperCAmelCase       :       Dict							   =self.dummy_sample
             _UpperCAmelCase       :       int							   =0.1 * sample

             _UpperCAmelCase       :       List[str]							   =self.get_scheduler_config()
             _UpperCAmelCase       :       Tuple							   =scheduler_class(**snake_case)

             scheduler.set_timesteps(snake_case)

             # before power of 3 fix, would error on first step, so we only need to do two
             for i, t in enumerate(scheduler.prk_timesteps[:2]):
                _UpperCAmelCase       :       Dict							   =scheduler.step_prk(snake_case  ,							snake_case  ,							snake_case).prev_sample






       def       lowerCAmelCase		(      self)						->  str:




          '''simple docstring'''
          with self.assertRaises(snake_case):
             _UpperCAmelCase       :       Optional[Any]							   =self.scheduler_classes[0]
             _UpperCAmelCase       :       Dict							   =self.get_scheduler_config()
             _UpperCAmelCase       :       str							   =scheduler_class(**snake_case)

             scheduler.step_plms(self.dummy_sample  ,							1  ,							self.dummy_sample).prev_sample






       def       lowerCAmelCase		(      self)						->  Tuple:




          '''simple docstring'''
          _UpperCAmelCase       :       str							   =self.full_loop()
          _UpperCAmelCase       :       Tuple							   =torch.sum(torch.abs(snake_case))
          _UpperCAmelCase       :       Optional[Any]							   =torch.mean(torch.abs(snake_case))

          assert abs(result_sum.item() - 1_98.13_18) < 1E-2
          assert abs(result_mean.item() - 0.25_80) < 1E-3






       def       lowerCAmelCase		(      self)						->  Union[str, Any]:




          '''simple docstring'''
          _UpperCAmelCase       :       Any							   =self.full_loop(prediction_type='v_prediction')
          _UpperCAmelCase       :       Optional[int]							   =torch.sum(torch.abs(snake_case))
          _UpperCAmelCase       :       Union[str, Any]							   =torch.mean(torch.abs(snake_case))

          assert abs(result_sum.item() - 67.39_86) < 1E-2
          assert abs(result_mean.item() - 0.08_78) < 1E-3






       def       lowerCAmelCase		(      self)						->  Union[str, Any]:




          '''simple docstring'''
          # We specify different beta, so that the first alpha is 0.99
          _UpperCAmelCase       :       Optional[Any]							   =self.full_loop(set_alpha_to_one=snake_case  ,							beta_start=0.01)
          _UpperCAmelCase       :       str							   =torch.sum(torch.abs(snake_case))
          _UpperCAmelCase       :       int							   =torch.mean(torch.abs(snake_case))

          assert abs(result_sum.item() - 2_30.03_99) < 1E-2
          assert abs(result_mean.item() - 0.29_95) < 1E-3






       def       lowerCAmelCase		(      self)						->  List[str]:




          '''simple docstring'''
          # We specify different beta, so that the first alpha is 0.99
          _UpperCAmelCase       :       Union[str, Any]							   =self.full_loop(set_alpha_to_one=snake_case  ,							beta_start=0.01)
          _UpperCAmelCase       :       Optional[Any]							   =torch.sum(torch.abs(snake_case))
          _UpperCAmelCase       :       List[Any]							   =torch.mean(torch.abs(snake_case))

          assert abs(result_sum.item() - 1_86.94_82) < 1E-2
          assert abs(result_mean.item() - 0.24_34) < 1E-3





'''simple docstring'''






import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import is_speech_available, is_vision_available
from transformers.testing_utils import require_torch


if is_vision_available():
 from transformers import TvltImageProcessor

if is_speech_available():
 from transformers import TvltFeatureExtractor

from transformers import TvltProcessor






@require_torch
class  __magic_name__							(		unittest.TestCase    ):






       def       lowerCAmelCase		(      self)						->  str:




          '''simple docstring'''
          _UpperCAmelCase       :       Dict							   ='ZinengTang/tvlt-base'
          _UpperCAmelCase       :       Dict							   =tempfile.mkdtemp()






       def       lowerCAmelCase		(      self  ,							**snake_case)						->  Union[str, Any]:




          '''simple docstring'''
          return TvltImageProcessor.from_pretrained(self.checkpoint  ,							**snake_case)






       def       lowerCAmelCase		(      self  ,							**snake_case)						->  Dict:




          '''simple docstring'''
          return TvltFeatureExtractor.from_pretrained(self.checkpoint  ,							**snake_case)






       def       lowerCAmelCase		(      self)						->  Any:




          '''simple docstring'''
          shutil.rmtree(self.tmpdirname)






       def       lowerCAmelCase		(      self)						->  Optional[int]:




          '''simple docstring'''
          _UpperCAmelCase       :       Any							   =self.get_image_processor()
          _UpperCAmelCase       :       Optional[Any]							   =self.get_feature_extractor()

          _UpperCAmelCase       :       str							   =TvltProcessor(image_processor=snake_case  ,							feature_extractor=snake_case)
          processor.save_pretrained(self.tmpdirname)
          _UpperCAmelCase       :       str							   =TvltProcessor.from_pretrained(self.tmpdirname)

          self.assertIsInstance(processor.feature_extractor  ,							snake_case)
          self.assertIsInstance(processor.image_processor  ,							snake_case)






       def       lowerCAmelCase		(      self)						->  Union[str, Any]:




          '''simple docstring'''
          _UpperCAmelCase       :       str							   =self.get_image_processor()
          _UpperCAmelCase       :       List[Any]							   =self.get_feature_extractor()

          _UpperCAmelCase       :       str							   =TvltProcessor(image_processor=snake_case  ,							feature_extractor=snake_case)

          _UpperCAmelCase       :       Optional[int]							   =np.ones([1_2_0_0_0])

          _UpperCAmelCase       :       str							   =feature_extractor(snake_case  ,							return_tensors='np')
          _UpperCAmelCase       :       Dict							   =processor(audio=snake_case  ,							return_tensors='np')

          for key in audio_dict.keys():
             self.assertAlmostEqual(audio_dict[key].sum()  ,							input_processor[key].sum()  ,							delta=1E-2)






       def       lowerCAmelCase		(      self)						->  int:




          '''simple docstring'''
          _UpperCAmelCase       :       Dict							   =self.get_image_processor()
          _UpperCAmelCase       :       int							   =self.get_feature_extractor()

          _UpperCAmelCase       :       int							   =TvltProcessor(image_processor=snake_case  ,							feature_extractor=snake_case)

          _UpperCAmelCase       :       Union[str, Any]							   =np.ones([3, 2_2_4, 2_2_4])

          _UpperCAmelCase       :       Optional[Any]							   =image_processor(snake_case  ,							return_tensors='np')
          _UpperCAmelCase       :       List[Any]							   =processor(images=snake_case  ,							return_tensors='np')

          for key in image_dict.keys():
             self.assertAlmostEqual(image_dict[key].sum()  ,							input_processor[key].sum()  ,							delta=1E-2)






       def       lowerCAmelCase		(      self)						->  List[str]:




          '''simple docstring'''
          _UpperCAmelCase       :       List[str]							   =self.get_image_processor()
          _UpperCAmelCase       :       Dict							   =self.get_feature_extractor()

          _UpperCAmelCase       :       Optional[int]							   =TvltProcessor(image_processor=snake_case  ,							feature_extractor=snake_case)

          _UpperCAmelCase       :       Optional[int]							   =np.ones([1_2_0_0_0])
          _UpperCAmelCase       :       str							   =np.ones([3, 2_2_4, 2_2_4])

          _UpperCAmelCase       :       Optional[int]							   =processor(audio=snake_case  ,							images=snake_case)

          self.assertListEqual(list(inputs.keys())  ,							['audio_values', 'audio_mask', 'pixel_values', 'pixel_mask'])

          # test if it raises when no input is passed
          with pytest.raises(snake_case):
             processor()






       def       lowerCAmelCase		(      self)						->  Optional[int]:




          '''simple docstring'''
          _UpperCAmelCase       :       Optional[int]							   =self.get_image_processor()
          _UpperCAmelCase       :       Tuple							   =self.get_feature_extractor()

          _UpperCAmelCase       :       Dict							   =TvltProcessor(image_processor=snake_case  ,							feature_extractor=snake_case)

          self.assertListEqual(
              processor.model_input_names  ,							image_processor.model_input_names + feature_extractor.model_input_names  ,							msg='`processor` and `image_processor`+`feature_extractor` model input names do not match'  ,							)

