






"""simple docstring"""


from io import BytesIO
from typing import List, Union

import requests

from ..utils import add_end_docstrings, is_decord_available, is_torch_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_decord_available():
   import numpy as np
   from decord import VideoReader


if is_torch_available():
   from ..models.auto.modeling_auto import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING

_UpperCamelCase    :   str   			=						logging.get_logger(__name__)






@add_end_docstrings(UpperCAmelCase	)
class 						snake_case  (  UpperCAmelCase	):





    def __init__(							self     : List[str]    ,		*A     : List[Any]    ,		**A     : str      ):


      '''simple docstring'''




      super().__init__(*A    ,		**A      )
      requires_backends(self    ,		'decord'      )
      self.check_model_type(A      )





    def     lowerCamelCase__				(							self     : Tuple    ,		A     : int=None    ,		A     : str=None    ,		A     : List[str]=None      ):


      '''simple docstring'''




      a					:      int    =       {}
      if frame_sampling_rate is not None:
        a					:      Optional[Any]    =       frame_sampling_rate
      if num_frames is not None:
        a					:      int    =       num_frames

      a					:      Union[str, Any]    =       {}
      if top_k is not None:
        a					:      str    =       top_k
      return preprocess_params, {}, postprocess_params





    def __call__(							self     : Union[str, Any]    ,		A     : Union[str, List[str]]    ,		**A     : Union[str, Any]      ):


      '''simple docstring'''




      return super().__call__(A    ,		**A      )





    def     lowerCamelCase__				(							self     : Optional[Any]    ,		A     : Any    ,		A     : Any=None    ,		A     : int=1      ):


      '''simple docstring'''




      if num_frames is None:
        a					:      List[str]    =       self.model.config.num_frames

      if video.startswith('http://'      ) or video.startswith('https://'      ):
        a					:      Optional[int]    =       BytesIO(requests.get(A      ).content      )

      a					:      Dict    =       VideoReader(A      )
      videoreader.seek(0      )

      a					:      Dict    =       0
      a					:      int    =       num_frames * frame_sampling_rate - 1
      a					:      str    =       np.linspace(A    ,		A    ,		num=A    ,		dtype=np.intaa      )

      a					:      Optional[int]    =       videoreader.get_batch(A      ).asnumpy()
      a					:      List[Any]    =       list(A      )

      a					:      int    =       self.image_processor(A    ,		return_tensors=self.framework      )
      return model_inputs





    def     lowerCamelCase__				(							self     : Any    ,		A     : List[Any]      ):


      '''simple docstring'''




      a					:      Any    =       self.model(**A      )
      return model_outputs





    def     lowerCamelCase__				(							self     : Optional[int]    ,		A     : Any    ,		A     : str=5      ):


      '''simple docstring'''




      if top_k > self.model.config.num_labels:
        a					:      List[str]    =       self.model.config.num_labels

      if self.framework == "pt":
        a					:      Any    =       model_outputs.logits.softmax(-1      )[0]
        a,					a					:      Tuple    =       probs.topk(A      )
      else:
        raise ValueError(F'''Unsupported framework: {self.framework}'''      )

      a					:      int    =       scores.tolist()
      a					:      Tuple    =       ids.tolist()
      return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(A    ,		A      )]












"""simple docstring"""


import inspect
import unittest

from transformers import SegformerConfig, is_torch_available, is_vision_available
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
   import torch

   from transformers import (
       MODEL_MAPPING,
       SegformerForImageClassification,
       SegformerForSemanticSegmentation,
       SegformerModel,
   )
   from transformers.models.segformer.modeling_segformer import SEGFORMER_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
   from PIL import Image

   from transformers import SegformerImageProcessor






class 						snake_case  (  UpperCAmelCase	):





    def     lowerCamelCase__				(							self     : Dict      ):


      '''simple docstring'''




      a					:      Any    =       self.config_class(**self.inputs_dict      )
      self.parent.assertTrue(hasattr(A    ,		'hidden_sizes'      )      )
      self.parent.assertTrue(hasattr(A    ,		'num_attention_heads'      )      )
      self.parent.assertTrue(hasattr(A    ,		'num_encoder_blocks'      )      )






class 						snake_case  :





    def __init__(							self     : List[Any]    ,		A     : Dict    ,		A     : List[Any]=1_3    ,		A     : str=6_4    ,		A     : Union[str, Any]=3    ,		A     : Union[str, Any]=4    ,		A     : Union[str, Any]=[2, 2, 2, 2]    ,		A     : List[str]=[8, 4, 2, 1]    ,		A     : Optional[Any]=[1_6, 3_2, 6_4, 1_2_8]    ,		A     : Optional[Any]=[1, 4, 8, 1_6]    ,		A     : Tuple=[1, 2, 4, 8]    ,		A     : Optional[Any]=True    ,		A     : Any=True    ,		A     : Optional[Any]="gelu"    ,		A     : Optional[int]=0.1    ,		A     : List[Any]=0.1    ,		A     : List[str]=0.02    ,		A     : List[Any]=3    ,		A     : str=None    ,		):


      '''simple docstring'''




      a					:      Optional[Any]    =       parent
      a					:      Optional[Any]    =       batch_size
      a					:      Optional[Any]    =       image_size
      a					:      Optional[int]    =       num_channels
      a					:      List[str]    =       num_encoder_blocks
      a					:      Optional[Any]    =       sr_ratios
      a					:      Any    =       depths
      a					:      Any    =       hidden_sizes
      a					:      Union[str, Any]    =       downsampling_rates
      a					:      Any    =       num_attention_heads
      a					:      int    =       is_training
      a					:      Dict    =       use_labels
      a					:      str    =       hidden_act
      a					:      Optional[int]    =       hidden_dropout_prob
      a					:      Union[str, Any]    =       attention_probs_dropout_prob
      a					:      Optional[Any]    =       initializer_range
      a					:      Dict    =       num_labels
      a					:      Union[str, Any]    =       scope





    def     lowerCamelCase__				(							self     : Optional[int]      ):


      '''simple docstring'''




      a					:      Any    =       floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]      )

      a					:      int    =       None
      if self.use_labels:
        a					:      Dict    =       ids_tensor([self.batch_size, self.image_size, self.image_size]    ,		self.num_labels      )

      a					:      str    =       self.get_config()
      return config, pixel_values, labels





    def     lowerCamelCase__				(							self     : List[Any]      ):


      '''simple docstring'''




      return SegformerConfig(
          image_size=self.image_size    ,		num_channels=self.num_channels    ,		num_encoder_blocks=self.num_encoder_blocks    ,		depths=self.depths    ,		hidden_sizes=self.hidden_sizes    ,		num_attention_heads=self.num_attention_heads    ,		hidden_act=self.hidden_act    ,		hidden_dropout_prob=self.hidden_dropout_prob    ,		attention_probs_dropout_prob=self.attention_probs_dropout_prob    ,		initializer_range=self.initializer_range    ,		)





    def     lowerCamelCase__				(							self     : int    ,		A     : str    ,		A     : List[Any]    ,		A     : List[Any]      ):


      '''simple docstring'''




      a					:      Optional[Any]    =       SegformerModel(config=A      )
      model.to(A      )
      model.eval()
      a					:      Union[str, Any]    =       model(A      )
      a					:      Optional[int]    =       self.image_size // (self.downsampling_rates[-1] * 2)
      self.parent.assertEqual(
          result.last_hidden_state.shape    ,		(self.batch_size, self.hidden_sizes[-1], expected_height, expected_width)      )





    def     lowerCamelCase__				(							self     : Optional[int]    ,		A     : Union[str, Any]    ,		A     : str    ,		A     : Optional[Any]      ):


      '''simple docstring'''




      a					:      List[Any]    =       self.num_labels
      a					:      Optional[int]    =       SegformerForSemanticSegmentation(A      )
      model.to(A      )
      model.eval()
      a					:      str    =       model(A      )
      self.parent.assertEqual(
          result.logits.shape    ,		(self.batch_size, self.num_labels, self.image_size // 4, self.image_size // 4)      )
      a					:      int    =       model(A    ,		labels=A      )
      self.parent.assertEqual(
          result.logits.shape    ,		(self.batch_size, self.num_labels, self.image_size // 4, self.image_size // 4)      )
      self.parent.assertGreater(result.loss    ,		0.0      )





    def     lowerCamelCase__				(							self     : Dict    ,		A     : Dict    ,		A     : Any    ,		A     : Optional[Any]      ):


      '''simple docstring'''




      a					:      Optional[int]    =       1
      a					:      List[Any]    =       SegformerForSemanticSegmentation(config=A      )
      model.to(A      )
      model.eval()
      a					:      Any    =       torch.randint(0    ,		1    ,		(self.batch_size, self.image_size, self.image_size)      ).to(A      )
      a					:      Dict    =       model(A    ,		labels=A      )
      self.parent.assertGreater(result.loss    ,		0.0      )





    def     lowerCamelCase__				(							self     : Union[str, Any]      ):


      '''simple docstring'''




      a					:      str    =       self.prepare_config_and_inputs()
      a,					a,					a					:      str    =       config_and_inputs
      a					:      Dict    =       {'pixel_values': pixel_values}
      return config, inputs_dict






@require_torch
class 						snake_case  (  UpperCAmelCase       ,			UpperCAmelCase       ,			unittest.TestCase	):
    __magic_name__   					=					(
        (
            SegformerModel,
            SegformerForSemanticSegmentation,
            SegformerForImageClassification,
        )
        if is_torch_available()
        else ()
    )
    __magic_name__   					=					(
        {
            '''feature-extraction''': SegformerModel,
            '''image-classification''': SegformerForImageClassification,
            '''image-segmentation''': SegformerForSemanticSegmentation,
        }
        if is_torch_available()
        else {}
    )

    __magic_name__   					=					True
    __magic_name__   					=					False
    __magic_name__   					=					False
    __magic_name__   					=					False





    def     lowerCamelCase__				(							self     : Any      ):


      '''simple docstring'''




      a					:      Union[str, Any]    =       SegformerModelTester(self      )
      a					:      Tuple    =       SegformerConfigTester(self    ,		config_class=A      )





    def     lowerCamelCase__				(							self     : Tuple      ):


      '''simple docstring'''




      self.config_tester.run_common_tests()





    def     lowerCamelCase__				(							self     : Optional[int]      ):


      '''simple docstring'''




      a					:      Optional[int]    =       self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_model(*A      )





    def     lowerCamelCase__				(							self     : Union[str, Any]      ):


      '''simple docstring'''




      a					:      Optional[int]    =       self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_binary_image_segmentation(*A      )





    def     lowerCamelCase__				(							self     : int      ):


      '''simple docstring'''




      a					:      int    =       self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_image_segmentation(*A      )





    @unittest.skip('SegFormer does not use inputs_embeds'      )
    def     lowerCamelCase__				(							self     : List[str]      ):


      '''simple docstring'''




      pass





    @unittest.skip('SegFormer does not have get_input_embeddings method and get_output_embeddings methods'      )
    def     lowerCamelCase__				(							self     : List[str]      ):


      '''simple docstring'''




      pass





    def     lowerCamelCase__				(							self     : Dict      ):


      '''simple docstring'''




      a,					a					:      int    =       self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
        a					:      Dict    =       model_class(A      )
        a					:      Optional[int]    =       inspect.signature(model.forward      )
        # signature.parameters is an OrderedDict => so arg_names order is deterministic
        a					:      List[str]    =       [*signature.parameters.keys()]

        a					:      Optional[Any]    =       ['pixel_values']
        self.assertListEqual(arg_names[:1]    ,		A      )





    def     lowerCamelCase__				(							self     : Optional[int]      ):


      '''simple docstring'''




      a,					a					:      Any    =       self.model_tester.prepare_config_and_inputs_for_common()
      a					:      Any    =       True

      for model_class in self.all_model_classes:
        a					:      Optional[Any]    =       True
        a					:      Tuple    =       False
        a					:      int    =       True
        a					:      Any    =       model_class(A      )
        model.to(A      )
        model.eval()
        with torch.no_grad():
          a					:      Dict    =       model(**self._prepare_for_class(A    ,		A      )      )
        a					:      Union[str, Any]    =       outputs.attentions

        a					:      Tuple    =       sum(self.model_tester.depths      )
        self.assertEqual(len(A      )    ,		A      )

        # check that output_attentions also work using config
        del inputs_dict["output_attentions"]
        a					:      Tuple    =       True
        a					:      Optional[Any]    =       model_class(A      )
        model.to(A      )
        model.eval()
        with torch.no_grad():
          a					:      str    =       model(**self._prepare_for_class(A    ,		A      )      )
        a					:      Optional[int]    =       outputs.attentions

        self.assertEqual(len(A      )    ,		A      )

        # verify the first attentions (first block, first layer)
        a					:      Union[str, Any]    =       (self.model_tester.image_size // 4) ** 2
        a					:      List[str]    =       (self.model_tester.image_size // (4 * self.model_tester.sr_ratios[0])) ** 2
        self.assertListEqual(
            list(attentions[0].shape[-3:]      )    ,		[self.model_tester.num_attention_heads[0], expected_seq_len, expected_reduced_seq_len]    ,		)

        # verify the last attentions (last block, last layer)
        a					:      Tuple    =       (self.model_tester.image_size // 3_2) ** 2
        a					:      Tuple    =       (self.model_tester.image_size // (3_2 * self.model_tester.sr_ratios[-1])) ** 2
        self.assertListEqual(
            list(attentions[-1].shape[-3:]      )    ,		[self.model_tester.num_attention_heads[-1], expected_seq_len, expected_reduced_seq_len]    ,		)
        a					:      str    =       len(A      )

        # Check attention is always last and order is fine
        a					:      str    =       True
        a					:      Tuple    =       True
        a					:      List[str]    =       model_class(A      )
        model.to(A      )
        model.eval()
        with torch.no_grad():
          a					:      Dict    =       model(**self._prepare_for_class(A    ,		A      )      )

        self.assertEqual(out_len + 1    ,		len(A      )      )

        a					:      str    =       outputs.attentions

        self.assertEqual(len(A      )    ,		A      )
        # verify the first attentions (first block, first layer)
        a					:      Union[str, Any]    =       (self.model_tester.image_size // 4) ** 2
        a					:      Optional[int]    =       (self.model_tester.image_size // (4 * self.model_tester.sr_ratios[0])) ** 2
        self.assertListEqual(
            list(self_attentions[0].shape[-3:]      )    ,		[self.model_tester.num_attention_heads[0], expected_seq_len, expected_reduced_seq_len]    ,		)





    def     lowerCamelCase__				(							self     : int      ):


      '''simple docstring'''




      def check_hidden_states_output(A     : Optional[Any]    ,		A     : List[str]    ,		A     : Union[str, Any]      ):
        a					:      Optional[Any]    =       model_class(A      )
        model.to(A      )
        model.eval()

        with torch.no_grad():
          a					:      Optional[Any]    =       model(**self._prepare_for_class(A    ,		A      )      )

        a					:      Tuple    =       outputs.hidden_states

        a					:      Optional[Any]    =       self.model_tester.num_encoder_blocks
        self.assertEqual(len(A      )    ,		A      )

        # verify the first hidden states (first block)
        self.assertListEqual(
            list(hidden_states[0].shape[-3:]      )    ,		[
                self.model_tester.hidden_sizes[0],
                self.model_tester.image_size // 4,
                self.model_tester.image_size // 4,
            ]    ,		)

      a,					a					:      Any    =       self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
        a					:      List[str]    =       True
        check_hidden_states_output(A    ,		A    ,		A      )

        # check that output_hidden_states also work using config
        del inputs_dict["output_hidden_states"]
        a					:      str    =       True

        check_hidden_states_output(A    ,		A    ,		A      )





    def     lowerCamelCase__				(							self     : Optional[int]      ):


      '''simple docstring'''




      if not self.model_tester.is_training:
        return

      a,					a					:      List[str]    =       self.model_tester.prepare_config_and_inputs_for_common()
      a					:      List[Any]    =       True

      for model_class in self.all_model_classes:
        if model_class in get_values(A      ):
          continue

        a					:      List[Any]    =       model_class(A      )
        model.to(A      )
        model.train()
        a					:      Tuple    =       self._prepare_for_class(A    ,		A    ,		return_labels=A      )
        a					:      Any    =       model(**A      ).loss
        loss.backward()





    @unittest.skip('Will be fixed soon by reducing the size of the model used for common tests.'      )
    def     lowerCamelCase__				(							self     : str      ):


      '''simple docstring'''




      pass





    @slow
    def     lowerCamelCase__				(							self     : int      ):


      '''simple docstring'''




      for model_name in SEGFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
        a					:      Dict    =       SegformerModel.from_pretrained(A      )
        self.assertIsNotNone(A      )




def       snake_case  ():







  '''simple docstring'''

  a					:      Dict    =       Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'				)
  return image






@require_torch
class 						snake_case  (  unittest.TestCase	):





    @slow
    def     lowerCamelCase__				(							self     : Dict      ):


      '''simple docstring'''




      a					:      int    =       SegformerImageProcessor(
          image_scale=(5_1_2, 5_1_2)    ,		keep_ratio=A    ,		align=A    ,		do_random_crop=A      )
      a					:      Dict    =       SegformerForSemanticSegmentation.from_pretrained('nvidia/segformer-b0-finetuned-ade-512-512'      ).to(
          A      )

      a					:      str    =       prepare_img()
      a					:      List[str]    =       image_processor(images=A    ,		return_tensors='pt'      )
      a					:      List[str]    =       encoded_inputs.pixel_values.to(A      )

      with torch.no_grad():
        a					:      Optional[int]    =       model(A      )

      a					:      Any    =       torch.Size((1, model.config.num_labels, 1_2_8, 1_2_8)      )
      self.assertEqual(outputs.logits.shape    ,		A      )

      a					:      str    =       torch.tensor(
          [
              [[-4.63_10, -5.52_32, -6.23_56], [-5.19_21, -6.14_44, -6.59_96], [-5.44_24, -6.27_90, -6.75_74]],
              [[-12.13_91, -13.31_22, -13.95_54], [-12.87_32, -13.93_52, -14.35_63], [-12.94_38, -13.82_26, -14.25_13]],
              [[-12.51_34, -13.46_86, -14.49_15], [-12.86_69, -14.43_43, -14.77_58], [-13.25_23, -14.58_19, -15.06_94]],
          ]      ).to(A      )
      self.assertTrue(torch.allclose(outputs.logits[0, :3, :3, :3]    ,		A    ,		atol=1E-4      )      )





    @slow
    def     lowerCamelCase__				(							self     : Optional[Any]      ):


      '''simple docstring'''




      a					:      Optional[Any]    =       SegformerImageProcessor(
          image_scale=(5_1_2, 5_1_2)    ,		keep_ratio=A    ,		align=A    ,		do_random_crop=A      )
      a					:      Optional[Any]    =       SegformerForSemanticSegmentation.from_pretrained(
          'nvidia/segformer-b1-finetuned-cityscapes-1024-1024'      ).to(A      )

      a					:      List[Any]    =       prepare_img()
      a					:      Optional[Any]    =       image_processor(images=A    ,		return_tensors='pt'      )
      a					:      int    =       encoded_inputs.pixel_values.to(A      )

      with torch.no_grad():
        a					:      Optional[Any]    =       model(A      )

      a					:      Tuple    =       torch.Size((1, model.config.num_labels, 1_2_8, 1_2_8)      )
      self.assertEqual(outputs.logits.shape    ,		A      )

      a					:      Optional[Any]    =       torch.tensor(
          [
              [[-13.57_48, -13.91_11, -12.65_00], [-14.35_00, -15.36_83, -14.23_28], [-14.75_32, -16.04_24, -15.60_87]],
              [[-17.16_51, -15.87_25, -12.96_53], [-17.25_80, -17.37_18, -14.82_23], [-16.60_58, -16.87_83, -16.74_52]],
              [[-3.64_56, -3.02_09, -1.42_03], [-3.07_97, -3.19_59, -2.00_00], [-1.87_57, -1.92_17, -1.69_97]],
          ]      ).to(A      )
      self.assertTrue(torch.allclose(outputs.logits[0, :3, :3, :3]    ,		A    ,		atol=1E-1      )      )





    @slow
    def     lowerCamelCase__				(							self     : int      ):


      '''simple docstring'''




      a					:      str    =       SegformerImageProcessor(
          image_scale=(5_1_2, 5_1_2)    ,		keep_ratio=A    ,		align=A    ,		do_random_crop=A      )
      a					:      Optional[int]    =       SegformerForSemanticSegmentation.from_pretrained('nvidia/segformer-b0-finetuned-ade-512-512'      ).to(
          A      )

      a					:      int    =       prepare_img()
      a					:      Any    =       image_processor(images=A    ,		return_tensors='pt'      )
      a					:      List[Any]    =       encoded_inputs.pixel_values.to(A      )

      with torch.no_grad():
        a					:      str    =       model(A      )

      a					:      str    =       outputs.logits.detach().cpu()

      a					:      Optional[Any]    =       image_processor.post_process_semantic_segmentation(outputs=A    ,		target_sizes=[(5_0_0, 3_0_0)]      )
      a					:      Dict    =       torch.Size((5_0_0, 3_0_0)      )
      self.assertEqual(segmentation[0].shape    ,		A      )

      a					:      int    =       image_processor.post_process_semantic_segmentation(outputs=A      )
      a					:      Any    =       torch.Size((1_2_8, 1_2_8)      )
      self.assertEqual(segmentation[0].shape    ,		A      )




