



import collections
import tempfile
import unittest

import numpy as np

from transformers.testing_utils import (
    is_pt_flax_cross_test,
    require_flax,
    require_torch,
    require_vision,
    slow,
    torch_device,
)
from transformers.utils import is_flax_available, is_torch_available, is_vision_available

from ...test_modeling_flax_common import floats_tensor, ids_tensor, random_attention_mask
from ..bert.test_modeling_flax_bert import FlaxBertModelTester
from ..clip.test_modeling_flax_clip import FlaxCLIPVisionModelTester
from ..vit.test_modeling_flax_vit import FlaxViTModelTester


if is_flax_available():
				from transformers import (
				    FlaxBertModel,
				    FlaxCLIPVisionModel,
				    FlaxVisionTextDualEncoderModel,
				    FlaxViTModel,
				    VisionTextDualEncoderConfig,
				    VisionTextDualEncoderProcessor,
				)
				from transformers.modeling_flax_pytorch_utils import (
				    convert_pytorch_state_dict_to_flax,
				    load_flax_weights_in_pytorch_model,
				)


if is_torch_available():
				import torch

				from transformers import VisionTextDualEncoderModel

if is_vision_available():
				from PIL import Image






def  lowerCamelCase_	(      lowerCAmelCase__	:	Any				)       ->							str:

							'''simple docstring'''






							if isinstance(lowerCAmelCase__			,	collections.abc.Iterable				):
														return x
							return (x, x)



@require_flax
class        lowerCAmelCase__					:


				def __UpperCamelCase						(							self      :  Dict						, __UpperCamelCase      :  List[Any]						, __UpperCamelCase      :  Optional[int]       )		->     str:
											pass


				def __UpperCamelCase						(							self      :  str       )		->     Tuple:
											pass


				def __UpperCamelCase						(							self      :  Any       )		->     List[Any]:
											pass


				def __UpperCamelCase						(							self      :  Any						, __UpperCamelCase      :  np.ndarray						, __UpperCamelCase      :  np.ndarray						, __UpperCamelCase      :  float       )		->     int:
											A      						=    np.abs((a - b)       ).max()
											self.assertLessEqual(__UpperCamelCase						, __UpperCamelCase						, f'''Difference between torch and flax is {diff} (>= {tol}).'''       )


				def __UpperCamelCase						(							self      :  int						, __UpperCamelCase      :  Dict						, __UpperCamelCase      :  Optional[Any]						, __UpperCamelCase      :  Union[str, Any]						, __UpperCamelCase      :  int						, __UpperCamelCase      :  int=None						, **__UpperCamelCase      :  Optional[int]       )		->     Optional[int]:
											A      						=    VisionTextDualEncoderConfig.from_vision_text_configs(__UpperCamelCase						, __UpperCamelCase       )

											A      						=    FlaxVisionTextDualEncoderModel(__UpperCamelCase       )

											A      						=    model(input_ids=__UpperCamelCase						, pixel_values=__UpperCamelCase						, attention_mask=__UpperCamelCase       )

											self.assertEqual(output['text_embeds'].shape						, (input_ids.shape[0], config.projection_dim)       )
											self.assertEqual(output['image_embeds'].shape						, (pixel_values.shape[0], config.projection_dim)       )


				def __UpperCamelCase						(							self      :  List[str]						, __UpperCamelCase      :  Dict						, __UpperCamelCase      :  Union[str, Any]						, __UpperCamelCase      :  Dict						, __UpperCamelCase      :  Optional[Any]						, __UpperCamelCase      :  str=None						, **__UpperCamelCase      :  int       )		->     List[str]:
											A						,   A      						=    self.get_vision_text_model(__UpperCamelCase						, __UpperCamelCase       )
											A      						=    {'vision_model': vision_model, 'text_model': text_model}
											A      						=    FlaxVisionTextDualEncoderModel.from_vision_text_pretrained(**__UpperCamelCase       )

											A      						=    model(input_ids=__UpperCamelCase						, pixel_values=__UpperCamelCase						, attention_mask=__UpperCamelCase       )

											self.assertEqual(output['text_embeds'].shape						, (input_ids.shape[0], model.config.projection_dim)       )
											self.assertEqual(output['image_embeds'].shape						, (pixel_values.shape[0], model.config.projection_dim)       )


				def __UpperCamelCase						(							self      :  str						, __UpperCamelCase      :  Dict						, __UpperCamelCase      :  Optional[int]						, __UpperCamelCase      :  Dict						, __UpperCamelCase      :  Dict						, __UpperCamelCase      :  Dict=None						, **__UpperCamelCase      :  Optional[int]       )		->     int:
											A						,   A      						=    self.get_vision_text_model(__UpperCamelCase						, __UpperCamelCase       )
											A      						=    {'vision_model': vision_model, 'text_model': text_model}
											A      						=    FlaxVisionTextDualEncoderModel.from_vision_text_pretrained(**__UpperCamelCase       )

											A      						=    model(input_ids=__UpperCamelCase						, pixel_values=__UpperCamelCase						, attention_mask=__UpperCamelCase       )
											A      						=    output[0]

											with tempfile.TemporaryDirectory() as tmpdirname:
																		model.save_pretrained(__UpperCamelCase       )
																		A      						=    FlaxVisionTextDualEncoderModel.from_pretrained(__UpperCamelCase       )

																		A      						=    model(input_ids=__UpperCamelCase						, pixel_values=__UpperCamelCase						, attention_mask=__UpperCamelCase       )
																		A      						=    after_output[0]
																		A      						=    np.amax(np.abs(out_a - out_a       )       )
																		self.assertLessEqual(__UpperCamelCase						, 1e-3       )


				def __UpperCamelCase						(							self      :  Optional[int]						, __UpperCamelCase      :  Dict						, __UpperCamelCase      :  Union[str, Any]						, __UpperCamelCase      :  int						, __UpperCamelCase      :  List[Any]						, __UpperCamelCase      :  Optional[int]=None						, **__UpperCamelCase      :  Tuple       )		->     str:
											A						,   A      						=    self.get_vision_text_model(__UpperCamelCase						, __UpperCamelCase       )
											A      						=    {'vision_model': vision_model, 'text_model': text_model}
											A      						=    FlaxVisionTextDualEncoderModel.from_vision_text_pretrained(**__UpperCamelCase       )

											A      						=    model(
											    input_ids=__UpperCamelCase						, pixel_values=__UpperCamelCase						, attention_mask=__UpperCamelCase						, output_attentions=__UpperCamelCase       )

											A      						=    output.vision_model_output.attentions
											self.assertEqual(len(__UpperCamelCase       )						, vision_config.num_hidden_layers       )

											# in ViT, the seq_len equals the number of patches + 1 (we add 1 for the [CLS] token)
											A      						=    to_atuple(vision_model.config.image_size       )
											A      						=    to_atuple(vision_model.config.patch_size       )
											A      						=    (image_size[1] // patch_size[1]) * (image_size[0] // patch_size[0])
											A      						=    num_patches + 1
											self.assertEqual(vision_attentions[0].shape[-3:]						, (vision_config.num_attention_heads, seq_len, seq_len)       )

											A      						=    output.text_model_output.attentions
											self.assertEqual(len(__UpperCamelCase       )						, text_config.num_hidden_layers       )

											self.assertEqual(
											    text_attentions[0].shape[-3:]						, (text_config.num_attention_heads, input_ids.shape[-1], input_ids.shape[-1])						, )


				def __UpperCamelCase						(							self      :  Optional[Any]						, __UpperCamelCase      :  Tuple						, __UpperCamelCase      :  Optional[Any]						, __UpperCamelCase      :  Any       )		->     Dict:
											pt_model.to(__UpperCamelCase       )
											pt_model.eval()

											# prepare inputs
											A      						=    inputs_dict
											A      						=    {k: torch.tensor(v.tolist()       ) for k, v in flax_inputs.items()}

											with torch.no_grad():
																		A      						=    pt_model(**__UpperCamelCase       ).to_tuple()

											A      						=    fx_model(**__UpperCamelCase       ).to_tuple()
											self.assertEqual(len(__UpperCamelCase       )						, len(__UpperCamelCase       )						, 'Output lengths differ between Flax and PyTorch'       )
											for fx_output, pt_output in zip(fx_outputs[:4]						, pt_outputs[:4]       ):
																		self.assert_almost_equals(__UpperCamelCase						, pt_output.numpy()						, 4e-2       )

											# PT -> Flax
											with tempfile.TemporaryDirectory() as tmpdirname:
																		pt_model.save_pretrained(__UpperCamelCase       )
																		A      						=    FlaxVisionTextDualEncoderModel.from_pretrained(__UpperCamelCase						, from_pt=__UpperCamelCase       )

											A      						=    fx_model_loaded(**__UpperCamelCase       ).to_tuple()
											self.assertEqual(len(__UpperCamelCase       )						, len(__UpperCamelCase       )						, 'Output lengths differ between Flax and PyTorch'       )
											for fx_output_loaded, pt_output in zip(fx_outputs_loaded[:4]						, pt_outputs[:4]       ):
																		self.assert_almost_equals(__UpperCamelCase						, pt_output.numpy()						, 4e-2       )

											# Flax -> PT
											with tempfile.TemporaryDirectory() as tmpdirname:
																		fx_model.save_pretrained(__UpperCamelCase       )
																		A      						=    VisionTextDualEncoderModel.from_pretrained(__UpperCamelCase						, from_flax=__UpperCamelCase       )

											pt_model_loaded.to(__UpperCamelCase       )
											pt_model_loaded.eval()

											with torch.no_grad():
																		A      						=    pt_model_loaded(**__UpperCamelCase       ).to_tuple()

											self.assertEqual(len(__UpperCamelCase       )						, len(__UpperCamelCase       )						, 'Output lengths differ between Flax and PyTorch'       )
											for fx_output, pt_output_loaded in zip(fx_outputs[:4]						, pt_outputs_loaded[:4]       ):
																		self.assert_almost_equals(__UpperCamelCase						, pt_output_loaded.numpy()						, 4e-2       )


				def __UpperCamelCase						(							self      :  Any						, __UpperCamelCase      :  Optional[int]						, __UpperCamelCase      :  Union[str, Any]						, __UpperCamelCase      :  Any       )		->     int:
											A      						=    VisionTextDualEncoderConfig.from_vision_text_configs(__UpperCamelCase						, __UpperCamelCase       )

											A      						=    VisionTextDualEncoderModel(__UpperCamelCase       )
											A      						=    FlaxVisionTextDualEncoderModel(__UpperCamelCase       )

											A      						=    convert_pytorch_state_dict_to_flax(pt_model.state_dict()						, __UpperCamelCase       )
											A      						=    fx_state

											self.check_pt_flax_equivalence(__UpperCamelCase						, __UpperCamelCase						, __UpperCamelCase       )


				def __UpperCamelCase						(							self      :  Any						, __UpperCamelCase      :  Tuple						, __UpperCamelCase      :  Any						, __UpperCamelCase      :  Optional[Any]       )		->     Optional[Any]:
											A      						=    VisionTextDualEncoderConfig.from_vision_text_configs(__UpperCamelCase						, __UpperCamelCase       )

											A      						=    VisionTextDualEncoderModel(__UpperCamelCase       )
											A      						=    FlaxVisionTextDualEncoderModel(__UpperCamelCase       )

											A      						=    load_flax_weights_in_pytorch_model(__UpperCamelCase						, fx_model.params       )

											self.check_pt_flax_equivalence(__UpperCamelCase						, __UpperCamelCase						, __UpperCamelCase       )


				def __UpperCamelCase						(							self      :  Dict       )		->     Optional[int]:
											A      						=    self.prepare_config_and_inputs()
											self.check_model_from_pretrained_configs(**__UpperCamelCase       )


				def __UpperCamelCase						(							self      :  Tuple       )		->     Tuple:
											A      						=    self.prepare_config_and_inputs()
											self.check_vision_text_dual_encoder_from_pretrained(**__UpperCamelCase       )


				def __UpperCamelCase						(							self      :  Optional[int]       )		->     Tuple:
											A      						=    self.prepare_config_and_inputs()
											self.check_save_load(**__UpperCamelCase       )


				def __UpperCamelCase						(							self      :  List[Any]       )		->     Optional[Any]:
											A      						=    self.prepare_config_and_inputs()
											self.check_vision_text_output_attention(**__UpperCamelCase       )


				@is_pt_flax_cross_test
				def __UpperCamelCase						(							self      :  Optional[int]       )		->     Tuple:
											A      						=    self.prepare_config_and_inputs()
											A      						=    config_inputs_dict.pop('vision_config'       )
											A      						=    config_inputs_dict.pop('text_config'       )

											A      						=    config_inputs_dict

											self.check_equivalence_pt_to_flax(__UpperCamelCase						, __UpperCamelCase						, __UpperCamelCase       )
											self.check_equivalence_flax_to_pt(__UpperCamelCase						, __UpperCamelCase						, __UpperCamelCase       )


				@slow
				def __UpperCamelCase						(							self      :  int       )		->     int:
											A						,   A      						=    self.get_pretrained_model_and_inputs()

											A      						=    model_a(**__UpperCamelCase       )
											A      						=    outputs[0]

											with tempfile.TemporaryDirectory() as tmp_dirname:
																		model_a.save_pretrained(__UpperCamelCase       )
																		A      						=    FlaxVisionTextDualEncoderModel.from_pretrained(__UpperCamelCase       )

																		A      						=    model_a(**__UpperCamelCase       )
																		A      						=    after_outputs[0]
																		A      						=    np.amax(np.abs(out_a - out_a       )       )
																		self.assertLessEqual(__UpperCamelCase						, 1e-5       )





@require_flax
class        lowerCAmelCase__					(  _lowerCamelCase ,				unittest.TestCase       ):


				def __UpperCamelCase						(							self      :  Any       )		->     str:
											A      						=    FlaxVisionTextDualEncoderModel.from_vision_text_pretrained(
											    'hf-internal-testing/tiny-random-vit'						, 'hf-internal-testing/tiny-bert'						, vision_from_pt=__UpperCamelCase						, text_from_pt=__UpperCamelCase						, )
											A      						=    13
											A      						=    floats_tensor(
											    [
											        batch_size,
											        model.config.vision_config.num_channels,
											        model.config.vision_config.image_size,
											        model.config.vision_config.image_size,
											    ]       )
											A      						=    ids_tensor([batch_size, 4]						, model.config.text_config.vocab_size       )
											A      						=    random_attention_mask([batch_size, 4]       )
											A      						=    {'pixel_values': pixel_values, 'input_ids': input_ids, 'attention_mask': attention_mask}

											return model, inputs


				def __UpperCamelCase						(							self      :  Tuple						, __UpperCamelCase      :  Optional[Any]						, __UpperCamelCase      :  str       )		->     List[str]:
											A      						=    FlaxViTModel(__UpperCamelCase       )
											A      						=    FlaxBertModel(__UpperCamelCase       )
											return vision_model, text_model


				def __UpperCamelCase						(							self      :  int       )		->     List[str]:
											A      						=    FlaxViTModelTester(self       )
											A      						=    FlaxBertModelTester(self       )
											A      						=    vit_model_tester.prepare_config_and_inputs()
											A      						=    bert_model_tester.prepare_config_and_inputs()

											A						,   A      						=    vision_config_and_inputs

											A						,   A						,   A						,   A      						=    text_config_and_inputs

											# make sure that cross attention layers are added
											return {
											    "text_config": text_config,
											    "vision_config": vision_config,
											    "pixel_values": pixel_values,
											    "attention_mask": attention_mask,
											    "input_ids": input_ids,
											    "token_type_ids": token_type_ids,
											}





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


				def __UpperCamelCase						(							self      :  str       )		->     Any:
											A      						=    FlaxVisionTextDualEncoderModel.from_vision_text_pretrained(
											    'hf-internal-testing/tiny-random-clip'						, 'hf-internal-testing/tiny-bert'						, vision_from_pt=__UpperCamelCase						, text_from_pt=__UpperCamelCase						, )
											A      						=    13
											A      						=    floats_tensor(
											    [
											        batch_size,
											        model.config.vision_config.num_channels,
											        model.config.vision_config.image_size,
											        model.config.vision_config.image_size,
											    ]       )
											A      						=    ids_tensor([batch_size, 4]						, model.config.text_config.vocab_size       )
											A      						=    random_attention_mask([batch_size, 4]       )
											A      						=    {'pixel_values': pixel_values, 'input_ids': input_ids, 'attention_mask': attention_mask}

											return model, inputs


				def __UpperCamelCase						(							self      :  Tuple						, __UpperCamelCase      :  Optional[int]						, __UpperCamelCase      :  Union[str, Any]       )		->     Any:
											A      						=    FlaxCLIPVisionModel(__UpperCamelCase       )
											A      						=    FlaxBertModel(__UpperCamelCase       )
											return vision_model, text_model


				def __UpperCamelCase						(							self      :  Optional[Any]       )		->     Optional[int]:
											A      						=    FlaxCLIPVisionModelTester(self       )
											A      						=    FlaxBertModelTester(self       )
											A      						=    clip_model_tester.prepare_config_and_inputs()
											A      						=    bert_model_tester.prepare_config_and_inputs()

											A						,   A      						=    vision_config_and_inputs

											A						,   A						,   A						,   A      						=    text_config_and_inputs

											# make sure that cross attention layers are added
											return {
											    "text_config": text_config,
											    "vision_config": vision_config,
											    "pixel_values": pixel_values,
											    "attention_mask": attention_mask,
											    "input_ids": input_ids,
											    "token_type_ids": token_type_ids,
											}





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


				@slow
				def __UpperCamelCase						(							self      :  Dict       )		->     str:
											A      						=    FlaxVisionTextDualEncoderModel.from_pretrained('clip-italian/clip-italian'						, logit_scale_init_value=1.0       )
											A      						=    VisionTextDualEncoderProcessor.from_pretrained('clip-italian/clip-italian'       )

											A      						=    Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'       )
											A      						=    processor(
											    text=['una foto di un gatto', 'una foto di un cane']						, images=__UpperCamelCase						, padding=__UpperCamelCase						, return_tensors='np'       )

											A      						=    model(**__UpperCamelCase       )

											# verify the logits
											self.assertEqual(outputs.logits_per_image.shape						, (inputs.pixel_values.shape[0], inputs.input_ids.shape[0])       )
											self.assertEqual(
											    outputs.logits_per_text.shape						, (inputs.input_ids.shape[0], inputs.pixel_values.shape[0])						, )

											A      						=    np.array([[1.2_2_8_4_7_2_7, 0.3_1_0_4_1_2_2]]       )

											self.assertTrue(np.allclose(outputs.logits_per_image						, __UpperCamelCase						, atol=1e-3       )       )




import logging
import os
import threading
import time


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

try:
				import msvcrt
except ImportError:
				__snake_case     :Union[str, Any]			       =None

try:
				import fcntl
except ImportError:
				__snake_case     :str			       =None


# Backward compatibility
# ------------------------------------------------
try:
				TimeoutError
except NameError:
				__snake_case     :str			       =OSError


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

__snake_case     :str			       ='3.0.12'


__snake_case     :str			       =None






def  lowerCamelCase_	(      )       ->							List[str]:

							'''simple docstring'''






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



class        lowerCAmelCase__					(  _lowerCamelCase       ):


				def __init__(							self      :  Tuple						, __UpperCamelCase      :  Union[str, Any]       )		->     List[Any]:
											A      						=    lock_file
											return None


				def __str__(							self      :  List[Any]       )		->     int:
											A      						=    f'''The file lock \'{self.lock_file}\' could not be acquired.'''
											return temp





class        lowerCAmelCase__					:


				def __init__(							self      :  int						, __UpperCamelCase      :  Union[str, Any]       )		->     List[str]:
											A      						=    lock
											return None


				def __enter__(							self      :  Dict       )		->     Dict:
											return self.lock


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





class        lowerCAmelCase__					:


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

											# The default timeout value.
											A      						=    timeout

											# We use this lock primarily for the lock counter.
											A      						=    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.
											A      						=    0
											return None


				@property
				def __UpperCamelCase						(							self      :  str       )		->     Union[str, Any]:
											return self._lock_file


				@property
				def __UpperCamelCase						(							self      :  Tuple       )		->     Union[str, Any]:
											return self._timeout


				@timeout.setter
				def __UpperCamelCase						(							self      :  Union[str, Any]						, __UpperCamelCase      :  Any       )		->     Tuple:
											A      						=    float(__UpperCamelCase       )
											return None


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


				def __UpperCamelCase						(							self      :  int       )		->     str:
											raise NotImplementedError()


				@property
				def __UpperCamelCase						(							self      :  List[Any]       )		->     Optional[Any]:
											return self._lock_file_fd is not None


				def __UpperCamelCase						(							self      :  int						, __UpperCamelCase      :  List[str]=None						, __UpperCamelCase      :  Any=0.0_5       )		->     Any:
											# Use the default timeout, if no timeout is provided.
											if timeout is None:
																		A      						=    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

											A      						=    id(self       )
											A      						=    self._lock_file
											A      						=    time.time()
											try:
																		while True:
																									with self._thread_lock:
																																if not self.is_locked:
																																							logger().debug(f'''Attempting to acquire lock {lock_id} on {lock_filename}'''       )
																																							self._acquire()

																									if self.is_locked:
																																logger().debug(f'''Lock {lock_id} acquired on {lock_filename}'''       )
																																break
																									elif timeout >= 0 and time.time() - start_time > timeout:
																																logger().debug(f'''Timeout on acquiring lock {lock_id} on {lock_filename}'''       )
																																raise Timeout(self._lock_file       )
																									else:
																																logger().debug(
																																    f'''Lock {lock_id} not acquired on {lock_filename}, waiting {poll_intervall} seconds ...'''       )
																																time.sleep(__UpperCamelCase       )
											except:  # noqa
																		# Something did go wrong, so decrement the counter.
																		with self._thread_lock:
																									A      						=    max(0						, self._lock_counter - 1       )

																		raise
											return _Acquire_ReturnProxy(lock=self       )


				def __UpperCamelCase						(							self      :  Optional[int]						, __UpperCamelCase      :  Tuple=False       )		->     Tuple:
											with self._thread_lock:
																		if self.is_locked:
																									self._lock_counter -= 1

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

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

											return None


				def __enter__(							self      :  int       )		->     Dict:
											self.acquire()
											return self


				def __exit__(							self      :  List[str]						, __UpperCamelCase      :  Dict						, __UpperCamelCase      :  Optional[Any]						, __UpperCamelCase      :  str       )		->     Dict:
											self.release()
											return None


				def __del__(							self      :  Union[str, Any]       )		->     Optional[int]:
											self.release(force=__UpperCamelCase       )
											return None


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





class        lowerCAmelCase__					(  _lowerCamelCase       ):


				def __init__(							self      :  Tuple						, __UpperCamelCase      :  List[str]						, __UpperCamelCase      :  Tuple=-1						, __UpperCamelCase      :  Optional[Any]=None       )		->     Union[str, Any]:
											from .file_utils import relative_to_absolute_path

											super().__init__(__UpperCamelCase						, timeout=__UpperCamelCase						, max_filename_length=__UpperCamelCase       )
											A      						=    '\\\\?\\' + relative_to_absolute_path(self.lock_file       )


				def __UpperCamelCase						(							self      :  Any       )		->     Any:
											A      						=    os.O_RDWR | os.O_CREAT | os.O_TRUNC

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


				def __UpperCamelCase						(							self      :  Optional[Any]       )		->     List[Any]:
											A      						=    self._lock_file_fd
											A      						=    None
											msvcrt.locking(__UpperCamelCase						, msvcrt.LK_UNLCK						, 1       )
											os.close(__UpperCamelCase       )

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





class        lowerCAmelCase__					(  _lowerCamelCase       ):


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


				def __UpperCamelCase						(							self      :  Any       )		->     int:
											A      						=    os.O_RDWR | os.O_CREAT | os.O_TRUNC
											A      						=    os.open(self._lock_file						, __UpperCamelCase       )

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


				def __UpperCamelCase						(							self      :  Optional[int]       )		->     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
											A      						=    self._lock_file_fd
											A      						=    None
											fcntl.flock(__UpperCamelCase						, fcntl.LOCK_UN       )
											os.close(__UpperCamelCase       )
											return None





class        lowerCAmelCase__					(  _lowerCamelCase       ):


				def __UpperCamelCase						(							self      :  int       )		->     Optional[int]:
											A      						=    os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
											try:
																		A      						=    os.open(self._lock_file						, __UpperCamelCase       )
											except OSError:
																		pass
											else:
																		A      						=    fd
											return None


				def __UpperCamelCase						(							self      :  Optional[Any]       )		->     List[str]:
											os.close(self._lock_file_fd       )
											A      						=    None

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





__snake_case     :List[str]			       =None

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

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