





"""simple docstring"""






import os
import tempfile
import unittest

from transformers import NezhaConfig, is_torch_available
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, require_torch_gpu, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
		import torch

		from transformers import (
		    MODEL_FOR_PRETRAINING_MAPPING,
		    NezhaForMaskedLM,
		    NezhaForMultipleChoice,
		    NezhaForNextSentencePrediction,
		    NezhaForPreTraining,
		    NezhaForQuestionAnswering,
		    NezhaForSequenceClassification,
		    NezhaForTokenClassification,
		    NezhaModel,
		)
		from transformers.models.nezha.modeling_nezha import NEZHA_PRETRAINED_MODEL_ARCHIVE_LIST




class 					__lowercase      :







					'''simple docstring'''




					def __init__(    self						,      _UpperCAmelCase						,      _UpperCAmelCase=13						,      _UpperCAmelCase=7						,      _UpperCAmelCase=True						,      _UpperCAmelCase=True						,      _UpperCAmelCase=True						,      _UpperCAmelCase=True						,      _UpperCAmelCase=99						,      _UpperCAmelCase=32						,      _UpperCAmelCase=5						,      _UpperCAmelCase=4						,      _UpperCAmelCase=37						,      _UpperCAmelCase="gelu"						,      _UpperCAmelCase=0.1						,      _UpperCAmelCase=0.1						,      _UpperCAmelCase=128						,      _UpperCAmelCase=32						,      _UpperCAmelCase=16						,      _UpperCAmelCase=2						,      _UpperCAmelCase=0.0_2						,      _UpperCAmelCase=3						,      _UpperCAmelCase=4						,      _UpperCAmelCase=None						,      ):
									__a			:    Optional[Any]      							=	parent
									__a			:    Optional[int]      							=	batch_size
									__a			:    Tuple      							=	seq_length
									__a			:    List[Any]      							=	is_training
									__a			:    Optional[int]      							=	use_input_mask
									__a			:    int      							=	use_token_type_ids
									__a			:    Optional[int]      							=	use_labels
									__a			:    Any      							=	vocab_size
									__a			:    Union[str, Any]      							=	hidden_size
									__a			:    List[Any]      							=	num_hidden_layers
									__a			:    Optional[Any]      							=	num_attention_heads
									__a			:    List[str]      							=	intermediate_size
									__a			:    Dict      							=	hidden_act
									__a			:    List[str]      							=	hidden_dropout_prob
									__a			:    Dict      							=	attention_probs_dropout_prob
									__a			:    Any      							=	max_position_embeddings
									__a			:    Any      							=	type_vocab_size
									__a			:    Union[str, Any]      							=	type_sequence_label_size
									__a			:    Optional[Any]      							=	initializer_range
									__a			:    List[str]      							=	num_labels
									__a			:    Tuple      							=	num_choices
									__a			:    str      							=	scope







					def 			_lowerCamelCase		(    self  ):
									__a			:    Optional[Any]      							=	ids_tensor([self.batch_size, self.seq_length]						,      self.vocab_size  )

									__a			:    List[str]      							=	None
									if self.use_input_mask:
													__a			:    List[str]      							=	random_attention_mask([self.batch_size, self.seq_length]  )

									__a			:    List[str]      							=	None
									if self.use_token_type_ids:
													__a			:    Union[str, Any]      							=	ids_tensor([self.batch_size, self.seq_length]						,      self.type_vocab_size  )

									__a			:    str      							=	None
									__a			:    List[str]      							=	None
									__a			:    Union[str, Any]      							=	None
									if self.use_labels:
													__a			:    List[Any]      							=	ids_tensor([self.batch_size]						,      self.type_sequence_label_size  )
													__a			:    Dict      							=	ids_tensor([self.batch_size, self.seq_length]						,      self.num_labels  )
													__a			:    str      							=	ids_tensor([self.batch_size]						,      self.num_choices  )

									__a			:    List[str]      							=	self.get_config()

									return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels







					def 			_lowerCamelCase		(    self  ):
									return NezhaConfig(
									    vocab_size=self.vocab_size						,      hidden_size=self.hidden_size						,      num_hidden_layers=self.num_hidden_layers						,      num_attention_heads=self.num_attention_heads						,      intermediate_size=self.intermediate_size						,      hidden_act=self.hidden_act						,      hidden_dropout_prob=self.hidden_dropout_prob						,      attention_probs_dropout_prob=self.attention_probs_dropout_prob						,      max_position_embeddings=self.max_position_embeddings						,      type_vocab_size=self.type_vocab_size						,      is_decoder=_UpperCAmelCase						,      initializer_range=self.initializer_range						,      )







					def 			_lowerCamelCase		(    self  ):
									(
									    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    
									)			:    Any      							=	self.prepare_config_and_inputs()

									__a			:    str      							=	True
									__a			:    str      							=	floats_tensor([self.batch_size, self.seq_length, self.hidden_size]  )
									__a			:    int      							=	ids_tensor([self.batch_size, self.seq_length]						,      vocab_size=2  )

									return (
									    config,
									    input_ids,
									    token_type_ids,
									    input_mask,
									    sequence_labels,
									    token_labels,
									    choice_labels,
									    encoder_hidden_states,
									    encoder_attention_mask,
									)







					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  ):
									__a			:    Any      							=	NezhaModel(config=_UpperCAmelCase  )
									model.to(_UpperCAmelCase  )
									model.eval()
									__a			:    List[Any]      							=	model(_UpperCAmelCase						,      attention_mask=_UpperCAmelCase						,      token_type_ids=_UpperCAmelCase  )
									__a			:    str      							=	model(_UpperCAmelCase						,      token_type_ids=_UpperCAmelCase  )
									__a			:    int      							=	model(_UpperCAmelCase  )
									self.parent.assertEqual(result.last_hidden_state.shape						,      (self.batch_size, self.seq_length, self.hidden_size)  )
									self.parent.assertEqual(result.pooler_output.shape						,      (self.batch_size, self.hidden_size)  )







					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      ):
									__a			:    Dict      							=	True
									__a			:    str      							=	NezhaModel(_UpperCAmelCase  )
									model.to(_UpperCAmelCase  )
									model.eval()
									__a			:    List[Any]      							=	model(
									    _UpperCAmelCase						,      attention_mask=_UpperCAmelCase						,      token_type_ids=_UpperCAmelCase						,      encoder_hidden_states=_UpperCAmelCase						,      encoder_attention_mask=_UpperCAmelCase						,      )
									__a			:    Union[str, Any]      							=	model(
									    _UpperCAmelCase						,      attention_mask=_UpperCAmelCase						,      token_type_ids=_UpperCAmelCase						,      encoder_hidden_states=_UpperCAmelCase						,      )
									__a			:    Optional[int]      							=	model(_UpperCAmelCase						,      attention_mask=_UpperCAmelCase						,      token_type_ids=_UpperCAmelCase  )
									self.parent.assertEqual(result.last_hidden_state.shape						,      (self.batch_size, self.seq_length, self.hidden_size)  )
									self.parent.assertEqual(result.pooler_output.shape						,      (self.batch_size, self.hidden_size)  )







					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  ):
									__a			:    List[str]      							=	NezhaForMaskedLM(config=_UpperCAmelCase  )
									model.to(_UpperCAmelCase  )
									model.eval()
									__a			:    List[str]      							=	model(_UpperCAmelCase						,      attention_mask=_UpperCAmelCase						,      token_type_ids=_UpperCAmelCase						,      labels=_UpperCAmelCase  )
									self.parent.assertEqual(result.logits.shape						,      (self.batch_size, self.seq_length, self.vocab_size)  )







					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  ):
									__a			:    List[Any]      							=	NezhaForNextSentencePrediction(config=_UpperCAmelCase  )
									model.to(_UpperCAmelCase  )
									model.eval()
									__a			:    Tuple      							=	model(
									    _UpperCAmelCase						,      attention_mask=_UpperCAmelCase						,      token_type_ids=_UpperCAmelCase						,      labels=_UpperCAmelCase						,      )
									self.parent.assertEqual(result.logits.shape						,      (self.batch_size, 2)  )







					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  ):
									__a			:    List[Any]      							=	NezhaForPreTraining(config=_UpperCAmelCase  )
									model.to(_UpperCAmelCase  )
									model.eval()
									__a			:    str      							=	model(
									    _UpperCAmelCase						,      attention_mask=_UpperCAmelCase						,      token_type_ids=_UpperCAmelCase						,      labels=_UpperCAmelCase						,      next_sentence_label=_UpperCAmelCase						,      )
									self.parent.assertEqual(result.prediction_logits.shape						,      (self.batch_size, self.seq_length, self.vocab_size)  )
									self.parent.assertEqual(result.seq_relationship_logits.shape						,      (self.batch_size, 2)  )







					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  ):
									__a			:    Dict      							=	NezhaForQuestionAnswering(config=_UpperCAmelCase  )
									model.to(_UpperCAmelCase  )
									model.eval()
									__a			:    Dict      							=	model(
									    _UpperCAmelCase						,      attention_mask=_UpperCAmelCase						,      token_type_ids=_UpperCAmelCase						,      start_positions=_UpperCAmelCase						,      end_positions=_UpperCAmelCase						,      )
									self.parent.assertEqual(result.start_logits.shape						,      (self.batch_size, self.seq_length)  )
									self.parent.assertEqual(result.end_logits.shape						,      (self.batch_size, self.seq_length)  )







					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  ):
									__a			:    Dict      							=	self.num_labels
									__a			:    Any      							=	NezhaForSequenceClassification(_UpperCAmelCase  )
									model.to(_UpperCAmelCase  )
									model.eval()
									__a			:    Optional[Any]      							=	model(_UpperCAmelCase						,      attention_mask=_UpperCAmelCase						,      token_type_ids=_UpperCAmelCase						,      labels=_UpperCAmelCase  )
									self.parent.assertEqual(result.logits.shape						,      (self.batch_size, self.num_labels)  )







					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  ):
									__a			:    Dict      							=	self.num_labels
									__a			:    Tuple      							=	NezhaForTokenClassification(config=_UpperCAmelCase  )
									model.to(_UpperCAmelCase  )
									model.eval()
									__a			:    Tuple      							=	model(_UpperCAmelCase						,      attention_mask=_UpperCAmelCase						,      token_type_ids=_UpperCAmelCase						,      labels=_UpperCAmelCase  )
									self.parent.assertEqual(result.logits.shape						,      (self.batch_size, self.seq_length, self.num_labels)  )







					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase  ):
									__a			:    Union[str, Any]      							=	self.num_choices
									__a			:    Optional[int]      							=	NezhaForMultipleChoice(config=_UpperCAmelCase  )
									model.to(_UpperCAmelCase  )
									model.eval()
									__a			:    Optional[int]      							=	input_ids.unsqueeze(1  ).expand(-1						,      self.num_choices						,      -1  ).contiguous()
									__a			:    str      							=	token_type_ids.unsqueeze(1  ).expand(-1						,      self.num_choices						,      -1  ).contiguous()
									__a			:    Any      							=	input_mask.unsqueeze(1  ).expand(-1						,      self.num_choices						,      -1  ).contiguous()
									__a			:    List[str]      							=	model(
									    _UpperCAmelCase						,      attention_mask=_UpperCAmelCase						,      token_type_ids=_UpperCAmelCase						,      labels=_UpperCAmelCase						,      )
									self.parent.assertEqual(result.logits.shape						,      (self.batch_size, self.num_choices)  )







					def 			_lowerCamelCase		(    self  ):
									__a			:    str      							=	self.prepare_config_and_inputs()
									(
									    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    
									)			:    Optional[int]      							=	config_and_inputs
									__a			:    Optional[Any]      							=	{'''input_ids''': input_ids, '''token_type_ids''': token_type_ids, '''attention_mask''': input_mask}
									return config, inputs_dict






@require_torch
class 					__lowercase      (      _UpperCamelCase ,   _UpperCamelCase ,   _UpperCamelCase ,   unittest.TestCase				):







					'''simple docstring'''




					__lowerCAmelCase	=	(
					    (
					        NezhaModel,
					        NezhaForMaskedLM,
					        NezhaForMultipleChoice,
					        NezhaForNextSentencePrediction,
					        NezhaForPreTraining,
					        NezhaForQuestionAnswering,
					        NezhaForSequenceClassification,
					        NezhaForTokenClassification,
					    )
					    if is_torch_available()
					    else ()
					)
					__lowerCAmelCase	=	(
					    {
					        '''feature-extraction''': NezhaModel,
					        '''fill-mask''': NezhaForMaskedLM,
					        '''question-answering''': NezhaForQuestionAnswering,
					        '''text-classification''': NezhaForSequenceClassification,
					        '''token-classification''': NezhaForTokenClassification,
					        '''zero-shot''': NezhaForSequenceClassification,
					    }
					    if is_torch_available()
					    else {}
					)
					__lowerCAmelCase	=	True







					def 			_lowerCamelCase		(    self						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase=False  ):
									__a			:    Dict      							=	super()._prepare_for_class(_UpperCAmelCase						,      _UpperCAmelCase						,      return_labels=_UpperCAmelCase  )

									if return_labels:
													if model_class in get_values(_UpperCAmelCase  ):
																	__a			:    List[str]      							=	torch.zeros(
																	    (self.model_tester.batch_size, self.model_tester.seq_length)						,      dtype=torch.long						,      device=_UpperCAmelCase  )
																	__a			:    Union[str, Any]      							=	torch.zeros(
																	    self.model_tester.batch_size						,      dtype=torch.long						,      device=_UpperCAmelCase  )
									return inputs_dict







					def 			_lowerCamelCase		(    self  ):
									__a			:    List[Any]      							=	NezhaModelTester(self  )
									__a			:    Optional[int]      							=	ConfigTester(self						,      config_class=_UpperCAmelCase						,      hidden_size=37  )







					def 			_lowerCamelCase		(    self  ):
									self.config_tester.run_common_tests()







					def 			_lowerCamelCase		(    self  ):
									__a			:    int      							=	self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_model(*_UpperCAmelCase  )







					def 			_lowerCamelCase		(    self  ):
									__a			:    List[str]      							=	self.model_tester.prepare_config_and_inputs_for_decoder()
									self.model_tester.create_and_check_model_as_decoder(*_UpperCAmelCase  )







					def 			_lowerCamelCase		(    self  ):
									# This regression test was failing with PyTorch < 1.3
									(
									    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    (
									    __a
									)							,    
									)			:    int      							=	self.model_tester.prepare_config_and_inputs_for_decoder()

									__a			:    List[Any]      							=	None

									self.model_tester.create_and_check_model_as_decoder(
									    _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      )







					def 			_lowerCamelCase		(    self  ):
									__a			:    List[Any]      							=	self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_masked_lm(*_UpperCAmelCase  )







					def 			_lowerCamelCase		(    self  ):
									__a			:    Any      							=	self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_multiple_choice(*_UpperCAmelCase  )







					def 			_lowerCamelCase		(    self  ):
									__a			:    Dict      							=	self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_next_sequence_prediction(*_UpperCAmelCase  )







					def 			_lowerCamelCase		(    self  ):
									__a			:    Optional[Any]      							=	self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_pretraining(*_UpperCAmelCase  )







					def 			_lowerCamelCase		(    self  ):
									__a			:    str      							=	self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_question_answering(*_UpperCAmelCase  )







					def 			_lowerCamelCase		(    self  ):
									__a			:    Any      							=	self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_sequence_classification(*_UpperCAmelCase  )







					def 			_lowerCamelCase		(    self  ):
									__a			:    str      							=	self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_token_classification(*_UpperCAmelCase  )







					@slow
					def 			_lowerCamelCase		(    self  ):
									for model_name in NEZHA_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
													__a			:    Tuple      							=	NezhaModel.from_pretrained(_UpperCAmelCase  )
													self.assertIsNotNone(_UpperCAmelCase  )







					@slow
					@require_torch_gpu
					def 			_lowerCamelCase		(    self  ):
									__a							,    __a			:    str      							=	self.model_tester.prepare_config_and_inputs_for_common()
									for model_class in self.all_model_classes:
													# NezhaForMultipleChoice behaves incorrectly in JIT environments.
													if model_class == NezhaForMultipleChoice:
																	return

													__a			:    str      							=	True
													__a			:    str      							=	model_class(config=_UpperCAmelCase  )

													__a			:    List[Any]      							=	self._prepare_for_class(_UpperCAmelCase						,      _UpperCAmelCase  )
													__a			:    int      							=	torch.jit.trace(
													    _UpperCAmelCase						,      (inputs_dict['''input_ids'''].to('''cpu'''  ), inputs_dict['''attention_mask'''].to('''cpu'''  ))  )

													with tempfile.TemporaryDirectory() as tmp:
																	torch.jit.save(_UpperCAmelCase						,      os.path.join(_UpperCAmelCase						,      '''bert.pt'''  )  )
																	__a			:    int      							=	torch.jit.load(os.path.join(_UpperCAmelCase						,      '''bert.pt'''  )						,      map_location=_UpperCAmelCase  )
																	loaded(inputs_dict['''input_ids'''].to(_UpperCAmelCase  )						,      inputs_dict['''attention_mask'''].to(_UpperCAmelCase  )  )






@require_torch
class 					__lowercase      (      unittest.TestCase				):







					'''simple docstring'''




					@slow
					def 			_lowerCamelCase		(    self  ):
									__a			:    str      							=	NezhaModel.from_pretrained('''sijunhe/nezha-cn-base'''  )
									__a			:    Dict      							=	torch.tensor([[0, 1, 2, 3, 4, 5]]  )
									__a			:    Optional[Any]      							=	torch.tensor([[0, 1, 1, 1, 1, 1]]  )
									with torch.no_grad():
													__a			:    Optional[int]      							=	model(_UpperCAmelCase						,      attention_mask=_UpperCAmelCase  )[0]
									__a			:    Optional[int]      							=	torch.Size((1, 6, 768)  )
									self.assertEqual(output.shape						,      _UpperCAmelCase  )
									__a			:    Optional[int]      							=	torch.tensor([[[0.0_6_8_5, 0.2_4_4_1, 0.1_1_0_2], [0.0_6_0_0, 0.1_9_0_6, 0.1_3_4_9], [0.0_2_2_1, 0.0_8_1_9, 0.0_5_8_6]]]  )

									self.assertTrue(torch.allclose(output[:, 1:4, 1:4]						,      _UpperCAmelCase						,      atol=1e-4  )  )







					@slow
					def 			_lowerCamelCase		(    self  ):
									__a			:    Optional[int]      							=	NezhaForMaskedLM.from_pretrained('''sijunhe/nezha-cn-base'''  )
									__a			:    Optional[Any]      							=	torch.tensor([[0, 1, 2, 3, 4, 5]]  )
									__a			:    Union[str, Any]      							=	torch.tensor([[1, 1, 1, 1, 1, 1]]  )
									with torch.no_grad():
													__a			:    Optional[Any]      							=	model(_UpperCAmelCase						,      attention_mask=_UpperCAmelCase  )[0]
									__a			:    List[Any]      							=	torch.Size((1, 6, 21128)  )
									self.assertEqual(output.shape						,      _UpperCAmelCase  )
									__a			:    List[Any]      							=	torch.tensor(
									    [[-2.7_9_3_9, -1.7_9_0_2, -2.2_1_8_9], [-2.8_5_8_5, -1.8_9_0_8, -2.3_7_2_3], [-2.6_4_9_9, -1.7_7_5_0, -2.2_5_5_8]]  )

									self.assertTrue(torch.allclose(output[:, 1:4, 1:4]						,      _UpperCAmelCase						,      atol=1e-4  )  )
"""simple docstring"""




import logging
import os
import threading
import time


try:
			import warnings
except ImportError:
			__lowerCAmelCase     :				Optional[int]    				=    None

try:
			import msvcrt
except ImportError:
			__lowerCAmelCase     :				List[Any]    				=    None

try:
			import fcntl
except ImportError:
			__lowerCAmelCase     :				Optional[int]    				=    None


# Backward compatibility
# ------------------------------------------------
try:
			TimeoutError
except NameError:
			__lowerCAmelCase     :				int    				=    OSError


# Data
# ------------------------------------------------
__lowerCAmelCase     :				Dict    				=    [
    "Timeout",
    "BaseFileLock",
    "WindowsFileLock",
    "UnixFileLock",
    "SoftFileLock",
    "FileLock",
]

__lowerCAmelCase     :				str    				=    "3.0.12"


__lowerCAmelCase     :				Optional[int]    				=    None







def 				_UpperCAmelCase		(							):





		"""simple docstring"""




		global _logger
		lowerCAmelCase__   = _logger or logging.getLogger(__name__  )
		return _logger

class    a_   (  __UpperCamelCase      ):




					def __init__(						self    :  str	,    snake_case__    :  Optional[int]  ):
							lowerCAmelCase__   = lock_file
							return None




					def __str__(						self    :  Any  ):
							lowerCAmelCase__   = F"""The file lock '{self.lock_file}' could not be acquired."""
							return temp

class    a_   :




					def __init__(						self    :  Tuple	,    snake_case__    :  Optional[int]  ):
							lowerCAmelCase__   = lock
							return None




					def __enter__(						self    :  Union[str, Any]  ):
							return self.lock




					def __exit__(						self    :  List[Any]	,    snake_case__    :  Dict	,    snake_case__    :  str	,    snake_case__    :  List[Any]  ):
							self.lock.release()
							return None

class    a_   :




					def __init__(						self    :  Dict	,    snake_case__    :  Tuple	,    snake_case__    :  Dict=-1	,    snake_case__    :  Dict=None  ):
							lowerCAmelCase__   = max_filename_length if max_filename_length is not None else 255
							# Hash the filename if it's too long
							lowerCAmelCase__   = self.hash_filename_if_too_long(snake_case__	,    snake_case__  )
							# The path to the lock file.
							lowerCAmelCase__   = 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.
							lowerCAmelCase__   = None

							# The default timeout value.
							lowerCAmelCase__   = timeout

							# We use this lock primarily for the lock counter.
							lowerCAmelCase__   = 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.
							lowerCAmelCase__   = 0
							return None




					@property
					def     _SCREAMING_SNAKE_CASE    (						self    :  Tuple  ):
							return self._lock_file




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




					@timeout.setter
					def     _SCREAMING_SNAKE_CASE    (						self    :  str	,    snake_case__    :  str  ):
							lowerCAmelCase__   = float(snake_case__  )
							return None




					def     _SCREAMING_SNAKE_CASE    (						self    :  List[str]  ):
							raise NotImplementedError()




					def     _SCREAMING_SNAKE_CASE    (						self    :  str  ):
							raise NotImplementedError()




					@property
					def     _SCREAMING_SNAKE_CASE    (						self    :  str  ):
							return self._lock_file_fd is not None




					def     _SCREAMING_SNAKE_CASE    (						self    :  List[Any]	,    snake_case__    :  Union[str, Any]=None	,    snake_case__    :  Optional[int]=0.05  ):
							# Use the default timeout, if no timeout is provided.
							if timeout is None:
									lowerCAmelCase__   = 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

							lowerCAmelCase__   = id(self  )
							lowerCAmelCase__   = self._lock_file
							lowerCAmelCase__   = 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(snake_case__  )
							except:  # noqa
									# Something did go wrong, so decrement the counter.
									with self._thread_lock:
											lowerCAmelCase__   = max(0	,    self._lock_counter - 1  )

									raise
							return _Acquire_ReturnProxy(lock=self  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[int]	,    snake_case__    :  Tuple=False  ):
							with self._thread_lock:
									if self.is_locked:
											self._lock_counter -= 1

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

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

							return None




					def __enter__(						self    :  Optional[Any]  ):
							self.acquire()
							return self




					def __exit__(						self    :  Any	,    snake_case__    :  Optional[Any]	,    snake_case__    :  Optional[Any]	,    snake_case__    :  Optional[Any]  ):
							self.release()
							return None




					def __del__(						self    :  Optional[int]  ):
							self.release(force=snake_case__  )
							return None




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[int]	,    snake_case__    :  str	,    snake_case__    :  int  ):
							lowerCAmelCase__   = os.path.basename(snake_case__  )
							if len(snake_case__  ) > max_length and max_length > 0:
									lowerCAmelCase__   = os.path.dirname(snake_case__  )
									lowerCAmelCase__   = str(hash(snake_case__  )  )
									lowerCAmelCase__   = filename[: max_length - len(snake_case__  ) - 8] + """...""" + hashed_filename + """.lock"""
									return os.path.join(snake_case__	,    snake_case__  )
							else:
									return path

class    a_   (  __UpperCamelCase      ):




					def __init__(						self    :  List[Any]	,    snake_case__    :  int	,    snake_case__    :  str=-1	,    snake_case__    :  Tuple=None  ):
							from .file_utils import relative_to_absolute_path

							super().__init__(snake_case__	,    timeout=snake_case__	,    max_filename_length=snake_case__  )
							lowerCAmelCase__   = """\\\\?\\""" + relative_to_absolute_path(self.lock_file  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Tuple  ):
							lowerCAmelCase__   = os.O_RDWR | os.O_CREAT | os.O_TRUNC

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




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[int]  ):
							lowerCAmelCase__   = self._lock_file_fd
							lowerCAmelCase__   = None
							msvcrt.locking(snake_case__	,    msvcrt.LK_UNLCK	,    1  )
							os.close(snake_case__  )

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

class    a_   (  __UpperCamelCase      ):




					def __init__(						self    :  Union[str, Any]	,    snake_case__    :  List[str]	,    snake_case__    :  str=-1	,    snake_case__    :  List[str]=None  ):
							lowerCAmelCase__   = os.statvfs(os.path.dirname(snake_case__  )  ).f_namemax
							super().__init__(snake_case__	,    timeout=snake_case__	,    max_filename_length=snake_case__  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[Any]  ):
							lowerCAmelCase__   = os.O_RDWR | os.O_CREAT | os.O_TRUNC
							lowerCAmelCase__   = os.open(self._lock_file	,    snake_case__  )

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




					def     _SCREAMING_SNAKE_CASE    (						self    :  str  ):
							# 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
							lowerCAmelCase__   = self._lock_file_fd
							lowerCAmelCase__   = None
							fcntl.flock(snake_case__	,    fcntl.LOCK_UN  )
							os.close(snake_case__  )
							return None

class    a_   (  __UpperCamelCase      ):




					def     _SCREAMING_SNAKE_CASE    (						self    :  Tuple  ):
							lowerCAmelCase__   = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
							try:
									lowerCAmelCase__   = os.open(self._lock_file	,    snake_case__  )
							except OSError:
									pass
							else:
									lowerCAmelCase__   = fd
							return None




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[int]  ):
							os.close(self._lock_file_fd  )
							lowerCAmelCase__   = None

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

if msvcrt:
			__lowerCAmelCase     :				Tuple    				=    WindowsFileLock
elif fcntl:
			__lowerCAmelCase     :				Optional[int]    				=    UnixFileLock
else:
			__lowerCAmelCase     :				Tuple    				=    SoftFileLock

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




