



from typing import List, Optional, Tuple, Union

import torch
from torch import nn
from torch.nn import CrossEntropyLoss

from ... import AutoBackbone
from ...modeling_outputs import SemanticSegmenterOutput
from ...modeling_utils import PreTrainedModel
from ...utils import add_start_docstrings, add_start_docstrings_to_model_forward, replace_return_docstrings
from ...utils.backbone_utils import BackboneMixin
from .configuration_upernet import UperNetConfig


a_      :      int    =     [
    'openmmlab/upernet-convnext-tiny',
    # See all UperNet models at https://huggingface.co/models?filter=upernet
]

# General docstring
a_      :      Optional[int]    =     'UperNetConfig'
class    __UpperCamelCase			(       nn.Module				):

 """simple docstring"""



 def __init__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE = 0		,				SCREAMING_SNAKE_CASE = False		,				SCREAMING_SNAKE_CASE = 1		,				) ->			None:
  super().__init__()
  a__												= nn.Convad(
      in_channels=SCREAMING_SNAKE_CASE		,				out_channels=SCREAMING_SNAKE_CASE		,				kernel_size=SCREAMING_SNAKE_CASE		,				padding=SCREAMING_SNAKE_CASE		,				bias=SCREAMING_SNAKE_CASE		,				dilation=SCREAMING_SNAKE_CASE		,				)
  a__												= nn.BatchNormad(SCREAMING_SNAKE_CASE      )
  a__												= nn.ReLU()



 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			torch.Tensor:
  a__												= self.conv(SCREAMING_SNAKE_CASE      )
  a__												= self.batch_norm(SCREAMING_SNAKE_CASE      )
  a__												= self.activation(SCREAMING_SNAKE_CASE      )

  return output





class    __UpperCamelCase			(       nn.Module				):

 """simple docstring"""



 def __init__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ) ->			None:
  super().__init__()
  a__												= [
      nn.AdaptiveAvgPoolad(SCREAMING_SNAKE_CASE      ),
      UperNetConvModule(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				kernel_size=1      ),
  ]
  for i, layer in enumerate(self.layers      ):
   self.add_module(str(SCREAMING_SNAKE_CASE      )		,				SCREAMING_SNAKE_CASE      )



 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			torch.Tensor:
  a__												= input
  for layer in self.layers:
   a__												= layer(SCREAMING_SNAKE_CASE      )
  return hidden_state





class    __UpperCamelCase			(       nn.Module				):

 """simple docstring"""



 def __init__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ) ->			None:
  super().__init__()
  a__												= pool_scales
  a__												= align_corners
  a__												= in_channels
  a__												= channels
  a__												= []
  for i, pool_scale in enumerate(SCREAMING_SNAKE_CASE      ):
   a__												= UperNetPyramidPoolingBlock(pool_scale=SCREAMING_SNAKE_CASE		,				in_channels=SCREAMING_SNAKE_CASE		,				channels=SCREAMING_SNAKE_CASE      )
   self.blocks.append(SCREAMING_SNAKE_CASE      )
   self.add_module(str(SCREAMING_SNAKE_CASE      )		,				SCREAMING_SNAKE_CASE      )



 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			List[torch.Tensor]:
  a__												= []
  for ppm in self.blocks:
   a__												= ppm(SCREAMING_SNAKE_CASE      )
   a__												= nn.functional.interpolate(
       SCREAMING_SNAKE_CASE		,				size=x.size()[2:]		,				mode='''bilinear'''		,				align_corners=self.align_corners      )
   ppm_outs.append(SCREAMING_SNAKE_CASE      )
  return ppm_outs





class    __UpperCamelCase			(       nn.Module				):

 """simple docstring"""



 def __init__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ) ->			Any:
  super().__init__()

  a__												= config
  a__												= config.pool_scales  # e.g. (1, 2, 3, 6)
  a__												= in_channels
  a__												= config.hidden_size
  a__												= False
  a__												= nn.Convad(self.channels		,				config.num_labels		,				kernel_size=1      )

  # PSP Module
  a__												= UperNetPyramidPoolingModule(
      self.pool_scales		,				self.in_channels[-1]		,				self.channels		,				align_corners=self.align_corners		,				)
  a__												= UperNetConvModule(
      self.in_channels[-1] + len(self.pool_scales      ) * self.channels		,				self.channels		,				kernel_size=3		,				padding=1		,				)
  # FPN Module
  a__												= nn.ModuleList()
  a__												= nn.ModuleList()
  for in_channels in self.in_channels[:-1]:  # skip the top layer
   a__												= UperNetConvModule(SCREAMING_SNAKE_CASE		,				self.channels		,				kernel_size=1      )
   a__												= UperNetConvModule(self.channels		,				self.channels		,				kernel_size=3		,				padding=1      )
   self.lateral_convs.append(SCREAMING_SNAKE_CASE      )
   self.fpn_convs.append(SCREAMING_SNAKE_CASE      )

  a__												= UperNetConvModule(
      len(self.in_channels      ) * self.channels		,				self.channels		,				kernel_size=3		,				padding=1		,				)

 def 						_UpperCAmelCase       (	self      ) ->			Optional[Any]:
  self.apply(self._init_weights      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			Optional[int]:
  if isinstance(SCREAMING_SNAKE_CASE		,				nn.Convad      ):
   module.weight.data.normal_(mean=0.0		,				std=self.config.initializer_range      )
   if module.bias is not None:
    module.bias.data.zero_()

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			str:
  a__												= inputs[-1]
  a__												= [x]
  psp_outs.extend(self.psp_modules(SCREAMING_SNAKE_CASE      )      )
  a__												= torch.cat(SCREAMING_SNAKE_CASE		,				dim=1      )
  a__												= self.bottleneck(SCREAMING_SNAKE_CASE      )

  return output



 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			torch.Tensor:
  # build laterals
  a__												= [lateral_conv(encoder_hidden_states[i]      ) for i, lateral_conv in enumerate(self.lateral_convs      )]

  laterals.append(self.psp_forward(SCREAMING_SNAKE_CASE      )      )

  # build top-down path
  a__												= len(SCREAMING_SNAKE_CASE      )
  for i in range(used_backbone_levels - 1		,				0		,				-1      ):
   a__												= laterals[i - 1].shape[2:]
   a__												= laterals[i - 1] + nn.functional.interpolate(
       laterals[i]		,				size=SCREAMING_SNAKE_CASE		,				mode='''bilinear'''		,				align_corners=self.align_corners      )

  # build outputs
  a__												= [self.fpn_convs[i](laterals[i]      ) for i in range(used_backbone_levels - 1      )]
  # append psp feature
  fpn_outs.append(laterals[-1]      )

  for i in range(used_backbone_levels - 1		,				0		,				-1      ):
   a__												= nn.functional.interpolate(
       fpn_outs[i]		,				size=fpn_outs[0].shape[2:]		,				mode='''bilinear'''		,				align_corners=self.align_corners      )
  a__												= torch.cat(SCREAMING_SNAKE_CASE		,				dim=1      )
  a__												= self.fpn_bottleneck(SCREAMING_SNAKE_CASE      )
  a__												= self.classifier(SCREAMING_SNAKE_CASE      )

  return output





class    __UpperCamelCase			(       nn.Module				):

 """simple docstring"""



 def __init__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE = 2		,				SCREAMING_SNAKE_CASE = 3		,				SCREAMING_SNAKE_CASE = 1      ) ->			None:
  super().__init__()

  a__												= config
  a__												= config.auxiliary_in_channels
  a__												= config.auxiliary_channels
  a__												= config.auxiliary_num_convs
  a__												= config.auxiliary_concat_input
  a__												= in_index

  a__												= (kernel_size // 2) * dilation
  a__												= []
  convs.append(
      UperNetConvModule(
          self.in_channels		,				self.channels		,				kernel_size=SCREAMING_SNAKE_CASE		,				padding=SCREAMING_SNAKE_CASE		,				dilation=SCREAMING_SNAKE_CASE      )      )
  for i in range(self.num_convs - 1      ):
   convs.append(
       UperNetConvModule(
           self.channels		,				self.channels		,				kernel_size=SCREAMING_SNAKE_CASE		,				padding=SCREAMING_SNAKE_CASE		,				dilation=SCREAMING_SNAKE_CASE      )      )
  if self.num_convs == 0:
   a__												= nn.Identity()
  else:
   a__												= nn.Sequential(*SCREAMING_SNAKE_CASE      )
  if self.concat_input:
   a__												= UperNetConvModule(
       self.in_channels + self.channels		,				self.channels		,				kernel_size=SCREAMING_SNAKE_CASE		,				padding=kernel_size // 2      )

  a__												= nn.Convad(self.channels		,				config.num_labels		,				kernel_size=1      )

 def 						_UpperCAmelCase       (	self      ) ->			Union[str, Any]:
  self.apply(self._init_weights      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			Optional[int]:
  if isinstance(SCREAMING_SNAKE_CASE		,				nn.Convad      ):
   module.weight.data.normal_(mean=0.0		,				std=self.config.initializer_range      )
   if module.bias is not None:
    module.bias.data.zero_()



 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			torch.Tensor:
  # just take the relevant feature maps
  a__												= encoder_hidden_states[self.in_index]
  a__												= self.convs(SCREAMING_SNAKE_CASE      )
  if self.concat_input:
   a__												= self.conv_cat(torch.cat([hidden_states, output]		,				dim=1      )      )
  a__												= self.classifier(SCREAMING_SNAKE_CASE      )
  return output





class    __UpperCamelCase			(       _lowercase				):

 """simple docstring"""



 _lowercase   :    Tuple      						=					UperNetConfig
 _lowercase   :    List[str]      						=					'''pixel_values'''
 _lowercase   :    Any      						=					True

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			Any:
  if isinstance(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ):
   module.backbone.init_weights()
   module.decode_head.init_weights()
   module.auxiliary_head.init_weights()

 def 						_UpperCAmelCase       (	self      ) ->			Dict:
  self.backbone.init_weights()
  self.decode_head.init_weights()
  self.auxiliary_head.init_weights()



 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE=False      ) ->			Tuple:
  if isinstance(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ):
   a__												= value





a_      :      str    =     R'\n    Parameters:\n    This model is a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) sub-class. Use\n    it as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage and\n    behavior.\n        config ([`UperNetConfig`]): Model configuration class with all the parameters of the model.\n            Initializing with a config file does not load the weights associated with the model, only the\n            configuration. Check out the [`~PreTrainedModel.from_pretrained`] method to load the model weights.\n'

a_      :      Tuple    =     R'\n    Args:\n        pixel_values (`torch.FloatTensor` of shape `(batch_size, num_channels, height, width)`):\n            Pixel values. Padding will be ignored by default should you provide it. Pixel values can be obtained using\n            [`AutoImageProcessor`]. See [`SegformerImageProcessor.__call__`] for details.\n        output_attentions (`bool`, *optional*):\n            Whether or not to return the attentions tensors of all attention layers in case the backbone has them. See\n            `attentions` under returned tensors for more detail.\n        output_hidden_states (`bool`, *optional*):\n            Whether or not to return the hidden states of all layers of the backbone. See `hidden_states` under\n            returned tensors for more detail.\n        return_dict (`bool`, *optional*):\n            Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple.\n'
@add_start_docstrings(
    '''UperNet framework leveraging any vision backbone e.g. for ADE20k, CityScapes.'''	,	_lowercase	,	)
class    __UpperCamelCase			(       _lowercase				):

 """simple docstring"""



 def __init__(	self		,				SCREAMING_SNAKE_CASE      ) ->			Optional[int]:
  super().__init__(SCREAMING_SNAKE_CASE      )

  a__												= AutoBackbone.from_config(config.backbone_config      )

  # Semantic segmentation head(s)
  a__												= UperNetHead(SCREAMING_SNAKE_CASE		,				in_channels=self.backbone.channels      )
  a__												= UperNetFCNHead(SCREAMING_SNAKE_CASE      ) if config.use_auxiliary_head else None

  # Initialize weights and apply final processing
  self.post_init()



 @add_start_docstrings_to_model_forward(UPERNET_INPUTS_DOCSTRING.format('''batch_size, sequence_length'''      )      )
 @replace_return_docstrings(output_type=SCREAMING_SNAKE_CASE		,				config_class=_CONFIG_FOR_DOC      )
 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE = None		,				SCREAMING_SNAKE_CASE = None		,				SCREAMING_SNAKE_CASE = None		,				SCREAMING_SNAKE_CASE = None		,				SCREAMING_SNAKE_CASE = None		,				) ->			Union[tuple, SemanticSegmenterOutput]:
  a__												= return_dict if return_dict is not None else self.config.use_return_dict
  a__												= (
      output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
  )
  a__												= output_attentions if output_attentions is not None else self.config.output_attentions

  a__												= self.backbone.forward_with_filtered_kwargs(
      SCREAMING_SNAKE_CASE		,				output_hidden_states=SCREAMING_SNAKE_CASE		,				output_attentions=SCREAMING_SNAKE_CASE      )
  a__												= outputs.feature_maps

  a__												= self.decode_head(SCREAMING_SNAKE_CASE      )
  a__												= nn.functional.interpolate(SCREAMING_SNAKE_CASE		,				size=pixel_values.shape[2:]		,				mode='''bilinear'''		,				align_corners=SCREAMING_SNAKE_CASE      )

  a__												= None
  if self.auxiliary_head is not None:
   a__												= self.auxiliary_head(SCREAMING_SNAKE_CASE      )
   a__												= nn.functional.interpolate(
       SCREAMING_SNAKE_CASE		,				size=pixel_values.shape[2:]		,				mode='''bilinear'''		,				align_corners=SCREAMING_SNAKE_CASE      )

  a__												= None
  if labels is not None:
   if self.config.num_labels == 1:
    raise ValueError('''The number of labels should be greater than one'''      )
   else:
    # compute weighted loss
    a__												= CrossEntropyLoss(ignore_index=self.config.loss_ignore_index      )
    a__												= loss_fct(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      )
    a__												= loss_fct(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      )
    a__												= main_loss + self.config.auxiliary_loss_weight * auxiliary_loss

  if not return_dict:
   if output_hidden_states:
    a__												= (logits,) + outputs[1:]
   else:
    a__												= (logits,) + outputs[2:]
   return ((loss,) + output) if loss is not None else output

  return SemanticSegmenterOutput(
      loss=SCREAMING_SNAKE_CASE		,				logits=SCREAMING_SNAKE_CASE		,				hidden_states=outputs.hidden_states		,				attentions=outputs.attentions		,				)










import logging
import os

from .state import PartialState
class    __UpperCamelCase			(       logging.LoggerAdapter				):

 """simple docstring"""



 @staticmethod
 def 						_UpperCAmelCase       (	SCREAMING_SNAKE_CASE      ) ->			Optional[Any]:
  a__												= PartialState()
  return not main_process_only or (main_process_only and state.is_main_process)



 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				*SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE      ) ->			Optional[int]:
  if PartialState._shared_state == {}:
   raise RuntimeError(
       '''You must initialize the accelerate state by calling either `PartialState()` or `Accelerator()` before using the logging utility.'''      )
  a__												= kwargs.pop('''main_process_only'''		,				SCREAMING_SNAKE_CASE      )
  a__												= kwargs.pop('''in_order'''		,				SCREAMING_SNAKE_CASE      )

  if self.isEnabledFor(SCREAMING_SNAKE_CASE      ):
   if self._should_log(SCREAMING_SNAKE_CASE      ):
    a__							,			a__												= self.process(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      )
    self.logger.log(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				*SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE      )

   elif in_order:
    a__												= PartialState()
    for i in range(state.num_processes      ):
     if i == state.process_index:
      a__							,			a__												= self.process(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      )
      self.logger.log(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				*SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE      )
     state.wait_for_everyone()






def     __a					(		__UpperCAmelCase							,  __UpperCAmelCase = None       ):
 if log_level is None:
  a__												= os.environ.get('''ACCELERATE_LOG_LEVEL'''							,  __UpperCAmelCase       )
 a__												= logging.getLogger(__UpperCAmelCase       )
 if log_level is not None:
  logger.setLevel(log_level.upper()       )
  logger.root.setLevel(log_level.upper()       )
 return MultiProcessAdapter(__UpperCAmelCase							,  {}       )





