

'''simple docstring'''



import copy
import inspect
import unittest

import numpy as np
from huggingface_hub import hf_hub_download

from transformers import VideoMAEConfig
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
  import torch
  from torch import nn

  from transformers import (
      MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING,
      VideoMAEForPreTraining,
      VideoMAEForVideoClassification,
      VideoMAEModel,
  )
  from transformers.models.videomae.modeling_videomae import VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
  from transformers import VideoMAEImageProcessor







class 						lowerCAmelCase__   :



 """simple docstring"""



 def __init__(		self					:       Any					,							__SCREAMING_SNAKE_CASE					:       List[Any]					,							__SCREAMING_SNAKE_CASE					:       List[str]=13					,							__SCREAMING_SNAKE_CASE					:       str=10					,							__SCREAMING_SNAKE_CASE					:       Union[str, Any]=3					,							__SCREAMING_SNAKE_CASE					:       int=2					,							__SCREAMING_SNAKE_CASE					:       List[Any]=2					,							__SCREAMING_SNAKE_CASE					:       Tuple=2					,							__SCREAMING_SNAKE_CASE					:       Dict=True					,							__SCREAMING_SNAKE_CASE					:       Dict=True					,							__SCREAMING_SNAKE_CASE					:       Union[str, Any]=32					,							__SCREAMING_SNAKE_CASE					:       int=5					,							__SCREAMING_SNAKE_CASE					:       List[str]=4					,							__SCREAMING_SNAKE_CASE					:       List[Any]=37					,							__SCREAMING_SNAKE_CASE					:       Optional[int]="gelu"					,							__SCREAMING_SNAKE_CASE					:       str=0.1					,							__SCREAMING_SNAKE_CASE					:       Any=0.1					,							__SCREAMING_SNAKE_CASE					:       Tuple=10					,							__SCREAMING_SNAKE_CASE					:       Optional[int]=0.02					,							__SCREAMING_SNAKE_CASE					:       Any=0.9					,							__SCREAMING_SNAKE_CASE					:       List[Any]=None					,							)  -> int:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      parent
      __SCREAMING_SNAKE_CASE     =      batch_size
      __SCREAMING_SNAKE_CASE     =      image_size
      __SCREAMING_SNAKE_CASE     =      num_channels
      __SCREAMING_SNAKE_CASE     =      patch_size
      __SCREAMING_SNAKE_CASE     =      tubelet_size
      __SCREAMING_SNAKE_CASE     =      num_frames
      __SCREAMING_SNAKE_CASE     =      is_training
      __SCREAMING_SNAKE_CASE     =      use_labels
      __SCREAMING_SNAKE_CASE     =      hidden_size
      __SCREAMING_SNAKE_CASE     =      num_hidden_layers
      __SCREAMING_SNAKE_CASE     =      num_attention_heads
      __SCREAMING_SNAKE_CASE     =      intermediate_size
      __SCREAMING_SNAKE_CASE     =      hidden_act
      __SCREAMING_SNAKE_CASE     =      hidden_dropout_prob
      __SCREAMING_SNAKE_CASE     =      attention_probs_dropout_prob
      __SCREAMING_SNAKE_CASE     =      type_sequence_label_size
      __SCREAMING_SNAKE_CASE     =      initializer_range
      __SCREAMING_SNAKE_CASE     =      mask_ratio
      __SCREAMING_SNAKE_CASE     =      scope

      # in VideoMAE, the number of tokens equals num_frames/tubelet_size * num_patches per frame
      __SCREAMING_SNAKE_CASE     =      (image_size // patch_size) ** 2
      __SCREAMING_SNAKE_CASE     =      (num_frames // tubelet_size) * self.num_patches_per_frame

      # use this variable to define bool_masked_pos
      __SCREAMING_SNAKE_CASE     =      int(mask_ratio * self.seq_length			)
 def       UpperCAmelCase__		(		self					:       Optional[Any]			)  -> Optional[int]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      floats_tensor(
          [self.batch_size, self.num_frames, self.num_channels, self.image_size, self.image_size]			)

      __SCREAMING_SNAKE_CASE     =      None
      if self.use_labels:
           __SCREAMING_SNAKE_CASE     =      ids_tensor([self.batch_size]					,							self.type_sequence_label_size			)

      __SCREAMING_SNAKE_CASE     =      self.get_config()

      return config, pixel_values, labels
 def       UpperCAmelCase__		(		self					:       Tuple			)  -> str:


      """simple docstring"""




      return VideoMAEConfig(
          image_size=self.image_size					,							patch_size=self.patch_size					,							num_channels=self.num_channels					,							num_frames=self.num_frames					,							tubelet_size=self.tubelet_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					,							is_decoder=__SCREAMING_SNAKE_CASE					,							initializer_range=self.initializer_range					,							)
 def       UpperCAmelCase__		(		self					:       Any					,							__SCREAMING_SNAKE_CASE					:       int					,							__SCREAMING_SNAKE_CASE					:       Tuple					,							__SCREAMING_SNAKE_CASE					:       List[str]			)  -> str:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      VideoMAEModel(config=__SCREAMING_SNAKE_CASE			)
      model.to(__SCREAMING_SNAKE_CASE			)
      model.eval()
      __SCREAMING_SNAKE_CASE     =      model(__SCREAMING_SNAKE_CASE			)
      self.parent.assertEqual(result.last_hidden_state.shape					,							(self.batch_size, self.seq_length, self.hidden_size)			)
 def       UpperCAmelCase__		(		self					:       List[str]					,							__SCREAMING_SNAKE_CASE					:       int					,							__SCREAMING_SNAKE_CASE					:       Union[str, Any]					,							__SCREAMING_SNAKE_CASE					:       Dict			)  -> str:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      VideoMAEForPreTraining(__SCREAMING_SNAKE_CASE			)
      model.to(__SCREAMING_SNAKE_CASE			)
      model.eval()
      # important: each video needs to have the same number of masked patches
      # hence we define a single mask, which we then repeat for each example in the batch
      __SCREAMING_SNAKE_CASE     =      torch.ones((self.num_masks,)			)
      __SCREAMING_SNAKE_CASE     =      torch.cat([mask, torch.zeros(self.seq_length - mask.size(0			)			)]			)
      __SCREAMING_SNAKE_CASE     =      mask.expand(self.batch_size					,							-1			).bool()

      __SCREAMING_SNAKE_CASE     =      model(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)
      # model only returns predictions for masked patches
      __SCREAMING_SNAKE_CASE     =      mask.sum().item()
      __SCREAMING_SNAKE_CASE     =      3 * self.tubelet_size * self.patch_size**2
      self.parent.assertEqual(result.logits.shape					,							(self.batch_size, num_masked_patches, decoder_num_labels)			)


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


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      self.prepare_config_and_inputs()
      __SCREAMING_SNAKE_CASE   ,			__SCREAMING_SNAKE_CASE   ,			__SCREAMING_SNAKE_CASE     =      config_and_inputs
      __SCREAMING_SNAKE_CASE     =      {"""pixel_values""": pixel_values}
      return config, inputs_dict







@require_torch
class 						lowerCAmelCase__   (      a ,					a ,					unittest.TestCase					):



 """simple docstring"""



 lowerCAmelCase__  					=      (
     (VideoMAEModel, VideoMAEForPreTraining, VideoMAEForVideoClassification) if is_torch_available() else ()
 )
 lowerCAmelCase__  					=      (
     {"feature-extraction": VideoMAEModel, "video-classification": VideoMAEForVideoClassification}
     if is_torch_available()
     else {}
 )

 lowerCAmelCase__  					=      False
 lowerCAmelCase__  					=      False
 lowerCAmelCase__  					=      False
 lowerCAmelCase__  					=      False
 def       UpperCAmelCase__		(		self					:       Any			)  -> int:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      VideoMAEModelTester(self			)
      __SCREAMING_SNAKE_CASE     =      ConfigTester(self					,							config_class=__SCREAMING_SNAKE_CASE					,							has_text_modality=__SCREAMING_SNAKE_CASE					,							hidden_size=37			)
 def       UpperCAmelCase__		(		self					:       Any					,							__SCREAMING_SNAKE_CASE					:       List[str]					,							__SCREAMING_SNAKE_CASE					:       Dict					,							__SCREAMING_SNAKE_CASE					:       Union[str, Any]=False			)  -> int:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      copy.deepcopy(__SCREAMING_SNAKE_CASE			)

      if model_class == VideoMAEForPreTraining:
           # important: each video needs to have the same number of masked patches
           # hence we define a single mask, which we then repeat for each example in the batch
           __SCREAMING_SNAKE_CASE     =      torch.ones((self.model_tester.num_masks,)			)
           __SCREAMING_SNAKE_CASE     =      torch.cat([mask, torch.zeros(self.model_tester.seq_length - mask.size(0			)			)]			)
           __SCREAMING_SNAKE_CASE     =      mask.expand(self.model_tester.batch_size					,							-1			).bool()
           __SCREAMING_SNAKE_CASE     =      bool_masked_pos.to(__SCREAMING_SNAKE_CASE			)

      if return_labels:
           if model_class in [
               *get_values(__SCREAMING_SNAKE_CASE			),
           ]:
                __SCREAMING_SNAKE_CASE     =      torch.zeros(
                    self.model_tester.batch_size					,							dtype=torch.long					,							device=__SCREAMING_SNAKE_CASE			)

      return inputs_dict
 def       UpperCAmelCase__		(		self					:       Any			)  -> Union[str, Any]:


      """simple docstring"""




      self.config_tester.run_common_tests()
 @unittest.skip(reason="""VideoMAE does not use inputs_embeds"""			)
 def       UpperCAmelCase__		(		self					:       Optional[int]			)  -> Optional[Any]:


      """simple docstring"""




      pass
 def       UpperCAmelCase__		(		self					:       Dict			)  -> List[str]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE   ,			__SCREAMING_SNAKE_CASE     =      self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
           __SCREAMING_SNAKE_CASE     =      model_class(__SCREAMING_SNAKE_CASE			)
           self.assertIsInstance(model.get_input_embeddings()					,							(nn.Module)			)
           __SCREAMING_SNAKE_CASE     =      model.get_output_embeddings()
           self.assertTrue(x is None or isinstance(__SCREAMING_SNAKE_CASE					,							nn.Linear			)			)
 def       UpperCAmelCase__		(		self					:       Tuple			)  -> int:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE   ,			__SCREAMING_SNAKE_CASE     =      self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
           __SCREAMING_SNAKE_CASE     =      model_class(__SCREAMING_SNAKE_CASE			)
           __SCREAMING_SNAKE_CASE     =      inspect.signature(model.forward			)
           # signature.parameters is an OrderedDict => so arg_names order is deterministic
           __SCREAMING_SNAKE_CASE     =      [*signature.parameters.keys()]

           __SCREAMING_SNAKE_CASE     =      ["""pixel_values"""]
           self.assertListEqual(arg_names[:1]					,							__SCREAMING_SNAKE_CASE			)
 def       UpperCAmelCase__		(		self					:       Any			)  -> Optional[int]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_model(*__SCREAMING_SNAKE_CASE			)
 def       UpperCAmelCase__		(		self					:       int			)  -> List[str]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_pretraining(*__SCREAMING_SNAKE_CASE			)
 @slow
 def       UpperCAmelCase__		(		self					:       int			)  -> List[str]:


      """simple docstring"""




      for model_name in VIDEOMAE_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
           __SCREAMING_SNAKE_CASE     =      VideoMAEModel.from_pretrained(__SCREAMING_SNAKE_CASE			)
           self.assertIsNotNone(__SCREAMING_SNAKE_CASE			)
 def       UpperCAmelCase__		(		self					:       Optional[Any]			)  -> Union[str, Any]:


      """simple docstring"""




      if not self.has_attentions:
           pass

      else:
           __SCREAMING_SNAKE_CASE   ,			__SCREAMING_SNAKE_CASE     =      self.model_tester.prepare_config_and_inputs_for_common()
           __SCREAMING_SNAKE_CASE     =      True

           for model_class in self.all_model_classes:
                __SCREAMING_SNAKE_CASE     =      self.model_tester.seq_length - self.model_tester.num_masks
                __SCREAMING_SNAKE_CASE     =      (
                    num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length
                )

                __SCREAMING_SNAKE_CASE     =      True
                __SCREAMING_SNAKE_CASE     =      False
                __SCREAMING_SNAKE_CASE     =      True
                __SCREAMING_SNAKE_CASE     =      model_class(__SCREAMING_SNAKE_CASE			)
                model.to(__SCREAMING_SNAKE_CASE			)
                model.eval()
                with torch.no_grad():
                     __SCREAMING_SNAKE_CASE     =      model(**self._prepare_for_class(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)			)
                __SCREAMING_SNAKE_CASE     =      outputs.attentions
                self.assertEqual(len(__SCREAMING_SNAKE_CASE			)					,							self.model_tester.num_hidden_layers			)

                # check that output_attentions also work using config
                del inputs_dict["output_attentions"]
                __SCREAMING_SNAKE_CASE     =      True
                __SCREAMING_SNAKE_CASE     =      model_class(__SCREAMING_SNAKE_CASE			)
                model.to(__SCREAMING_SNAKE_CASE			)
                model.eval()
                with torch.no_grad():
                     __SCREAMING_SNAKE_CASE     =      model(**self._prepare_for_class(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)			)
                __SCREAMING_SNAKE_CASE     =      outputs.attentions
                self.assertEqual(len(__SCREAMING_SNAKE_CASE			)					,							self.model_tester.num_hidden_layers			)

                self.assertListEqual(
                    list(attentions[0].shape[-3:]			)					,							[self.model_tester.num_attention_heads, seq_len, seq_len]					,							)
                __SCREAMING_SNAKE_CASE     =      len(__SCREAMING_SNAKE_CASE			)

                # Check attention is always last and order is fine
                __SCREAMING_SNAKE_CASE     =      True
                __SCREAMING_SNAKE_CASE     =      True
                __SCREAMING_SNAKE_CASE     =      model_class(__SCREAMING_SNAKE_CASE			)
                model.to(__SCREAMING_SNAKE_CASE			)
                model.eval()
                with torch.no_grad():
                     __SCREAMING_SNAKE_CASE     =      model(**self._prepare_for_class(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)			)

                self.assertEqual(out_len + 1					,							len(__SCREAMING_SNAKE_CASE			)			)

                __SCREAMING_SNAKE_CASE     =      outputs.attentions

                self.assertEqual(len(__SCREAMING_SNAKE_CASE			)					,							self.model_tester.num_hidden_layers			)
                self.assertListEqual(
                    list(self_attentions[0].shape[-3:]			)					,							[self.model_tester.num_attention_heads, seq_len, seq_len]					,							)
 def       UpperCAmelCase__		(		self					:       str			)  -> Dict:


      """simple docstring"""




      def check_hidden_states_output(__SCREAMING_SNAKE_CASE					:       Optional[Any]					,							__SCREAMING_SNAKE_CASE					:       Any					,							__SCREAMING_SNAKE_CASE					:       Any			):
           __SCREAMING_SNAKE_CASE     =      model_class(__SCREAMING_SNAKE_CASE			)
           model.to(__SCREAMING_SNAKE_CASE			)
           model.eval()

           with torch.no_grad():
                __SCREAMING_SNAKE_CASE     =      model(**self._prepare_for_class(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)			)

           __SCREAMING_SNAKE_CASE     =      outputs.hidden_states
           __SCREAMING_SNAKE_CASE     =      self.model_tester.num_hidden_layers + 1
           self.assertEqual(len(__SCREAMING_SNAKE_CASE			)					,							__SCREAMING_SNAKE_CASE			)

           __SCREAMING_SNAKE_CASE     =      self.model_tester.seq_length - self.model_tester.num_masks
           __SCREAMING_SNAKE_CASE     =      num_visible_patches if model_class == VideoMAEForPreTraining else self.model_tester.seq_length

           self.assertListEqual(
               list(hidden_states[0].shape[-2:]			)					,							[seq_length, self.model_tester.hidden_size]					,							)

      __SCREAMING_SNAKE_CASE   ,			__SCREAMING_SNAKE_CASE     =      self.model_tester.prepare_config_and_inputs_for_common()

      for model_class in self.all_model_classes:
           __SCREAMING_SNAKE_CASE     =      True
           check_hidden_states_output(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)

           # check that output_hidden_states also work using config
           del inputs_dict["output_hidden_states"]
           __SCREAMING_SNAKE_CASE     =      True

           check_hidden_states_output(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)


 @unittest.skip("""Will be fixed soon by reducing the size of the model used for common tests."""			)
 def       UpperCAmelCase__		(		self					:       int			)  -> Union[str, Any]:


      """simple docstring"""




      pass







def    a__       (  ):




     """simple docstring"""


     __SCREAMING_SNAKE_CASE     =      hf_hub_download(
         repo_id="""hf-internal-testing/spaghetti-video"""      ,    filename="""eating_spaghetti.npy"""      ,    repo_type="""dataset"""						)
     __SCREAMING_SNAKE_CASE     =      np.load(a__						)
     return list(a__						)







@require_torch
@require_vision
class 						lowerCAmelCase__   (      unittest.TestCase					):



 """simple docstring"""



 @cached_property
 def       UpperCAmelCase__		(		self					:       List[Any]			)  -> Optional[int]:


      """simple docstring"""




      return (
          VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]					,							image_std=[0.5, 0.5, 0.5]			)
          if is_vision_available()
          else None
      )
 @slow
 def       UpperCAmelCase__		(		self					:       Optional[Any]			)  -> Tuple:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      VideoMAEForVideoClassification.from_pretrained("""MCG-NJU/videomae-base-finetuned-kinetics"""			).to(
          __SCREAMING_SNAKE_CASE			)

      __SCREAMING_SNAKE_CASE     =      self.default_image_processor
      __SCREAMING_SNAKE_CASE     =      prepare_video()
      __SCREAMING_SNAKE_CASE     =      image_processor(__SCREAMING_SNAKE_CASE					,							return_tensors="""pt"""			).to(__SCREAMING_SNAKE_CASE			)

      # forward pass
      with torch.no_grad():
           __SCREAMING_SNAKE_CASE     =      model(**__SCREAMING_SNAKE_CASE			)

      # verify the logits
      __SCREAMING_SNAKE_CASE     =      torch.Size((1, 400)			)
      self.assertEqual(outputs.logits.shape					,							__SCREAMING_SNAKE_CASE			)

      __SCREAMING_SNAKE_CASE     =      torch.tensor([0.3669, -0.0688, -0.2421]			).to(__SCREAMING_SNAKE_CASE			)

      self.assertTrue(torch.allclose(outputs.logits[0, :3]					,							__SCREAMING_SNAKE_CASE					,							atol=1E-4			)			)


 @slow
 def       UpperCAmelCase__		(		self					:       Dict			)  -> List[Any]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      VideoMAEForPreTraining.from_pretrained("""MCG-NJU/videomae-base-short"""			).to(__SCREAMING_SNAKE_CASE			)

      __SCREAMING_SNAKE_CASE     =      self.default_image_processor
      __SCREAMING_SNAKE_CASE     =      prepare_video()
      __SCREAMING_SNAKE_CASE     =      image_processor(__SCREAMING_SNAKE_CASE					,							return_tensors="""pt"""			).to(__SCREAMING_SNAKE_CASE			)

      # add boolean mask, indicating which patches to mask
      __SCREAMING_SNAKE_CASE     =      hf_hub_download(repo_id="""hf-internal-testing/bool-masked-pos"""					,							filename="""bool_masked_pos.pt"""			)
      __SCREAMING_SNAKE_CASE     =      torch.load(__SCREAMING_SNAKE_CASE			)

      # forward pass
      with torch.no_grad():
           __SCREAMING_SNAKE_CASE     =      model(**__SCREAMING_SNAKE_CASE			)

      # verify the logits
      __SCREAMING_SNAKE_CASE     =      torch.Size([1, 1_408, 1_536]			)
      __SCREAMING_SNAKE_CASE     =      torch.tensor(
          [[0.7994, 0.9612, 0.8508], [0.7401, 0.8958, 0.8302], [0.5862, 0.7468, 0.7325]]					,							device=__SCREAMING_SNAKE_CASE			)
      self.assertEqual(outputs.logits.shape					,							__SCREAMING_SNAKE_CASE			)
      self.assertTrue(torch.allclose(outputs.logits[0, :3, :3]					,							__SCREAMING_SNAKE_CASE					,							atol=1E-4			)			)

      # verify the loss (`config.norm_pix_loss` = `True`)
      __SCREAMING_SNAKE_CASE     =      torch.tensor([0.5142]					,							device=__SCREAMING_SNAKE_CASE			)
      self.assertTrue(torch.allclose(outputs.loss					,							__SCREAMING_SNAKE_CASE					,							atol=1E-4			)			)

      # verify the loss (`config.norm_pix_loss` = `False`)
      __SCREAMING_SNAKE_CASE     =      VideoMAEForPreTraining.from_pretrained("""MCG-NJU/videomae-base-short"""					,							norm_pix_loss=__SCREAMING_SNAKE_CASE			).to(
          __SCREAMING_SNAKE_CASE			)

      with torch.no_grad():
           __SCREAMING_SNAKE_CASE     =      model(**__SCREAMING_SNAKE_CASE			)

      __SCREAMING_SNAKE_CASE     =      torch.tensor(torch.tensor([0.6469]			)					,							device=__SCREAMING_SNAKE_CASE			)
      self.assertTrue(torch.allclose(outputs.loss					,							__SCREAMING_SNAKE_CASE					,							atol=1E-4			)			)









'''simple docstring'''



import logging
import os
import threading
import time


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

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

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


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


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

UpperCAmelCase :       Tuple			   =     '3.0.12'


UpperCAmelCase :       str			   =     None







def    a__       (  ):




     """simple docstring"""


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







class 						lowerCAmelCase__   (      a					):



 """simple docstring"""



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


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      lock_file
      return None


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


      """simple docstring"""




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







class 						lowerCAmelCase__   :



 """simple docstring"""



 def __init__(		self					:       Tuple					,							__SCREAMING_SNAKE_CASE					:       List[Any]			)  -> Union[str, Any]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      lock
      return None
 def __enter__(		self					:       List[str]			)  -> List[Any]:


      """simple docstring"""




      return self.lock


 def __exit__(		self					:       Tuple					,							__SCREAMING_SNAKE_CASE					:       List[str]					,							__SCREAMING_SNAKE_CASE					:       Dict					,							__SCREAMING_SNAKE_CASE					:       Optional[Any]			)  -> List[str]:


      """simple docstring"""




      self.lock.release()
      return None







class 						lowerCAmelCase__   :



 """simple docstring"""



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


      """simple docstring"""




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

      # The default timeout value.
      __SCREAMING_SNAKE_CASE     =      timeout

      # We use this lock primarily for the lock counter.
      __SCREAMING_SNAKE_CASE     =      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.
      __SCREAMING_SNAKE_CASE     =      0
      return None
 @property
 def       UpperCAmelCase__		(		self					:       List[Any]			)  -> List[str]:


      """simple docstring"""




      return self._lock_file
 @property
 def       UpperCAmelCase__		(		self					:       Optional[Any]			)  -> Optional[int]:


      """simple docstring"""




      return self._timeout
 @timeout.setter
 def       UpperCAmelCase__		(		self					:       List[str]					,							__SCREAMING_SNAKE_CASE					:       Dict			)  -> Optional[int]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      float(__SCREAMING_SNAKE_CASE			)
      return None
 def       UpperCAmelCase__		(		self					:       Optional[Any]			)  -> List[Any]:


      """simple docstring"""




      raise NotImplementedError()
 def       UpperCAmelCase__		(		self					:       Union[str, Any]			)  -> Any:


      """simple docstring"""




      raise NotImplementedError()
 @property
 def       UpperCAmelCase__		(		self					:       Union[str, Any]			)  -> int:


      """simple docstring"""




      return self._lock_file_fd is not None
 def       UpperCAmelCase__		(		self					:       Optional[Any]					,							__SCREAMING_SNAKE_CASE					:       Optional[Any]=None					,							__SCREAMING_SNAKE_CASE					:       Optional[int]=0.05			)  -> Optional[Any]:


      """simple docstring"""




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

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

           raise
      return _Acquire_ReturnProxy(lock=self			)
 def       UpperCAmelCase__		(		self					:       Optional[Any]					,							__SCREAMING_SNAKE_CASE					:       Optional[Any]=False			)  -> Dict:


      """simple docstring"""




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

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

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

      return None
 def __enter__(		self					:       Union[str, Any]			)  -> Optional[int]:


      """simple docstring"""




      self.acquire()
      return self
 def __exit__(		self					:       Optional[Any]					,							__SCREAMING_SNAKE_CASE					:       Tuple					,							__SCREAMING_SNAKE_CASE					:       Union[str, Any]					,							__SCREAMING_SNAKE_CASE					:       Any			)  -> Tuple:


      """simple docstring"""




      self.release()
      return None
 def __del__(		self					:       str			)  -> Union[str, Any]:


      """simple docstring"""




      self.release(force=__SCREAMING_SNAKE_CASE			)
      return None


 def       UpperCAmelCase__		(		self					:       Optional[int]					,							__SCREAMING_SNAKE_CASE					:       str					,							__SCREAMING_SNAKE_CASE					:       int			)  -> str:


      """simple docstring"""




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







class 						lowerCAmelCase__   (      a					):



 """simple docstring"""



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


      """simple docstring"""




      from .file_utils import relative_to_absolute_path

      super().__init__(__SCREAMING_SNAKE_CASE					,							timeout=__SCREAMING_SNAKE_CASE					,							max_filename_length=__SCREAMING_SNAKE_CASE			)
      __SCREAMING_SNAKE_CASE     =      """\\\\?\\""" + relative_to_absolute_path(self.lock_file			)
 def       UpperCAmelCase__		(		self					:       Dict			)  -> Optional[int]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      os.O_RDWR | os.O_CREAT | os.O_TRUNC

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


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


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      self._lock_file_fd
      __SCREAMING_SNAKE_CASE     =      None
      msvcrt.locking(__SCREAMING_SNAKE_CASE					,							msvcrt.LK_UNLCK					,							1			)
      os.close(__SCREAMING_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 						lowerCAmelCase__   (      a					):



 """simple docstring"""



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


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      os.statvfs(os.path.dirname(__SCREAMING_SNAKE_CASE			)			).f_namemax
      super().__init__(__SCREAMING_SNAKE_CASE					,							timeout=__SCREAMING_SNAKE_CASE					,							max_filename_length=__SCREAMING_SNAKE_CASE			)
 def       UpperCAmelCase__		(		self					:       Optional[int]			)  -> List[Any]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      os.O_RDWR | os.O_CREAT | os.O_TRUNC
      __SCREAMING_SNAKE_CASE     =      os.open(self._lock_file					,							__SCREAMING_SNAKE_CASE			)

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


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


      """simple docstring"""




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







class 						lowerCAmelCase__   (      a					):



 """simple docstring"""



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


      """simple docstring"""




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


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


      """simple docstring"""




      os.close(self._lock_file_fd			)
      __SCREAMING_SNAKE_CASE     =      None

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


UpperCAmelCase :       Dict			   =     None

if msvcrt:
  UpperCAmelCase :       Optional[int]			   =     WindowsFileLock
elif fcntl:
  UpperCAmelCase :       Optional[Any]			   =     UnixFileLock
else:
  UpperCAmelCase :       int			   =     SoftFileLock

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






