

'''simple docstring'''



import argparse

import fairseq
import torch
from torch import nn

from transformers import (
    MBartaaTokenizer,
    MBartConfig,
    MBartForCausalLM,
    SpeechEncoderDecoderConfig,
    SpeechEncoderDecoderModel,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaModel,
    logging,
)


logging.set_verbosity_info()
UpperCAmelCase :       Any			   =     logging.get_logger(__name__)

UpperCAmelCase :       int			   =     {
    'post_extract_proj': 'feature_projection.projection',
    'encoder.pos_conv.0': 'encoder.pos_conv_embed.conv',
    'self_attn.k_proj': 'encoder.layers.*.attention.k_proj',
    'self_attn.v_proj': 'encoder.layers.*.attention.v_proj',
    'self_attn.q_proj': 'encoder.layers.*.attention.q_proj',
    'self_attn.out_proj': 'encoder.layers.*.attention.out_proj',
    'self_attn_layer_norm': 'encoder.layers.*.layer_norm',
    'fc1': 'encoder.layers.*.feed_forward.intermediate_dense',
    'fc2': 'encoder.layers.*.feed_forward.output_dense',
    'final_layer_norm': 'encoder.layers.*.final_layer_norm',
    'encoder.layer_norm': 'encoder.layer_norm',
    'w2v_model.layer_norm': 'feature_projection.layer_norm',
    'quantizer.weight_proj': 'quantizer.weight_proj',
    'quantizer.vars': 'quantizer.codevectors',
    'project_q': 'project_q',
    'final_proj': 'project_hid',
    'w2v_encoder.proj': 'lm_head',
    'mask_emb': 'masked_spec_embed',
}
UpperCAmelCase :       Union[str, Any]			   =     [
    'lm_head',
    'quantizer.weight_proj',
    'quantizer.codevectors',
    'project_q',
    'project_hid',
]







def    a__       (  a__      ,    a__      ,    a__      ,    a__      ,    a__						):




     """simple docstring"""


     for attribute in key.split("""."""						):
          __SCREAMING_SNAKE_CASE     =      getattr(a__      ,    a__						)

     if weight_type is not None:
          __SCREAMING_SNAKE_CASE     =      getattr(a__      ,    a__						).shape
     else:
          __SCREAMING_SNAKE_CASE     =      hf_pointer.shape

     assert hf_shape == value.shape, (
         F'Shape of hf {key + "." + weight_type if weight_type is not None else ""} is {hf_shape}, but should be'
         F' {value.shape} for {full_name}'
     )

     if weight_type == "weight":
          __SCREAMING_SNAKE_CASE     =      value
     elif weight_type == "weight_g":
          __SCREAMING_SNAKE_CASE     =      value
     elif weight_type == "weight_v":
          __SCREAMING_SNAKE_CASE     =      value
     elif weight_type == "bias":
          __SCREAMING_SNAKE_CASE     =      value
     else:
          __SCREAMING_SNAKE_CASE     =      value

     logger.info(F'{key + "." + weight_type if weight_type is not None else ""} was initialized from {full_name}.'						)







def    a__       (  a__      ,    a__						):




     """simple docstring"""


     __SCREAMING_SNAKE_CASE     =      []
     __SCREAMING_SNAKE_CASE     =      fairseq_model.state_dict()

     __SCREAMING_SNAKE_CASE     =      hf_model.feature_extractor
     __SCREAMING_SNAKE_CASE     =      hf_model.adapter

     for name, value in fairseq_dict.items():
          __SCREAMING_SNAKE_CASE     =      False
          if "conv_layers" in name:
               load_conv_layer(
                   a__      ,    a__      ,    a__      ,    a__      ,    hf_model.config.feat_extract_norm == """group"""      ,    )
               __SCREAMING_SNAKE_CASE     =      True
          elif any(x in name for x in ["""adaptor""", """w2v_encoder.proj.""", """w2v_proj_ln."""]						):
               load_adapter(a__      ,    a__      ,    a__      ,    a__						)
               __SCREAMING_SNAKE_CASE     =      True
          else:
               for key, mapped_key in MAPPING.items():
                    if key in name or key.split("""w2v_model."""						)[-1] == name.split("""."""						)[0]:
                         __SCREAMING_SNAKE_CASE     =      True
                         if "*" in mapped_key:
                              __SCREAMING_SNAKE_CASE     =      name.split(a__						)[0].split("""."""						)[-2]
                              __SCREAMING_SNAKE_CASE     =      mapped_key.replace("""*"""      ,    a__						)
                         if "weight_g" in name:
                              __SCREAMING_SNAKE_CASE     =      """weight_g"""
                         elif "weight_v" in name:
                              __SCREAMING_SNAKE_CASE     =      """weight_v"""
                         elif "bias" in name:
                              __SCREAMING_SNAKE_CASE     =      """bias"""
                         elif "weight" in name:
                              __SCREAMING_SNAKE_CASE     =      """weight"""
                         else:
                              __SCREAMING_SNAKE_CASE     =      None
                         set_recursively(a__      ,    a__      ,    a__      ,    a__      ,    a__						)
                    continue
          if not is_used:
               unused_weights.append(a__						)

     logger.warning(F'Unused weights: {unused_weights}'						)







def    a__       (  a__      ,    a__      ,    a__      ,    a__      ,    a__						):




     """simple docstring"""


     __SCREAMING_SNAKE_CASE     =      full_name.split("""conv_layers."""						)[-1]
     __SCREAMING_SNAKE_CASE     =      name.split("""."""						)
     __SCREAMING_SNAKE_CASE     =      int(items[0]						)
     __SCREAMING_SNAKE_CASE     =      int(items[1]						)

     if type_id == 0:
          if "bias" in name:
               assert value.shape == feature_extractor.conv_layers[layer_id].conv.bias.data.shape, (
                   F'{full_name} has size {value.shape}, but'
                   F' {feature_extractor.conv_layers[layer_id].conv.bias.data.shape} was found.'
               )
               __SCREAMING_SNAKE_CASE     =      value
               logger.info(F'Feat extract conv layer {layer_id} was initialized from {full_name}.'						)
          elif "weight" in name:
               assert value.shape == feature_extractor.conv_layers[layer_id].conv.weight.data.shape, (
                   F'{full_name} has size {value.shape}, but'
                   F' {feature_extractor.conv_layers[layer_id].conv.weight.data.shape} was found.'
               )
               __SCREAMING_SNAKE_CASE     =      value
               logger.info(F'Feat extract conv layer {layer_id} was initialized from {full_name}.'						)
     elif (type_id == 2 and not use_group_norm) or (type_id == 2 and layer_id == 0 and use_group_norm):
          if "bias" in name:
               assert value.shape == feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape, (
                   F'{full_name} has size {value.shape}, but {feature_extractor[layer_id].layer_norm.bias.data.shape} was'
                   " found."
               )
               __SCREAMING_SNAKE_CASE     =      value
               logger.info(F'Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'						)
          elif "weight" in name:
               assert value.shape == feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape, (
                   F'{full_name} has size {value.shape}, but'
                   F' {feature_extractor[layer_id].layer_norm.weight.data.shape} was found.'
               )
               __SCREAMING_SNAKE_CASE     =      value
               logger.info(F'Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'						)
     else:
          unused_weights.append(a__						)







def    a__       (  a__      ,    a__      ,    a__      ,    a__						):




     """simple docstring"""


     __SCREAMING_SNAKE_CASE     =      full_name.split("""adaptor."""						)[-1]
     __SCREAMING_SNAKE_CASE     =      name.split("""."""						)

     if items[1].isdigit():
          __SCREAMING_SNAKE_CASE     =      int(items[1]						)
     else:
          __SCREAMING_SNAKE_CASE     =      None

     if "adaptor" not in full_name:
          if "proj_ln" in full_name:
               # has to be layer norm
               if "bias" in name:
                    assert (
                        value.shape == adapter.proj_layer_norm.bias.data.shape
                    ), F'{full_name} has size {value.shape}, but {adapter.proj_layer_norm.bias.data.shape} was found.'
                    __SCREAMING_SNAKE_CASE     =      value
                    logger.info(F'Adapter proj layer norm bias was initialized from {full_name}.'						)
               if "weight" in name:
                    assert (
                        value.shape == adapter.proj_layer_norm.weight.data.shape
                    ), F'{full_name} has size {value.shape}, but {adapter.proj_layer_norm.weight.data.shape} was found.'
                    __SCREAMING_SNAKE_CASE     =      value
          else:
               # has to be projection layer
               if "bias" in name:
                    assert (
                        value.shape == adapter.proj.bias.data.shape
                    ), F'{full_name} has size {value.shape}, but {adapter.proj.bias.data.shape} was found.'
                    __SCREAMING_SNAKE_CASE     =      value
                    logger.info(F'Adapter proj layer bias was initialized from {full_name}.'						)
               if "weight" in name:
                    assert (
                        value.shape == adapter.proj.weight.data.shape
                    ), F'{full_name} has size {value.shape}, but {adapter.proj.weight.data.shape} was found.'
                    __SCREAMING_SNAKE_CASE     =      value
                    logger.info(F'Adapter proj layer weight was initialized from {full_name}.'						)
     elif isinstance(a__      ,    a__						):
          if "bias" in name:
               assert (
                   value.shape == adapter.layers[layer_id].conv.bias.data.shape
               ), F'{full_name} has size {value.shape}, but {adapter.layers[layer_id].conv.bias.data.shape} was found.'
               __SCREAMING_SNAKE_CASE     =      value
               logger.info(F'Adapter layer {layer_id} bias was initialized from {full_name}.'						)
          elif "weight" in name:
               assert (
                   value.shape == adapter.layers[layer_id].conv.weight.data.shape
               ), F'{full_name} has size {value.shape}, but {adapter.layers[layer_id].conv.weight.data.shape} was found.'
               __SCREAMING_SNAKE_CASE     =      value
               logger.info(F'Adapter layer {layer_id} bias was initialized from {full_name}.'						)
     else:
          unused_weights.append(a__						)







def    a__       (  a__						):




     """simple docstring"""


     __SCREAMING_SNAKE_CASE   ,			__SCREAMING_SNAKE_CASE     =      emb.weight.shape
     __SCREAMING_SNAKE_CASE     =      nn.Linear(a__      ,    a__      ,    bias=a__						)
     __SCREAMING_SNAKE_CASE     =      emb.weight.data
     return lin_layer







@torch.no_grad()
def    a__       (  a__      ,    a__      ,    a__      ,    a__      ,    a__      ,    a__      ,    a__      ,    a__      ,    a__      ,    a__      ,    a__      ,    ):




     """simple docstring"""


     __SCREAMING_SNAKE_CASE     =      WavaVecaConfig.from_pretrained(
         a__      ,    add_adapter=a__      ,    adapter_stride=a__      ,    adapter_kernel_size=a__      ,    use_auth_token=a__      ,    output_hidden_size=a__      ,    )
     __SCREAMING_SNAKE_CASE     =      MBartConfig.from_pretrained(a__						)

     # load model
     __SCREAMING_SNAKE_CASE   ,			__SCREAMING_SNAKE_CASE   ,			__SCREAMING_SNAKE_CASE     =      fairseq.checkpoint_utils.load_model_ensemble_and_task(
         [checkpoint_path]      ,    arg_overrides={
             """config_yaml""": config_yaml_path,
             """data""": """/""".join(dict_path.split("""/"""						)[:-1]						),
             """w2v_path""": checkpoint_path,
             """load_pretrained_decoder_from""": None,
         }      ,    )
     __SCREAMING_SNAKE_CASE     =      model[0].eval()

     # load feature extractor
     __SCREAMING_SNAKE_CASE     =      WavaVecaFeatureExtractor.from_pretrained(a__      ,    use_auth_token=a__						)

     # set weights for wav2vec2 encoder
     __SCREAMING_SNAKE_CASE     =      WavaVecaModel(a__						)

     recursively_load_weights_wavaveca(model.encoder      ,    a__						)

     # load decoder weights
     __SCREAMING_SNAKE_CASE     =      MBartForCausalLM(a__						)
     __SCREAMING_SNAKE_CASE   ,			__SCREAMING_SNAKE_CASE     =      hf_decoder.model.decoder.load_state_dict(model.decoder.state_dict()      ,    strict=a__						)
     logger.warning(F'The following keys are missing when loading the decoder weights: {missing_keys}'						)
     logger.warning(F'The following keys are unexpected when loading the decoder weights: {unexpected_keys}'						)

     __SCREAMING_SNAKE_CASE     =      SpeechEncoderDecoderModel(encoder=a__      ,    decoder=a__						)
     __SCREAMING_SNAKE_CASE     =      False

     __SCREAMING_SNAKE_CASE     =      MBartaaTokenizer(a__						)
     tokenizer.save_pretrained(a__						)

     __SCREAMING_SNAKE_CASE     =      hf_wavavec.config.to_dict()
     __SCREAMING_SNAKE_CASE     =      tokenizer.pad_token_id
     __SCREAMING_SNAKE_CASE     =      tokenizer.bos_token_id
     __SCREAMING_SNAKE_CASE     =      tokenizer.eos_token_id
     __SCREAMING_SNAKE_CASE     =      """mbart50"""
     __SCREAMING_SNAKE_CASE     =      """wav2vec2"""

     __SCREAMING_SNAKE_CASE     =      tokenizer.eos_token_id
     __SCREAMING_SNAKE_CASE     =      25_00_04
     __SCREAMING_SNAKE_CASE     =      tokenizer.eos_token_id

     __SCREAMING_SNAKE_CASE     =      SpeechEncoderDecoderConfig.from_dict(a__						)

     hf_wavavec.save_pretrained(a__						)
     feature_extractor.save_pretrained(a__						)


if __name__ == "__main__":
  UpperCAmelCase :       Any			   =     argparse.ArgumentParser()
  parser.add_argument('--pytorch_dump_folder_path', default=None, type=str, help='Path to the output PyTorch model.')
  parser.add_argument('--checkpoint_path', default=None, type=str, help='Path to fairseq checkpoint')
  parser.add_argument('--dict_path', default=None, type=str, help='Path to dict of fine-tuned model')
  parser.add_argument('--config_yaml_path', default=None, type=str, help='Path to yaml file of fine-tuned model')
  parser.add_argument(
      '--encoder_config_path',
      default='facebook/wav2vec2-xls-r-1b',
      type=str,
      help='Path to hf encoder wav2vec2 checkpoint config',
  )
  parser.add_argument(
      '--decoder_config_path',
      default='facebook/mbart-large-50-one-to-many-mmt',
      type=str,
      help='Path to hf decoder checkpoint config',
  )
  parser.add_argument('--add_adapter', default=True, type=bool, help='whethere to add model adapter layers')
  parser.add_argument('--adapter_stride', default=2, type=int, help='stride of adapter layers')
  parser.add_argument('--adapter_kernel_size', default=3, type=int, help='kernel size of adapter layers')
  parser.add_argument('--encoder_output_dim', default=1_0_2_4, type=int, help='encoder output dim')
  parser.add_argument('--start_token_id', default=2_5_0_0_0_4, type=int, help='`decoder_start_token_id` of model config')

  UpperCAmelCase :       List[str]			   =     parser.parse_args()
  convert_wavaveca_checkpoint(
      args.checkpoint_path,
      args.pytorch_dump_folder_path,
      args.dict_path,
      args.config_yaml_path,
      encoder_config_path=args.encoder_config_path,
      decoder_config_path=args.decoder_config_path,
      add_adapter=args.add_adapter,
      adapter_kernel_size=args.adapter_kernel_size,
      adapter_stride=args.adapter_stride,
      decoder_start_token_id=args.start_token_id,
      encoder_output_dim=args.encoder_output_dim,
  )









'''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')






