


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	:       Dict     =	[
    'openmmlab/upernet-convnext-tiny',
    # See all UperNet models at https://huggingface.co/models?filter=upernet
]

# General docstring
__lowerCAmelCase	:       int     =	'UperNetConfig'




class 					snake_case__							(nn.Module     ):
				"""simple docstring"""



				def __init__(   self       :  Dict      ,   __lowerCamelCase       :  Optional[Any]      ,   __lowerCamelCase       :  Tuple      ,   __lowerCamelCase       :  Optional[Any]      ,   __lowerCamelCase       :  int = 0      ,   __lowerCamelCase       :  Tuple = False      ,   __lowerCamelCase       :  Optional[int] = 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       :  Tuple      ,   __lowerCamelCase       :  Optional[Any]							)      ->  torch.Tensor:
							a					=    self.conv(_SCREAMING_SNAKE_CASE							)
							a					=    self.batch_norm(_SCREAMING_SNAKE_CASE							)
							a					=    self.activation(_SCREAMING_SNAKE_CASE							)

							return output




class 					snake_case__							(nn.Module     ):
				"""simple docstring"""



				def __init__(   self       :  Any      ,   __lowerCamelCase       :  Tuple      ,   __lowerCamelCase       :  int      ,   __lowerCamelCase       :  List[Any]							)      ->  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       :  Any      ,   __lowerCamelCase       :  Optional[int]							)      ->  torch.Tensor:
							a					=    input
							for layer in self.layers:
										a					=    layer(_SCREAMING_SNAKE_CASE							)
							return hidden_state




class 					snake_case__							(nn.Module     ):
				"""simple docstring"""



				def __init__(   self       :  Union[str, Any]      ,   __lowerCamelCase       :  Any      ,   __lowerCamelCase       :  Tuple      ,   __lowerCamelCase       :  Union[str, Any]      ,   __lowerCamelCase       :  List[str]							)      ->  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       :  Tuple      ,   __lowerCamelCase       :  Optional[Any]							)      ->  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 					snake_case__							(nn.Module     ):
				"""simple docstring"""



				def __init__(   self       :  Optional[int]      ,   __lowerCamelCase       :  Optional[Any]      ,   __lowerCamelCase       :  List[Any]							)      ->  Dict:
							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       :  str							)      ->  List[str]:
							self.apply(self._init_weights							)





				def 							__UpperCAmelCase							(   self       :  int      ,   __lowerCamelCase       :  Dict							)      ->  Dict:
							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       :  Dict      ,   __lowerCamelCase       :  Optional[int]							)      ->  List[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       :  Optional[int]      ,   __lowerCamelCase       :  str							)      ->  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 					snake_case__							(nn.Module     ):
				"""simple docstring"""



				def __init__(   self       :  Tuple      ,   __lowerCamelCase       :  Tuple      ,   __lowerCamelCase       :  str = 2      ,   __lowerCamelCase       :  Dict = 3      ,   __lowerCamelCase       :  Union[str, Any] = 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       :  Tuple							)      ->  str:
							self.apply(self._init_weights							)





				def 							__UpperCAmelCase							(   self       :  int      ,   __lowerCamelCase       :  str							)      ->  Optional[Any]:
							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       :  Union[str, Any]      ,   __lowerCamelCase       :  str							)      ->  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 					snake_case__							(lowerCAmelCase__     ):
				"""simple docstring"""



				SCREAMING_SNAKE_CASE_ :							Tuple											=      UperNetConfig
				SCREAMING_SNAKE_CASE_ :							Tuple											=      "pixel_values"
				SCREAMING_SNAKE_CASE_ :							Tuple											=      True





				def 							__UpperCAmelCase							(   self       :  Optional[Any]      ,   __lowerCamelCase       :  int							)      ->  Tuple:
							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       :  Optional[int]							)      ->  Union[str, Any]:
							self.backbone.init_weights()
							self.decode_head.init_weights()
							self.auxiliary_head.init_weights()





				def 							__UpperCAmelCase							(   self       :  str      ,   __lowerCamelCase       :  List[Any]      ,   __lowerCamelCase       :  str=False							)      ->  Optional[int]:
							if isinstance(_SCREAMING_SNAKE_CASE      ,   _SCREAMING_SNAKE_CASE							):
										a					=    value

__lowerCAmelCase	:       List[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	:       List[Any]     =	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."""					, lowerCAmelCase__					, )
class 					snake_case__							(lowerCAmelCase__     ):
				"""simple docstring"""



				def __init__(   self       :  Dict      ,   __lowerCamelCase       :  List[Any]							)      ->  Dict:
							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       :  Any      ,   __lowerCamelCase       :  List[str] = None      ,   __lowerCamelCase       :  str = None      ,   __lowerCamelCase       :  Union[str, Any] = None      ,   __lowerCamelCase       :  Any = None      ,   __lowerCamelCase       :  Tuple = 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      ,   )












from __future__ import annotations

import time

import numpy as np

__lowerCAmelCase	:       List[str]     =	[8, 5, 9, 7]
__lowerCAmelCase	:       str     =	[
    [2, 0, 1, 1],
    [0, 1, 2, 1],
    [4, 0, 0, 3],
    [0, 2, 1, 0],
    [1, 0, 3, 0],
]
__lowerCAmelCase	:       Optional[Any]     =	[
    [3, 2, 1, 4],
    [0, 2, 5, 2],
    [5, 1, 0, 5],
    [1, 5, 3, 0],
    [3, 0, 3, 3],
]




class 					snake_case__							:
				"""simple docstring"""



				def __init__(   self       :  Any      ,   __lowerCamelCase       :  list[int]      ,   __lowerCamelCase       :  list[list[int]]      ,   __lowerCamelCase       :  list[list[int]]      ,   )      ->  None:
							a					=    claim_vector
							a					=    allocated_resources_table
							a					=    maximum_claim_table





				def 							__UpperCAmelCase							(   self       :  List[str]							)      ->  list[int]:
							return [
							    sum(p_item[i] for p_item in self.__allocated_resources_table							)
							    for i in range(len(self.__allocated_resources_table[0]							)							)
							]





				def 							__UpperCAmelCase							(   self       :  str							)      ->  list[int]:
							return np.array(self.__claim_vector							) - np.array(
							    self.__processes_resource_summation()							)





				def 							__UpperCAmelCase							(   self       :  Dict							)      ->  list[list[int]]:
							return [
							    list(np.array(self.__maximum_claim_table[i]							) - np.array(__lowerCamelCase							)							)
							    for i, allocated_resource in enumerate(self.__allocated_resources_table							)
							]





				def 							__UpperCAmelCase							(   self       :  Dict							)      ->  dict[int, list[int]]:
							return {self.__need().index(__lowerCamelCase							): i for i in self.__need()}





				def 							__UpperCAmelCase							(   self       :  Optional[Any]      ,   **__lowerCamelCase       :  Any							)      ->  None:
							a					=    self.__need()
							a					=    self.__allocated_resources_table
							a					=    self.__available_resources()
							a					=    self.__need_index_manager()
							for kw, val in kwargs.items():
										if kw and val is True:
													self.__pretty_data()
							print("_" * 50 + "\n"							)
							while need_list:
										a					=    False
										for each_need in need_list:
													a					=    True
													for index, need in enumerate(__lowerCamelCase							):
																if need > available_resources[index]:
																			a					=    False
																			break
													if execution:
																a					=    True
																# get the original index of the process from ind_ctrl db
																for original_need_index, need_clone in need_index_manager.items():
																			if each_need == need_clone:
																						a					=    original_need_index
																print(f"""Process {process_number + 1} is executing."""							)
																# remove the process run from stack
																need_list.remove(__lowerCamelCase							)
																# update available/freed resources stack
																a					=    np.array(__lowerCamelCase							) + np.array(
																    alloc_resources_table[process_number]							)
																print(
																    "Updated available resource stack for processes: "
																    + " ".join([str(__lowerCamelCase							) for x in available_resources]							)							)
																break
										if safe:
													print("The process is in a safe state.\n"							)
										else:
													print("System in unsafe state. Aborting...\n"							)
													break





				def 							__UpperCAmelCase							(   self       :  Any							)      ->  str:
							print(" " * 9 + "Allocated Resource Table"							)
							for item in self.__allocated_resources_table:
										print(
										    f"""P{self.__allocated_resources_table.index(__lowerCamelCase							) + 1}"""
										    + " ".join(f"""{it:>8}""" for it in item							)
										    + "\n"							)
							print(" " * 9 + "System Resource Table"							)
							for item in self.__maximum_claim_table:
										print(
										    f"""P{self.__maximum_claim_table.index(__lowerCamelCase							) + 1}"""
										    + " ".join(f"""{it:>8}""" for it in item							)
										    + "\n"							)
							print(
							    "Current Usage by Active Processes: "
							    + " ".join(str(__lowerCamelCase							) for x in self.__claim_vector							)							)
							print(
							    "Initial Available Resources:       "
							    + " ".join(str(__lowerCamelCase							) for x in self.__available_resources()							)							)
							time.sleep(1							)


if __name__ == "__main__":
		import doctest

		doctest.testmod()






