



"""simple docstring"""


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

snake_case_    					=	logging.get_logger(__name__)

def 			_lowerCAmelCase				(       lowercase_		,     lowercase_		,     lowercase_		,     lowercase_=None		,     lowercase_=None						):
			# Recurse if needed
			if "." in tensor_name:
						UpperCAmelCase					=			tensor_name.split('.'						)
						for split in splits[:-1]:
									UpperCAmelCase					=			getattr(lowercase_		,     lowercase_						)
									if new_module is None:
												raise ValueError(F"""{module} has no attribute {split}."""						)
									UpperCAmelCase					=			new_module
						UpperCAmelCase					=			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}."""						)
			UpperCAmelCase					=			tensor_name in module._buffers
			UpperCAmelCase					=			getattr(lowercase_		,     lowercase_						)

			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}."""						)

			UpperCAmelCase					=			False
			UpperCAmelCase					=			False
			if is_buffer or not is_bitsandbytes_available():
						UpperCAmelCase					=			False
						UpperCAmelCase					=			False
			else:
						UpperCAmelCase					=			hasattr(bnb.nn		,     'Params4bit'						) and isinstance(module._parameters[tensor_name]		,     bnb.nn.Paramsabit						)
						UpperCAmelCase					=			isinstance(module._parameters[tensor_name]		,     bnb.nn.IntaParams						)

			if is_abit or is_abit:
						UpperCAmelCase					=			module._parameters[tensor_name]
						if param.device.type != "cuda":
									if value is None:
												UpperCAmelCase					=			old_value.to(lowercase_						)
									elif isinstance(lowercase_		,     torch.Tensor						):
												UpperCAmelCase					=			value.to('cpu'						)
												if value.dtype == torch.inta:
															UpperCAmelCase					=			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:
												UpperCAmelCase					=			torch.tensor(lowercase_		,     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		,     lowercase_						) and fpaa_statistics is None:
												UpperCAmelCase					=			new_value.T

									UpperCAmelCase					=			old_value.__dict__
									if is_abit:
												UpperCAmelCase					=			bnb.nn.IntaParams(lowercase_		,     requires_grad=lowercase_		,     **lowercase_						).to(lowercase_						)
									elif is_abit:
												UpperCAmelCase					=			bnb.nn.Paramsabit(lowercase_		,     requires_grad=lowercase_		,     **lowercase_						).to(lowercase_						)

									UpperCAmelCase					=			new_value
									if fpaa_statistics is not None:
												setattr(module.weight		,     'SCB'		,     fpaa_statistics.to(lowercase_						)						)

			else:
						if value is None:
									UpperCAmelCase					=			old_value.to(lowercase_						)
						elif isinstance(lowercase_		,     torch.Tensor						):
									UpperCAmelCase					=			value.to(lowercase_						)
						else:
									UpperCAmelCase					=			torch.tensor(lowercase_		,     device=lowercase_						)

						if is_buffer:
									UpperCAmelCase					=			new_value
						else:
									UpperCAmelCase					=			nn.Parameter(lowercase_		,     requires_grad=old_value.requires_grad						)
									UpperCAmelCase					=			new_value

def 			_lowerCAmelCase				(       lowercase_		,     lowercase_=None		,     lowercase_=None		,     lowercase_=None		,     lowercase_=False						):
			for name, module in model.named_children():
						if current_key_name is None:
									UpperCAmelCase					=			[]
						current_key_name.append(lowercase_						)

						if (isinstance(lowercase_		,     nn.Linear						) or isinstance(lowercase_		,     lowercase_						)) 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(lowercase_						) for key in modules_to_not_convert						):
												with init_empty_weights():
															if isinstance(lowercase_		,     lowercase_						):
																		UpperCAmelCase				,   UpperCAmelCase					=			module.weight.shape
															else:
																		UpperCAmelCase					=			module.in_features
																		UpperCAmelCase					=			module.out_features

															if quantization_config.quantization_method() == "llm_int8":
																		UpperCAmelCase					=			bnb.nn.LinearabitLt(
																		    lowercase_		,     lowercase_		,     module.bias is not None		,     has_fpaa_weights=quantization_config.llm_inta_has_fpaa_weight		,     threshold=quantization_config.llm_inta_threshold		,     )
																		UpperCAmelCase					=			True
															else:
																		if (
																		    quantization_config.llm_inta_skip_modules is not None
																		    and name in quantization_config.llm_inta_skip_modules
																		):
																					pass
																		else:
																					UpperCAmelCase					=			bnb.nn.Linearabit(
																					    lowercase_		,     lowercase_		,     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		,     )
																					UpperCAmelCase					=			True
                    # Store the module class in case we need to transpose the weight later
															UpperCAmelCase					=			type(lowercase_						)
															# Force requires grad to False to avoid unexpected errors
															model._modules[name].requires_grad_(lowercase_						)
						if len(list(module.children()						)						) > 0:
									UpperCAmelCase				,   UpperCAmelCase					=			_replace_with_bnb_linear(
									    lowercase_		,     lowercase_		,     lowercase_		,     lowercase_		,     has_been_replaced=lowercase_		,     )
						# Remove the last key for recursion
						current_key_name.pop(-1						)
			return model, has_been_replaced

def 			_lowerCAmelCase				(       lowercase_		,     lowercase_=None		,     lowercase_=None		,     lowercase_=None						):
			UpperCAmelCase					=			['lm_head'] if modules_to_not_convert is None else modules_to_not_convert
			UpperCAmelCase				,   UpperCAmelCase					=			_replace_with_bnb_linear(
			    lowercase_		,     lowercase_		,     lowercase_		,     lowercase_						)

			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 			_lowerCAmelCase				(       *lowercase_		,     **lowercase_						):
			warnings.warn(
			    '`replace_8bit_linear` will be deprecated in a future version, please use `replace_with_bnb_linear` instead'		,     lowercase_		,     )
			return replace_with_bnb_linear(*lowercase_		,     **lowercase_						)

def 			_lowerCAmelCase				(       *lowercase_		,     **lowercase_						):
			warnings.warn(
			    '`set_module_8bit_tensor_to_device` will be deprecated in a future version, please use `set_module_quantized_tensor_to_device` instead'		,     lowercase_		,     )
			return set_module_quantized_tensor_to_device(*lowercase_		,     **lowercase_						)






def 			_lowerCAmelCase				(       lowercase_						):
			UpperCAmelCase					=			deepcopy(lowercase_						)  # this has 0 cost since it is done inside `init_empty_weights` context manager`
			tied_model.tie_weights()

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

			# Check if it is a base model
			UpperCAmelCase					=			not hasattr(lowercase_		,     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
			UpperCAmelCase					=			list(model.named_children()						)
			UpperCAmelCase					=			[list_modules[-1][0]]

			# add last module together with tied weights
			UpperCAmelCase					=			set(lowercase_						) - set(lowercase_						)
			UpperCAmelCase					=			list(set(lowercase_						)						) + list(lowercase_						)

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

			return filtered_module_names










"""simple docstring"""


import logging
import os
import threading
import time


try:
				import warnings
except ImportError:
				snake_case_    					=	None

try:
				import msvcrt
except ImportError:
				snake_case_    					=	None

try:
				import fcntl
except ImportError:
				snake_case_    					=	None


# Backward compatibility
# ------------------------------------------------
try:
				TimeoutError
except NameError:
				snake_case_    					=	OSError


# Data
# ------------------------------------------------
snake_case_    					=	[
    """Timeout""",
    """BaseFileLock""",
    """WindowsFileLock""",
    """UnixFileLock""",
    """SoftFileLock""",
    """FileLock""",
]

snake_case_    					=	"""3.0.12"""


snake_case_    					=	None

def 			_lowerCAmelCase				(       ):
						global _logger
						UpperCAmelCase					=			_logger or logging.getLogger(__name__						)
						return _logger







class 				A_      (					SCREAMING_SNAKE_CASE_    ):



						"""simple docstring"""






						def __init__( self			:int       ,  lowercase_			:List[Any] )    ->      Optional[int]:
												UpperCAmelCase					=			lock_file
												return None






						def __str__( self			:Dict )    ->      Optional[int]:
												UpperCAmelCase					=			f"""The file lock '{self.lock_file}' could not be acquired."""
												return temp







class 				A_      :



						"""simple docstring"""






						def __init__( self			:List[Any]       ,  lowercase_			:Dict )    ->      Dict:
												UpperCAmelCase					=			lock
												return None






						def __enter__( self			:Optional[int] )    ->      Optional[int]:
												return self.lock






						def __exit__( self			:Optional[int]       ,  lowercase_			:Tuple       ,  lowercase_			:Any       ,  lowercase_			:Union[str, Any] )    ->      List[Any]:
												self.lock.release()
												return None







class 				A_      :



						"""simple docstring"""






						def __init__( self			:List[str]       ,  lowercase_			:str       ,  lowercase_			:str=-1       ,  lowercase_			:List[Any]=None )    ->      Dict:
												UpperCAmelCase					=			max_filename_length if max_filename_length is not None else 2_55
												# Hash the filename if it's too long
												UpperCAmelCase					=			self.hash_filename_if_too_long(lowercase_       ,  lowercase_ )
												# The path to the lock file.
												UpperCAmelCase					=			lock_file

												# The file descriptor for the *_lock_file* as it is returned by the
												# os.open() function.
												# This file lock is only NOT None, if the object currently holds the
												# lock.
												UpperCAmelCase					=			None

												# The default timeout value.
												UpperCAmelCase					=			timeout

												# We use this lock primarily for the lock counter.
												UpperCAmelCase					=			threading.Lock()

												# The lock counter is used for implementing the nested locking
												# mechanism. Whenever the lock is acquired, the counter is increased and
												# the lock is only released, when this value is 0 again.
												UpperCAmelCase					=			0
												return None






						@property
						def      UpperCAmelCase__					( self			:Optional[Any] )    ->      int:
												return self._lock_file






						@property
						def      UpperCAmelCase__					( self			:Union[str, Any] )    ->      Any:
												return self._timeout






						@timeout.setter
						def      UpperCAmelCase__					( self			:Union[str, Any]       ,  lowercase_			:Any )    ->      Tuple:
												UpperCAmelCase					=			float(lowercase_ )
												return None






						def      UpperCAmelCase__					( self			:Any )    ->      List[str]:
												raise NotImplementedError()






						def      UpperCAmelCase__					( self			:Union[str, Any] )    ->      List[str]:
												raise NotImplementedError()






						@property
						def      UpperCAmelCase__					( self			:Optional[Any] )    ->      Union[str, Any]:
												return self._lock_file_fd is not None






						def      UpperCAmelCase__					( self			:Union[str, Any]       ,  lowercase_			:Dict=None       ,  lowercase_			:int=0.05 )    ->      Union[str, Any]:
												# Use the default timeout, if no timeout is provided.
												if timeout is None:
																		UpperCAmelCase					=			self.timeout

												# Increment the number right at the beginning.
												# We can still undo it, if something fails.
												with self._thread_lock:
																		self._lock_counter += 1

												UpperCAmelCase					=			id(self )
												UpperCAmelCase					=			self._lock_file
												UpperCAmelCase					=			time.time()
												try:
																		while True:
																								with self._thread_lock:
																														if not self.is_locked:
																																				logger().debug(f"""Attempting to acquire lock {lock_id} on {lock_filename}""" )
																																				self._acquire()

																								if self.is_locked:
																														logger().debug(f"""Lock {lock_id} acquired on {lock_filename}""" )
																														break
																								elif timeout >= 0 and time.time() - start_time > timeout:
																														logger().debug(f"""Timeout on acquiring lock {lock_id} on {lock_filename}""" )
																														raise Timeout(self._lock_file )
																								else:
																														logger().debug(
																														    f"""Lock {lock_id} not acquired on {lock_filename}, waiting {poll_intervall} seconds ...""" )
																														time.sleep(lowercase_ )
												except:  # noqa
																		# Something did go wrong, so decrement the counter.
																		with self._thread_lock:
																								UpperCAmelCase					=			max(0       ,  self._lock_counter - 1 )

																		raise
												return _Acquire_ReturnProxy(lock=self )






						def      UpperCAmelCase__					( self			:Optional[int]       ,  lowercase_			:List[str]=False )    ->      int:
												with self._thread_lock:
																		if self.is_locked:
																								self._lock_counter -= 1

																								if self._lock_counter == 0 or force:
																														UpperCAmelCase					=			id(self )
																														UpperCAmelCase					=			self._lock_file

																														logger().debug(f"""Attempting to release lock {lock_id} on {lock_filename}""" )
																														self._release()
																														UpperCAmelCase					=			0
																														logger().debug(f"""Lock {lock_id} released on {lock_filename}""" )

												return None






						def __enter__( self			:Any )    ->      Tuple:
												self.acquire()
												return self






						def __exit__( self			:Dict       ,  lowercase_			:Optional[int]       ,  lowercase_			:Union[str, Any]       ,  lowercase_			:Union[str, Any] )    ->      Any:
												self.release()
												return None






						def __del__( self			:Any )    ->      int:
												self.release(force=lowercase_ )
												return None






						def      UpperCAmelCase__					( self			:List[str]       ,  lowercase_			:str       ,  lowercase_			:int )    ->      str:
												UpperCAmelCase					=			os.path.basename(lowercase_ )
												if len(lowercase_ ) > max_length and max_length > 0:
																		UpperCAmelCase					=			os.path.dirname(lowercase_ )
																		UpperCAmelCase					=			str(hash(lowercase_ ) )
																		UpperCAmelCase					=			filename[: max_length - len(lowercase_ ) - 8] + '...' + hashed_filename + '.lock'
																		return os.path.join(lowercase_       ,  lowercase_ )
												else:
																		return path







class 				A_      (					SCREAMING_SNAKE_CASE_    ):



						"""simple docstring"""






						def __init__( self			:Tuple       ,  lowercase_			:Tuple       ,  lowercase_			:List[Any]=-1       ,  lowercase_			:Union[str, Any]=None )    ->      Optional[Any]:
												from .file_utils import relative_to_absolute_path

												super().__init__(lowercase_       ,  timeout=lowercase_       ,  max_filename_length=lowercase_ )
												UpperCAmelCase					=			'\\\\?\\' + relative_to_absolute_path(self.lock_file )






						def      UpperCAmelCase__					( self			:List[str] )    ->      str:
												UpperCAmelCase					=			os.O_RDWR | os.O_CREAT | os.O_TRUNC

												try:
																		UpperCAmelCase					=			os.open(self._lock_file       ,  lowercase_ )
												except OSError:
																		pass
												else:
																		try:
																								msvcrt.locking(lowercase_       ,  msvcrt.LK_NBLCK       ,  1 )
																		except OSError:
																								os.close(lowercase_ )
																		else:
																								UpperCAmelCase					=			fd
												return None






						def      UpperCAmelCase__					( self			:int )    ->      Union[str, Any]:
												UpperCAmelCase					=			self._lock_file_fd
												UpperCAmelCase					=			None
												msvcrt.locking(lowercase_       ,  msvcrt.LK_UNLCK       ,  1 )
												os.close(lowercase_ )

												try:
																		os.remove(self._lock_file )
												# Probably another instance of the application
												# that acquired the file lock.
												except OSError:
																		pass
												return None







class 				A_      (					SCREAMING_SNAKE_CASE_    ):



						"""simple docstring"""






						def __init__( self			:Optional[Any]       ,  lowercase_			:str       ,  lowercase_			:List[str]=-1       ,  lowercase_			:str=None )    ->      str:
												UpperCAmelCase					=			os.statvfs(os.path.dirname(lowercase_ ) ).f_namemax
												super().__init__(lowercase_       ,  timeout=lowercase_       ,  max_filename_length=lowercase_ )






						def      UpperCAmelCase__					( self			:str )    ->      Dict:
												UpperCAmelCase					=			os.O_RDWR | os.O_CREAT | os.O_TRUNC
												UpperCAmelCase					=			os.open(self._lock_file       ,  lowercase_ )

												try:
																		fcntl.flock(lowercase_       ,  fcntl.LOCK_EX | fcntl.LOCK_NB )
												except OSError:
																		os.close(lowercase_ )
												else:
																		UpperCAmelCase					=			fd
												return None






						def      UpperCAmelCase__					( self			:List[Any] )    ->      List[Any]:
												# Do not remove the lockfile:
												#
												#   https://github.com/benediktschmitt/py-filelock/issues/31
												#   https://stackoverflow.com/questions/17708885/flock-removing-locked-file-without-race-condition
												UpperCAmelCase					=			self._lock_file_fd
												UpperCAmelCase					=			None
												fcntl.flock(lowercase_       ,  fcntl.LOCK_UN )
												os.close(lowercase_ )
												return None







class 				A_      (					SCREAMING_SNAKE_CASE_    ):



						"""simple docstring"""






						def      UpperCAmelCase__					( self			:List[Any] )    ->      List[str]:
												UpperCAmelCase					=			os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
												try:
																		UpperCAmelCase					=			os.open(self._lock_file       ,  lowercase_ )
												except OSError:
																		pass
												else:
																		UpperCAmelCase					=			fd
												return None






						def      UpperCAmelCase__					( self			:Optional[int] )    ->      List[Any]:
												os.close(self._lock_file_fd )
												UpperCAmelCase					=			None

												try:
																		os.remove(self._lock_file )
												# The file is already deleted and that's what we want.
												except OSError:
																		pass
												return None


snake_case_    					=	None

if msvcrt:
				snake_case_    					=	WindowsFileLock
elif fcntl:
				snake_case_    					=	UnixFileLock
else:
				snake_case_    					=	SoftFileLock

				if warnings is not None:
								warnings.warn("""only soft file lock is available""")





