

from __future__ import annotations

import os
import tempfile
import unittest

from transformers import ConvBertConfig, is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
					import tensorflow as tf

					from transformers import (
					    TFConvBertForMaskedLM,
					    TFConvBertForMultipleChoice,
					    TFConvBertForQuestionAnswering,
					    TFConvBertForSequenceClassification,
					    TFConvBertForTokenClassification,
					    TFConvBertModel,
					)





class       snake_case_  :







			'''simple docstring'''

			def __init__(	self					:							Tuple				,  _UpperCamelCase					:							Optional[int]				,  _UpperCamelCase					:							Union[str, Any]=1_3				,  _UpperCamelCase					:							Dict=7				,  _UpperCamelCase					:							List[str]=True				,  _UpperCamelCase					:							Any=True				,  _UpperCamelCase					:							Union[str, Any]=True				,  _UpperCamelCase					:							List[Any]=True				,  _UpperCamelCase					:							str=9_9				,  _UpperCamelCase					:							str=3_2				,  _UpperCamelCase					:							Union[str, Any]=2				,  _UpperCamelCase					:							Optional[Any]=4				,  _UpperCamelCase					:							Any=3_7				,  _UpperCamelCase					:							Tuple="gelu"				,  _UpperCamelCase					:							Optional[Any]=0.1				,  _UpperCamelCase					:							List[Any]=0.1				,  _UpperCamelCase					:							Union[str, Any]=5_1_2				,  _UpperCamelCase					:							Optional[int]=1_6				,  _UpperCamelCase					:							Tuple=2				,  _UpperCamelCase					:							Optional[Any]=0.02				,  _UpperCamelCase					:							Union[str, Any]=3				,  _UpperCamelCase					:							Optional[Any]=4				,  _UpperCamelCase					:							List[str]=None				,  ) ->Tuple:
									snake_case_  =      parent
									snake_case_  =      1_3
									snake_case_  =      7
									snake_case_  =      True
									snake_case_  =      True
									snake_case_  =      True
									snake_case_  =      True
									snake_case_  =      9_9
									snake_case_  =      3_8_4
									snake_case_  =      2
									snake_case_  =      4
									snake_case_  =      3_7
									snake_case_  =      '''gelu'''
									snake_case_  =      0.1
									snake_case_  =      0.1
									snake_case_  =      5_1_2
									snake_case_  =      1_6
									snake_case_  =      2
									snake_case_  =      0.02
									snake_case_  =      3
									snake_case_  =      4
									snake_case_  =      1_2_8
									snake_case_  =      2
									snake_case_  =      9
									snake_case_  =      1
									snake_case_  =      None







			def 							snake_case__(	self					:							Optional[Any]				) ->Optional[int]:
									snake_case_  =      ids_tensor([self.batch_size, self.seq_length]				,  self.vocab_size				)

									snake_case_  =      None
									if self.use_input_mask:
															snake_case_  =      random_attention_mask([self.batch_size, self.seq_length]				)

									snake_case_  =      None
									if self.use_token_type_ids:
															snake_case_  =      ids_tensor([self.batch_size, self.seq_length]				,  self.type_vocab_size				)

									snake_case_  =      None
									snake_case_  =      None
									snake_case_  =      None
									if self.use_labels:
															snake_case_  =      ids_tensor([self.batch_size]				,  self.type_sequence_label_size				)
															snake_case_  =      ids_tensor([self.batch_size, self.seq_length]				,  self.num_labels				)
															snake_case_  =      ids_tensor([self.batch_size]				,  self.num_choices				)

									snake_case_  =      ConvBertConfig(
									    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				,  initializer_range=self.initializer_range				,  return_dict=_UpperCamelCase				,  )

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







			def 							snake_case__(	self					:							Optional[int]				,  _UpperCamelCase					:							Any				,  _UpperCamelCase					:							Union[str, Any]				,  _UpperCamelCase					:							List[Any]				,  _UpperCamelCase					:							Any				,  _UpperCamelCase					:							Dict				,  _UpperCamelCase					:							Optional[Any]				,  _UpperCamelCase					:							int				) ->Optional[int]:
									snake_case_  =      TFConvBertModel(config=_UpperCamelCase				)
									snake_case_  =      {'''input_ids''': input_ids, '''attention_mask''': input_mask, '''token_type_ids''': token_type_ids}

									snake_case_  =      [input_ids, input_mask]
									snake_case_  =      model(_UpperCamelCase				)

									snake_case_  =      model(_UpperCamelCase				)

									self.parent.assertEqual(result.last_hidden_state.shape				,  (self.batch_size, self.seq_length, self.hidden_size)				)







			def 							snake_case__(	self					:							str				,  _UpperCamelCase					:							List[Any]				,  _UpperCamelCase					:							Optional[Any]				,  _UpperCamelCase					:							Tuple				,  _UpperCamelCase					:							str				,  _UpperCamelCase					:							Optional[int]				,  _UpperCamelCase					:							Tuple				,  _UpperCamelCase					:							Union[str, Any]				) ->Optional[int]:
									snake_case_  =      TFConvBertForMaskedLM(config=_UpperCamelCase				)
									snake_case_  =      {
									    '''input_ids''': input_ids,
									    '''attention_mask''': input_mask,
									    '''token_type_ids''': token_type_ids,
									}
									snake_case_  =      model(_UpperCamelCase				)
									self.parent.assertEqual(result.logits.shape				,  (self.batch_size, self.seq_length, self.vocab_size)				)







			def 							snake_case__(	self					:							Tuple				,  _UpperCamelCase					:							Any				,  _UpperCamelCase					:							Tuple				,  _UpperCamelCase					:							Union[str, Any]				,  _UpperCamelCase					:							int				,  _UpperCamelCase					:							str				,  _UpperCamelCase					:							int				,  _UpperCamelCase					:							List[Any]				) ->Dict:
									snake_case_  =      self.num_labels
									snake_case_  =      TFConvBertForSequenceClassification(config=_UpperCamelCase				)
									snake_case_  =      {
									    '''input_ids''': input_ids,
									    '''attention_mask''': input_mask,
									    '''token_type_ids''': token_type_ids,
									}

									snake_case_  =      model(_UpperCamelCase				)
									self.parent.assertEqual(result.logits.shape				,  (self.batch_size, self.num_labels)				)







			def 							snake_case__(	self					:							List[Any]				,  _UpperCamelCase					:							Dict				,  _UpperCamelCase					:							Optional[int]				,  _UpperCamelCase					:							int				,  _UpperCamelCase					:							int				,  _UpperCamelCase					:							List[Any]				,  _UpperCamelCase					:							int				,  _UpperCamelCase					:							List[Any]				) ->str:
									snake_case_  =      self.num_choices
									snake_case_  =      TFConvBertForMultipleChoice(config=_UpperCamelCase				)
									snake_case_  =      tf.tile(tf.expand_dims(_UpperCamelCase				,  1				)				,  (1, self.num_choices, 1)				)
									snake_case_  =      tf.tile(tf.expand_dims(_UpperCamelCase				,  1				)				,  (1, self.num_choices, 1)				)
									snake_case_  =      tf.tile(tf.expand_dims(_UpperCamelCase				,  1				)				,  (1, self.num_choices, 1)				)
									snake_case_  =      {
									    '''input_ids''': multiple_choice_inputs_ids,
									    '''attention_mask''': multiple_choice_input_mask,
									    '''token_type_ids''': multiple_choice_token_type_ids,
									}
									snake_case_  =      model(_UpperCamelCase				)
									self.parent.assertEqual(result.logits.shape				,  (self.batch_size, self.num_choices)				)







			def 							snake_case__(	self					:							Any				,  _UpperCamelCase					:							int				,  _UpperCamelCase					:							Optional[int]				,  _UpperCamelCase					:							Optional[int]				,  _UpperCamelCase					:							Optional[Any]				,  _UpperCamelCase					:							List[Any]				,  _UpperCamelCase					:							Dict				,  _UpperCamelCase					:							Optional[int]				) ->str:
									snake_case_  =      self.num_labels
									snake_case_  =      TFConvBertForTokenClassification(config=_UpperCamelCase				)
									snake_case_  =      {
									    '''input_ids''': input_ids,
									    '''attention_mask''': input_mask,
									    '''token_type_ids''': token_type_ids,
									}
									snake_case_  =      model(_UpperCamelCase				)
									self.parent.assertEqual(result.logits.shape				,  (self.batch_size, self.seq_length, self.num_labels)				)







			def 							snake_case__(	self					:							Optional[Any]				,  _UpperCamelCase					:							Tuple				,  _UpperCamelCase					:							Dict				,  _UpperCamelCase					:							Dict				,  _UpperCamelCase					:							List[Any]				,  _UpperCamelCase					:							str				,  _UpperCamelCase					:							Optional[int]				,  _UpperCamelCase					:							Optional[Any]				) ->Tuple:
									snake_case_  =      TFConvBertForQuestionAnswering(config=_UpperCamelCase				)
									snake_case_  =      {
									    '''input_ids''': input_ids,
									    '''attention_mask''': input_mask,
									    '''token_type_ids''': token_type_ids,
									}

									snake_case_  =      model(_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 							snake_case__(	self					:							List[str]				) ->List[Any]:
									snake_case_  =      self.prepare_config_and_inputs()
									(
									    (
									    snake_case_
									),       (
									    snake_case_
									),       (
									    snake_case_
									),       (
									    snake_case_
									),       (
									    snake_case_
									),       (
									    snake_case_
									),       (
									    snake_case_
									),       
									)  =      config_and_inputs
									snake_case_  =      {'''input_ids''': input_ids, '''token_type_ids''': token_type_ids, '''attention_mask''': input_mask}
									return config, inputs_dict






@require_tf
class       snake_case_  (						__A      ,			__A      ,			unittest.TestCase   ):







			'''simple docstring'''

			SCREAMING_SNAKE_CASE	:  int		     =							(
			    (
			        TFConvBertModel,
			        TFConvBertForMaskedLM,
			        TFConvBertForQuestionAnswering,
			        TFConvBertForSequenceClassification,
			        TFConvBertForTokenClassification,
			        TFConvBertForMultipleChoice,
			    )
			    if is_tf_available()
			    else ()
			)
			SCREAMING_SNAKE_CASE	:  Dict		     =							(
			    {
			        "feature-extraction": TFConvBertModel,
			        "fill-mask": TFConvBertForMaskedLM,
			        "question-answering": TFConvBertForQuestionAnswering,
			        "text-classification": TFConvBertForSequenceClassification,
			        "token-classification": TFConvBertForTokenClassification,
			        "zero-shot": TFConvBertForSequenceClassification,
			    }
			    if is_tf_available()
			    else {}
			)
			SCREAMING_SNAKE_CASE	:  Dict		     =							False
			SCREAMING_SNAKE_CASE	:  Union[str, Any]		     =							False
			SCREAMING_SNAKE_CASE	:  int		     =							False







			def 							snake_case__(	self					:							str				) ->Union[str, Any]:
									snake_case_  =      TFConvBertModelTester(self				)
									snake_case_  =      ConfigTester(self				,  config_class=_UpperCamelCase				,  hidden_size=3_7				)







			def 							snake_case__(	self					:							List[str]				) ->Optional[Any]:
									self.config_tester.run_common_tests()







			def 							snake_case__(	self					:							str				) ->Optional[Any]:
									snake_case_  =      self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_model(*_UpperCamelCase				)







			def 							snake_case__(	self					:							Optional[Any]				) ->List[str]:
									snake_case_  =      self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_masked_lm(*_UpperCamelCase				)







			def 							snake_case__(	self					:							str				) ->Dict:
									snake_case_  =      self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_multiple_choice(*_UpperCamelCase				)







			def 							snake_case__(	self					:							int				) ->str:
									snake_case_  =      self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_question_answering(*_UpperCamelCase				)







			def 							snake_case__(	self					:							Optional[int]				) ->str:
									snake_case_  =      self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_sequence_classification(*_UpperCamelCase				)







			def 							snake_case__(	self					:							str				) ->Union[str, Any]:
									snake_case_  =      self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_token_classification(*_UpperCamelCase				)







			@slow
			def 							snake_case__(	self					:							str				) ->str:
									snake_case_,       snake_case_  =      self.model_tester.prepare_config_and_inputs_for_common()
									snake_case_  =      True
									snake_case_  =      True

									if hasattr(_UpperCamelCase				,  '''use_cache'''				):
															snake_case_  =      True

									snake_case_  =      getattr(self.model_tester				,  '''encoder_seq_length'''				,  self.model_tester.seq_length				)
									snake_case_  =      getattr(self.model_tester				,  '''key_length'''				,  _UpperCamelCase				)

									for model_class in self.all_model_classes:
															snake_case_  =      self._prepare_for_class(_UpperCamelCase				,  _UpperCamelCase				)
															snake_case_  =      model_class(_UpperCamelCase				)
															snake_case_  =      len(model(_UpperCamelCase				)				)

															with tempfile.TemporaryDirectory() as tmpdirname:
																					model.save_pretrained(_UpperCamelCase				,  saved_model=_UpperCamelCase				)
																					snake_case_  =      os.path.join(_UpperCamelCase				,  '''saved_model'''				,  '''1'''				)
																					snake_case_  =      tf.keras.models.load_model(_UpperCamelCase				)
																					snake_case_  =      model(_UpperCamelCase				)

																					if self.is_encoder_decoder:
																											snake_case_  =      outputs['''encoder_hidden_states''']
																											snake_case_  =      outputs['''encoder_attentions''']
																					else:
																											snake_case_  =      outputs['''hidden_states''']
																											snake_case_  =      outputs['''attentions''']

																					self.assertEqual(len(_UpperCamelCase				)				,  _UpperCamelCase				)

																					snake_case_  =      getattr(
																					    self.model_tester				,  '''expected_num_hidden_layers'''				,  self.model_tester.num_hidden_layers + 1				)

																					self.assertEqual(len(_UpperCamelCase				)				,  _UpperCamelCase				)
																					self.assertListEqual(
																					    list(output_hidden_states[0].shape[-2:]				)				,  [self.model_tester.seq_length, self.model_tester.hidden_size]				,  )

																					self.assertEqual(len(_UpperCamelCase				)				,  self.model_tester.num_hidden_layers				)
																					self.assertListEqual(
																					    list(output_attentions[0].shape[-3:]				)				,  [self.model_tester.num_attention_heads / 2, encoder_seq_length, encoder_key_length]				,  )







			@slow
			def 							snake_case__(	self					:							Union[str, Any]				) ->Tuple:
									snake_case_  =      TFConvBertModel.from_pretrained('''YituTech/conv-bert-base'''				)
									self.assertIsNotNone(_UpperCamelCase				)







			def 							snake_case__(	self					:							List[str]				) ->Any:
									snake_case_,       snake_case_  =      self.model_tester.prepare_config_and_inputs_for_common()
									snake_case_  =      True
									snake_case_  =      getattr(self.model_tester				,  '''decoder_seq_length'''				,  self.model_tester.seq_length				)
									snake_case_  =      getattr(self.model_tester				,  '''encoder_seq_length'''				,  self.model_tester.seq_length				)
									snake_case_  =      getattr(self.model_tester				,  '''key_length'''				,  _UpperCamelCase				)
									snake_case_  =      getattr(self.model_tester				,  '''key_length'''				,  _UpperCamelCase				)

									def check_decoder_attentions_output(_UpperCamelCase					:							List[Any]				):
															snake_case_  =      len(_UpperCamelCase				)
															self.assertEqual(out_len % 2				,  0				)
															snake_case_  =      outputs.decoder_attentions
															self.assertEqual(len(_UpperCamelCase				)				,  self.model_tester.num_hidden_layers				)
															self.assertListEqual(
															    list(decoder_attentions[0].shape[-3:]				)				,  [self.model_tester.num_attention_heads / 2, decoder_seq_length, decoder_key_length]				,  )

									def check_encoder_attentions_output(_UpperCamelCase					:							Dict				):
															snake_case_  =      [
															    t.numpy() for t in (outputs.encoder_attentions if config.is_encoder_decoder else outputs.attentions)
															]
															self.assertEqual(len(_UpperCamelCase				)				,  self.model_tester.num_hidden_layers				)
															self.assertListEqual(
															    list(attentions[0].shape[-3:]				)				,  [self.model_tester.num_attention_heads / 2, encoder_seq_length, encoder_key_length]				,  )

									for model_class in self.all_model_classes:
															snake_case_  =      True
															snake_case_  =      False
															snake_case_  =      model_class(_UpperCamelCase				)
															snake_case_  =      model(self._prepare_for_class(_UpperCamelCase				,  _UpperCamelCase				)				)
															snake_case_  =      len(_UpperCamelCase				)
															self.assertEqual(config.output_hidden_states				,  _UpperCamelCase				)
															check_encoder_attentions_output(_UpperCamelCase				)

															if self.is_encoder_decoder:
																					snake_case_  =      model_class(_UpperCamelCase				)
																					snake_case_  =      model(self._prepare_for_class(_UpperCamelCase				,  _UpperCamelCase				)				)
																					self.assertEqual(config.output_hidden_states				,  _UpperCamelCase				)
																					check_decoder_attentions_output(_UpperCamelCase				)

															# Check that output attentions can also be changed via the config
															del inputs_dict["output_attentions"]
															snake_case_  =      True
															snake_case_  =      model_class(_UpperCamelCase				)
															snake_case_  =      model(self._prepare_for_class(_UpperCamelCase				,  _UpperCamelCase				)				)
															self.assertEqual(config.output_hidden_states				,  _UpperCamelCase				)
															check_encoder_attentions_output(_UpperCamelCase				)

															# Check attention is always last and order is fine
															snake_case_  =      True
															snake_case_  =      True
															snake_case_  =      model_class(_UpperCamelCase				)
															snake_case_  =      model(self._prepare_for_class(_UpperCamelCase				,  _UpperCamelCase				)				)

															self.assertEqual(out_len + (2 if self.is_encoder_decoder else 1)				,  len(_UpperCamelCase				)				)
															self.assertEqual(model.config.output_hidden_states				,  _UpperCamelCase				)
															check_encoder_attentions_output(_UpperCamelCase				)






@require_tf
class       snake_case_  (						unittest.TestCase   ):







			'''simple docstring'''

			@slow
			def 							snake_case__(	self					:							Any				) ->int:
									snake_case_  =      TFConvBertModel.from_pretrained('''YituTech/conv-bert-base'''				)
									snake_case_  =      tf.constant([[0, 1, 2, 3, 4, 5]]				)
									snake_case_  =      model(_UpperCamelCase				)[0]

									snake_case_  =      [1, 6, 7_6_8]
									self.assertEqual(output.shape				,  _UpperCamelCase				)

									snake_case_  =      tf.constant(
									    [
									        [
									            [-0.03475493, -0.4686034, -0.30638832],
									            [0.22637248, -0.26988646, -0.7423424],
									            [0.10324868, -0.45013508, -0.58280784],
									        ]
									    ]				)
									tf.debugging.assert_near(output[:, :3, :3]				,  _UpperCamelCase				,  atol=1e-4				)






import logging
import os
import threading
import time


try:
      import warnings
except ImportError:
      a__      :     Optional[Any]         =		None

try:
      import msvcrt
except ImportError:
      a__      :     List[str]         =		None

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


# Backward compatibility
# ------------------------------------------------
try:
      TimeoutError
except NameError:
      a__      :     List[str]         =		OSError


# Data
# ------------------------------------------------
a__      :     Union[str, Any]         =		[
    'Timeout',
    'BaseFileLock',
    'WindowsFileLock',
    'UnixFileLock',
    'SoftFileLock',
    'FileLock',
]

a__      :     str         =		'3.0.12'


a__      :     Optional[int]         =		None

def      UpperCAmelCase_    (    )	->				Tuple:
     '''simple docstring'''


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





class UpperCAmelCase_				( _SCREAMING_SNAKE_CASE ):


  def __init__(							self ,__snake_case ):



       """simple docstring"""





       A_          =       lock_file
       return None




  def __str__(							self ):



       """simple docstring"""





       A_          =       f'The file lock \'{self.lock_file}\' could not be acquired.'
       return temp





class UpperCAmelCase_				:


  def __init__(							self ,__snake_case ):



       """simple docstring"""





       A_          =       lock
       return None


  def __enter__(							self ):



       """simple docstring"""





       return self.lock




  def __exit__(							self ,__snake_case ,__snake_case ,__snake_case ):



       """simple docstring"""





       self.lock.release()
       return None





class UpperCAmelCase_				:


  def __init__(							self ,__snake_case ,__snake_case=-1 ,__snake_case=None ):



       """simple docstring"""





       A_          =       max_filename_length if max_filename_length is not None else 2_5_5
       # Hash the filename if it's too long
       A_          =       self.hash_filename_if_too_long(__snake_case ,__snake_case )
       # The path to the lock file.
       A_          =       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.
       A_          =       None

       # The default timeout value.
       A_          =       timeout

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


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return self._lock_file


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return self._timeout


  @timeout.setter
  def 						__UpperCAmelCase     (							self ,__snake_case ):



       """simple docstring"""





       A_          =       float(__snake_case )
       return None


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       raise NotImplementedError()


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       raise NotImplementedError()


  @property
  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       return self._lock_file_fd is not None


  def 						__UpperCAmelCase     (							self ,__snake_case=None ,__snake_case=0.05 ):



       """simple docstring"""





       if timeout is None:
            A_          =       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

       A_          =       id(self )
       A_          =       self._lock_file
       A_          =       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:
                 A_          =       max(0 ,self._lock_counter - 1 )

            raise
       return _Acquire_ReturnProxy(lock=self )


  def 						__UpperCAmelCase     (							self ,__snake_case=False ):



       """simple docstring"""





       with self._thread_lock:
            if self.is_locked:
                 self._lock_counter -= 1

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

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

       return None


  def __enter__(							self ):



       """simple docstring"""





       self.acquire()
       return self


  def __exit__(							self ,__snake_case ,__snake_case ,__snake_case ):



       """simple docstring"""





       self.release()
       return None


  def __del__(							self ):



       """simple docstring"""





       self.release(force=__snake_case )
       return None




  def 						__UpperCAmelCase     (							self ,__snake_case ,__snake_case ):



       """simple docstring"""





       A_          =       os.path.basename(__snake_case )
       if len(__snake_case ) > max_length and max_length > 0:
            A_          =       os.path.dirname(__snake_case )
            A_          =       str(hash(__snake_case ) )
            A_          =       filename[: max_length - len(__snake_case ) - 8] + '''...''' + hashed_filename + '''.lock'''
            return os.path.join(__snake_case ,__snake_case )
       else:
            return path





class UpperCAmelCase_				( _SCREAMING_SNAKE_CASE ):


  def __init__(							self ,__snake_case ,__snake_case=-1 ,__snake_case=None ):



       """simple docstring"""





       from .file_utils import relative_to_absolute_path

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


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       os.O_RDWR | os.O_CREAT | os.O_TRUNC

       try:
            A_          =       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:
                 A_          =       fd
       return None




  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       self._lock_file_fd
       A_          =       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 UpperCAmelCase_				( _SCREAMING_SNAKE_CASE ):


  def __init__(							self ,__snake_case ,__snake_case=-1 ,__snake_case=None ):



       """simple docstring"""





       A_          =       os.statvfs(os.path.dirname(__snake_case ) ).f_namemax
       super().__init__(__snake_case ,timeout=__snake_case ,max_filename_length=__snake_case )


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       os.O_RDWR | os.O_CREAT | os.O_TRUNC
       A_          =       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:
            A_          =       fd
       return None




  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       self._lock_file_fd
       A_          =       None
       fcntl.flock(__snake_case ,fcntl.LOCK_UN )
       os.close(__snake_case )
       return None





class UpperCAmelCase_				( _SCREAMING_SNAKE_CASE ):


  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       A_          =       os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
       try:
            A_          =       os.open(self._lock_file ,__snake_case )
       except OSError:
            pass
       else:
            A_          =       fd
       return None




  def 						__UpperCAmelCase     (							self ):



       """simple docstring"""





       os.close(self._lock_file_fd )
       A_          =       None

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




a__      :     Union[str, Any]         =		None

if msvcrt:
      a__      :     Dict         =		WindowsFileLock
elif fcntl:
      a__      :     Dict         =		UnixFileLock
else:
      a__      :     Optional[int]         =		SoftFileLock

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



