






from __future__ import annotations

import unittest

from transformers import is_tf_available
from transformers.testing_utils import require_sentencepiece, require_tf, require_tokenizers, 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 numpy as np
	import tensorflow as tf

	from transformers import (
	    TF_FLAUBERT_PRETRAINED_MODEL_ARCHIVE_LIST,
	    FlaubertConfig,
	    TFFlaubertForMultipleChoice,
	    TFFlaubertForQuestionAnsweringSimple,
	    TFFlaubertForSequenceClassification,
	    TFFlaubertForTokenClassification,
	    TFFlaubertModel,
	    TFFlaubertWithLMHeadModel,
	)

class 		lowerCAmelCase     :

					'''simple docstring'''

					def __init__(  self				:	Optional[int]		,     __a				:	Tuple		,     )			->  int:


												"""simple docstring"""

												__lowercase      :							List[Any]				=     parent
												__lowercase      :							Union[str, Any]				=     13
												__lowercase      :							Optional[Any]				=     7
												__lowercase      :							Any				=     True
												__lowercase      :							Tuple				=     True
												__lowercase      :							List[str]				=     True
												__lowercase      :							Optional[int]				=     True
												__lowercase      :							Optional[Any]				=     True
												__lowercase      :							Any				=     False
												__lowercase      :							List[str]				=     False
												__lowercase      :							Tuple				=     False
												__lowercase      :							Any				=     2
												__lowercase      :							Any				=     99
												__lowercase      :							Optional[int]				=     0
												__lowercase      :							int				=     32
												__lowercase      :							Any				=     2
												__lowercase      :							Optional[int]				=     4
												__lowercase      :							List[str]				=     0.1
												__lowercase      :							Union[str, Any]				=     0.1
												__lowercase      :							Optional[Any]				=     512
												__lowercase      :							str				=     16
												__lowercase      :							List[Any]				=     2
												__lowercase      :							List[str]				=     0.02
												__lowercase      :							Tuple				=     3
												__lowercase      :							Optional[Any]				=     4
												__lowercase      :							Tuple				=     """last"""
												__lowercase      :							str				=     True
												__lowercase      :							List[str]				=     None
												__lowercase      :							Optional[Any]				=     0



					def       lowerCAmelCase				(  self				:	Optional[Any]							)			->  Union[str, Any]:


												"""simple docstring"""

												__lowercase      :							Optional[Any]				=     ids_tensor([self.batch_size, self.seq_length]		,     self.vocab_size							)
												__lowercase      :							int				=     random_attention_mask([self.batch_size, self.seq_length]		,     dtype=tf.floataa							)

												__lowercase      :							List[Any]				=     None
												if self.use_input_lengths:
																			__lowercase      :							Tuple				=     (
																			    ids_tensor([self.batch_size]		,     vocab_size=2							) + self.seq_length - 2
																			)  # small variation of seq_length

												__lowercase      :							Optional[int]				=     None
												if self.use_token_type_ids:
																			__lowercase      :							Optional[int]				=     ids_tensor([self.batch_size, self.seq_length]		,     self.n_langs							)

												__lowercase      :							Tuple				=     None
												__lowercase      :							List[Any]				=     None
												__lowercase      :							Any				=     None
												if self.use_labels:
																			__lowercase      :							int				=     ids_tensor([self.batch_size]		,     self.type_sequence_label_size							)
																			__lowercase      :							Dict				=     ids_tensor([self.batch_size, self.seq_length]		,     self.num_labels							)
																			__lowercase      :							Dict				=     ids_tensor([self.batch_size]		,     2		,     dtype=tf.floataa							)
																			__lowercase      :							Optional[Any]				=     ids_tensor([self.batch_size]		,     self.num_choices							)

												__lowercase      :							List[Any]				=     FlaubertConfig(
												    vocab_size=self.vocab_size		,     n_special=self.n_special		,     emb_dim=self.hidden_size		,     n_layers=self.num_hidden_layers		,     n_heads=self.num_attention_heads		,     dropout=self.hidden_dropout_prob		,     attention_dropout=self.attention_probs_dropout_prob		,     gelu_activation=self.gelu_activation		,     sinusoidal_embeddings=self.sinusoidal_embeddings		,     asm=self.asm		,     causal=self.causal		,     n_langs=self.n_langs		,     max_position_embeddings=self.max_position_embeddings		,     initializer_range=self.initializer_range		,     summary_type=self.summary_type		,     use_proj=self.use_proj		,     bos_token_id=self.bos_token_id		,     )

												return (
												    config,
												    input_ids,
												    token_type_ids,
												    input_lengths,
												    sequence_labels,
												    token_labels,
												    is_impossible_labels,
												    choice_labels,
												    input_mask,
												)



					def       lowerCAmelCase				(  self				:	List[str]		,     __a				:	Optional[int]		,     __a				:	Optional[Any]		,     __a				:	Any		,     __a				:	Union[str, Any]		,     __a				:	Union[str, Any]		,     __a				:	Optional[int]		,     __a				:	Dict		,     __a				:	str		,     __a				:	Union[str, Any]		,     )			->  str:


												"""simple docstring"""

												__lowercase      :							List[str]				=     TFFlaubertModel(config=lowerCAmelCase_							)
												__lowercase      :							str				=     {"""input_ids""": input_ids, """lengths""": input_lengths, """langs""": token_type_ids}
												__lowercase      :							Dict				=     model(lowerCAmelCase_							)

												__lowercase      :							Dict				=     [input_ids, input_mask]
												__lowercase      :							List[Any]				=     model(lowerCAmelCase_							)
												self.parent.assertEqual(result.last_hidden_state.shape		,     (self.batch_size, self.seq_length, self.hidden_size)							)



					def       lowerCAmelCase				(  self				:	Optional[int]		,     __a				:	List[Any]		,     __a				:	Tuple		,     __a				:	List[str]		,     __a				:	List[Any]		,     __a				:	List[str]		,     __a				:	str		,     __a				:	Optional[Any]		,     __a				:	Dict		,     __a				:	str		,     )			->  Optional[Any]:


												"""simple docstring"""

												__lowercase      :							List[Any]				=     TFFlaubertWithLMHeadModel(lowerCAmelCase_							)

												__lowercase      :							str				=     {"""input_ids""": input_ids, """lengths""": input_lengths, """langs""": token_type_ids}
												__lowercase      :							List[str]				=     model(lowerCAmelCase_							)

												self.parent.assertEqual(result.logits.shape		,     (self.batch_size, self.seq_length, self.vocab_size)							)



					def       lowerCAmelCase				(  self				:	Optional[Any]		,     __a				:	Optional[int]		,     __a				:	List[str]		,     __a				:	List[Any]		,     __a				:	Union[str, Any]		,     __a				:	Optional[Any]		,     __a				:	Optional[int]		,     __a				:	Union[str, Any]		,     __a				:	Optional[int]		,     __a				:	Any		,     )			->  List[Any]:


												"""simple docstring"""

												__lowercase      :							Optional[int]				=     TFFlaubertForQuestionAnsweringSimple(lowerCAmelCase_							)

												__lowercase      :							List[str]				=     {"""input_ids""": input_ids, """lengths""": input_lengths}

												__lowercase      :							Any				=     model(lowerCAmelCase_							)

												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				:	str		,     __a				:	List[str]		,     __a				:	Optional[Any]		,     __a				:	str		,     __a				:	List[Any]		,     __a				:	Union[str, Any]		,     __a				:	Dict		,     __a				:	int		,     __a				:	Any		,     __a				:	str		,     )			->  Tuple:


												"""simple docstring"""

												__lowercase      :							Optional[Any]				=     TFFlaubertForSequenceClassification(lowerCAmelCase_							)

												__lowercase      :							Any				=     {"""input_ids""": input_ids, """lengths""": input_lengths}

												__lowercase      :							List[Any]				=     model(lowerCAmelCase_							)

												self.parent.assertEqual(result.logits.shape		,     (self.batch_size, self.type_sequence_label_size)							)



					def       lowerCAmelCase				(  self				:	Any		,     __a				:	int		,     __a				:	str		,     __a				:	List[str]		,     __a				:	Optional[Any]		,     __a				:	Optional[Any]		,     __a				:	int		,     __a				:	Union[str, Any]		,     __a				:	int		,     __a				:	Tuple		,     )			->  List[str]:


												"""simple docstring"""

												__lowercase      :							List[Any]				=     self.num_labels
												__lowercase      :							str				=     TFFlaubertForTokenClassification(config=lowerCAmelCase_							)
												__lowercase      :							Any				=     {"""input_ids""": input_ids, """attention_mask""": input_mask, """token_type_ids""": token_type_ids}
												__lowercase      :							Optional[Any]				=     model(lowerCAmelCase_							)
												self.parent.assertEqual(result.logits.shape		,     (self.batch_size, self.seq_length, self.num_labels)							)



					def       lowerCAmelCase				(  self				:	Optional[int]		,     __a				:	str		,     __a				:	Optional[Any]		,     __a				:	int		,     __a				:	Any		,     __a				:	Union[str, Any]		,     __a				:	int		,     __a				:	int		,     __a				:	int		,     __a				:	int		,     )			->  Union[str, Any]:


												"""simple docstring"""

												__lowercase      :							Tuple				=     self.num_choices
												__lowercase      :							str				=     TFFlaubertForMultipleChoice(config=lowerCAmelCase_							)
												__lowercase      :							Union[str, Any]				=     tf.tile(tf.expand_dims(lowerCAmelCase_		,     1							)		,     (1, self.num_choices, 1)							)
												__lowercase      :							List[Any]				=     tf.tile(tf.expand_dims(lowerCAmelCase_		,     1							)		,     (1, self.num_choices, 1)							)
												__lowercase      :							str				=     tf.tile(tf.expand_dims(lowerCAmelCase_		,     1							)		,     (1, self.num_choices, 1)							)
												__lowercase      :							Tuple				=     {
												    """input_ids""": multiple_choice_inputs_ids,
												    """attention_mask""": multiple_choice_input_mask,
												    """token_type_ids""": multiple_choice_token_type_ids,
												}
												__lowercase      :							Dict				=     model(lowerCAmelCase_							)
												self.parent.assertEqual(result.logits.shape		,     (self.batch_size, self.num_choices)							)



					def       lowerCAmelCase				(  self				:	Optional[Any]							)			->  Optional[Any]:


												"""simple docstring"""

												__lowercase      :							Union[str, Any]				=     self.prepare_config_and_inputs()
												(
												    (
												    __lowercase    
												)			,			(
												    __lowercase    
												)			,			(
												    __lowercase    
												)			,			(
												    __lowercase    
												)			,			(
												    __lowercase    
												)			,			(
												    __lowercase    
												)			,			(
												    __lowercase    
												)			,			(
												    __lowercase    
												)			,			(
												    __lowercase    
												)			,			    
												)      :							List[Any]				=     config_and_inputs
												__lowercase      :							Tuple				=     {
												    """input_ids""": input_ids,
												    """token_type_ids""": token_type_ids,
												    """langs""": token_type_ids,
												    """lengths""": input_lengths,
												}
												return config, inputs_dict







@require_tf
class 		lowerCAmelCase     ( __UpperCAmelCase      ,       __UpperCAmelCase      ,       unittest.TestCase			):

					'''simple docstring'''

					_A						:			Union[str, Any]									=      (
					    (
					        TFFlaubertModel,
					        TFFlaubertWithLMHeadModel,
					        TFFlaubertForSequenceClassification,
					        TFFlaubertForQuestionAnsweringSimple,
					        TFFlaubertForTokenClassification,
					        TFFlaubertForMultipleChoice,
					    )
					    if is_tf_available()
					    else ()
					)
					_A						:			List[str]									=      (
					    (TFFlaubertWithLMHeadModel,) if is_tf_available() else ()
					)  # TODO (PVP): Check other models whether language generation is also applicable
					_A						:			Union[str, Any]									=      (
					    {
					        '''feature-extraction''': TFFlaubertModel,
					        '''fill-mask''': TFFlaubertWithLMHeadModel,
					        '''question-answering''': TFFlaubertForQuestionAnsweringSimple,
					        '''text-classification''': TFFlaubertForSequenceClassification,
					        '''token-classification''': TFFlaubertForTokenClassification,
					        '''zero-shot''': TFFlaubertForSequenceClassification,
					    }
					    if is_tf_available()
					    else {}
					)
					_A						:			Union[str, Any]									=      False
					_A						:			Union[str, Any]									=      False



					def       lowerCAmelCase				(  self				:	Optional[int]		,     __a				:	List[str]		,     __a				:	Union[str, Any]		,     __a				:	str		,     __a				:	Any		,     __a				:	Any							)			->  Union[str, Any]:


												"""simple docstring"""

												if (
												    pipeline_test_casse_name == "QAPipelineTests"
												    and tokenizer_name is not None
												    and not tokenizer_name.endswith("""Fast"""							)
												):
																			# `QAPipelineTests` fails for a few models when the slower tokenizer are used.
																			# (The slower tokenizers were never used for pipeline tests before the pipeline testing rework)
																			# TODO: check (and possibly fix) the `QAPipelineTests` with slower tokenizer
																			return True

												return False



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


												"""simple docstring"""

												__lowercase      :							Union[str, Any]				=     TFFlaubertModelTester(self							)
												__lowercase      :							int				=     ConfigTester(self		,     config_class=lowerCAmelCase_		,     emb_dim=37							)



					def       lowerCAmelCase				(  self				:	int							)			->  Dict:


												"""simple docstring"""

												self.config_tester.run_common_tests()



					def       lowerCAmelCase				(  self				:	List[Any]							)			->  str:


												"""simple docstring"""

												__lowercase      :							str				=     self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_flaubert_model(*lowerCAmelCase_							)



					def       lowerCAmelCase				(  self				:	Union[str, Any]							)			->  Dict:


												"""simple docstring"""

												__lowercase      :							Tuple				=     self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_flaubert_lm_head(*lowerCAmelCase_							)



					def       lowerCAmelCase				(  self				:	List[str]							)			->  List[str]:


												"""simple docstring"""

												__lowercase      :							Dict				=     self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_flaubert_qa(*lowerCAmelCase_							)



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


												"""simple docstring"""

												__lowercase      :							int				=     self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_flaubert_sequence_classif(*lowerCAmelCase_							)



					def       lowerCAmelCase				(  self				:	str							)			->  List[str]:


												"""simple docstring"""

												__lowercase      :							List[str]				=     self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_flaubert_for_token_classification(*lowerCAmelCase_							)



					def       lowerCAmelCase				(  self				:	int							)			->  str:


												"""simple docstring"""

												__lowercase      :							str				=     self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_flaubert_for_multiple_choice(*lowerCAmelCase_							)



					@slow
					def       lowerCAmelCase				(  self				:	List[Any]							)			->  Union[str, Any]:


												"""simple docstring"""

												for model_name in TF_FLAUBERT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																			__lowercase      :							Any				=     TFFlaubertModel.from_pretrained(lowerCAmelCase_							)
																			self.assertIsNotNone(lowerCAmelCase_							)







@require_tf
@require_sentencepiece
@require_tokenizers
class 		lowerCAmelCase     ( unittest.TestCase			):

					'''simple docstring'''

					@slow
					def       lowerCAmelCase				(  self				:	Union[str, Any]							)			->  Any:


												"""simple docstring"""

												__lowercase      :							Any				=     TFFlaubertModel.from_pretrained("""jplu/tf-flaubert-small-cased"""							)

												__lowercase      :							Any				=     tf.convert_to_tensor(
												    [[0, 158, 735, 2592, 1424, 6727, 82, 1]]		,     dtype=tf.intaa		,     )  # "J'aime flaubert !"

												__lowercase      :							int				=     model(lowerCAmelCase_							)[0]
												__lowercase      :							Tuple				=     tf.TensorShape((1, 8, 512)							)
												self.assertEqual(output.shape		,     lowerCAmelCase_							)
												# compare the actual values for a slice.
												__lowercase      :							int				=     tf.convert_to_tensor(
												    [
												        [
												            [-1.8768773, -1.566555, 0.27072418],
												            [-1.6920038, -0.5873505, 1.9329599],
												            [-2.9563985, -1.6993835, 1.7972052],
												        ]
												    ]		,     dtype=tf.floataa		,     )

												self.assertTrue(np.allclose(output[:, :3, :3].numpy()		,     expected_slice.numpy()		,     atol=1E-4							)							)







import logging
import os
import threading
import time


try:
	import warnings
except ImportError:
	lowerCamelCase : Any						   =       None

try:
	import msvcrt
except ImportError:
	lowerCamelCase : str						   =       None

try:
	import fcntl
except ImportError:
	lowerCamelCase : Optional[Any]						   =       None


# Backward compatibility
# ------------------------------------------------
try:
	TimeoutError
except NameError:
	lowerCamelCase : Union[str, Any]						   =       OSError


# Data
# ------------------------------------------------
lowerCamelCase : Tuple						   =       [
    '''Timeout''',
    '''BaseFileLock''',
    '''WindowsFileLock''',
    '''UnixFileLock''',
    '''SoftFileLock''',
    '''FileLock''',
]

lowerCamelCase : Tuple						   =       '''3.0.12'''


lowerCamelCase : Any						   =       None
def       snake_case_		(     ):
							global _logger
							__lowercase      :							List[str]				=     _logger or logging.getLogger(__name__      )
							return _logger

class 		lowerCAmelCase     ( __a			):

					'''simple docstring'''

					def __init__(  self				:	Any		,     __a				:	Any							)			->  List[Any]:


												"""simple docstring"""

												__lowercase      :							List[str]				=     lock_file
												return None



					def __str__(  self				:	str							)			->  Any:


												"""simple docstring"""

												__lowercase      :							Any				=     F"The file lock '{self.lock_file}' could not be acquired."
												return temp







class 		lowerCAmelCase     :

					'''simple docstring'''

					def __init__(  self				:	List[Any]		,     __a				:	Optional[int]							)			->  int:


												"""simple docstring"""

												__lowercase      :							Optional[Any]				=     lock
												return None



					def __enter__(  self				:	Dict							)			->  Dict:


												"""simple docstring"""

												return self.lock



					def __exit__(  self				:	Optional[int]		,     __a				:	Dict		,     __a				:	Any		,     __a				:	Tuple							)			->  Optional[Any]:


												"""simple docstring"""

												self.lock.release()
												return None







class 		lowerCAmelCase     :

					'''simple docstring'''

					def __init__(  self				:	Tuple		,     __a				:	Any		,     __a				:	Dict=-1		,     __a				:	Optional[Any]=None							)			->  Any:


												"""simple docstring"""

												__lowercase      :							Optional[int]				=     max_filename_length if max_filename_length is not None else 255
												# Hash the filename if it's too long
												__lowercase      :							Dict				=     self.hash_filename_if_too_long(__a		,     __a							)
												# The path to the lock file.
												__lowercase      :							Optional[Any]				=     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.
												__lowercase      :							int				=     None

												# The default timeout value.
												__lowercase      :							Optional[int]				=     timeout

												# We use this lock primarily for the lock counter.
												__lowercase      :							Optional[Any]				=     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.
												__lowercase      :							Union[str, Any]				=     0
												return None



					@property
					def       lowerCAmelCase				(  self				:	Optional[Any]							)			->  Optional[int]:


												"""simple docstring"""

												return self._lock_file



					@property
					def       lowerCAmelCase				(  self				:	Optional[int]							)			->  Optional[Any]:


												"""simple docstring"""

												return self._timeout



					@timeout.setter
					def       lowerCAmelCase				(  self				:	Tuple		,     __a				:	Tuple							)			->  Dict:


												"""simple docstring"""

												__lowercase      :							Tuple				=     float(__a							)
												return None



					def       lowerCAmelCase				(  self				:	Dict							)			->  Tuple:


												"""simple docstring"""

												raise NotImplementedError()



					def       lowerCAmelCase				(  self				:	Dict							)			->  Any:


												"""simple docstring"""

												raise NotImplementedError()



					@property
					def       lowerCAmelCase				(  self				:	Dict							)			->  List[Any]:


												"""simple docstring"""

												return self._lock_file_fd is not None



					def       lowerCAmelCase				(  self				:	Any		,     __a				:	Optional[Any]=None		,     __a				:	Union[str, Any]=0.05							)			->  List[str]:


												"""simple docstring"""

												if timeout is None:
																			__lowercase      :							Union[str, Any]				=     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

												__lowercase      :							int				=     id(self							)
												__lowercase      :							Optional[Any]				=     self._lock_file
												__lowercase      :							List[str]				=     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(__a							)
												except:  # noqa
																			# Something did go wrong, so decrement the counter.
																			with self._thread_lock:
																										__lowercase      :							Optional[int]				=     max(0		,     self._lock_counter - 1							)

																			raise
												return _Acquire_ReturnProxy(lock=self							)



					def       lowerCAmelCase				(  self				:	Union[str, Any]		,     __a				:	Optional[Any]=False							)			->  Optional[Any]:


												"""simple docstring"""

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

																										if self._lock_counter == 0 or force:
																																	__lowercase      :							Optional[Any]				=     id(self							)
																																	__lowercase      :							str				=     self._lock_file

																																	logger().debug(F"Attempting to release lock {lock_id} on {lock_filename}"							)
																																	self._release()
																																	__lowercase      :							List[str]				=     0
																																	logger().debug(F"Lock {lock_id} released on {lock_filename}"							)

												return None



					def __enter__(  self				:	Any							)			->  Optional[Any]:


												"""simple docstring"""

												self.acquire()
												return self



					def __exit__(  self				:	List[str]		,     __a				:	str		,     __a				:	int		,     __a				:	List[Any]							)			->  Tuple:


												"""simple docstring"""

												self.release()
												return None



					def __del__(  self				:	Optional[int]							)			->  Optional[int]:


												"""simple docstring"""

												self.release(force=__a							)
												return None



					def       lowerCAmelCase				(  self				:	Tuple		,     __a				:	str		,     __a				:	int							)			->  str:


												"""simple docstring"""

												__lowercase      :							List[Any]				=     os.path.basename(__a							)
												if len(__a							) > max_length and max_length > 0:
																			__lowercase      :							int				=     os.path.dirname(__a							)
																			__lowercase      :							List[str]				=     str(hash(__a							)							)
																			__lowercase      :							Optional[Any]				=     filename[: max_length - len(__a							) - 8] + """...""" + hashed_filename + """.lock"""
																			return os.path.join(__a		,     __a							)
												else:
																			return path







class 		lowerCAmelCase     ( __a			):

					'''simple docstring'''

					def __init__(  self				:	Union[str, Any]		,     __a				:	List[Any]		,     __a				:	Optional[int]=-1		,     __a				:	Tuple=None							)			->  List[Any]:


												"""simple docstring"""

												from .file_utils import relative_to_absolute_path

												super().__init__(__a		,     timeout=__a		,     max_filename_length=__a							)
												__lowercase      :							Tuple				=     """\\\\?\\""" + relative_to_absolute_path(self.lock_file							)



					def       lowerCAmelCase				(  self				:	Dict							)			->  Union[str, Any]:


												"""simple docstring"""

												__lowercase      :							List[str]				=     os.O_RDWR | os.O_CREAT | os.O_TRUNC

												try:
																			__lowercase      :							Tuple				=     os.open(self._lock_file		,     __a							)
												except OSError:
																			pass
												else:
																			try:
																										msvcrt.locking(__a		,     msvcrt.LK_NBLCK		,     1							)
																			except OSError:
																										os.close(__a							)
																			else:
																										__lowercase      :							Union[str, Any]				=     fd
												return None



					def       lowerCAmelCase				(  self				:	Any							)			->  List[Any]:


												"""simple docstring"""

												__lowercase      :							Optional[Any]				=     self._lock_file_fd
												__lowercase      :							int				=     None
												msvcrt.locking(__a		,     msvcrt.LK_UNLCK		,     1							)
												os.close(__a							)

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







class 		lowerCAmelCase     ( __a			):

					'''simple docstring'''

					def __init__(  self				:	List[str]		,     __a				:	Optional[Any]		,     __a				:	str=-1		,     __a				:	List[str]=None							)			->  Any:


												"""simple docstring"""

												__lowercase      :							Dict				=     os.statvfs(os.path.dirname(__a							)							).f_namemax
												super().__init__(__a		,     timeout=__a		,     max_filename_length=__a							)



					def       lowerCAmelCase				(  self				:	Optional[Any]							)			->  Dict:


												"""simple docstring"""

												__lowercase      :							List[Any]				=     os.O_RDWR | os.O_CREAT | os.O_TRUNC
												__lowercase      :							List[str]				=     os.open(self._lock_file		,     __a							)

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



					def       lowerCAmelCase				(  self				:	List[Any]							)			->  Optional[int]:


												"""simple docstring"""

												__lowercase      :							Any				=     self._lock_file_fd
												__lowercase      :							List[str]				=     None
												fcntl.flock(__a		,     fcntl.LOCK_UN							)
												os.close(__a							)
												return None







class 		lowerCAmelCase     ( __a			):

					'''simple docstring'''

					def       lowerCAmelCase				(  self				:	List[str]							)			->  Union[str, Any]:


												"""simple docstring"""

												__lowercase      :							Tuple				=     os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
												try:
																			__lowercase      :							Union[str, Any]				=     os.open(self._lock_file		,     __a							)
												except OSError:
																			pass
												else:
																			__lowercase      :							Optional[int]				=     fd
												return None



					def       lowerCAmelCase				(  self				:	List[str]							)			->  Tuple:


												"""simple docstring"""

												os.close(self._lock_file_fd							)
												__lowercase      :							int				=     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[Any]						   =       None

if msvcrt:
	lowerCamelCase : List[Any]						   =       WindowsFileLock
elif fcntl:
	lowerCamelCase : List[Any]						   =       UnixFileLock
else:
	lowerCamelCase : Union[str, Any]						   =       SoftFileLock

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