





'''simple docstring'''






import unittest

from parameterized import parameterized

from transformers import OpenLlamaConfig, is_torch_available, set_seed
from transformers.testing_utils import require_torch, torch_device

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


if is_torch_available():
      import torch

      from transformers import OpenLlamaForCausalLM, OpenLlamaForSequenceClassification, OpenLlamaModel




class 			UpperCAmelCase					:




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




      def     lowercase__   (      self      :			Tuple  )    ->				int:
            _lowerCAmelCase												=							ids_tensor([self.batch_size, self.seq_length]      ,      self.vocab_size  )

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

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

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

            _lowerCAmelCase												=							self.get_config()

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




      def     lowercase__   (      self      :			List[str]  )    ->				Optional[Any]:
            return OpenLlamaConfig(
                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=__snake_case      ,      initializer_range=self.initializer_range      ,      use_stable_embedding=__snake_case      ,      )




      def     lowercase__   (      self      :			Optional[Any]      ,      __snake_case      :			Optional[Any]      ,      __snake_case      :			Tuple      ,      __snake_case      :			Dict      ,      __snake_case      :			str      ,      __snake_case      :			List[Any]      ,      __snake_case      :			Dict      ,      __snake_case      :			List[Any]  )    ->				Dict:
            _lowerCAmelCase												=							OpenLlamaModel(config=__snake_case  )
            model.to(__snake_case  )
            model.eval()
            _lowerCAmelCase												=							model(__snake_case      ,      attention_mask=__snake_case  )
            _lowerCAmelCase												=							model(__snake_case  )
            self.parent.assertEqual(result.last_hidden_state.shape      ,      (self.batch_size, self.seq_length, self.hidden_size)  )




      def     lowercase__   (      self      :			List[Any]      ,      __snake_case      :			Optional[int]      ,      __snake_case      :			Any      ,      __snake_case      :			List[str]      ,      __snake_case      :			Union[str, Any]      ,      __snake_case      :			Optional[int]      ,      __snake_case      :			Optional[int]      ,      __snake_case      :			Tuple      ,      __snake_case      :			Any      ,      __snake_case      :			Optional[int]      ,      )    ->				Tuple:
            _lowerCAmelCase												=							True
            _lowerCAmelCase												=							OpenLlamaModel(__snake_case  )
            model.to(__snake_case  )
            model.eval()
            _lowerCAmelCase												=							model(
                __snake_case      ,      attention_mask=__snake_case      ,      encoder_hidden_states=__snake_case      ,      encoder_attention_mask=__snake_case      ,      )
            _lowerCAmelCase												=							model(
                __snake_case      ,      attention_mask=__snake_case      ,      encoder_hidden_states=__snake_case      ,      )
            _lowerCAmelCase												=							model(__snake_case      ,      attention_mask=__snake_case  )
            self.parent.assertEqual(result.last_hidden_state.shape      ,      (self.batch_size, self.seq_length, self.hidden_size)  )




      def     lowercase__   (      self      :			Tuple      ,      __snake_case      :			str      ,      __snake_case      :			Optional[Any]      ,      __snake_case      :			Dict      ,      __snake_case      :			Dict      ,      __snake_case      :			Optional[int]      ,      __snake_case      :			Union[str, Any]      ,      __snake_case      :			Tuple      ,      __snake_case      :			List[Any]      ,      __snake_case      :			Optional[Any]      ,      )    ->				int:
            _lowerCAmelCase												=							OpenLlamaForCausalLM(config=__snake_case  )
            model.to(__snake_case  )
            model.eval()
            _lowerCAmelCase												=							model(__snake_case      ,      attention_mask=__snake_case      ,      labels=__snake_case  )
            self.parent.assertEqual(result.logits.shape      ,      (self.batch_size, self.seq_length, self.vocab_size)  )




      def     lowercase__   (      self      :			Union[str, Any]      ,      __snake_case      :			Tuple      ,      __snake_case      :			Union[str, Any]      ,      __snake_case      :			Optional[Any]      ,      __snake_case      :			Optional[Any]      ,      __snake_case      :			Union[str, Any]      ,      __snake_case      :			str      ,      __snake_case      :			List[str]      ,      __snake_case      :			List[str]      ,      __snake_case      :			List[Any]      ,      )    ->				List[Any]:
            _lowerCAmelCase												=							True
            _lowerCAmelCase												=							True
            _lowerCAmelCase												=							OpenLlamaForCausalLM(config=__snake_case  )
            model.to(__snake_case  )
            model.eval()

            # first forward pass
            _lowerCAmelCase												=							model(
                __snake_case      ,      attention_mask=__snake_case      ,      encoder_hidden_states=__snake_case      ,      encoder_attention_mask=__snake_case      ,      use_cache=__snake_case      ,      )
            _lowerCAmelCase												=							outputs.past_key_values

            # create hypothetical multiple next token and extent to next_input_ids
            _lowerCAmelCase												=							ids_tensor((self.batch_size, 3)      ,      config.vocab_size  )
            _lowerCAmelCase												=							ids_tensor((self.batch_size, 3)      ,      vocab_size=2  )

            # append to next input_ids and
            _lowerCAmelCase												=							torch.cat([input_ids, next_tokens]      ,      dim=-1  )
            _lowerCAmelCase												=							torch.cat([input_mask, next_mask]      ,      dim=-1  )

            _lowerCAmelCase												=							model(
                __snake_case      ,      attention_mask=__snake_case      ,      encoder_hidden_states=__snake_case      ,      encoder_attention_mask=__snake_case      ,      output_hidden_states=__snake_case      ,      )["""hidden_states"""][0]
            _lowerCAmelCase												=							model(
                __snake_case      ,      attention_mask=__snake_case      ,      encoder_hidden_states=__snake_case      ,      encoder_attention_mask=__snake_case      ,      past_key_values=__snake_case      ,      output_hidden_states=__snake_case      ,      )["""hidden_states"""][0]

            # select random slice
            _lowerCAmelCase												=							ids_tensor((1,)      ,      output_from_past.shape[-1]  ).item()
            _lowerCAmelCase												=							output_from_no_past[:, -3:, random_slice_idx].detach()
            _lowerCAmelCase												=							output_from_past[:, :, random_slice_idx].detach()

            self.parent.assertTrue(output_from_past_slice.shape[1] == next_tokens.shape[1]  )

            # test that outputs are equal for slice
            self.parent.assertTrue(torch.allclose(__snake_case      ,      __snake_case      ,      atol=1E-3  )  )




      def     lowercase__   (      self      :			Optional[int]  )    ->				Union[str, Any]:
            _lowerCAmelCase												=							self.prepare_config_and_inputs()
            (
                (
                _lowerCAmelCase
            )					,							(
                _lowerCAmelCase
            )					,							(
                _lowerCAmelCase
            )					,							(
                _lowerCAmelCase
            )					,							(
                _lowerCAmelCase
            )					,							(
                _lowerCAmelCase
            )					,							(
                _lowerCAmelCase
            )					,							
            )												=							config_and_inputs
            _lowerCAmelCase												=							{"""input_ids""": input_ids, """attention_mask""": input_mask}
            return config, inputs_dict




@require_torch
class 			UpperCAmelCase					(      snake_case_							,		snake_case_							,		snake_case_							,		unittest.TestCase       ):
      _lowercase:      Optional[int]											=       (
          (OpenLlamaModel, OpenLlamaForCausalLM, OpenLlamaForSequenceClassification) if is_torch_available() else ()
      )
      _lowercase:      Any											=       (OpenLlamaForCausalLM,) if is_torch_available() else ()
      _lowercase:      Optional[int]											=       (
          {
              '''feature-extraction''': OpenLlamaModel,
              '''text-classification''': OpenLlamaForSequenceClassification,
              '''text-generation''': OpenLlamaForCausalLM,
              '''zero-shot''': OpenLlamaForSequenceClassification,
          }
          if is_torch_available()
          else {}
      )
      _lowercase:      Tuple											=       False
      _lowercase:      Any											=       False




      def     lowercase__   (      self      :			int  )    ->				str:
            _lowerCAmelCase												=							OpenLlamaModelTester(self  )
            _lowerCAmelCase												=							ConfigTester(self      ,      config_class=__snake_case      ,      hidden_size=37  )




      def     lowercase__   (      self      :			List[str]  )    ->				Union[str, Any]:
            self.config_tester.run_common_tests()




      def     lowercase__   (      self      :			Optional[Any]  )    ->				Optional[int]:
            _lowerCAmelCase												=							self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_model(*__snake_case  )




      def     lowercase__   (      self      :			Dict  )    ->				str:
            _lowerCAmelCase												=							self.model_tester.prepare_config_and_inputs()
            for type in ["absolute", "relative_key", "relative_key_query"]:
                  _lowerCAmelCase												=							type
                  self.model_tester.create_and_check_model(*__snake_case  )




      def     lowercase__   (      self      :			List[Any]  )    ->				Dict:
            _lowerCAmelCase					,							_lowerCAmelCase												=							self.model_tester.prepare_config_and_inputs_for_common()
            _lowerCAmelCase												=							3
            _lowerCAmelCase												=							input_dict["""input_ids"""]
            _lowerCAmelCase												=							input_ids.ne(1  ).to(__snake_case  )
            _lowerCAmelCase												=							ids_tensor([self.model_tester.batch_size]      ,      self.model_tester.type_sequence_label_size  )
            _lowerCAmelCase												=							OpenLlamaForSequenceClassification(__snake_case  )
            model.to(__snake_case  )
            model.eval()
            _lowerCAmelCase												=							model(__snake_case      ,      attention_mask=__snake_case      ,      labels=__snake_case  )
            self.assertEqual(result.logits.shape      ,      (self.model_tester.batch_size, self.model_tester.num_labels)  )




      def     lowercase__   (      self      :			Tuple  )    ->				List[Any]:
            _lowerCAmelCase					,							_lowerCAmelCase												=							self.model_tester.prepare_config_and_inputs_for_common()
            _lowerCAmelCase												=							3
            _lowerCAmelCase												=							"""single_label_classification"""
            _lowerCAmelCase												=							input_dict["""input_ids"""]
            _lowerCAmelCase												=							input_ids.ne(1  ).to(__snake_case  )
            _lowerCAmelCase												=							ids_tensor([self.model_tester.batch_size]      ,      self.model_tester.type_sequence_label_size  )
            _lowerCAmelCase												=							OpenLlamaForSequenceClassification(__snake_case  )
            model.to(__snake_case  )
            model.eval()
            _lowerCAmelCase												=							model(__snake_case      ,      attention_mask=__snake_case      ,      labels=__snake_case  )
            self.assertEqual(result.logits.shape      ,      (self.model_tester.batch_size, self.model_tester.num_labels)  )




      def     lowercase__   (      self      :			Tuple  )    ->				int:
            _lowerCAmelCase					,							_lowerCAmelCase												=							self.model_tester.prepare_config_and_inputs_for_common()
            _lowerCAmelCase												=							3
            _lowerCAmelCase												=							"""multi_label_classification"""
            _lowerCAmelCase												=							input_dict["""input_ids"""]
            _lowerCAmelCase												=							input_ids.ne(1  ).to(__snake_case  )
            _lowerCAmelCase												=							ids_tensor(
                [self.model_tester.batch_size, config.num_labels]      ,      self.model_tester.type_sequence_label_size  ).to(torch.float  )
            _lowerCAmelCase												=							OpenLlamaForSequenceClassification(__snake_case  )
            model.to(__snake_case  )
            model.eval()
            _lowerCAmelCase												=							model(__snake_case      ,      attention_mask=__snake_case      ,      labels=__snake_case  )
            self.assertEqual(result.logits.shape      ,      (self.model_tester.batch_size, self.model_tester.num_labels)  )




      @unittest.skip("""Open-Llama buffers include complex numbers, which breaks this test"""  )
      def     lowercase__   (      self      :			Optional[Any]  )    ->				int:
            pass




      @parameterized.expand([("""linear""",), ("""dynamic""",)]  )
      def     lowercase__   (      self      :			Dict      ,      __snake_case      :			Optional[Any]  )    ->				int:
            _lowerCAmelCase					,							_lowerCAmelCase												=							self.model_tester.prepare_config_and_inputs_for_common()
            _lowerCAmelCase												=							ids_tensor([1, 10]      ,      config.vocab_size  )
            _lowerCAmelCase												=							ids_tensor([1, int(config.max_position_embeddings * 1.5  )]      ,      config.vocab_size  )

            set_seed(42  )  # Fixed seed at init time so the two models get the same random weights
            _lowerCAmelCase												=							OpenLlamaModel(__snake_case  )
            original_model.to(__snake_case  )
            original_model.eval()
            _lowerCAmelCase												=							original_model(__snake_case  ).last_hidden_state
            _lowerCAmelCase												=							original_model(__snake_case  ).last_hidden_state

            set_seed(42  )  # Fixed seed at init time so the two models get the same random weights
            _lowerCAmelCase												=							{"""type""": scaling_type, """factor""": 10.0}
            _lowerCAmelCase												=							OpenLlamaModel(__snake_case  )
            scaled_model.to(__snake_case  )
            scaled_model.eval()
            _lowerCAmelCase												=							scaled_model(__snake_case  ).last_hidden_state
            _lowerCAmelCase												=							scaled_model(__snake_case  ).last_hidden_state

            # Dynamic scaling does not change the RoPE embeddings until it receives an input longer than the original
            # maximum sequence length, so the outputs for the short input should match.
            if scaling_type == "dynamic":
                  self.assertTrue(torch.allclose(__snake_case      ,      __snake_case      ,      atol=1E-5  )  )
            else:
                  self.assertFalse(torch.allclose(__snake_case      ,      __snake_case      ,      atol=1E-5  )  )

            # The output should be different for long inputs
            self.assertFalse(torch.allclose(__snake_case      ,      __snake_case      ,      atol=1E-5  )  )













'''simple docstring'''






import logging
import os
import threading
import time


try:
      import warnings
except ImportError:
      A__    :       Tuple              =None

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

try:
      import fcntl
except ImportError:
      A__    :       Dict              =None


# Backward compatibility
# ------------------------------------------------
try:
      TimeoutError
except NameError:
      A__    :       Tuple              =OSError


# Data
# ------------------------------------------------
A__    :       Optional[Any]              =[
    '''Timeout''',
    '''BaseFileLock''',
    '''WindowsFileLock''',
    '''UnixFileLock''',
    '''SoftFileLock''',
    '''FileLock''',
]

A__    :       Union[str, Any]              ='''3.0.12'''


A__    :       Optional[Any]              =None
def        UpperCamelCase__				(		):





      """simple docstring"""


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




class 			UpperCAmelCase					(      snake_case_       ):




      def __init__(      self      :			Optional[int]      ,      __snake_case      :			str  )    ->				Optional[Any]:
            _lowerCAmelCase												=							lock_file
            return None




      def __str__(      self      :			Union[str, Any]  )    ->				Any:
            _lowerCAmelCase												=							f"The file lock '{self.lock_file}' could not be acquired."
            return temp




class 			UpperCAmelCase					:




      def __init__(      self      :			str      ,      __snake_case      :			str  )    ->				Tuple:
            _lowerCAmelCase												=							lock
            return None




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




      def __exit__(      self      :			Tuple      ,      __snake_case      :			List[Any]      ,      __snake_case      :			int      ,      __snake_case      :			Optional[int]  )    ->				Union[str, Any]:
            self.lock.release()
            return None




class 			UpperCAmelCase					:




      def __init__(      self      :			List[Any]      ,      __snake_case      :			str      ,      __snake_case      :			Tuple=-1      ,      __snake_case      :			List[str]=None  )    ->				Any:
            _lowerCAmelCase												=							max_filename_length if max_filename_length is not None else 2_55
            # 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     lowercase__   (      self      :			int  )    ->				List[str]:
            return self._lock_file




      @property
      def     lowercase__   (      self      :			Dict  )    ->				List[Any]:
            return self._timeout




      @timeout.setter
      def     lowercase__   (      self      :			Tuple      ,      __snake_case      :			int  )    ->				Optional[Any]:
            _lowerCAmelCase												=							float(__snake_case  )
            return None




      def     lowercase__   (      self      :			Optional[int]  )    ->				Union[str, Any]:
            raise NotImplementedError()




      def     lowercase__   (      self      :			Optional[Any]  )    ->				Any:
            raise NotImplementedError()




      @property
      def     lowercase__   (      self      :			str  )    ->				Dict:
            return self._lock_file_fd is not None




      def     lowercase__   (      self      :			Union[str, Any]      ,      __snake_case      :			Union[str, Any]=None      ,      __snake_case      :			List[str]=0.05  )    ->				List[str]:
            # 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     lowercase__   (      self      :			Dict      ,      __snake_case      :			Optional[int]=False  )    ->				Union[str, Any]:
            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      :			List[str]  )    ->				Union[str, Any]:
            self.acquire()
            return self




      def __exit__(      self      :			Union[str, Any]      ,      __snake_case      :			str      ,      __snake_case      :			Tuple      ,      __snake_case      :			Any  )    ->				int:
            self.release()
            return None




      def __del__(      self      :			Optional[Any]  )    ->				Any:
            self.release(force=__snake_case  )
            return None




      def     lowercase__   (      self      :			Optional[Any]      ,      __snake_case      :			str      ,      __snake_case      :			int  )    ->				str:
            _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 			UpperCAmelCase					(      snake_case_       ):




      def __init__(      self      :			str      ,      __snake_case      :			Union[str, Any]      ,      __snake_case      :			Optional[int]=-1      ,      __snake_case      :			str=None  )    ->				Dict:
            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     lowercase__   (      self      :			List[str]  )    ->				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     lowercase__   (      self      :			List[str]  )    ->				List[str]:
            _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 			UpperCAmelCase					(      snake_case_       ):




      def __init__(      self      :			Dict      ,      __snake_case      :			List[Any]      ,      __snake_case      :			List[Any]=-1      ,      __snake_case      :			List[str]=None  )    ->				Dict:
            _lowerCAmelCase												=							os.statvfs(os.path.dirname(__snake_case  )  ).f_namemax
            super().__init__(__snake_case      ,      timeout=__snake_case      ,      max_filename_length=__snake_case  )




      def     lowercase__   (      self      :			Optional[int]  )    ->				List[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     lowercase__   (      self      :			Optional[int]  )    ->				Optional[int]:
            # 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 			UpperCAmelCase					(      snake_case_       ):




      def     lowercase__   (      self      :			Union[str, Any]  )    ->				Dict:
            _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     lowercase__   (      self      :			Any  )    ->				Optional[Any]:
            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




A__    :       Tuple              =None

if msvcrt:
      A__    :       List[Any]              =WindowsFileLock
elif fcntl:
      A__    :       Tuple              =UnixFileLock
else:
      A__    :       Tuple              =SoftFileLock

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






