
import unittest

from transformers import is_vision_available
from transformers.pipelines import pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
     from PIL import Image
else:




     class 					_SCREAMING_SNAKE_CASE  :
           @staticmethod
           def A_			(			*lowerCamelCase		,	**lowerCamelCase ):
               pass




@is_pipeline_test
@require_vision
class 					_SCREAMING_SNAKE_CASE  (      unittest.TestCase ):
      @require_torch
      def A_			(			self ):
          snake_case__						=				pipeline(
              model="hf-internal-testing/tiny-random-clip-zero-shot-image-classification"		,	)
          snake_case__						=				Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png" )
          snake_case__						=				image_classifier(lowerCamelCase		,	candidate_labels=["a", "b", "c"] )

          # The floating scores are so close, we enter floating error approximation and the order is not guaranteed across
          # python and torch versions.
          self.assertIn(
              nested_simplify(lowerCamelCase )		,	[
                  [{"score": 0.3_3_3, "label": "a"}, {"score": 0.3_3_3, "label": "b"}, {"score": 0.3_3_3, "label": "c"}],
                  [{"score": 0.3_3_3, "label": "a"}, {"score": 0.3_3_3, "label": "c"}, {"score": 0.3_3_3, "label": "b"}],
              ]		,	)

          snake_case__						=				image_classifier([image] * 5		,	candidate_labels=["A", "B", "C"]		,	batch_size=2 )
          self.assertEqual(
              nested_simplify(lowerCamelCase )		,	[
                  [
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                  ],
                  [
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                  ],
                  [
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                  ],
                  [
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                  ],
                  [
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                  ],
              ]		,	)
      @require_tf
      def A_			(			self ):
          snake_case__						=				pipeline(
              model="hf-internal-testing/tiny-random-clip-zero-shot-image-classification"		,	framework="tf" )
          snake_case__						=				Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png" )
          snake_case__						=				image_classifier(lowerCamelCase		,	candidate_labels=["a", "b", "c"] )

          self.assertEqual(
              nested_simplify(lowerCamelCase )		,	[{"score": 0.3_3_3, "label": "a"}, {"score": 0.3_3_3, "label": "b"}, {"score": 0.3_3_3, "label": "c"}]		,	)

          snake_case__						=				image_classifier([image] * 5		,	candidate_labels=["A", "B", "C"]		,	batch_size=2 )
          self.assertEqual(
              nested_simplify(lowerCamelCase )		,	[
                  [
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                  ],
                  [
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                  ],
                  [
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                  ],
                  [
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                  ],
                  [
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                      {"score": 0.3_3_3, "label": ANY(lowerCamelCase )},
                  ],
              ]		,	)
      @slow
      @require_torch
      def A_			(			self ):
          snake_case__						=				pipeline(
              task="zero-shot-image-classification"		,	model="openai/clip-vit-base-patch32"		,	)
          # This is an image of 2 cats with remotes and no planes
          snake_case__						=				Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png" )
          snake_case__						=				image_classifier(lowerCamelCase		,	candidate_labels=["cat", "plane", "remote"] )

          self.assertEqual(
              nested_simplify(lowerCamelCase )		,	[
                  {"score": 0.5_1_1, "label": "remote"},
                  {"score": 0.4_8_5, "label": "cat"},
                  {"score": 0.0_0_4, "label": "plane"},
              ]		,	)

          snake_case__						=				image_classifier([image] * 5		,	candidate_labels=["cat", "plane", "remote"]		,	batch_size=2 )
          self.assertEqual(
              nested_simplify(lowerCamelCase )		,	[
                  [
                      {"score": 0.5_1_1, "label": "remote"},
                      {"score": 0.4_8_5, "label": "cat"},
                      {"score": 0.0_0_4, "label": "plane"},
                  ],
              ]
              * 5		,	)



      @slow
      @require_tf
      def A_			(			self ):
          snake_case__						=				pipeline(
              task="zero-shot-image-classification"		,	model="openai/clip-vit-base-patch32"		,	framework="tf" )
          # This is an image of 2 cats with remotes and no planes
          snake_case__						=				Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png" )
          snake_case__						=				image_classifier(lowerCamelCase		,	candidate_labels=["cat", "plane", "remote"] )
          self.assertEqual(
              nested_simplify(lowerCamelCase )		,	[
                  {"score": 0.5_1_1, "label": "remote"},
                  {"score": 0.4_8_5, "label": "cat"},
                  {"score": 0.0_0_4, "label": "plane"},
              ]		,	)

          snake_case__						=				image_classifier([image] * 5		,	candidate_labels=["cat", "plane", "remote"]		,	batch_size=2 )
          self.assertEqual(
              nested_simplify(lowerCamelCase )		,	[
                  [
                      {"score": 0.5_1_1, "label": "remote"},
                      {"score": 0.4_8_5, "label": "cat"},
                      {"score": 0.0_0_4, "label": "plane"},
                  ],
              ]
              * 5		,	)





import torch

from diffusers import KDPMaDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest




class 					_SCREAMING_SNAKE_CASE  (      __UpperCamelCase ):
      _A						:		Any    	=     (KDPMaDiscreteScheduler,)
      _A						:		Dict    	=     10
      def A_			(			self		,	**lowerCamelCase ):
          snake_case__						=				{
              "num_train_timesteps": 11_00,
              "beta_start": 0.0_0_0_1,
              "beta_end": 0.0_2,
              "beta_schedule": "linear",
          }

          config.update(**lowerCamelCase )
          return config
      def A_			(			self ):
          for timesteps in [10, 50, 1_00, 10_00]:
              self.check_over_configs(num_train_timesteps=lowerCamelCase )
      def A_			(			self ):
          for beta_start, beta_end in zip([0.0_0_0_0_1, 0.0_0_0_1, 0.0_0_1]		,	[0.0_0_0_2, 0.0_0_2, 0.0_2] ):
              self.check_over_configs(beta_start=lowerCamelCase		,	beta_end=lowerCamelCase )
      def A_			(			self ):
          for schedule in ["linear", "scaled_linear"]:
              self.check_over_configs(beta_schedule=lowerCamelCase )
      def A_			(			self ):
          for prediction_type in ["epsilon", "v_prediction"]:
              self.check_over_configs(prediction_type=lowerCamelCase )
      def A_			(			self ):
          snake_case__						=				self.scheduler_classes[0]
          snake_case__						=				self.get_scheduler_config(prediction_type="v_prediction" )
          snake_case__						=				scheduler_class(**lowerCamelCase )

          scheduler.set_timesteps(self.num_inference_steps )

          snake_case__						=				self.dummy_model()
          snake_case__						=				self.dummy_sample_deter * scheduler.init_noise_sigma
          snake_case__						=				sample.to(lowerCamelCase )

          for i, t in enumerate(scheduler.timesteps ):
              snake_case__						=				scheduler.scale_model_input(lowerCamelCase		,	lowerCamelCase )

              snake_case__						=				model(lowerCamelCase		,	lowerCamelCase )

              snake_case__						=				scheduler.step(lowerCamelCase		,	lowerCamelCase		,	lowerCamelCase )
              snake_case__						=				output.prev_sample

          snake_case__						=				torch.sum(torch.abs(lowerCamelCase ) )
          snake_case__						=				torch.mean(torch.abs(lowerCamelCase ) )

          if torch_device in ["cpu", "mps"]:
              assert abs(result_sum.item() - 4.69_34e-07 ) < 1e-2
              assert abs(result_mean.item() - 6.11_12e-10 ) < 1e-3
          else:
              # CUDA
              assert abs(result_sum.item() - 4.6_93_42_86_50_17_09_72e-07 ) < 1e-2
              assert abs(result_mean.item() - 0.0_0_0_2 ) < 1e-3
      def A_			(			self ):
          if torch_device == "mps":
              return
          snake_case__						=				self.scheduler_classes[0]
          snake_case__						=				self.get_scheduler_config()
          snake_case__						=				scheduler_class(**lowerCamelCase )

          scheduler.set_timesteps(self.num_inference_steps )

          snake_case__						=				self.dummy_model()
          snake_case__						=				self.dummy_sample_deter * scheduler.init_noise_sigma
          snake_case__						=				sample.to(lowerCamelCase )

          for i, t in enumerate(scheduler.timesteps ):
              snake_case__						=				scheduler.scale_model_input(lowerCamelCase		,	lowerCamelCase )

              snake_case__						=				model(lowerCamelCase		,	lowerCamelCase )

              snake_case__						=				scheduler.step(lowerCamelCase		,	lowerCamelCase		,	lowerCamelCase )
              snake_case__						=				output.prev_sample

          snake_case__						=				torch.sum(torch.abs(lowerCamelCase ) )
          snake_case__						=				torch.mean(torch.abs(lowerCamelCase ) )

          if torch_device in ["cpu", "mps"]:
              assert abs(result_sum.item() - 2_0.4_1_2_5 ) < 1e-2
              assert abs(result_mean.item() - 0.0_2_6_6 ) < 1e-3
          else:
              # CUDA
              assert abs(result_sum.item() - 2_0.4_1_2_5 ) < 1e-2
              assert abs(result_mean.item() - 0.0_2_6_6 ) < 1e-3



      def A_			(			self ):
          if torch_device == "mps":
              return
          snake_case__						=				self.scheduler_classes[0]
          snake_case__						=				self.get_scheduler_config()
          snake_case__						=				scheduler_class(**lowerCamelCase )

          scheduler.set_timesteps(self.num_inference_steps		,	device=lowerCamelCase )

          snake_case__						=				self.dummy_model()
          snake_case__						=				self.dummy_sample_deter.to(lowerCamelCase ) * scheduler.init_noise_sigma

          for t in scheduler.timesteps:
              snake_case__						=				scheduler.scale_model_input(lowerCamelCase		,	lowerCamelCase )

              snake_case__						=				model(lowerCamelCase		,	lowerCamelCase )

              snake_case__						=				scheduler.step(lowerCamelCase		,	lowerCamelCase		,	lowerCamelCase )
              snake_case__						=				output.prev_sample

          snake_case__						=				torch.sum(torch.abs(lowerCamelCase ) )
          snake_case__						=				torch.mean(torch.abs(lowerCamelCase ) )

          if str(lowerCamelCase ).startswith("cpu" ):
              # The following sum varies between 148 and 156 on mps. Why?
              assert abs(result_sum.item() - 2_0.4_1_2_5 ) < 1e-2
              assert abs(result_mean.item() - 0.0_2_6_6 ) < 1e-3
          else:
              # CUDA
              assert abs(result_sum.item() - 2_0.4_1_2_5 ) < 1e-2
              assert abs(result_mean.item() - 0.0_2_6_6 ) < 1e-3



