



'''simple docstring'''





import inspect
import unittest

from huggingface_hub import hf_hub_download

from transformers import ConvNextConfig, UperNetConfig
from transformers.testing_utils import require_torch, require_torch_multi_gpu, require_vision, slow, torch_device
from transformers.utils import is_torch_available, is_vision_available

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


if is_torch_available():
   import torch

   from transformers import UperNetForSemanticSegmentation
   from transformers.models.upernet.modeling_upernet import UPERNET_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
   from PIL import Image

   from transformers import AutoImageProcessor






class 					lowerCamelCase_    :
  """simple docstring"""




  def __init__( self      :	int			,	_a      :	List[Any]			,	_a      :	Optional[int]=13			,	_a      :	int=32			,	_a      :	Tuple=3			,	_a      :	int=4			,	_a      :	List[Any]=[10, 20, 30, 40]			,	_a      :	List[str]=[2, 2, 3, 2]			,	_a      :	List[Any]=True			,	_a      :	str=True			,	_a      :	Optional[Any]=37			,	_a      :	Union[str, Any]="gelu"			,	_a      :	Optional[Any]=10			,	_a      :	Optional[int]=0.02			,	_a      :	List[str]=["stage2", "stage3", "stage4"]			,	_a      :	str=3			,	_a      :	Tuple=None			,	)					->					List[str]:
    __lowerCamelCase       :       Tuple           = parent
    __lowerCamelCase       :       int           = batch_size
    __lowerCamelCase       :       str           = image_size
    __lowerCamelCase       :       Any           = num_channels
    __lowerCamelCase       :       Union[str, Any]           = num_stages
    __lowerCamelCase       :       Union[str, Any]           = hidden_sizes
    __lowerCamelCase       :       Optional[Any]           = depths
    __lowerCamelCase       :       Dict           = is_training
    __lowerCamelCase       :       Optional[int]           = use_labels
    __lowerCamelCase       :       List[str]           = intermediate_size
    __lowerCamelCase       :       List[str]           = hidden_act
    __lowerCamelCase       :       Optional[Any]           = type_sequence_label_size
    __lowerCamelCase       :       Tuple           = initializer_range
    __lowerCamelCase       :       List[str]           = out_features
    __lowerCamelCase       :       Dict           = num_labels
    __lowerCamelCase       :       Optional[int]           = scope
    __lowerCamelCase       :       Union[str, Any]           = num_stages




  def 						_lowercase     ( self      :	Optional[Any]   )					->					str:
    __lowerCamelCase       :       List[Any]           = floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]   )

    __lowerCamelCase       :       List[str]           = None
    if self.use_labels:
      __lowerCamelCase       :       List[Any]           = ids_tensor([self.batch_size]			,	self.type_sequence_label_size   )

    __lowerCamelCase       :       str           = self.get_config()

    return config, pixel_values, labels




  def 						_lowercase     ( self      :	Union[str, Any]   )					->					List[str]:
    return ConvNextConfig(
        num_channels=self.num_channels			,	num_stages=self.num_stages			,	hidden_sizes=self.hidden_sizes			,	depths=self.depths			,	is_training=self.is_training			,	intermediate_size=self.intermediate_size			,	hidden_act=self.hidden_act			,	out_features=self.out_features			,	)




  def 						_lowercase     ( self      :	List[str]   )					->					Any:
    return UperNetConfig(
        backbone_config=self.get_backbone_config()			,	hidden_size=512			,	pool_scales=[1, 2, 3, 6]			,	use_auxiliary_head=_a			,	auxiliary_loss_weight=0.4			,	auxiliary_in_channels=40			,	auxiliary_channels=256			,	auxiliary_num_convs=1			,	auxiliary_concat_input=_a			,	loss_ignore_index=255			,	num_labels=self.num_labels			,	)




  def 						_lowercase     ( self      :	Tuple			,	_a      :	Union[str, Any]			,	_a      :	List[str]			,	_a      :	List[Any]   )					->					List[str]:
    __lowerCamelCase       :       int           = UperNetForSemanticSegmentation(config=_a   )
    model.to(_a   )
    model.eval()
    __lowerCamelCase       :       List[Any]           = model(_a   )
    self.parent.assertEqual(
        result.logits.shape			,	(self.batch_size, self.num_labels, self.image_size, self.image_size)   )




  def 						_lowercase     ( self      :	Optional[int]   )					->					List[str]:
    __lowerCamelCase       :       Union[str, Any]           = self.prepare_config_and_inputs()
    (
        (
        __lowerCamelCase
    )       ,(
        __lowerCamelCase
    )       ,(
        __lowerCamelCase
    )       ,
    )       :       List[Any]           = config_and_inputs
    __lowerCamelCase       :       Optional[Any]           = {'pixel_values': pixel_values}
    return config, inputs_dict






@require_torch
class 					lowerCamelCase_    (				SCREAMING_SNAKE_CASE_   ,   SCREAMING_SNAKE_CASE_   ,   unittest.TestCase     ):
  """simple docstring"""




  a_												=(UperNetForSemanticSegmentation,) if is_torch_available() else ()
  a_												={"""image-segmentation""": UperNetForSemanticSegmentation} if is_torch_available() else {}
  a_												=False
  a_												=False
  a_												=False
  a_												=False
  a_												=False
  a_												=False




  def 						_lowercase     ( self      :	Tuple   )					->					int:
    __lowerCamelCase       :       Optional[int]           = UperNetModelTester(self   )
    __lowerCamelCase       :       str           = ConfigTester(self			,	config_class=_a			,	has_text_modality=_a			,	hidden_size=37   )




  def 						_lowercase     ( self      :	Dict   )					->					List[Any]:
    self.create_and_test_config_common_properties()
    self.config_tester.create_and_test_config_to_json_string()
    self.config_tester.create_and_test_config_to_json_file()
    self.config_tester.create_and_test_config_from_and_save_pretrained()
    self.config_tester.create_and_test_config_with_num_labels()
    self.config_tester.check_config_can_be_init_without_params()
    self.config_tester.check_config_arguments_init()




  def 						_lowercase     ( self      :	Union[str, Any]   )					->					List[str]:
    return




  def 						_lowercase     ( self      :	List[Any]   )					->					Tuple:
    __lowerCamelCase       ,__lowerCamelCase       :       str           = self.model_tester.prepare_config_and_inputs_for_common()

    for model_class in self.all_model_classes:
      __lowerCamelCase       :       Union[str, Any]           = model_class(_a   )
      __lowerCamelCase       :       Optional[Any]           = inspect.signature(model.forward   )
      # signature.parameters is an OrderedDict => so arg_names order is deterministic
      __lowerCamelCase       :       Optional[int]           = [*signature.parameters.keys()]

      __lowerCamelCase       :       List[Any]           = ['pixel_values']
      self.assertListEqual(arg_names[:1]			,	_a   )




  def 						_lowercase     ( self      :	Optional[Any]   )					->					Dict:
    __lowerCamelCase       :       int           = self.model_tester.prepare_config_and_inputs()
    self.model_tester.create_and_check_for_semantic_segmentation(*_a   )




  @unittest.skip(reason='UperNet does not use inputs_embeds'   )
  def 						_lowercase     ( self      :	int   )					->					Any:
    pass




  @unittest.skip(reason='UperNet does not support input and output embeddings'   )
  def 						_lowercase     ( self      :	Dict   )					->					List[str]:
    pass




  @unittest.skip(reason='UperNet does not have a base model'   )
  def 						_lowercase     ( self      :	Optional[Any]   )					->					List[Any]:
    pass




  @unittest.skip(reason='UperNet does not have a base model'   )
  def 						_lowercase     ( self      :	List[str]   )					->					int:
    pass




  @require_torch_multi_gpu
  @unittest.skip(reason='UperNet has some layers using `add_module` which doesn\'t work well with `nn.DataParallel`'   )
  def 						_lowercase     ( self      :	List[str]   )					->					Optional[int]:
    pass




  @unittest.skip('Will be fixed soon by reducing the size of the model used for common tests.'   )
  def 						_lowercase     ( self      :	Union[str, Any]   )					->					List[Any]:
    pass




  def 						_lowercase     ( self      :	Optional[Any]   )					->					int:
    def check_hidden_states_output(_a      :	int			,	_a      :	Optional[int]			,	_a      :	Any   ):
      __lowerCamelCase       :       Dict           = model_class(_a   )
      model.to(_a   )
      model.eval()

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

      __lowerCamelCase       :       Dict           = outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

      __lowerCamelCase       :       List[str]           = self.model_tester.num_stages
      self.assertEqual(len(_a   )			,	expected_num_stages + 1   )

      # ConvNext's feature maps are of shape (batch_size, num_channels, height, width)
      self.assertListEqual(
          list(hidden_states[0].shape[-2:]   )			,	[self.model_tester.image_size // 4, self.model_tester.image_size // 4]			,	)

    __lowerCamelCase       ,__lowerCamelCase       :       str           = self.model_tester.prepare_config_and_inputs_for_common()

    for model_class in self.all_model_classes:
      __lowerCamelCase       :       int           = True
      check_hidden_states_output(_a			,	_a			,	_a   )

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

      check_hidden_states_output(_a			,	_a			,	_a   )




  def 						_lowercase     ( self      :	int   )					->					List[Any]:
    __lowerCamelCase       ,__lowerCamelCase       :       int           = self.model_tester.prepare_config_and_inputs_for_common()

    __lowerCamelCase       :       Tuple           = _config_zero_init(_a   )
    __lowerCamelCase       :       List[Any]           = _config_zero_init(configs_no_init.backbone_config   )
    for model_class in self.all_model_classes:
      __lowerCamelCase       :       str           = model_class(config=_a   )
      for name, param in model.named_parameters():
        if param.requires_grad:
          self.assertIn(
              ((param.data.mean() * 1e9).round() / 1e9).item()			,	[0.0, 1.0]			,	msg=f'Parameter {name} of model {model_class} seems not properly initialized'			,	)




  @unittest.skip(reason='UperNet does not have tied weights'   )
  def 						_lowercase     ( self      :	Union[str, Any]   )					->					Optional[Any]:
    pass




  @slow
  def 						_lowercase     ( self      :	int   )					->					int:
    for model_name in UPERNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
      __lowerCamelCase       :       List[str]           = UperNetForSemanticSegmentation.from_pretrained(_a   )
      self.assertIsNotNone(_a   )




def        a_     (							)      ->      Tuple:
  __lowerCamelCase       :       Optional[int]           = hf_hub_download(
      repo_id='hf-internal-testing/fixtures_ade20k'       ,repo_type='dataset'       ,filename='ADE_val_00000001.jpg'       )
  __lowerCamelCase       :       List[Any]           = Image.open(_lowerCAmelCase       ).convert('RGB'       )
  return image






@require_torch
@require_vision
@slow
class 					lowerCamelCase_    (				unittest.TestCase     ):
  """simple docstring"""




  def 						_lowercase     ( self      :	Dict   )					->					Optional[Any]:
    __lowerCamelCase       :       List[Any]           = AutoImageProcessor.from_pretrained('openmmlab/upernet-swin-tiny'   )
    __lowerCamelCase       :       Any           = UperNetForSemanticSegmentation.from_pretrained('openmmlab/upernet-swin-tiny'   ).to(_a   )

    __lowerCamelCase       :       Tuple           = prepare_img()
    __lowerCamelCase       :       int           = processor(images=_a			,	return_tensors='pt'   ).to(_a   )

    with torch.no_grad():
      __lowerCamelCase       :       List[str]           = model(**_a   )

    __lowerCamelCase       :       Union[str, Any]           = torch.Size((1, model.config.num_labels, 512, 512)   )
    self.assertEqual(outputs.logits.shape			,	_a   )

    __lowerCamelCase       :       Any           = torch.tensor(
        [[-7.5958, -7.5958, -7.4302], [-7.5958, -7.5958, -7.4302], [-7.4797, -7.4797, -7.3068]]   ).to(_a   )
    self.assertTrue(torch.allclose(outputs.logits[0, 0, :3, :3]			,	_a			,	atol=1e-4   )   )




  def 						_lowercase     ( self      :	Optional[Any]   )					->					Optional[Any]:
    __lowerCamelCase       :       Union[str, Any]           = AutoImageProcessor.from_pretrained('openmmlab/upernet-convnext-tiny'   )
    __lowerCamelCase       :       str           = UperNetForSemanticSegmentation.from_pretrained('openmmlab/upernet-convnext-tiny'   ).to(_a   )

    __lowerCamelCase       :       List[str]           = prepare_img()
    __lowerCamelCase       :       Tuple           = processor(images=_a			,	return_tensors='pt'   ).to(_a   )

    with torch.no_grad():
      __lowerCamelCase       :       int           = model(**_a   )

    __lowerCamelCase       :       Optional[int]           = torch.Size((1, model.config.num_labels, 512, 512)   )
    self.assertEqual(outputs.logits.shape			,	_a   )

    __lowerCamelCase       :       Tuple           = torch.tensor(
        [[-8.8110, -8.8110, -8.6521], [-8.8110, -8.8110, -8.6521], [-8.7746, -8.7746, -8.6130]]   ).to(_a   )
    self.assertTrue(torch.allclose(outputs.logits[0, 0, :3, :3]			,	_a			,	atol=1e-4   )   )











'''simple docstring'''





import datetime
import platform
import subprocess
from typing import Optional, Tuple, Union

import numpy as np

def        a_     (							_lowerCAmelCase       ,_lowerCAmelCase       )      ->      np.array:
  __lowerCamelCase       :       Any           = F'{sampling_rate}'
  __lowerCamelCase       :       List[str]           = '1'
  __lowerCamelCase       :       int           = 'f32le'
  __lowerCamelCase       :       Dict           = [
      'ffmpeg',
      '-i',
      'pipe:0',
      '-ac',
      ac,
      '-ar',
      ar,
      '-f',
      format_for_conversion,
      '-hide_banner',
      '-loglevel',
      'quiet',
      'pipe:1',
  ]

  try:
    with subprocess.Popen(_lowerCAmelCase       ,stdin=subprocess.PIPE       ,stdout=subprocess.PIPE       ) as ffmpeg_process:
      __lowerCamelCase       :       Tuple           = ffmpeg_process.communicate(_lowerCAmelCase       )
  except FileNotFoundError as error:
    raise ValueError('ffmpeg was not found but is required to load audio files from filename'       ) from error
  __lowerCamelCase       :       Any           = output_stream[0]
  __lowerCamelCase       :       Union[str, Any]           = np.frombuffer(_lowerCAmelCase       ,np.floataa       )
  if audio.shape[0] == 0:
    raise ValueError('Malformed soundfile'       )
  return audio

def        a_     (							_lowerCAmelCase       ,_lowerCAmelCase       ,_lowerCAmelCase = "f32le"       ,)      ->      Dict:
  __lowerCamelCase       :       Optional[Any]           = F'{sampling_rate}'
  __lowerCamelCase       :       Optional[int]           = '1'
  if format_for_conversion == "s16le":
    __lowerCamelCase       :       List[Any]           = 2
  elif format_for_conversion == "f32le":
    __lowerCamelCase       :       Tuple           = 4
  else:
    raise ValueError(F'Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'       )

  __lowerCamelCase       :       Any           = platform.system()
  if system == "Linux":
    __lowerCamelCase       :       Tuple           = 'alsa'
    __lowerCamelCase       :       Optional[Any]           = 'default'
  elif system == "Darwin":
    __lowerCamelCase       :       Union[str, Any]           = 'avfoundation'
    __lowerCamelCase       :       Tuple           = ':0'
  elif system == "Windows":
    __lowerCamelCase       :       List[str]           = 'dshow'
    __lowerCamelCase       :       Optional[Any]           = 'default'

  __lowerCamelCase       :       Optional[int]           = [
      'ffmpeg',
      '-f',
      format_,
      '-i',
      input_,
      '-ac',
      ac,
      '-ar',
      ar,
      '-f',
      format_for_conversion,
      '-fflags',
      'nobuffer',
      '-hide_banner',
      '-loglevel',
      'quiet',
      'pipe:1',
  ]
  __lowerCamelCase       :       List[str]           = int(round(sampling_rate * chunk_length_s       )       ) * size_of_sample
  __lowerCamelCase       :       int           = _ffmpeg_stream(_lowerCAmelCase       ,_lowerCAmelCase       )
  for item in iterator:
    yield item

def        a_     (							_lowerCAmelCase       ,_lowerCAmelCase       ,_lowerCAmelCase = None       ,_lowerCAmelCase = None       ,_lowerCAmelCase = "f32le"       ,)      ->      List[str]:
  if stream_chunk_s is not None:
    __lowerCamelCase       :       int           = stream_chunk_s
  else:
    __lowerCamelCase       :       List[Any]           = chunk_length_s

  __lowerCamelCase       :       Dict           = ffmpeg_microphone(_lowerCAmelCase       ,_lowerCAmelCase       ,format_for_conversion=_lowerCAmelCase       )
  if format_for_conversion == "s16le":
    __lowerCamelCase       :       List[str]           = np.intaa
    __lowerCamelCase       :       Union[str, Any]           = 2
  elif format_for_conversion == "f32le":
    __lowerCamelCase       :       Union[str, Any]           = np.floataa
    __lowerCamelCase       :       Optional[Any]           = 4
  else:
    raise ValueError(F'Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'       )

  if stride_length_s is None:
    __lowerCamelCase       :       Any           = chunk_length_s / 6
  __lowerCamelCase       :       List[str]           = int(round(sampling_rate * chunk_length_s       )       ) * size_of_sample
  if isinstance(_lowerCAmelCase       ,(int, float)       ):
    __lowerCamelCase       :       Tuple           = [stride_length_s, stride_length_s]

  __lowerCamelCase       :       Union[str, Any]           = int(round(sampling_rate * stride_length_s[0]       )       ) * size_of_sample
  __lowerCamelCase       :       Optional[Any]           = int(round(sampling_rate * stride_length_s[1]       )       ) * size_of_sample
  __lowerCamelCase       :       Dict           = datetime.datetime.now()
  __lowerCamelCase       :       Any           = datetime.timedelta(seconds=_lowerCAmelCase       )
  for item in chunk_bytes_iter(_lowerCAmelCase       ,_lowerCAmelCase       ,stride=(stride_left, stride_right)       ,stream=_lowerCAmelCase       ):
    # Put everything back in numpy scale
    __lowerCamelCase       :       Optional[int]           = np.frombuffer(item['raw']       ,dtype=_lowerCAmelCase       )
    __lowerCamelCase       :       Tuple           = (
        item['stride'][0] // size_of_sample,
        item['stride'][1] // size_of_sample,
    )
    __lowerCamelCase       :       Optional[int]           = sampling_rate
    audio_time += delta
    if datetime.datetime.now() > audio_time + 10 * delta:
      # We're late !! SKIP
      continue
    yield item

def        a_     (							_lowerCAmelCase       ,_lowerCAmelCase       ,_lowerCAmelCase       ,_lowerCAmelCase = False       )      ->      str:
  __lowerCamelCase       :       Optional[int]           = b''
  __lowerCamelCase       ,__lowerCamelCase       :       Any           = stride
  if stride_left + stride_right >= chunk_len:
    raise ValueError(
        F'Stride needs to be strictly smaller than chunk_len: ({stride_left}, {stride_right}) vs {chunk_len}'       )
  __lowerCamelCase       :       str           = 0
  for raw in iterator:
    acc += raw
    if stream and len(_lowerCAmelCase       ) < chunk_len:
      __lowerCamelCase       :       Any           = (_stride_left, 0)
      yield {"raw": acc[:chunk_len], "stride": stride, "partial": True}
    else:
      while len(_lowerCAmelCase       ) >= chunk_len:
        # We are flushing the accumulator
        __lowerCamelCase       :       Any           = (_stride_left, stride_right)
        __lowerCamelCase       :       Optional[int]           = {'raw': acc[:chunk_len], 'stride': stride}
        if stream:
          __lowerCamelCase       :       List[str]           = False
        yield item
        __lowerCamelCase       :       Tuple           = stride_left
        __lowerCamelCase       :       Union[str, Any]           = acc[chunk_len - stride_left - stride_right :]
    # Last chunk
  if len(_lowerCAmelCase       ) > stride_left:
    __lowerCamelCase       :       Tuple           = {'raw': acc, 'stride': (_stride_left, 0)}
    if stream:
      __lowerCamelCase       :       List[str]           = False
    yield item


def        a_     (							_lowerCAmelCase       ,_lowerCAmelCase       )      ->      Tuple:
  __lowerCamelCase       :       int           = 2**24  # 16Mo
  try:
    with subprocess.Popen(_lowerCAmelCase       ,stdout=subprocess.PIPE       ,bufsize=_lowerCAmelCase       ) as ffmpeg_process:
      while True:
        __lowerCamelCase       :       Union[str, Any]           = ffmpeg_process.stdout.read(_lowerCAmelCase       )
        if raw == b"":
          break
        yield raw
  except FileNotFoundError as error:
    raise ValueError('ffmpeg was not found but is required to stream audio files from filename'       ) from error






