

'''simple docstring'''


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


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

# General docstring
lowerCAmelCase__						:       Optional[Any]										=						'UperNetConfig'




class     a			(				nn.Module       ):






      """simple docstring"""




      def __init__(							self					:      Any			,			snake_case_					:      int			,			snake_case_					:      int			,			snake_case_					:      Union[int, Tuple[int, int]]			,			snake_case_					:      Union[int, Tuple[int, int], str] = 0			,			snake_case_					:      bool = False			,			snake_case_					:      Union[int, Tuple[int, int]] = 1			,			):



        '''simple docstring'''



        super().__init__()
        snake_case__						:       Union[str, Any]       						=  nn.Convad(
            in_channels=UpperCamelCase__			,			out_channels=UpperCamelCase__			,			kernel_size=UpperCamelCase__			,			padding=UpperCamelCase__			,			bias=UpperCamelCase__			,			dilation=UpperCamelCase__			,			)
        snake_case__						:       Optional[int]       						=  nn.BatchNormad(UpperCamelCase__  )
        snake_case__						:       Optional[int]       						=  nn.ReLU()







      def       __magic_name__							(							self					:      Tuple			,			snake_case_					:      torch.Tensor  ):



        '''simple docstring'''



        snake_case__						:       List[Any]       						=  self.conv(UpperCamelCase__  )
        snake_case__						:       Union[str, Any]       						=  self.batch_norm(UpperCamelCase__  )
        snake_case__						:       List[str]       						=  self.activation(UpperCamelCase__  )

        return output





class     a			(				nn.Module       ):






      """simple docstring"""




      def __init__(							self					:      Optional[int]			,			snake_case_					:      int			,			snake_case_					:      int			,			snake_case_					:      int  ):



        '''simple docstring'''



        super().__init__()
        snake_case__						:       Dict       						=  [
            nn.AdaptiveAvgPoolad(UpperCamelCase__  ),
            UperNetConvModule(UpperCamelCase__			,			UpperCamelCase__			,			kernel_size=1  ),
        ]
        for i, layer in enumerate(self.layers  ):
          self.add_module(str(UpperCamelCase__  )			,			UpperCamelCase__  )







      def       __magic_name__							(							self					:      List[Any]			,			snake_case_					:      torch.Tensor  ):



        '''simple docstring'''



        snake_case__						:       List[str]       						=  input
        for layer in self.layers:
          snake_case__						:       int       						=  layer(UpperCamelCase__  )
        return hidden_state





class     a			(				nn.Module       ):






      """simple docstring"""




      def __init__(							self					:      str			,			snake_case_					:      Tuple[int, ...]			,			snake_case_					:      int			,			snake_case_					:      int			,			snake_case_					:      bool  ):



        '''simple docstring'''



        super().__init__()
        snake_case__						:       Any       						=  pool_scales
        snake_case__						:       List[Any]       						=  align_corners
        snake_case__						:       List[Any]       						=  in_channels
        snake_case__						:       str       						=  channels
        snake_case__						:       List[Any]       						=  []
        for i, pool_scale in enumerate(UpperCamelCase__  ):
          snake_case__						:       Union[str, Any]       						=  UperNetPyramidPoolingBlock(pool_scale=UpperCamelCase__			,			in_channels=UpperCamelCase__			,			channels=UpperCamelCase__  )
          self.blocks.append(UpperCamelCase__  )
          self.add_module(str(UpperCamelCase__  )			,			UpperCamelCase__  )







      def       __magic_name__							(							self					:      Dict			,			snake_case_					:      torch.Tensor  ):



        '''simple docstring'''



        snake_case__						:       Optional[int]       						=  []
        for ppm in self.blocks:
          snake_case__						:       Tuple       						=  ppm(UpperCamelCase__  )
          snake_case__						:       Any       						=  nn.functional.interpolate(
              UpperCamelCase__			,			size=x.size()[2:]			,			mode='''bilinear'''			,			align_corners=self.align_corners  )
          ppm_outs.append(UpperCamelCase__  )
        return ppm_outs





class     a			(				nn.Module       ):






      """simple docstring"""




      def __init__(							self					:      Union[str, Any]			,			snake_case_					:      Dict			,			snake_case_					:      Tuple  ):



        '''simple docstring'''



        super().__init__()

        snake_case__						:       str       						=  config
        snake_case__						:       Dict       						=  config.pool_scales  # e.g. (1, 2, 3, 6)
        snake_case__						:       Any       						=  in_channels
        snake_case__						:       Optional[Any]       						=  config.hidden_size
        snake_case__						:       Dict       						=  False
        snake_case__						:       int       						=  nn.Convad(self.channels			,			config.num_labels			,			kernel_size=1  )

        # PSP Module
        snake_case__						:       Tuple       						=  UperNetPyramidPoolingModule(
            self.pool_scales			,			self.in_channels[-1]			,			self.channels			,			align_corners=self.align_corners			,			)
        snake_case__						:       Optional[Any]       						=  UperNetConvModule(
            self.in_channels[-1] + len(self.pool_scales  ) * self.channels			,			self.channels			,			kernel_size=3			,			padding=1			,			)
        # FPN Module
        snake_case__						:       Tuple       						=  nn.ModuleList()
        snake_case__						:       Union[str, Any]       						=  nn.ModuleList()
        for in_channels in self.in_channels[:-1]:  # skip the top layer
          snake_case__						:       int       						=  UperNetConvModule(UpperCamelCase__			,			self.channels			,			kernel_size=1  )
          snake_case__						:       List[str]       						=  UperNetConvModule(self.channels			,			self.channels			,			kernel_size=3			,			padding=1  )
          self.lateral_convs.append(UpperCamelCase__  )
          self.fpn_convs.append(UpperCamelCase__  )

        snake_case__						:       Optional[Any]       						=  UperNetConvModule(
            len(self.in_channels  ) * self.channels			,			self.channels			,			kernel_size=3			,			padding=1			,			)







      def       __magic_name__							(							self					:      Optional[int]  ):



        '''simple docstring'''



        self.apply(self._init_weights  )







      def       __magic_name__							(							self					:      Union[str, Any]			,			snake_case_					:      Tuple  ):



        '''simple docstring'''



        if isinstance(UpperCamelCase__			,			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       __magic_name__							(							self					:      Any			,			snake_case_					:      Tuple  ):



        '''simple docstring'''



        snake_case__						:       Dict       						=  inputs[-1]
        snake_case__						:       List[Any]       						=  [x]
        psp_outs.extend(self.psp_modules(UpperCamelCase__  )  )
        snake_case__						:       Optional[Any]       						=  torch.cat(UpperCamelCase__			,			dim=1  )
        snake_case__						:       List[str]       						=  self.bottleneck(UpperCamelCase__  )

        return output







      def       __magic_name__							(							self					:      List[Any]			,			snake_case_					:      torch.Tensor  ):



        '''simple docstring'''



        snake_case__						:       Union[str, Any]       						=  [lateral_conv(encoder_hidden_states[i]  ) for i, lateral_conv in enumerate(self.lateral_convs  )]

        laterals.append(self.psp_forward(UpperCamelCase__  )  )

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

        # build outputs
        snake_case__						:       List[str]       						=  [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  ):
          snake_case__						:       List[Any]       						=  nn.functional.interpolate(
              fpn_outs[i]			,			size=fpn_outs[0].shape[2:]			,			mode='''bilinear'''			,			align_corners=self.align_corners  )
        snake_case__						:       Any       						=  torch.cat(UpperCamelCase__			,			dim=1  )
        snake_case__						:       Optional[Any]       						=  self.fpn_bottleneck(UpperCamelCase__  )
        snake_case__						:       List[Any]       						=  self.classifier(UpperCamelCase__  )

        return output





class     a			(				nn.Module       ):






      """simple docstring"""




      def __init__(							self					:      List[Any]			,			snake_case_					:      Union[str, Any]			,			snake_case_					:      int = 2			,			snake_case_					:      int = 3			,			snake_case_					:      Union[int, Tuple[int, int]] = 1  ):



        '''simple docstring'''



        super().__init__()

        snake_case__						:       Dict       						=  config
        snake_case__						:       List[str]       						=  config.auxiliary_in_channels
        snake_case__						:       Optional[Any]       						=  config.auxiliary_channels
        snake_case__						:       int       						=  config.auxiliary_num_convs
        snake_case__						:       Union[str, Any]       						=  config.auxiliary_concat_input
        snake_case__						:       List[str]       						=  in_index

        snake_case__						:       Tuple       						=  (kernel_size // 2) * dilation
        snake_case__						:       str       						=  []
        convs.append(
            UperNetConvModule(
                self.in_channels			,			self.channels			,			kernel_size=UpperCamelCase__			,			padding=UpperCamelCase__			,			dilation=UpperCamelCase__  )  )
        for i in range(self.num_convs - 1  ):
          convs.append(
              UperNetConvModule(
                  self.channels			,			self.channels			,			kernel_size=UpperCamelCase__			,			padding=UpperCamelCase__			,			dilation=UpperCamelCase__  )  )
        if self.num_convs == 0:
          snake_case__						:       Optional[int]       						=  nn.Identity()
        else:
          snake_case__						:       List[Any]       						=  nn.Sequential(*UpperCamelCase__  )
        if self.concat_input:
          snake_case__						:       Any       						=  UperNetConvModule(
              self.in_channels + self.channels			,			self.channels			,			kernel_size=UpperCamelCase__			,			padding=kernel_size // 2  )

        snake_case__						:       List[Any]       						=  nn.Convad(self.channels			,			config.num_labels			,			kernel_size=1  )







      def       __magic_name__							(							self					:      List[Any]  ):



        '''simple docstring'''



        self.apply(self._init_weights  )







      def       __magic_name__							(							self					:      Optional[Any]			,			snake_case_					:      Tuple  ):



        '''simple docstring'''



        if isinstance(UpperCamelCase__			,			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       __magic_name__							(							self					:      Dict			,			snake_case_					:      torch.Tensor  ):



        '''simple docstring'''



        snake_case__						:       int       						=  encoder_hidden_states[self.in_index]
        snake_case__						:       Union[str, Any]       						=  self.convs(UpperCamelCase__  )
        if self.concat_input:
          snake_case__						:       Optional[Any]       						=  self.conv_cat(torch.cat([hidden_states, output]			,			dim=1  )  )
        snake_case__						:       Optional[int]       						=  self.classifier(UpperCamelCase__  )
        return output





class     a			(				UpperCamelCase_       ):






      """simple docstring"""




      __UpperCAmelCase			     =    UperNetConfig
      __UpperCAmelCase			     =    """pixel_values"""
      __UpperCAmelCase			     =    True







      def       __magic_name__							(							self					:      Optional[int]			,			snake_case_					:      Dict  ):



        '''simple docstring'''



        if isinstance(UpperCamelCase__			,			UpperCamelCase__  ):
          module.backbone.init_weights()
          module.decode_head.init_weights()
          module.auxiliary_head.init_weights()







      def       __magic_name__							(							self					:      List[str]  ):



        '''simple docstring'''



        self.backbone.init_weights()
        self.decode_head.init_weights()
        self.auxiliary_head.init_weights()







      def       __magic_name__							(							self					:      List[Any]			,			snake_case_					:      Any			,			snake_case_					:      int=False  ):



        '''simple docstring'''



        if isinstance(UpperCamelCase__			,			UpperCamelCase__  ):
          snake_case__						:       Optional[Any]       						=  value





lowerCAmelCase__						:       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'

lowerCAmelCase__						:       int										=						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."""			,			UpperCamelCase_			,			)
class     a			(				UpperCamelCase_       ):






      """simple docstring"""




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



        '''simple docstring'''



        super().__init__(UpperCamelCase__  )

        snake_case__						:       Optional[Any]       						=  AutoBackbone.from_config(config.backbone_config  )

        # Semantic segmentation head(s)
        snake_case__						:       List[Any]       						=  UperNetHead(UpperCamelCase__			,			in_channels=self.backbone.channels  )
        snake_case__						:       List[Any]       						=  UperNetFCNHead(UpperCamelCase__  ) 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=UpperCamelCase__			,			config_class=_CONFIG_FOR_DOC  )
      def       __magic_name__							(							self					:      List[str]			,			snake_case_					:      Optional[torch.Tensor] = None			,			snake_case_					:      Optional[bool] = None			,			snake_case_					:      Optional[bool] = None			,			snake_case_					:      Optional[torch.Tensor] = None			,			snake_case_					:      Optional[bool] = None			,			):



        '''simple docstring'''



        snake_case__						:       int       						=  return_dict if return_dict is not None else self.config.use_return_dict
        snake_case__						:       str       						=  (
            output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
        )
        snake_case__						:       Dict       						=  output_attentions if output_attentions is not None else self.config.output_attentions

        snake_case__						:       Optional[Any]       						=  self.backbone.forward_with_filtered_kwargs(
            UpperCamelCase__			,			output_hidden_states=UpperCamelCase__			,			output_attentions=UpperCamelCase__  )
        snake_case__						:       Optional[int]       						=  outputs.feature_maps

        snake_case__						:       Optional[int]       						=  self.decode_head(UpperCamelCase__  )
        snake_case__						:       List[Any]       						=  nn.functional.interpolate(UpperCamelCase__			,			size=pixel_values.shape[2:]			,			mode='''bilinear'''			,			align_corners=UpperCamelCase__  )

        snake_case__						:       Optional[int]       						=  None
        if self.auxiliary_head is not None:
          snake_case__						:       Any       						=  self.auxiliary_head(UpperCamelCase__  )
          snake_case__						:       Optional[Any]       						=  nn.functional.interpolate(
              UpperCamelCase__			,			size=pixel_values.shape[2:]			,			mode='''bilinear'''			,			align_corners=UpperCamelCase__  )

        snake_case__						:       Optional[Any]       						=  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
            snake_case__						:       str       						=  CrossEntropyLoss(ignore_index=self.config.loss_ignore_index  )
            snake_case__						:       Union[str, Any]       						=  loss_fct(UpperCamelCase__			,			UpperCamelCase__  )
            snake_case__						:       Union[str, Any]       						=  loss_fct(UpperCamelCase__			,			UpperCamelCase__  )
            snake_case__						:       List[str]       						=  main_loss + self.config.auxiliary_loss_weight * auxiliary_loss

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

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






'''simple docstring'''


# Copyright 2022 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import os
import subprocess

from packaging.version import Version, parse

from accelerate.commands.config.config_args import default_config_file, load_config_from_file


lowerCAmelCase__						:       List[Any]										=						"""Run commands across TPU VMs for initial setup before running `accelerate launch`."""




def 	_a       (				__lowerCAmelCase		:      Any=None     ):

		"""simple docstring"""

		if subparsers is not None:
				snake_case__						:       Tuple       						=  subparsers.add_parser('''tpu-config'''   ,							description=_description     )
		else:
				snake_case__						:       str       						=  argparse.ArgumentParser('''Accelerate tpu-config command'''   ,							description=_description     )
		# Core arguments
		snake_case__						:       Any       						=  parser.add_argument_group(
		    '''Config Arguments'''   ,							'''Arguments that can be configured through `accelerate config`.'''     )
		config_args.add_argument(
		    '''--config_file'''   ,							type=__lowerCAmelCase   ,							default=__lowerCAmelCase   ,							help='''Path to the config file to use for accelerate.'''   ,							)
		config_args.add_argument(
		    '''--tpu_name'''   ,							default=__lowerCAmelCase   ,							help='''The name of the TPU to use. If not specified, will use the TPU specified in the config file.'''   ,							)
		config_args.add_argument(
		    '''--tpu_zone'''   ,							default=__lowerCAmelCase   ,							help='''The zone of the TPU to use. If not specified, will use the zone specified in the config file.'''   ,							)
		snake_case__						:       Optional[Any]       						=  parser.add_argument_group('''TPU Arguments'''   ,							'''Arguments for options ran inside the TPU.'''     )
		pod_args.add_argument(
		    '''--use_alpha'''   ,							action='''store_true'''   ,							help='''Whether to use `gcloud alpha` when running the TPU training script instead of `gcloud`.'''   ,							)
		pod_args.add_argument(
		    '''--command_file'''   ,							default=__lowerCAmelCase   ,							help='''The path to the file containing the commands to run on the pod on startup.'''   ,							)
		pod_args.add_argument(
		    '''--command'''   ,							action='''append'''   ,							nargs='''+'''   ,							help='''A command to run on the pod. Can be passed multiple times.'''   ,							)
		pod_args.add_argument(
		    '''--install_accelerate'''   ,							action='''store_true'''   ,							help='''Whether to install accelerate on the pod. Defaults to False.'''   ,							)
		pod_args.add_argument(
		    '''--accelerate_version'''   ,							default='''latest'''   ,							help='''The version of accelerate to install on the pod. If not specified, will use the latest pypi version. Specify \'dev\' to install from GitHub.'''   ,							)
		pod_args.add_argument(
		    '''--debug'''   ,							action='''store_true'''   ,							help='''If set, will print the command that would be run instead of running it.'''     )

		if subparsers is not None:
				parser.set_defaults(func=__lowerCAmelCase     )
		return parser




def 	_a       (				__lowerCAmelCase		:      Optional[Any]     ):

		"""simple docstring"""

		snake_case__						:       Optional[int]       						=  None

		# Get the default from the config file if it exists.
		if args.config_file is not None or os.path.isfile(__lowerCAmelCase     ):
				snake_case__						:       Tuple       						=  load_config_from_file(args.config_file     )
				if not args.command_file and defaults.command_file is not None and not args.command:
						snake_case__						:       Optional[int]       						=  defaults.command_file
				if not args.command and defaults.commands is not None:
						snake_case__						:       Union[str, Any]       						=  defaults.commands
				if not args.tpu_name:
						snake_case__						:       Union[str, Any]       						=  defaults.tpu_name
				if not args.tpu_zone:
						snake_case__						:       Optional[Any]       						=  defaults.tpu_zone
		if args.accelerate_version == "dev":
				snake_case__						:       Tuple       						=  '''git+https://github.com/huggingface/accelerate.git'''
		elif args.accelerate_version == "latest":
				snake_case__						:       Union[str, Any]       						=  '''accelerate -U'''
		elif isinstance(parse(args.accelerate_version     )   ,							__lowerCAmelCase     ):
				snake_case__						:       Any       						=  F"""accelerate=={args.accelerate_version}"""

		if not args.command_file and not args.command:
				raise ValueError('''You must specify either a command file or a command to run on the pod.'''     )

		if args.command_file:
				with open(args.command_file   ,							'''r'''     ) as f:
						snake_case__						:       str       						=  [f.read().splitlines()]

    # To turn list of lists into list of strings
		if isinstance(args.command[0]   ,							__lowerCAmelCase     ):
				snake_case__						:       List[Any]       						=  [line for cmd in args.command for line in cmd]
		# Default to the shared folder and install accelerate
		snake_case__						:       Optional[int]       						=  ['''cd /usr/share''']
		if args.install_accelerate:
				new_cmd += [F"""pip install {args.accelerate_version}"""]
		new_cmd += args.command
		snake_case__						:       Tuple       						=  '''; '''.join(__lowerCAmelCase     )

		# Then send it to gcloud
		# Eventually try to use google-api-core to do this instead of subprocess
		snake_case__						:       str       						=  ['''gcloud''']
		if args.use_alpha:
				cmd += ["alpha"]
		cmd += [
		    "compute",
		    "tpus",
		    "tpu-vm",
		    "ssh",
		    args.tpu_name,
		    "--zone",
		    args.tpu_zone,
		    "--command",
		    args.command,
		    "--worker",
		    "all",
		]
		if args.debug:
				print(F"""Running {' '.join(__lowerCAmelCase     )}"""     )
				return
		subprocess.run(__lowerCAmelCase     )
		print('''Successfully setup pod.'''     )





def 	_a       (				):

		"""simple docstring"""

		snake_case__						:       Dict       						=  tpu_command_parser()
		snake_case__						:       Dict       						=  parser.parse_args()

		tpu_command_launcher(__lowerCAmelCase     )



