"""simple docstring"""




import argparse
import json
from pathlib import Path

import requests
import torch
from huggingface_hub import cached_download, hf_hub_download, hf_hub_url
from PIL import Image

from transformers import DetaConfig, DetaForObjectDetection, DetaImageProcessor, SwinConfig
from transformers.utils import logging


logging.set_verbosity_info()
__lowerCAmelCase     :				Tuple    				=    logging.get_logger(__name__)







def 				_UpperCAmelCase		(							lowerCamelCase__  ):





		"""simple docstring"""




		lowerCAmelCase__   = SwinConfig(
		    embed_dim=192      ,	depths=(2, 2, 18, 2)      ,	num_heads=(6, 12, 24, 48)      ,	window_size=12      ,	out_features=["""stage2""", """stage3""", """stage4"""]      ,	)

		lowerCAmelCase__   = DetaConfig(
		    backbone_config=lowerCamelCase__      ,	num_queries=900      ,	encoder_ffn_dim=2048      ,	decoder_ffn_dim=2048      ,	num_feature_levels=5      ,	assign_first_stage=lowerCamelCase__      ,	with_box_refine=lowerCamelCase__      ,	two_stage=lowerCamelCase__      ,	)

		# set labels
		lowerCAmelCase__   = """huggingface/label-files"""
		if "o365" in model_name:
				lowerCAmelCase__   = 366
				lowerCAmelCase__   = """object365-id2label.json"""
		else:
				lowerCAmelCase__   = 91
				lowerCAmelCase__   = """coco-detection-id2label.json"""

		lowerCAmelCase__   = num_labels
		lowerCAmelCase__   = json.load(open(cached_download(hf_hub_url(lowerCamelCase__      ,	lowerCamelCase__      ,	repo_type="""dataset"""  )  )      ,	"""r"""  )  )
		lowerCAmelCase__   = {int(lowerCamelCase__  ): v for k, v in idalabel.items()}
		lowerCAmelCase__   = idalabel
		lowerCAmelCase__   = {v: k for k, v in idalabel.items()}

		return config







def 				_UpperCAmelCase		(							lowerCamelCase__  ):





		"""simple docstring"""




		lowerCAmelCase__   = []

		# stem
		# fmt: off
		rename_keys.append(("""backbone.0.body.patch_embed.proj.weight""", """model.backbone.model.embeddings.patch_embeddings.projection.weight""")  )
		rename_keys.append(("""backbone.0.body.patch_embed.proj.bias""", """model.backbone.model.embeddings.patch_embeddings.projection.bias""")  )
		rename_keys.append(("""backbone.0.body.patch_embed.norm.weight""", """model.backbone.model.embeddings.norm.weight""")  )
		rename_keys.append(("""backbone.0.body.patch_embed.norm.bias""", """model.backbone.model.embeddings.norm.bias""")  )
		# stages
		for i in range(len(config.backbone_config.depths  )  ):
				for j in range(config.backbone_config.depths[i]  ):
						rename_keys.append((f"""backbone.0.body.layers.{i}.blocks.{j}.norm1.weight""", f"""model.backbone.model.encoder.layers.{i}.blocks.{j}.layernorm_before.weight""")  )
						rename_keys.append((f"""backbone.0.body.layers.{i}.blocks.{j}.norm1.bias""", f"""model.backbone.model.encoder.layers.{i}.blocks.{j}.layernorm_before.bias""")  )
						rename_keys.append((f"""backbone.0.body.layers.{i}.blocks.{j}.attn.relative_position_bias_table""", f"""model.backbone.model.encoder.layers.{i}.blocks.{j}.attention.self.relative_position_bias_table""")  )
						rename_keys.append((f"""backbone.0.body.layers.{i}.blocks.{j}.attn.relative_position_index""", f"""model.backbone.model.encoder.layers.{i}.blocks.{j}.attention.self.relative_position_index""")  )
						rename_keys.append((f"""backbone.0.body.layers.{i}.blocks.{j}.attn.proj.weight""", f"""model.backbone.model.encoder.layers.{i}.blocks.{j}.attention.output.dense.weight""")  )
						rename_keys.append((f"""backbone.0.body.layers.{i}.blocks.{j}.attn.proj.bias""", f"""model.backbone.model.encoder.layers.{i}.blocks.{j}.attention.output.dense.bias""")  )
						rename_keys.append((f"""backbone.0.body.layers.{i}.blocks.{j}.norm2.weight""", f"""model.backbone.model.encoder.layers.{i}.blocks.{j}.layernorm_after.weight""")  )
						rename_keys.append((f"""backbone.0.body.layers.{i}.blocks.{j}.norm2.bias""", f"""model.backbone.model.encoder.layers.{i}.blocks.{j}.layernorm_after.bias""")  )
						rename_keys.append((f"""backbone.0.body.layers.{i}.blocks.{j}.mlp.fc1.weight""", f"""model.backbone.model.encoder.layers.{i}.blocks.{j}.intermediate.dense.weight""")  )
						rename_keys.append((f"""backbone.0.body.layers.{i}.blocks.{j}.mlp.fc1.bias""", f"""model.backbone.model.encoder.layers.{i}.blocks.{j}.intermediate.dense.bias""")  )
						rename_keys.append((f"""backbone.0.body.layers.{i}.blocks.{j}.mlp.fc2.weight""", f"""model.backbone.model.encoder.layers.{i}.blocks.{j}.output.dense.weight""")  )
						rename_keys.append((f"""backbone.0.body.layers.{i}.blocks.{j}.mlp.fc2.bias""", f"""model.backbone.model.encoder.layers.{i}.blocks.{j}.output.dense.bias""")  )

				if i < 3:
						rename_keys.append((f"""backbone.0.body.layers.{i}.downsample.reduction.weight""", f"""model.backbone.model.encoder.layers.{i}.downsample.reduction.weight""")  )
						rename_keys.append((f"""backbone.0.body.layers.{i}.downsample.norm.weight""", f"""model.backbone.model.encoder.layers.{i}.downsample.norm.weight""")  )
						rename_keys.append((f"""backbone.0.body.layers.{i}.downsample.norm.bias""", f"""model.backbone.model.encoder.layers.{i}.downsample.norm.bias""")  )

		rename_keys.append(("""backbone.0.body.norm1.weight""", """model.backbone.model.hidden_states_norms.stage2.weight""")  )
		rename_keys.append(("""backbone.0.body.norm1.bias""", """model.backbone.model.hidden_states_norms.stage2.bias""")  )
		rename_keys.append(("""backbone.0.body.norm2.weight""", """model.backbone.model.hidden_states_norms.stage3.weight""")  )
		rename_keys.append(("""backbone.0.body.norm2.bias""", """model.backbone.model.hidden_states_norms.stage3.bias""")  )
		rename_keys.append(("""backbone.0.body.norm3.weight""", """model.backbone.model.hidden_states_norms.stage4.weight""")  )
		rename_keys.append(("""backbone.0.body.norm3.bias""", """model.backbone.model.hidden_states_norms.stage4.bias""")  )

		# transformer encoder
		for i in range(config.encoder_layers  ):
				rename_keys.append((f"""transformer.encoder.layers.{i}.self_attn.sampling_offsets.weight""", f"""model.encoder.layers.{i}.self_attn.sampling_offsets.weight""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.self_attn.sampling_offsets.bias""", f"""model.encoder.layers.{i}.self_attn.sampling_offsets.bias""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.self_attn.attention_weights.weight""", f"""model.encoder.layers.{i}.self_attn.attention_weights.weight""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.self_attn.attention_weights.bias""", f"""model.encoder.layers.{i}.self_attn.attention_weights.bias""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.self_attn.value_proj.weight""", f"""model.encoder.layers.{i}.self_attn.value_proj.weight""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.self_attn.value_proj.bias""", f"""model.encoder.layers.{i}.self_attn.value_proj.bias""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.self_attn.output_proj.weight""", f"""model.encoder.layers.{i}.self_attn.output_proj.weight""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.self_attn.output_proj.bias""", f"""model.encoder.layers.{i}.self_attn.output_proj.bias""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.norm1.weight""", f"""model.encoder.layers.{i}.self_attn_layer_norm.weight""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.norm1.bias""", f"""model.encoder.layers.{i}.self_attn_layer_norm.bias""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.linear1.weight""", f"""model.encoder.layers.{i}.fc1.weight""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.linear1.bias""", f"""model.encoder.layers.{i}.fc1.bias""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.linear2.weight""", f"""model.encoder.layers.{i}.fc2.weight""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.linear2.bias""", f"""model.encoder.layers.{i}.fc2.bias""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.norm2.weight""", f"""model.encoder.layers.{i}.final_layer_norm.weight""")  )
				rename_keys.append((f"""transformer.encoder.layers.{i}.norm2.bias""", f"""model.encoder.layers.{i}.final_layer_norm.bias""")  )

		# transformer decoder
		for i in range(config.decoder_layers  ):
				rename_keys.append((f"""transformer.decoder.layers.{i}.cross_attn.sampling_offsets.weight""", f"""model.decoder.layers.{i}.encoder_attn.sampling_offsets.weight""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.cross_attn.sampling_offsets.bias""", f"""model.decoder.layers.{i}.encoder_attn.sampling_offsets.bias""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.cross_attn.attention_weights.weight""", f"""model.decoder.layers.{i}.encoder_attn.attention_weights.weight""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.cross_attn.attention_weights.bias""", f"""model.decoder.layers.{i}.encoder_attn.attention_weights.bias""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.cross_attn.value_proj.weight""", f"""model.decoder.layers.{i}.encoder_attn.value_proj.weight""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.cross_attn.value_proj.bias""", f"""model.decoder.layers.{i}.encoder_attn.value_proj.bias""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.cross_attn.output_proj.weight""", f"""model.decoder.layers.{i}.encoder_attn.output_proj.weight""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.cross_attn.output_proj.bias""", f"""model.decoder.layers.{i}.encoder_attn.output_proj.bias""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.norm1.weight""", f"""model.decoder.layers.{i}.encoder_attn_layer_norm.weight""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.norm1.bias""", f"""model.decoder.layers.{i}.encoder_attn_layer_norm.bias""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.self_attn.out_proj.weight""", f"""model.decoder.layers.{i}.self_attn.out_proj.weight""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.self_attn.out_proj.bias""", f"""model.decoder.layers.{i}.self_attn.out_proj.bias""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.norm2.weight""", f"""model.decoder.layers.{i}.self_attn_layer_norm.weight""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.norm2.bias""", f"""model.decoder.layers.{i}.self_attn_layer_norm.bias""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.linear1.weight""", f"""model.decoder.layers.{i}.fc1.weight""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.linear1.bias""", f"""model.decoder.layers.{i}.fc1.bias""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.linear2.weight""", f"""model.decoder.layers.{i}.fc2.weight""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.linear2.bias""", f"""model.decoder.layers.{i}.fc2.bias""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.norm3.weight""", f"""model.decoder.layers.{i}.final_layer_norm.weight""")  )
				rename_keys.append((f"""transformer.decoder.layers.{i}.norm3.bias""", f"""model.decoder.layers.{i}.final_layer_norm.bias""")  )

		# fmt: on

		return rename_keys







def 				_UpperCAmelCase		(							lowerCamelCase__      ,	lowerCamelCase__      ,	lowerCamelCase__  ):





		"""simple docstring"""




		lowerCAmelCase__   = dct.pop(lowerCamelCase__  )
		lowerCAmelCase__   = val







def 				_UpperCAmelCase		(							lowerCamelCase__      ,	lowerCamelCase__  ):





		"""simple docstring"""




		lowerCAmelCase__   = [int(backbone_config.embed_dim * 2**i  ) for i in range(len(backbone_config.depths  )  )]
		for i in range(len(backbone_config.depths  )  ):
				lowerCAmelCase__   = num_features[i]
				for j in range(backbone_config.depths[i]  ):
						# fmt: off
						# read in weights + bias of input projection layer (in original implementation, this is a single matrix + bias)
						lowerCAmelCase__   = state_dict.pop(f"""backbone.0.body.layers.{i}.blocks.{j}.attn.qkv.weight"""  )
						lowerCAmelCase__   = state_dict.pop(f"""backbone.0.body.layers.{i}.blocks.{j}.attn.qkv.bias"""  )
						# next, add query, keys and values (in that order) to the state dict
						lowerCAmelCase__   = in_proj_weight[:dim, :]
						lowerCAmelCase__   = in_proj_bias[: dim]
						lowerCAmelCase__   = in_proj_weight[
						    dim : dim * 2, :
						]
						lowerCAmelCase__   = in_proj_bias[
						    dim : dim * 2
						]
						lowerCAmelCase__   = in_proj_weight[
						    -dim :, :
						]
						lowerCAmelCase__   = in_proj_bias[-dim :]
						# fmt: on







def 				_UpperCAmelCase		(							lowerCamelCase__      ,	lowerCamelCase__  ):





		"""simple docstring"""




		lowerCAmelCase__   = config.d_model
		for i in range(config.decoder_layers  ):
				# read in weights + bias of input projection layer of self-attention
				lowerCAmelCase__   = state_dict.pop(f"""transformer.decoder.layers.{i}.self_attn.in_proj_weight"""  )
				lowerCAmelCase__   = state_dict.pop(f"""transformer.decoder.layers.{i}.self_attn.in_proj_bias"""  )
				# next, add query, keys and values (in that order) to the state dict
				lowerCAmelCase__   = in_proj_weight[:hidden_size, :]
				lowerCAmelCase__   = in_proj_bias[:hidden_size]
				lowerCAmelCase__   = in_proj_weight[
				    hidden_size : hidden_size * 2, :
				]
				lowerCAmelCase__   = in_proj_bias[hidden_size : hidden_size * 2]
				lowerCAmelCase__   = in_proj_weight[-hidden_size:, :]
				lowerCAmelCase__   = in_proj_bias[-hidden_size:]







def 				_UpperCAmelCase		(							):





		"""simple docstring"""




		lowerCAmelCase__   = """http://images.cocodataset.org/val2017/000000039769.jpg"""
		lowerCAmelCase__   = Image.open(requests.get(lowerCamelCase__      ,	stream=lowerCamelCase__  ).raw  )

		return im







@torch.no_grad()
def 				_UpperCAmelCase		(							lowerCamelCase__      ,	lowerCamelCase__      ,	lowerCamelCase__  ):





		"""simple docstring"""




		lowerCAmelCase__   = get_deta_config(lowerCamelCase__  )

		# load original state dict
		if model_name == "deta-swin-large":
				lowerCAmelCase__   = hf_hub_download(repo_id="""nielsr/deta-checkpoints"""      ,	filename="""adet_swin_ft.pth"""  )
		elif model_name == "deta-swin-large-o365":
				lowerCAmelCase__   = hf_hub_download(repo_id="""jozhang97/deta-swin-l-o365"""      ,	filename="""deta_swin_pt_o365.pth"""  )
		else:
				raise ValueError(f"""Model name {model_name} not supported"""  )

		lowerCAmelCase__   = torch.load(lowerCamelCase__      ,	map_location="""cpu"""  )["""model"""]

		# original state dict
		for name, param in state_dict.items():
				print(lowerCamelCase__      ,	param.shape  )

		# rename keys
		lowerCAmelCase__   = create_rename_keys(lowerCamelCase__  )
		for src, dest in rename_keys:
				rename_key(lowerCamelCase__      ,	lowerCamelCase__      ,	lowerCamelCase__  )
		read_in_swin_q_k_v(lowerCamelCase__      ,	config.backbone_config  )
		read_in_decoder_q_k_v(lowerCamelCase__      ,	lowerCamelCase__  )

		# fix some prefixes
		for key in state_dict.copy().keys():
				if "transformer.decoder.class_embed" in key or "transformer.decoder.bbox_embed" in key:
						lowerCAmelCase__   = state_dict.pop(lowerCamelCase__  )
						lowerCAmelCase__   = val
				if "input_proj" in key:
						lowerCAmelCase__   = state_dict.pop(lowerCamelCase__  )
						lowerCAmelCase__   = val
				if "level_embed" in key or "pos_trans" in key or "pix_trans" in key or "enc_output" in key:
						lowerCAmelCase__   = state_dict.pop(lowerCamelCase__  )
						lowerCAmelCase__   = val

    # finally, create HuggingFace model and load state dict
		lowerCAmelCase__   = DetaForObjectDetection(lowerCamelCase__  )
		model.load_state_dict(lowerCamelCase__  )
		model.eval()

		lowerCAmelCase__   = """cuda""" if torch.cuda.is_available() else """cpu"""
		model.to(lowerCamelCase__  )

		# load image processor
		lowerCAmelCase__   = DetaImageProcessor(format="""coco_detection"""  )

		# verify our conversion on image
		lowerCAmelCase__   = prepare_img()
		lowerCAmelCase__   = processor(images=lowerCamelCase__      ,	return_tensors="""pt"""  )
		lowerCAmelCase__   = encoding["""pixel_values"""]
		lowerCAmelCase__   = model(pixel_values.to(lowerCamelCase__  )  )

		# verify logits
		print("""Logits:"""      ,	outputs.logits[0, :3, :3]  )
		print("""Boxes:"""      ,	outputs.pred_boxes[0, :3, :3]  )
		if model_name == "deta-swin-large":
				lowerCAmelCase__   = torch.tensor(
				    [[-7.63_08, -2.84_85, -5.37_37], [-7.20_37, -4.55_05, -4.80_27], [-7.29_43, -4.26_11, -4.66_17]]  )
				lowerCAmelCase__   = torch.tensor([[0.49_87, 0.49_69, 0.99_99], [0.25_49, 0.54_98, 0.48_05], [0.54_98, 0.27_57, 0.05_69]]  )
		elif model_name == "deta-swin-large-o365":
				lowerCAmelCase__   = torch.tensor(
				    [[-8.01_22, -3.57_20, -4.97_17], [-8.15_47, -3.68_86, -4.63_89], [-7.66_10, -3.61_94, -5.01_34]]  )
				lowerCAmelCase__   = torch.tensor([[0.25_23, 0.55_49, 0.48_81], [0.77_15, 0.41_49, 0.46_01], [0.55_03, 0.27_53, 0.05_75]]  )
		assert torch.allclose(outputs.logits[0, :3, :3]      ,	expected_logits.to(lowerCamelCase__  )      ,	atol=1e-4  )
		assert torch.allclose(outputs.pred_boxes[0, :3, :3]      ,	expected_boxes.to(lowerCamelCase__  )      ,	atol=1e-4  )
		print("""Everything ok!"""  )

		if pytorch_dump_folder_path:
				# Save model and processor
				logger.info(f"""Saving PyTorch model and processor to {pytorch_dump_folder_path}..."""  )
				Path(lowerCamelCase__  ).mkdir(exist_ok=lowerCamelCase__  )
				model.save_pretrained(lowerCamelCase__  )
				processor.save_pretrained(lowerCamelCase__  )

		# Push to hub
		if push_to_hub:
				print("""Pushing model and processor to hub..."""  )
				model.push_to_hub(f"""jozhang97/{model_name}"""  )
				processor.push_to_hub(f"""jozhang97/{model_name}"""  )


if __name__ == "__main__":
			__lowerCAmelCase     :				Optional[int]    				=    argparse.ArgumentParser()

			parser.add_argument(
			    "--model_name",
			    type=str,
			    default="deta-swin-large",
			    choices=["deta-swin-large", "deta-swin-large-o365"],
			    help="Name of the model you'd like to convert.",
			)
			parser.add_argument(
			    "--pytorch_dump_folder_path",
			    default=None,
			    type=str,
			    help="Path to the folder to output PyTorch model.",
			)
			parser.add_argument(
			    "--push_to_hub", action="store_true", help="Whether or not to push the converted model to the 🤗 hub."
			)
			__lowerCAmelCase     :				Dict    				=    parser.parse_args()
			convert_deta_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)





"""simple docstring"""




import logging
import os
import threading
import time


try:
			import warnings
except ImportError:
			__lowerCAmelCase     :				Optional[int]    				=    None

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

try:
			import fcntl
except ImportError:
			__lowerCAmelCase     :				Optional[int]    				=    None


# Backward compatibility
# ------------------------------------------------
try:
			TimeoutError
except NameError:
			__lowerCAmelCase     :				int    				=    OSError


# Data
# ------------------------------------------------
__lowerCAmelCase     :				Dict    				=    [
    "Timeout",
    "BaseFileLock",
    "WindowsFileLock",
    "UnixFileLock",
    "SoftFileLock",
    "FileLock",
]

__lowerCAmelCase     :				str    				=    "3.0.12"


__lowerCAmelCase     :				Optional[int]    				=    None







def 				_UpperCAmelCase		(							):





		"""simple docstring"""




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

class    a_   (  __UpperCamelCase      ):




					def __init__(						self    :  str	,    snake_case__    :  Optional[int]  ):
							lowerCAmelCase__   = lock_file
							return None




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

class    a_   :




					def __init__(						self    :  Tuple	,    snake_case__    :  Optional[int]  ):
							lowerCAmelCase__   = lock
							return None




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




					def __exit__(						self    :  List[Any]	,    snake_case__    :  Dict	,    snake_case__    :  str	,    snake_case__    :  List[Any]  ):
							self.lock.release()
							return None

class    a_   :




					def __init__(						self    :  Dict	,    snake_case__    :  Tuple	,    snake_case__    :  Dict=-1	,    snake_case__    :  Dict=None  ):
							lowerCAmelCase__   = max_filename_length if max_filename_length is not None else 255
							# Hash the filename if it's too long
							lowerCAmelCase__   = self.hash_filename_if_too_long(snake_case__	,    snake_case__  )
							# The path to the lock file.
							lowerCAmelCase__   = lock_file

							# The file descriptor for the *_lock_file* as it is returned by the
							# os.open() function.
							# This file lock is only NOT None, if the object currently holds the
							# lock.
							lowerCAmelCase__   = None

							# The default timeout value.
							lowerCAmelCase__   = timeout

							# We use this lock primarily for the lock counter.
							lowerCAmelCase__   = threading.Lock()

							# The lock counter is used for implementing the nested locking
							# mechanism. Whenever the lock is acquired, the counter is increased and
							# the lock is only released, when this value is 0 again.
							lowerCAmelCase__   = 0
							return None




					@property
					def     _SCREAMING_SNAKE_CASE    (						self    :  Tuple  ):
							return self._lock_file




					@property
					def     _SCREAMING_SNAKE_CASE    (						self    :  Union[str, Any]  ):
							return self._timeout




					@timeout.setter
					def     _SCREAMING_SNAKE_CASE    (						self    :  str	,    snake_case__    :  str  ):
							lowerCAmelCase__   = float(snake_case__  )
							return None




					def     _SCREAMING_SNAKE_CASE    (						self    :  List[str]  ):
							raise NotImplementedError()




					def     _SCREAMING_SNAKE_CASE    (						self    :  str  ):
							raise NotImplementedError()




					@property
					def     _SCREAMING_SNAKE_CASE    (						self    :  str  ):
							return self._lock_file_fd is not None




					def     _SCREAMING_SNAKE_CASE    (						self    :  List[Any]	,    snake_case__    :  Union[str, Any]=None	,    snake_case__    :  Optional[int]=0.05  ):
							# Use the default timeout, if no timeout is provided.
							if timeout is None:
									lowerCAmelCase__   = self.timeout

							# Increment the number right at the beginning.
							# We can still undo it, if something fails.
							with self._thread_lock:
									self._lock_counter += 1

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

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

									raise
							return _Acquire_ReturnProxy(lock=self  )




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

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

													logger().debug(F"""Attempting to release lock {lock_id} on {lock_filename}"""  )
													self._release()
													lowerCAmelCase__   = 0
													logger().debug(F"""Lock {lock_id} released on {lock_filename}"""  )

							return None




					def __enter__(						self    :  Optional[Any]  ):
							self.acquire()
							return self




					def __exit__(						self    :  Any	,    snake_case__    :  Optional[Any]	,    snake_case__    :  Optional[Any]	,    snake_case__    :  Optional[Any]  ):
							self.release()
							return None




					def __del__(						self    :  Optional[int]  ):
							self.release(force=snake_case__  )
							return None




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[int]	,    snake_case__    :  str	,    snake_case__    :  int  ):
							lowerCAmelCase__   = os.path.basename(snake_case__  )
							if len(snake_case__  ) > max_length and max_length > 0:
									lowerCAmelCase__   = os.path.dirname(snake_case__  )
									lowerCAmelCase__   = str(hash(snake_case__  )  )
									lowerCAmelCase__   = filename[: max_length - len(snake_case__  ) - 8] + """...""" + hashed_filename + """.lock"""
									return os.path.join(snake_case__	,    snake_case__  )
							else:
									return path

class    a_   (  __UpperCamelCase      ):




					def __init__(						self    :  List[Any]	,    snake_case__    :  int	,    snake_case__    :  str=-1	,    snake_case__    :  Tuple=None  ):
							from .file_utils import relative_to_absolute_path

							super().__init__(snake_case__	,    timeout=snake_case__	,    max_filename_length=snake_case__  )
							lowerCAmelCase__   = """\\\\?\\""" + relative_to_absolute_path(self.lock_file  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Tuple  ):
							lowerCAmelCase__   = os.O_RDWR | os.O_CREAT | os.O_TRUNC

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




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[int]  ):
							lowerCAmelCase__   = self._lock_file_fd
							lowerCAmelCase__   = None
							msvcrt.locking(snake_case__	,    msvcrt.LK_UNLCK	,    1  )
							os.close(snake_case__  )

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

class    a_   (  __UpperCamelCase      ):




					def __init__(						self    :  Union[str, Any]	,    snake_case__    :  List[str]	,    snake_case__    :  str=-1	,    snake_case__    :  List[str]=None  ):
							lowerCAmelCase__   = os.statvfs(os.path.dirname(snake_case__  )  ).f_namemax
							super().__init__(snake_case__	,    timeout=snake_case__	,    max_filename_length=snake_case__  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[Any]  ):
							lowerCAmelCase__   = os.O_RDWR | os.O_CREAT | os.O_TRUNC
							lowerCAmelCase__   = os.open(self._lock_file	,    snake_case__  )

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




					def     _SCREAMING_SNAKE_CASE    (						self    :  str  ):
							# Do not remove the lockfile:
							#
							#   https://github.com/benediktschmitt/py-filelock/issues/31
							#   https://stackoverflow.com/questions/17708885/flock-removing-locked-file-without-race-condition
							lowerCAmelCase__   = self._lock_file_fd
							lowerCAmelCase__   = None
							fcntl.flock(snake_case__	,    fcntl.LOCK_UN  )
							os.close(snake_case__  )
							return None

class    a_   (  __UpperCamelCase      ):




					def     _SCREAMING_SNAKE_CASE    (						self    :  Tuple  ):
							lowerCAmelCase__   = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
							try:
									lowerCAmelCase__   = os.open(self._lock_file	,    snake_case__  )
							except OSError:
									pass
							else:
									lowerCAmelCase__   = fd
							return None




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[int]  ):
							os.close(self._lock_file_fd  )
							lowerCAmelCase__   = None

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

if msvcrt:
			__lowerCAmelCase     :				Tuple    				=    WindowsFileLock
elif fcntl:
			__lowerCAmelCase     :				Optional[int]    				=    UnixFileLock
else:
			__lowerCAmelCase     :				Tuple    				=    SoftFileLock

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




