





import importlib.metadata
import warnings
from copy import deepcopy

from packaging import version

from ..utils import logging
from .import_utils import is_accelerate_available, is_bitsandbytes_available


if is_bitsandbytes_available():
		import bitsandbytes as bnb
		import torch
		import torch.nn as nn

		from ..pytorch_utils import ConvaD

if is_accelerate_available():
		from accelerate import init_empty_weights
		from accelerate.utils import find_tied_parameters

_lowerCamelCase :    Union[str, Any]  							= logging.get_logger(__name__)




def _a (  SCREAMING_SNAKE_CASE__      :  Optional[Any]			,				SCREAMING_SNAKE_CASE__      :  List[str]			,				SCREAMING_SNAKE_CASE__      :  Dict			,				SCREAMING_SNAKE_CASE__      :  str=None			,				SCREAMING_SNAKE_CASE__      :  Tuple=None       )	->    List[str]:

							'''simple docstring'''






							if "." in tensor_name:
														SCREAMING_SNAKE_CASE__       :      Union[str, Any]     =						tensor_name.split("."       )
														for split in splits[:-1]:
																					SCREAMING_SNAKE_CASE__       :      Dict     =						getattr(SCREAMING_SNAKE_CASE__			,				SCREAMING_SNAKE_CASE__       )
																					if new_module is None:
																												raise ValueError(f'''{module} has no attribute {split}.'''       )
																					SCREAMING_SNAKE_CASE__       :      str     =						new_module
														SCREAMING_SNAKE_CASE__       :      Any     =						splits[-1]

							if tensor_name not in module._parameters and tensor_name not in module._buffers:
														raise ValueError(f'''{module} does not have a parameter or a buffer named {tensor_name}.'''       )
							SCREAMING_SNAKE_CASE__       :      int     =						tensor_name in module._buffers
							SCREAMING_SNAKE_CASE__       :      Optional[Any]     =						getattr(SCREAMING_SNAKE_CASE__			,				SCREAMING_SNAKE_CASE__       )

							if old_value.device == torch.device("meta"       ) and device not in ["meta", torch.device("meta"       )] and value is None:
														raise ValueError(f'''{tensor_name} is on the meta device, we need a `value` to put in on {device}.'''       )

							SCREAMING_SNAKE_CASE__       :      Optional[int]     =						False
							SCREAMING_SNAKE_CASE__       :      str     =						False
							if is_buffer or not is_bitsandbytes_available():
														SCREAMING_SNAKE_CASE__       :      Any     =						False
														SCREAMING_SNAKE_CASE__       :      Any     =						False
							else:
														SCREAMING_SNAKE_CASE__       :      str     =						hasattr(bnb.nn			,				"Params4bit"       ) and isinstance(module._parameters[tensor_name]			,				bnb.nn.Paramsabit       )
														SCREAMING_SNAKE_CASE__       :      Tuple     =						isinstance(module._parameters[tensor_name]			,				bnb.nn.IntaParams       )

							if is_abit or is_abit:
														SCREAMING_SNAKE_CASE__       :      int     =						module._parameters[tensor_name]
														if param.device.type != "cuda":
																					if value is None:
																												SCREAMING_SNAKE_CASE__       :      Optional[int]     =						old_value.to(SCREAMING_SNAKE_CASE__       )
																					elif isinstance(SCREAMING_SNAKE_CASE__			,				torch.Tensor       ):
																												SCREAMING_SNAKE_CASE__       :      int     =						value.to("cpu"       )
																												if value.dtype == torch.inta:
																																			SCREAMING_SNAKE_CASE__       :      Optional[Any]     =						version.parse(importlib.metadata.version("bitsandbytes"       )       ) > version.parse(
																																			    "0.37.2"       )
																																			if not is_abit_serializable:
																																										raise ValueError(
																																										    "Detected int8 weights but the version of bitsandbytes is not compatible with int8 serialization. "
																																										    "Make sure to download the latest `bitsandbytes` version. `pip install --upgrade bitsandbytes`."       )
																					else:
																												SCREAMING_SNAKE_CASE__       :      Any     =						torch.tensor(SCREAMING_SNAKE_CASE__			,				device="cpu"       )

																					# Support models using `Conv1D` in place of `nn.Linear` (e.g. gpt2) by transposing the weight matrix prior to quantization.
																					# Since weights are saved in the correct "orientation", we skip transposing when loading.
																					if issubclass(module.source_cls			,				SCREAMING_SNAKE_CASE__       ) and fpaa_statistics is None:
																												SCREAMING_SNAKE_CASE__       :      List[str]     =						new_value.T

																					SCREAMING_SNAKE_CASE__       :      Union[str, Any]     =						old_value.__dict__
																					if is_abit:
																												SCREAMING_SNAKE_CASE__       :      Any     =						bnb.nn.IntaParams(SCREAMING_SNAKE_CASE__			,				requires_grad=SCREAMING_SNAKE_CASE__			,				**SCREAMING_SNAKE_CASE__       ).to(SCREAMING_SNAKE_CASE__       )
																					elif is_abit:
																												SCREAMING_SNAKE_CASE__       :      Any     =						bnb.nn.Paramsabit(SCREAMING_SNAKE_CASE__			,				requires_grad=SCREAMING_SNAKE_CASE__			,				**SCREAMING_SNAKE_CASE__       ).to(SCREAMING_SNAKE_CASE__       )

																					SCREAMING_SNAKE_CASE__       :      str     =						new_value
																					if fpaa_statistics is not None:
																												setattr(module.weight			,				"SCB"			,				fpaa_statistics.to(SCREAMING_SNAKE_CASE__       )       )

							else:
														if value is None:
																					SCREAMING_SNAKE_CASE__       :      str     =						old_value.to(SCREAMING_SNAKE_CASE__       )
														elif isinstance(SCREAMING_SNAKE_CASE__			,				torch.Tensor       ):
																					SCREAMING_SNAKE_CASE__       :      Union[str, Any]     =						value.to(SCREAMING_SNAKE_CASE__       )
														else:
																					SCREAMING_SNAKE_CASE__       :      Optional[int]     =						torch.tensor(SCREAMING_SNAKE_CASE__			,				device=SCREAMING_SNAKE_CASE__       )

														if is_buffer:
																					SCREAMING_SNAKE_CASE__       :      List[Any]     =						new_value
														else:
																					SCREAMING_SNAKE_CASE__       :      int     =						nn.Parameter(SCREAMING_SNAKE_CASE__			,				requires_grad=old_value.requires_grad       )
																					SCREAMING_SNAKE_CASE__       :      int     =						new_value




def _a (  SCREAMING_SNAKE_CASE__      :  Any			,				SCREAMING_SNAKE_CASE__      :  int=None			,				SCREAMING_SNAKE_CASE__      :  Optional[Any]=None			,				SCREAMING_SNAKE_CASE__      :  Dict=None			,				SCREAMING_SNAKE_CASE__      :  Tuple=False       )	->    Union[str, Any]:

							'''simple docstring'''






							for name, module in model.named_children():
														if current_key_name is None:
																					SCREAMING_SNAKE_CASE__       :      Optional[Any]     =						[]
														current_key_name.append(SCREAMING_SNAKE_CASE__       )

														if (isinstance(SCREAMING_SNAKE_CASE__			,				nn.Linear       ) or isinstance(SCREAMING_SNAKE_CASE__			,				SCREAMING_SNAKE_CASE__       )) and name not in modules_to_not_convert:
																					# Check if the current key is not in the `modules_to_not_convert`
																					if not any(key in ".".join(SCREAMING_SNAKE_CASE__       ) for key in modules_to_not_convert       ):
																												with init_empty_weights():
																																			if isinstance(SCREAMING_SNAKE_CASE__			,				SCREAMING_SNAKE_CASE__       ):
																																										SCREAMING_SNAKE_CASE__   ,SCREAMING_SNAKE_CASE__       :      Optional[int]     =						module.weight.shape
																																			else:
																																										SCREAMING_SNAKE_CASE__       :      Dict     =						module.in_features
																																										SCREAMING_SNAKE_CASE__       :      Dict     =						module.out_features

																																			if quantization_config.quantization_method() == "llm_int8":
																																										SCREAMING_SNAKE_CASE__       :      Any     =						bnb.nn.LinearabitLt(
																																										    SCREAMING_SNAKE_CASE__			,				SCREAMING_SNAKE_CASE__			,				module.bias is not None			,				has_fpaa_weights=quantization_config.llm_inta_has_fpaa_weight			,				threshold=quantization_config.llm_inta_threshold			,				)
																																										SCREAMING_SNAKE_CASE__       :      int     =						True
																																			else:
																																										if (
																																										    quantization_config.llm_inta_skip_modules is not None
																																										    and name in quantization_config.llm_inta_skip_modules
																																										):
																																																	pass
																																										else:
																																																	SCREAMING_SNAKE_CASE__       :      List[str]     =						bnb.nn.Linearabit(
																																																	    SCREAMING_SNAKE_CASE__			,				SCREAMING_SNAKE_CASE__			,				module.bias is not None			,				quantization_config.bnb_abit_compute_dtype			,				compress_statistics=quantization_config.bnb_abit_use_double_quant			,				quant_type=quantization_config.bnb_abit_quant_type			,				)
																																																	SCREAMING_SNAKE_CASE__       :      Dict     =						True
                    # Store the module class in case we need to transpose the weight later
																																			SCREAMING_SNAKE_CASE__       :      Optional[Any]     =						type(SCREAMING_SNAKE_CASE__       )
																																			# Force requires grad to False to avoid unexpected errors
																																			model._modules[name].requires_grad_(SCREAMING_SNAKE_CASE__       )
														if len(list(module.children()       )       ) > 0:
																					SCREAMING_SNAKE_CASE__   ,SCREAMING_SNAKE_CASE__       :      List[str]     =						_replace_with_bnb_linear(
																					    SCREAMING_SNAKE_CASE__			,				SCREAMING_SNAKE_CASE__			,				SCREAMING_SNAKE_CASE__			,				SCREAMING_SNAKE_CASE__			,				has_been_replaced=SCREAMING_SNAKE_CASE__			,				)
														# Remove the last key for recursion
														current_key_name.pop(-1       )
							return model, has_been_replaced




def _a (  SCREAMING_SNAKE_CASE__      :  Tuple			,				SCREAMING_SNAKE_CASE__      :  Tuple=None			,				SCREAMING_SNAKE_CASE__      :  Optional[int]=None			,				SCREAMING_SNAKE_CASE__      :  Union[str, Any]=None       )	->    Optional[int]:

							'''simple docstring'''






							SCREAMING_SNAKE_CASE__       :      Union[str, Any]     =						["lm_head"] if modules_to_not_convert is None else modules_to_not_convert
							SCREAMING_SNAKE_CASE__   ,SCREAMING_SNAKE_CASE__       :      List[str]     =						_replace_with_bnb_linear(
							    SCREAMING_SNAKE_CASE__			,				SCREAMING_SNAKE_CASE__			,				SCREAMING_SNAKE_CASE__			,				SCREAMING_SNAKE_CASE__       )

							if not has_been_replaced:
														logger.warning(
														    "You are loading your model in 8bit or 4bit but no linear modules were found in your model."
														    " Please double check your model architecture, or submit an issue on github if you think this is"
														    " a bug."       )

							return model




def _a (  *SCREAMING_SNAKE_CASE__      :  Any			,				**SCREAMING_SNAKE_CASE__      :  List[str]       )	->    List[str]:

							'''simple docstring'''






							warnings.warn(
							    "`replace_8bit_linear` will be deprecated in a future version, please use `replace_with_bnb_linear` instead"			,				SCREAMING_SNAKE_CASE__			,				)
							return replace_with_bnb_linear(*SCREAMING_SNAKE_CASE__			,				**SCREAMING_SNAKE_CASE__       )




def _a (  *SCREAMING_SNAKE_CASE__      :  Dict			,				**SCREAMING_SNAKE_CASE__      :  Tuple       )	->    Tuple:

							'''simple docstring'''






							warnings.warn(
							    "`set_module_8bit_tensor_to_device` will be deprecated in a future version, please use `set_module_quantized_tensor_to_device` instead"			,				SCREAMING_SNAKE_CASE__			,				)
							return set_module_quantized_tensor_to_device(*SCREAMING_SNAKE_CASE__			,				**SCREAMING_SNAKE_CASE__       )







def _a (  SCREAMING_SNAKE_CASE__      :  str       )	->    List[str]:

							'''simple docstring'''






							SCREAMING_SNAKE_CASE__       :      List[Any]     =						deepcopy(SCREAMING_SNAKE_CASE__       )  # this has 0 cost since it is done inside `init_empty_weights` context manager`
							tied_model.tie_weights()

							SCREAMING_SNAKE_CASE__       :      List[str]     =						find_tied_parameters(SCREAMING_SNAKE_CASE__       )
							# For compatibility with Accelerate < 0.18
							if isinstance(SCREAMING_SNAKE_CASE__			,				SCREAMING_SNAKE_CASE__       ):
														SCREAMING_SNAKE_CASE__       :      Tuple     =						sum(list(tied_params.values()       )			,				[]       ) + list(tied_params.keys()       )
							else:
														SCREAMING_SNAKE_CASE__       :      List[str]     =						sum(SCREAMING_SNAKE_CASE__			,				[]       )
							SCREAMING_SNAKE_CASE__       :      List[Any]     =						len(SCREAMING_SNAKE_CASE__       ) > 0

							# Check if it is a base model
							SCREAMING_SNAKE_CASE__       :      Optional[Any]     =						not hasattr(SCREAMING_SNAKE_CASE__			,				model.base_model_prefix       )

							# Ignore this for base models (BertModel, GPT2Model, etc.)
							if (not has_tied_params) and is_base_model:
														return []

							# otherwise they have an attached head
							SCREAMING_SNAKE_CASE__       :      str     =						list(model.named_children()       )
							SCREAMING_SNAKE_CASE__       :      List[Any]     =						[list_modules[-1][0]]

							# add last module together with tied weights
							SCREAMING_SNAKE_CASE__       :      List[Any]     =						set(SCREAMING_SNAKE_CASE__       ) - set(SCREAMING_SNAKE_CASE__       )
							SCREAMING_SNAKE_CASE__       :      int     =						list(set(SCREAMING_SNAKE_CASE__       )       ) + list(SCREAMING_SNAKE_CASE__       )

							# remove ".weight" from the keys
							SCREAMING_SNAKE_CASE__       :      Optional[Any]     =						[".weight", ".bias"]
							SCREAMING_SNAKE_CASE__       :      Any     =						[]
							for name in list_untouched:
														for name_to_remove in names_to_remove:
																					if name_to_remove in name:
																												SCREAMING_SNAKE_CASE__       :      Any     =						name.replace(SCREAMING_SNAKE_CASE__			,				""       )
														filtered_module_names.append(SCREAMING_SNAKE_CASE__       )

							return filtered_module_names













import sys
from typing import Tuple

import numpy as np
import torch
from PIL import Image
from torch import nn

from transformers.image_utils import PILImageResampling
from utils import img_tensorize


class 						lowerCamelCase       :







							"""simple docstring"""




							def __init__( self     :					Dict,			_UpperCAmelCase     :					Any,			_UpperCAmelCase     :					List[Any]=sys.maxsize						)				->			Optional[int]:





														"""simple docstring"""
														SCREAMING_SNAKE_CASE__       :      Optional[int]     =						"bilinear"
														SCREAMING_SNAKE_CASE__       :      Optional[int]     =						max_size
														SCREAMING_SNAKE_CASE__       :      Optional[int]     =						short_edge_length






							def __call__( self     :					Optional[int],			_UpperCAmelCase     :					Union[str, Any]						)				->			str:





														"""simple docstring"""
														SCREAMING_SNAKE_CASE__       :      List[Any]     =						[]
														for img in imgs:
																					SCREAMING_SNAKE_CASE__   ,SCREAMING_SNAKE_CASE__       :      Tuple     =						img.shape[:2]
																					# later: provide list and randomly choose index for resize
																					SCREAMING_SNAKE_CASE__       :      List[str]     =						np.random.randint(self.short_edge_length[0],			self.short_edge_length[1] + 1						)
																					if size == 0:
																												return img
																					SCREAMING_SNAKE_CASE__       :      int     =						size * 1.0 / min(_UpperCAmelCase,			_UpperCAmelCase						)
																					if h < w:
																												SCREAMING_SNAKE_CASE__   ,SCREAMING_SNAKE_CASE__       :      Optional[int]     =						size, scale * w
																					else:
																												SCREAMING_SNAKE_CASE__   ,SCREAMING_SNAKE_CASE__       :      List[str]     =						scale * h, size
																					if max(_UpperCAmelCase,			_UpperCAmelCase						) > self.max_size:
																												SCREAMING_SNAKE_CASE__       :      str     =						self.max_size * 1.0 / max(_UpperCAmelCase,			_UpperCAmelCase						)
																												SCREAMING_SNAKE_CASE__       :      Any     =						newh * scale
																												SCREAMING_SNAKE_CASE__       :      List[str]     =						neww * scale
																					SCREAMING_SNAKE_CASE__       :      Any     =						int(neww + 0.5						)
																					SCREAMING_SNAKE_CASE__       :      List[Any]     =						int(newh + 0.5						)

																					if img.dtype == np.uinta:
																												SCREAMING_SNAKE_CASE__       :      List[Any]     =						Image.fromarray(_UpperCAmelCase						)
																												SCREAMING_SNAKE_CASE__       :      Union[str, Any]     =						pil_image.resize((neww, newh),			PILImageResampling.BILINEAR						)
																												SCREAMING_SNAKE_CASE__       :      List[str]     =						np.asarray(_UpperCAmelCase						)
																					else:
																												SCREAMING_SNAKE_CASE__       :      str     =						img.permute(2,			0,			1						).unsqueeze(0						)  # 3, 0, 1)  # hw(c) -> nchw
																												SCREAMING_SNAKE_CASE__       :      Tuple     =						nn.functional.interpolate(
																												    _UpperCAmelCase,			(newh, neww),			mode=self.interp_method,			align_corners=_UpperCAmelCase						).squeeze(0						)
																					img_augs.append(_UpperCAmelCase						)

														return img_augs


class 						lowerCamelCase       :







							"""simple docstring"""




							def __init__( self     :					Dict,			_UpperCAmelCase     :					Optional[Any]						)				->			Optional[int]:





														"""simple docstring"""
														SCREAMING_SNAKE_CASE__       :      Any     =						ResizeShortestEdge([cfg.INPUT.MIN_SIZE_TEST, cfg.INPUT.MIN_SIZE_TEST],			cfg.INPUT.MAX_SIZE_TEST						)
														SCREAMING_SNAKE_CASE__       :      Any     =						cfg.INPUT.FORMAT
														SCREAMING_SNAKE_CASE__       :      List[str]     =						cfg.SIZE_DIVISIBILITY
														SCREAMING_SNAKE_CASE__       :      List[Any]     =						cfg.PAD_VALUE
														SCREAMING_SNAKE_CASE__       :      Dict     =						cfg.INPUT.MAX_SIZE_TEST
														SCREAMING_SNAKE_CASE__       :      Optional[int]     =						cfg.MODEL.DEVICE
														SCREAMING_SNAKE_CASE__       :      int     =						torch.tensor(cfg.MODEL.PIXEL_STD						).to(self.device						).view(len(cfg.MODEL.PIXEL_STD						),			1,			1						)
														SCREAMING_SNAKE_CASE__       :      str     =						torch.tensor(cfg.MODEL.PIXEL_MEAN						).to(self.device						).view(len(cfg.MODEL.PIXEL_STD						),			1,			1						)
														SCREAMING_SNAKE_CASE__       :      Optional[Any]     =						lambda _UpperCAmelCase						: (x - self.pixel_mean) / self.pixel_std






							def  A_     ( self     :					str,			_UpperCAmelCase     :					Union[str, Any]						)				->			Union[str, Any]:





														"""simple docstring"""
														SCREAMING_SNAKE_CASE__       :      Tuple     =						tuple(max(_UpperCAmelCase						) for s in zip(*[img.shape for img in images]						)						)
														SCREAMING_SNAKE_CASE__       :      Optional[Any]     =						[im.shape[-2:] for im in images]
														SCREAMING_SNAKE_CASE__       :      Union[str, Any]     =						[
														    nn.functional.pad(
														        _UpperCAmelCase,			[0, max_size[-1] - size[1], 0, max_size[-2] - size[0]],			value=self.pad_value,			)
														    for size, im in zip(_UpperCAmelCase,			_UpperCAmelCase						)
														]

														return torch.stack(_UpperCAmelCase						), torch.tensor(_UpperCAmelCase						)






							def __call__( self     :					Any,			_UpperCAmelCase     :					Dict,			_UpperCAmelCase     :					List[str]=False						)				->			Optional[Any]:





														"""simple docstring"""
														with torch.no_grad():
																					if not isinstance(_UpperCAmelCase,			_UpperCAmelCase						):
																												SCREAMING_SNAKE_CASE__       :      str     =						[images]
																					if single_image:
																												assert len(_UpperCAmelCase						) == 1
																					for i in range(len(_UpperCAmelCase						)						):
																												if isinstance(images[i],			torch.Tensor						):
																																			images.insert(_UpperCAmelCase,			images.pop(_UpperCAmelCase						).to(self.device						).float()						)
																												elif not isinstance(images[i],			torch.Tensor						):
																																			images.insert(
																																			    _UpperCAmelCase,			torch.as_tensor(img_tensorize(images.pop(_UpperCAmelCase						),			input_format=self.input_format						)						)
																																			    .to(self.device						)
																																			    .float(),			)
            # resize smallest edge
																					SCREAMING_SNAKE_CASE__       :      List[str]     =						torch.tensor([im.shape[:2] for im in images]						)
																					SCREAMING_SNAKE_CASE__       :      Tuple     =						self.aug(_UpperCAmelCase						)
																					# transpose images and convert to torch tensors
																					# images = [torch.as_tensor(i.astype("float32")).permute(2, 0, 1).to(self.device) for i in images]
																					# now normalize before pad to avoid useless arithmetic
																					SCREAMING_SNAKE_CASE__       :      List[Any]     =						[self.normalizer(_UpperCAmelCase						) for x in images]
																					# now pad them to do the following operations
																					SCREAMING_SNAKE_CASE__   ,SCREAMING_SNAKE_CASE__       :      List[str]     =						self.pad(_UpperCAmelCase						)
																					# Normalize

																					if self.size_divisibility > 0:
																												raise NotImplementedError()
																					# pad
																					SCREAMING_SNAKE_CASE__       :      Union[str, Any]     =						torch.true_divide(_UpperCAmelCase,			_UpperCAmelCase						)
																					if single_image:
																												return images[0], sizes[0], scales_yx[0]
																					else:
																												return images, sizes, scales_yx




def _a (  SCREAMING_SNAKE_CASE__      :  int			,				SCREAMING_SNAKE_CASE__      :  Dict       )	->    Union[str, Any]:

							'''simple docstring'''






							boxes[:, 0::2] *= scale_yx[:, 1]
							boxes[:, 1::2] *= scale_yx[:, 0]
							return boxes







def _a (  SCREAMING_SNAKE_CASE__      :  List[Any]			,				SCREAMING_SNAKE_CASE__      :  Tuple[int, int]       )	->    List[Any]:

							'''simple docstring'''






							assert torch.isfinite(SCREAMING_SNAKE_CASE__       ).all(), "Box tensor contains infinite or NaN!"
							SCREAMING_SNAKE_CASE__   ,SCREAMING_SNAKE_CASE__       :      int     =						box_size
							tensor[:, 0].clamp_(min=0			,				max=SCREAMING_SNAKE_CASE__       )
							tensor[:, 1].clamp_(min=0			,				max=SCREAMING_SNAKE_CASE__       )
							tensor[:, 2].clamp_(min=0			,				max=SCREAMING_SNAKE_CASE__       )
							tensor[:, 3].clamp_(min=0			,				max=SCREAMING_SNAKE_CASE__       )






