
'''simple docstring'''

import unittest

from transformers import (
    MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING,
    TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING,
    PreTrainedTokenizer,
    is_vision_available,
)
from transformers.pipelines import ImageClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_torch_or_tf,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
   from PIL import Image
else:







   class 	__lowerCAmelCase    :



        @staticmethod
        def     snake_case_ (*lowerCAmelCase__						,	**lowerCAmelCase__	):
             pass







@is_pipeline_test
@require_torch_or_tf
@require_vision
class 	__lowerCAmelCase    (  unittest.TestCase				):
     snake_case   :    Any							      =      MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
     snake_case   :    Tuple							      =      TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING



     def     snake_case_ (self						,	lowerCAmelCase__						,	lowerCAmelCase__						,	lowerCAmelCase__	):
          _UpperCAmelCase					:  int         =    ImageClassificationPipeline(model=lowerCAmelCase__						,	image_processor=lowerCAmelCase__						,	top_k=2	)
          _UpperCAmelCase					:  Any         =    [
              Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""	),
              """http://images.cocodataset.org/val2017/000000039769.jpg""",
          ]
          return image_classifier, examples



     def     snake_case_ (self						,	lowerCAmelCase__						,	lowerCAmelCase__	):
          _UpperCAmelCase					:  str         =    image_classifier("""./tests/fixtures/tests_samples/COCO/000000039769.png"""	)

          self.assertEqual(
              lowerCAmelCase__						,	[
                  {"""score""": ANY(lowerCAmelCase__	), """label""": ANY(lowerCAmelCase__	)},
                  {"""score""": ANY(lowerCAmelCase__	), """label""": ANY(lowerCAmelCase__	)},
              ]						,	)

          import datasets

          _UpperCAmelCase					:  Union[str, Any]         =    datasets.load_dataset("""hf-internal-testing/fixtures_image_utils"""						,	"""image"""						,	split="""test"""	)

          # Accepts URL + PIL.Image + lists
          _UpperCAmelCase					:  Tuple         =    image_classifier(
              [
                  Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""	),
                  """http://images.cocodataset.org/val2017/000000039769.jpg""",
                  # RGBA
                  dataset[0]["""file"""],
                  # LA
                  dataset[1]["""file"""],
                  # L
                  dataset[2]["""file"""],
              ]	)
          self.assertEqual(
              lowerCAmelCase__						,	[
                  [
                      {"""score""": ANY(lowerCAmelCase__	), """label""": ANY(lowerCAmelCase__	)},
                      {"""score""": ANY(lowerCAmelCase__	), """label""": ANY(lowerCAmelCase__	)},
                  ],
                  [
                      {"""score""": ANY(lowerCAmelCase__	), """label""": ANY(lowerCAmelCase__	)},
                      {"""score""": ANY(lowerCAmelCase__	), """label""": ANY(lowerCAmelCase__	)},
                  ],
                  [
                      {"""score""": ANY(lowerCAmelCase__	), """label""": ANY(lowerCAmelCase__	)},
                      {"""score""": ANY(lowerCAmelCase__	), """label""": ANY(lowerCAmelCase__	)},
                  ],
                  [
                      {"""score""": ANY(lowerCAmelCase__	), """label""": ANY(lowerCAmelCase__	)},
                      {"""score""": ANY(lowerCAmelCase__	), """label""": ANY(lowerCAmelCase__	)},
                  ],
                  [
                      {"""score""": ANY(lowerCAmelCase__	), """label""": ANY(lowerCAmelCase__	)},
                      {"""score""": ANY(lowerCAmelCase__	), """label""": ANY(lowerCAmelCase__	)},
                  ],
              ]						,	)



     @require_torch
     def     snake_case_ (self	):
          _UpperCAmelCase					:  Any         =    """hf-internal-testing/tiny-random-vit"""
          _UpperCAmelCase					:  str         =    pipeline("""image-classification"""						,	model=lowerCAmelCase__	)

          _UpperCAmelCase					:  Optional[int]         =    image_classifier("""http://images.cocodataset.org/val2017/000000039769.jpg"""	)
          self.assertEqual(
              nested_simplify(lowerCAmelCase__						,	decimals=4	)						,	[{"""label""": """LABEL_1""", """score""": 0.5_7_4}, {"""label""": """LABEL_0""", """score""": 0.4_2_6}]						,	)

          _UpperCAmelCase					:  List[str]         =    image_classifier(
              [
                  """http://images.cocodataset.org/val2017/000000039769.jpg""",
                  """http://images.cocodataset.org/val2017/000000039769.jpg""",
              ]						,	top_k=2						,	)
          self.assertEqual(
              nested_simplify(lowerCAmelCase__						,	decimals=4	)						,	[
                  [{"""label""": """LABEL_1""", """score""": 0.5_7_4}, {"""label""": """LABEL_0""", """score""": 0.4_2_6}],
                  [{"""label""": """LABEL_1""", """score""": 0.5_7_4}, {"""label""": """LABEL_0""", """score""": 0.4_2_6}],
              ]						,	)



     @require_tf
     def     snake_case_ (self	):
          _UpperCAmelCase					:  Union[str, Any]         =    """hf-internal-testing/tiny-random-vit"""
          _UpperCAmelCase					:  List[Any]         =    pipeline("""image-classification"""						,	model=lowerCAmelCase__						,	framework="""tf"""	)

          _UpperCAmelCase					:  Any         =    image_classifier("""http://images.cocodataset.org/val2017/000000039769.jpg"""	)
          self.assertEqual(
              nested_simplify(lowerCAmelCase__						,	decimals=4	)						,	[{"""label""": """LABEL_1""", """score""": 0.5_7_4}, {"""label""": """LABEL_0""", """score""": 0.4_2_6}]						,	)

          _UpperCAmelCase					:  Dict         =    image_classifier(
              [
                  """http://images.cocodataset.org/val2017/000000039769.jpg""",
                  """http://images.cocodataset.org/val2017/000000039769.jpg""",
              ]						,	top_k=2						,	)
          self.assertEqual(
              nested_simplify(lowerCAmelCase__						,	decimals=4	)						,	[
                  [{"""label""": """LABEL_1""", """score""": 0.5_7_4}, {"""label""": """LABEL_0""", """score""": 0.4_2_6}],
                  [{"""label""": """LABEL_1""", """score""": 0.5_7_4}, {"""label""": """LABEL_0""", """score""": 0.4_2_6}],
              ]						,	)



     def     snake_case_ (self	):
          _UpperCAmelCase					:  List[Any]         =    PreTrainedTokenizer()

          # Assert that the pipeline can be initialized with a feature extractor that is not in any mapping
          _UpperCAmelCase					:  Union[str, Any]         =    pipeline(
              """image-classification"""						,	model="""hf-internal-testing/tiny-random-vit"""						,	tokenizer=lowerCAmelCase__	)

          self.assertIs(image_classifier.tokenizer						,	lowerCAmelCase__	)



     @slow
     @require_torch
     def     snake_case_ (self	):
          # Perceiver is not tested by `run_pipeline_test` properly.
          # That is because the type of feature_extractor and model preprocessor need to be kept
          # in sync, which is not the case in the current design
          _UpperCAmelCase					:  int         =    pipeline("""image-classification"""						,	model="""deepmind/vision-perceiver-conv"""	)
          _UpperCAmelCase					:  int         =    image_classifier("""http://images.cocodataset.org/val2017/000000039769.jpg"""	)
          self.assertEqual(
              nested_simplify(lowerCAmelCase__						,	decimals=4	)						,	[
                  {"""score""": 0.4_3_8_5, """label""": """tabby, tabby cat"""},
                  {"""score""": 0.3_2_1, """label""": """tiger cat"""},
                  {"""score""": 0.0_5_0_2, """label""": """Egyptian cat"""},
                  {"""score""": 0.0_1_3_7, """label""": """crib, cot"""},
                  {"""score""": 0.0_0_7, """label""": """radiator"""},
              ]						,	)

          _UpperCAmelCase					:  List[str]         =    pipeline("""image-classification"""						,	model="""deepmind/vision-perceiver-fourier"""	)
          _UpperCAmelCase					:  Union[str, Any]         =    image_classifier("""http://images.cocodataset.org/val2017/000000039769.jpg"""	)
          self.assertEqual(
              nested_simplify(lowerCAmelCase__						,	decimals=4	)						,	[
                  {"""score""": 0.5_6_5_8, """label""": """tabby, tabby cat"""},
                  {"""score""": 0.1_3_0_9, """label""": """tiger cat"""},
                  {"""score""": 0.0_7_2_2, """label""": """Egyptian cat"""},
                  {"""score""": 0.0_7_0_7, """label""": """remote control, remote"""},
                  {"""score""": 0.0_0_8_2, """label""": """computer keyboard, keypad"""},
              ]						,	)

          _UpperCAmelCase					:  Optional[Any]         =    pipeline("""image-classification"""						,	model="""deepmind/vision-perceiver-learned"""	)
          _UpperCAmelCase					:  Any         =    image_classifier("""http://images.cocodataset.org/val2017/000000039769.jpg"""	)
          self.assertEqual(
              nested_simplify(lowerCAmelCase__						,	decimals=4	)						,	[
                  {"""score""": 0.3_0_2_2, """label""": """tabby, tabby cat"""},
                  {"""score""": 0.2_3_6_2, """label""": """Egyptian cat"""},
                  {"""score""": 0.1_8_5_6, """label""": """tiger cat"""},
                  {"""score""": 0.0_3_2_4, """label""": """remote control, remote"""},
                  {"""score""": 0.0_0_9_6, """label""": """quilt, comforter, comfort, puff"""},
              ]						,	)




'''simple docstring'''

import torch

from diffusers import HeunDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest







class 	__lowerCAmelCase    (  __a				):
     snake_case   :    List[str]							      =      (HeunDiscreteScheduler,)
     snake_case   :    Optional[Any]							      =      1_0



     def     snake_case_ (self						,	**lowerCAmelCase__	):
          _UpperCAmelCase					:  Union[str, Any]         =    {
              """num_train_timesteps""": 1_1_0_0,
              """beta_start""": 0.0_0_0_1,
              """beta_end""": 0.0_2,
              """beta_schedule""": """linear""",
          }

          config.update(**lowerCAmelCase__	)
          return config



     def     snake_case_ (self	):
          for timesteps in [1_0, 5_0, 1_0_0, 1_0_0_0]:
               self.check_over_configs(num_train_timesteps=lowerCAmelCase__	)



     def     snake_case_ (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     snake_case_ (self	):
          for schedule in ["linear", "scaled_linear", "exp"]:
               self.check_over_configs(beta_schedule=lowerCAmelCase__	)



     def     snake_case_ (self	):
          for clip_sample_range in [1.0, 2.0, 3.0]:
               self.check_over_configs(clip_sample_range=lowerCAmelCase__						,	clip_sample=lowerCAmelCase__	)



     def     snake_case_ (self	):
          for prediction_type in ["epsilon", "v_prediction", "sample"]:
               self.check_over_configs(prediction_type=lowerCAmelCase__	)



     def     snake_case_ (self	):
          _UpperCAmelCase					:  Tuple         =    self.scheduler_classes[0]
          _UpperCAmelCase					:  Optional[int]         =    self.get_scheduler_config()
          _UpperCAmelCase					:  List[str]         =    scheduler_class(**lowerCAmelCase__	)

          scheduler.set_timesteps(self.num_inference_steps	)

          _UpperCAmelCase					:  str         =    self.dummy_model()
          _UpperCAmelCase					:  List[Any]         =    self.dummy_sample_deter * scheduler.init_noise_sigma
          _UpperCAmelCase					:  Tuple         =    sample.to(lowerCAmelCase__	)

          for i, t in enumerate(scheduler.timesteps	):
               _UpperCAmelCase					:  Union[str, Any]         =    scheduler.scale_model_input(lowerCAmelCase__						,	lowerCAmelCase__	)

               _UpperCAmelCase					:  Any         =    model(lowerCAmelCase__						,	lowerCAmelCase__	)

               _UpperCAmelCase					:  Tuple         =    scheduler.step(lowerCAmelCase__						,	lowerCAmelCase__						,	lowerCAmelCase__	)
               _UpperCAmelCase					:  Optional[Any]         =    output.prev_sample

          _UpperCAmelCase					:  Optional[Any]         =    torch.sum(torch.abs(lowerCAmelCase__	)	)
          _UpperCAmelCase					:  int         =    torch.mean(torch.abs(lowerCAmelCase__	)	)

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



     def     snake_case_ (self	):
          _UpperCAmelCase					:  Optional[int]         =    self.scheduler_classes[0]
          _UpperCAmelCase					:  Any         =    self.get_scheduler_config(prediction_type="""v_prediction"""	)
          _UpperCAmelCase					:  Any         =    scheduler_class(**lowerCAmelCase__	)

          scheduler.set_timesteps(self.num_inference_steps	)

          _UpperCAmelCase					:  Optional[Any]         =    self.dummy_model()
          _UpperCAmelCase					:  Any         =    self.dummy_sample_deter * scheduler.init_noise_sigma
          _UpperCAmelCase					:  Any         =    sample.to(lowerCAmelCase__	)

          for i, t in enumerate(scheduler.timesteps	):
               _UpperCAmelCase					:  List[str]         =    scheduler.scale_model_input(lowerCAmelCase__						,	lowerCAmelCase__	)

               _UpperCAmelCase					:  List[Any]         =    model(lowerCAmelCase__						,	lowerCAmelCase__	)

               _UpperCAmelCase					:  Optional[Any]         =    scheduler.step(lowerCAmelCase__						,	lowerCAmelCase__						,	lowerCAmelCase__	)
               _UpperCAmelCase					:  Union[str, Any]         =    output.prev_sample

          _UpperCAmelCase					:  Union[str, Any]         =    torch.sum(torch.abs(lowerCAmelCase__	)	)
          _UpperCAmelCase					:  str         =    torch.mean(torch.abs(lowerCAmelCase__	)	)

          if torch_device in ["cpu", "mps"]:
               assert abs(result_sum.item() - 4.6_934e-07	) < 1e-2
               assert abs(result_mean.item() - 6.1_112e-10	) < 1e-3
          else:
               # CUDA
               assert abs(result_sum.item() - 4.693_428_650_170_972e-07	) < 1e-2
               assert abs(result_mean.item() - 0.0_0_0_2	) < 1e-3



     def     snake_case_ (self	):
          _UpperCAmelCase					:  Tuple         =    self.scheduler_classes[0]
          _UpperCAmelCase					:  Any         =    self.get_scheduler_config()
          _UpperCAmelCase					:  Optional[int]         =    scheduler_class(**lowerCAmelCase__	)

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

          _UpperCAmelCase					:  List[Any]         =    self.dummy_model()
          _UpperCAmelCase					:  Dict         =    self.dummy_sample_deter.to(lowerCAmelCase__	) * scheduler.init_noise_sigma

          for t in scheduler.timesteps:
               _UpperCAmelCase					:  Union[str, Any]         =    scheduler.scale_model_input(lowerCAmelCase__						,	lowerCAmelCase__	)

               _UpperCAmelCase					:  Optional[Any]         =    model(lowerCAmelCase__						,	lowerCAmelCase__	)

               _UpperCAmelCase					:  Optional[int]         =    scheduler.step(lowerCAmelCase__						,	lowerCAmelCase__						,	lowerCAmelCase__	)
               _UpperCAmelCase					:  int         =    output.prev_sample

          _UpperCAmelCase					:  Any         =    torch.sum(torch.abs(lowerCAmelCase__	)	)
          _UpperCAmelCase					:  Optional[Any]         =    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() - 0.1_2_3_3	) < 1e-2
               assert abs(result_mean.item() - 0.0_0_0_2	) < 1e-3
          elif str(lowerCAmelCase__	).startswith("""mps"""	):
               # Larger tolerance on mps
               assert abs(result_mean.item() - 0.0_0_0_2	) < 1e-2
          else:
               # CUDA
               assert abs(result_sum.item() - 0.1_2_3_3	) < 1e-2
               assert abs(result_mean.item() - 0.0_0_0_2	) < 1e-3



     def     snake_case_ (self	):
          _UpperCAmelCase					:  int         =    self.scheduler_classes[0]
          _UpperCAmelCase					:  str         =    self.get_scheduler_config()
          _UpperCAmelCase					:  Union[str, Any]         =    scheduler_class(**lowerCAmelCase__						,	use_karras_sigmas=lowerCAmelCase__	)

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

          _UpperCAmelCase					:  Optional[int]         =    self.dummy_model()
          _UpperCAmelCase					:  Optional[Any]         =    self.dummy_sample_deter.to(lowerCAmelCase__	) * scheduler.init_noise_sigma
          _UpperCAmelCase					:  Union[str, Any]         =    sample.to(lowerCAmelCase__	)

          for t in scheduler.timesteps:
               _UpperCAmelCase					:  Dict         =    scheduler.scale_model_input(lowerCAmelCase__						,	lowerCAmelCase__	)

               _UpperCAmelCase					:  Optional[Any]         =    model(lowerCAmelCase__						,	lowerCAmelCase__	)

               _UpperCAmelCase					:  Optional[int]         =    scheduler.step(lowerCAmelCase__						,	lowerCAmelCase__						,	lowerCAmelCase__	)
               _UpperCAmelCase					:  Optional[Any]         =    output.prev_sample

          _UpperCAmelCase					:  Union[str, Any]         =    torch.sum(torch.abs(lowerCAmelCase__	)	)
          _UpperCAmelCase					:  List[Any]         =    torch.mean(torch.abs(lowerCAmelCase__	)	)

          assert abs(result_sum.item() - 0.0_0_0_1_5	) < 1e-2
          assert abs(result_mean.item() - 1.9_869_554_535_034_695e-07	) < 1e-2


