"""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

__lowerCamelCase							 =     logging.get_logger(__name__)






def      UpperCAmelCase    (		UpperCamelCase__				,			UpperCamelCase__				,			UpperCamelCase__				,			UpperCamelCase__=None				,			UpperCamelCase__=None ):






							"""simple docstring"""





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

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

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

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

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

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

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

														if is_buffer:
																					A__		   =							new_value
														else:
																					A__		   =							nn.Parameter(UpperCamelCase__				,			requires_grad=old_value.requires_grad )
																					A__		   =							new_value






def      UpperCAmelCase    (		UpperCamelCase__				,			UpperCamelCase__=None				,			UpperCamelCase__=None				,			UpperCamelCase__=None				,			UpperCamelCase__=False ):






							"""simple docstring"""





							for name, module in model.named_children():
														if current_key_name is None:
																					A__		   =							[]
														current_key_name.append(UpperCamelCase__ )

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

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






def      UpperCAmelCase    (		UpperCamelCase__				,			UpperCamelCase__=None				,			UpperCamelCase__=None				,			UpperCamelCase__=None ):






							"""simple docstring"""





							A__		   =							['lm_head'] if modules_to_not_convert is None else modules_to_not_convert
							A__   ,     A__		   =							_replace_with_bnb_linear(
							    UpperCamelCase__				,			UpperCamelCase__				,			UpperCamelCase__				,			UpperCamelCase__ )

							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      UpperCAmelCase    (		*UpperCamelCase__				,			**UpperCamelCase__ ):






							"""simple docstring"""





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






def      UpperCAmelCase    (		*UpperCamelCase__				,			**UpperCamelCase__ ):






							"""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'				,			UpperCamelCase__				,			)
							return set_module_quantized_tensor_to_device(*UpperCamelCase__				,			**UpperCamelCase__ )






def      UpperCAmelCase    (		UpperCamelCase__ ):






							"""simple docstring"""





							A__		   =							deepcopy(UpperCamelCase__ )  # this has 0 cost since it is done inside `init_empty_weights` context manager`
							tied_model.tie_weights()

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

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

							# add last module together with tied weights
							A__		   =							set(UpperCamelCase__ ) - set(UpperCamelCase__ )
							A__		   =							list(set(UpperCamelCase__ ) ) + list(UpperCamelCase__ )

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

							return filtered_module_names






"""simple docstring"""




from io import BytesIO
from typing import List, Union

import requests

from ..utils import add_end_docstrings, is_decord_available, is_torch_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_decord_available():
				import numpy as np
				from decord import VideoReader


if is_torch_available():
				from ..models.auto.modeling_auto import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING

__lowerCamelCase							 =     logging.get_logger(__name__)
@add_end_docstrings(__A       )
class 			UpperCamelCase__(	__A       ):





						def __init__(				self						,*__UpperCAmelCase						,**__UpperCAmelCase   )    -> Tuple:
													super().__init__(*__UpperCAmelCase						,**__UpperCAmelCase   )
													requires_backends(self						,'decord'   )
													self.check_model_type(__UpperCAmelCase   )





						def  snake_case__		(				self						,__UpperCAmelCase=None						,__UpperCAmelCase=None						,__UpperCAmelCase=None   )    -> int:
													A__		   =							{}
													if frame_sampling_rate is not None:
																				A__		   =							frame_sampling_rate
													if num_frames is not None:
																				A__		   =							num_frames

													A__		   =							{}
													if top_k is not None:
																				A__		   =							top_k
													return preprocess_params, {}, postprocess_params





						def __call__(				self						,__UpperCAmelCase						,**__UpperCAmelCase   )    -> Dict:
													return super().__call__(__UpperCAmelCase						,**__UpperCAmelCase   )





						def  snake_case__		(				self						,__UpperCAmelCase						,__UpperCAmelCase=None						,__UpperCAmelCase=1   )    -> Union[str, Any]:
													if num_frames is None:
																				A__		   =							self.model.config.num_frames

													if video.startswith('http://'   ) or video.startswith('https://'   ):
																				A__		   =							BytesIO(requests.get(__UpperCAmelCase   ).content   )

													A__		   =							VideoReader(__UpperCAmelCase   )
													videoreader.seek(0   )

													A__		   =							0
													A__		   =							num_frames * frame_sampling_rate - 1
													A__		   =							np.linspace(__UpperCAmelCase						,__UpperCAmelCase						,num=__UpperCAmelCase						,dtype=np.intaa   )

													A__		   =							videoreader.get_batch(__UpperCAmelCase   ).asnumpy()
													A__		   =							list(__UpperCAmelCase   )

													A__		   =							self.image_processor(__UpperCAmelCase						,return_tensors=self.framework   )
													return model_inputs





						def  snake_case__		(				self						,__UpperCAmelCase   )    -> Dict:
													A__		   =							self.model(**__UpperCAmelCase   )
													return model_outputs





						def  snake_case__		(				self						,__UpperCAmelCase						,__UpperCAmelCase=5   )    -> Union[str, Any]:
													if top_k > self.model.config.num_labels:
																				A__		   =							self.model.config.num_labels

													if self.framework == "pt":
																				A__		   =							model_outputs.logits.softmax(-1   )[0]
																				A__   ,     A__		   =							probs.topk(__UpperCAmelCase   )
													else:
																				raise ValueError(f'''Unsupported framework: {self.framework}'''   )

													A__		   =							scores.tolist()
													A__		   =							ids.tolist()
													return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(__UpperCAmelCase						,__UpperCAmelCase   )]





