


import logging
import os
import threading
import time


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

try:
    import msvcrt
except ImportError:
    SCREAMING_SNAKE_CASE       :		int         =				None

try:
    import fcntl
except ImportError:
    SCREAMING_SNAKE_CASE       :		int         =				None


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


# Data
# ------------------------------------------------
SCREAMING_SNAKE_CASE       :		List[str]         =				[
    "Timeout",
    "BaseFileLock",
    "WindowsFileLock",
    "UnixFileLock",
    "SoftFileLock",
    "FileLock",
]

SCREAMING_SNAKE_CASE       :		Any         =				"3.0.12"


SCREAMING_SNAKE_CASE       :		Tuple         =				None




def     UpperCamelCase    (      )					->							List[str]:
     '''simple docstring'''





     global _logger
     lowercase_       :List[Any]     =       _logger or logging.getLogger(__name__		)
     return _logger






class      UpperCamelCase					(		lowercase__     ):


   '''simple docstring'''







   def __init__(     self   ,							UpperCamelCase_		):
        lowercase_       :Optional[int]     =       lock_file
        return None




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






class      UpperCamelCase					:


   '''simple docstring'''







   def __init__(     self   ,							UpperCamelCase_		):
        lowercase_       :Dict     =       lock
        return None
   def __enter__(     self		):
        return self.lock




   def __exit__(     self   ,							UpperCamelCase_   ,							UpperCamelCase_   ,							UpperCamelCase_		):
        self.lock.release()
        return None






class      UpperCamelCase					:


   '''simple docstring'''







   def __init__(     self   ,							UpperCamelCase_   ,							UpperCamelCase_=-1   ,							UpperCamelCase_=None		):
        lowercase_       :int     =       max_filename_length if max_filename_length is not None else 255
        # Hash the filename if it's too long
        lowercase_       :str     =       self.hash_filename_if_too_long(UpperCamelCase_   ,							UpperCamelCase_		)
        # The path to the lock file.
        lowercase_       :List[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_       :List[str]     =       None

        # The default timeout value.
        lowercase_       :List[str]     =       timeout

        # We use this lock primarily for the lock counter.
        lowercase_       :str     =       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_       :Dict     =       0
        return None
   @property
   def 		UpperCamelCase  (     self		):
        return self._lock_file
   @property
   def 		UpperCamelCase  (     self		):
        return self._timeout
   @timeout.setter
   def 		UpperCamelCase  (     self   ,							UpperCamelCase_		):
        lowercase_       :Optional[Any]     =       float(UpperCamelCase_		)
        return None
   def 		UpperCamelCase  (     self		):
        raise NotImplementedError()
   def 		UpperCamelCase  (     self		):
        raise NotImplementedError()
   @property
   def 		UpperCamelCase  (     self		):
        return self._lock_file_fd is not None
   def 		UpperCamelCase  (     self   ,							UpperCamelCase_=None   ,							UpperCamelCase_=0.05		):
        # Use the default timeout, if no timeout is provided.
        if timeout is None:
             lowercase_       :str     =       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_       :Optional[Any]     =       id(self		)
        lowercase_       :str     =       self._lock_file
        lowercase_       :int     =       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(UpperCamelCase_		)
        except:  # noqa
             # Something did go wrong, so decrement the counter.
             with self._thread_lock:
                  lowercase_       :Optional[Any]     =       max(0   ,							self._lock_counter - 1		)

             raise
        return _Acquire_ReturnProxy(lock=self		)
   def 		UpperCamelCase  (     self   ,							UpperCamelCase_=False		):
        with self._thread_lock:
             if self.is_locked:
                  self._lock_counter -= 1

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

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

        return None
   def __enter__(     self		):
        self.acquire()
        return self
   def __exit__(     self   ,							UpperCamelCase_   ,							UpperCamelCase_   ,							UpperCamelCase_		):
        self.release()
        return None
   def __del__(     self		):
        self.release(force=UpperCamelCase_		)
        return None




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






class      UpperCamelCase					(		lowercase__     ):


   '''simple docstring'''







   def __init__(     self   ,							UpperCamelCase_   ,							UpperCamelCase_=-1   ,							UpperCamelCase_=None		):
        from .file_utils import relative_to_absolute_path

        super().__init__(UpperCamelCase_   ,							timeout=UpperCamelCase_   ,							max_filename_length=UpperCamelCase_		)
        lowercase_       :Dict     =       '''\\\\?\\''' + relative_to_absolute_path(self.lock_file		)
   def 		UpperCamelCase  (     self		):
        lowercase_       :int     =       os.O_RDWR | os.O_CREAT | os.O_TRUNC

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




   def 		UpperCamelCase  (     self		):
        lowercase_       :Union[str, Any]     =       self._lock_file_fd
        lowercase_       :List[str]     =       None
        msvcrt.locking(UpperCamelCase_   ,							msvcrt.LK_UNLCK   ,							1		)
        os.close(UpperCamelCase_		)

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






class      UpperCamelCase					(		lowercase__     ):


   '''simple docstring'''







   def __init__(     self   ,							UpperCamelCase_   ,							UpperCamelCase_=-1   ,							UpperCamelCase_=None		):
        lowercase_       :Dict     =       os.statvfs(os.path.dirname(UpperCamelCase_		)		).f_namemax
        super().__init__(UpperCamelCase_   ,							timeout=UpperCamelCase_   ,							max_filename_length=UpperCamelCase_		)
   def 		UpperCamelCase  (     self		):
        lowercase_       :Any     =       os.O_RDWR | os.O_CREAT | os.O_TRUNC
        lowercase_       :Tuple     =       os.open(self._lock_file   ,							UpperCamelCase_		)

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




   def 		UpperCamelCase  (     self		):
        # 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
        lowercase_       :List[str]     =       self._lock_file_fd
        lowercase_       :Optional[Any]     =       None
        fcntl.flock(UpperCamelCase_   ,							fcntl.LOCK_UN		)
        os.close(UpperCamelCase_		)
        return None






class      UpperCamelCase					(		lowercase__     ):


   '''simple docstring'''







   def 		UpperCamelCase  (     self		):
        lowercase_       :List[str]     =       os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
        try:
             lowercase_       :Tuple     =       os.open(self._lock_file   ,							UpperCamelCase_		)
        except OSError:
             pass
        else:
             lowercase_       :List[str]     =       fd
        return None




   def 		UpperCamelCase  (     self		):
        os.close(self._lock_file_fd		)
        lowercase_       :Optional[Any]     =       None

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

if msvcrt:
    SCREAMING_SNAKE_CASE       :		Union[str, Any]         =				WindowsFileLock
elif fcntl:
    SCREAMING_SNAKE_CASE       :		Tuple         =				UnixFileLock
else:
    SCREAMING_SNAKE_CASE       :		Optional[int]         =				SoftFileLock

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





import unittest

import numpy as np

from transformers import BertConfig, is_flax_available
from transformers.testing_utils import require_flax, slow

from ...test_modeling_flax_common import FlaxModelTesterMixin, floats_tensor, ids_tensor, random_attention_mask


if is_flax_available():
   from transformers.models.bert.modeling_flax_bert import (
       FlaxBertForMaskedLM,
       FlaxBertForMultipleChoice,
       FlaxBertForNextSentencePrediction,
       FlaxBertForPreTraining,
       FlaxBertForQuestionAnswering,
       FlaxBertForSequenceClassification,
       FlaxBertForTokenClassification,
       FlaxBertModel,
   )






class      UpperCamelCase					(		unittest.TestCase     ):


   '''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_=512   ,							UpperCamelCase_=16   ,							UpperCamelCase_=2   ,							UpperCamelCase_=0.02   ,							UpperCamelCase_=4   ,							):
        lowercase_       :Union[str, Any]     =       parent
        lowercase_       :int     =       batch_size
        lowercase_       :int     =       seq_length
        lowercase_       :str     =       is_training
        lowercase_       :Dict     =       use_attention_mask
        lowercase_       :List[Any]     =       use_token_type_ids
        lowercase_       :str     =       use_labels
        lowercase_       :str     =       vocab_size
        lowercase_       :Optional[int]     =       hidden_size
        lowercase_       :Dict     =       num_hidden_layers
        lowercase_       :List[str]     =       num_attention_heads
        lowercase_       :int     =       intermediate_size
        lowercase_       :Union[str, Any]     =       hidden_act
        lowercase_       :Optional[int]     =       hidden_dropout_prob
        lowercase_       :Tuple     =       attention_probs_dropout_prob
        lowercase_       :int     =       max_position_embeddings
        lowercase_       :List[Any]     =       type_vocab_size
        lowercase_       :Any     =       type_sequence_label_size
        lowercase_       :Tuple     =       initializer_range
        lowercase_       :Any     =       num_choices
   def 		UpperCamelCase  (     self		):
        lowercase_       :Optional[int]     =       ids_tensor([self.batch_size, self.seq_length]   ,							self.vocab_size		)

        lowercase_       :List[str]     =       None
        if self.use_attention_mask:
             lowercase_       :Any     =       random_attention_mask([self.batch_size, self.seq_length]		)

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

        lowercase_       :Union[str, Any]     =       BertConfig(
            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   ,							)

        return config, input_ids, token_type_ids, attention_mask
   def 		UpperCamelCase  (     self		):
        lowercase_       :Optional[Any]     =       self.prepare_config_and_inputs()
        lowercase_    ,      lowercase_    ,      lowercase_    ,      lowercase_       :List[str]     =       config_and_inputs
        lowercase_       :Union[str, Any]     =       {'''input_ids''': input_ids, '''token_type_ids''': token_type_ids, '''attention_mask''': attention_mask}
        return config, inputs_dict




   def 		UpperCamelCase  (     self		):
        lowercase_       :Dict     =       self.prepare_config_and_inputs()
        lowercase_    ,      lowercase_    ,      lowercase_    ,      lowercase_       :Dict     =       config_and_inputs

        lowercase_       :str     =       True
        lowercase_       :Dict     =       floats_tensor([self.batch_size, self.seq_length, self.hidden_size]		)
        lowercase_       :Optional[Any]     =       ids_tensor([self.batch_size, self.seq_length]   ,							vocab_size=2		)

        return (
            config,
            input_ids,
            attention_mask,
            encoder_hidden_states,
            encoder_attention_mask,
        )






@require_flax
class      UpperCamelCase					(		lowercase__						,      unittest.TestCase     ):


   '''simple docstring'''







   lowercase					:	int									=True

   lowercase					:	Dict									=(
       (
           FlaxBertModel,
           FlaxBertForPreTraining,
           FlaxBertForMaskedLM,
           FlaxBertForMultipleChoice,
           FlaxBertForQuestionAnswering,
           FlaxBertForNextSentencePrediction,
           FlaxBertForSequenceClassification,
           FlaxBertForTokenClassification,
           FlaxBertForQuestionAnswering,
       )
       if is_flax_available()
       else ()
   )
   def 		UpperCamelCase  (     self		):
        lowercase_       :Union[str, Any]     =       FlaxBertModelTester(self		)




   @slow
   def 		UpperCamelCase  (     self		):
        # Only check this for base model, not necessary for all model classes.
        # This will also help speed-up tests.
        lowercase_       :Dict     =       FlaxBertModel.from_pretrained('''bert-base-cased'''		)
        lowercase_       :str     =       model(np.ones((1, 1)		)		)
        self.assertIsNotNone(UpperCamelCase_		)

