






import time
from dataclasses import dataclass
from multiprocessing import Pool
from unittest import TestCase
from unittest.mock import patch

import multiprocess
import numpy as np
import pytest

from datasets.utils.py_utils import (
    NestedDataStructure,
    asdict,
    iflatmap_unordered,
    map_nested,
    temp_seed,
    temporary_assignment,
    zip_dict,
)

from .utils import require_tf, require_torch

def        UpperCAmelCase__		(						lowerCamelCase_       :       Union[str, Any]				):  # picklable for multiprocessing
			return x.sum()

def        UpperCAmelCase__		(						lowerCamelCase_       :       int				):  # picklable for multiprocessing
			return i + 1






@dataclass
class 			_UpperCamelCase:
		__SCREAMING_SNAKE_CASE  :       int
		__SCREAMING_SNAKE_CASE  :       str






class 			_UpperCamelCase(       __UpperCamelCase		):

		def    __lowerCAmelCase  (					self		:    List[str]    ):



					'''simple docstring'''



					__a						:  Tuple									=     {}
					__a						:  Optional[Any]									=     []
					__a						:  List[str]									=     1
					__a						:  int									=     [1, 2]
					__a						:  Tuple									=     {'a': 1, 'b': 2}
					__a						:  List[Any]									=     {'a': [1, 2], 'b': [3, 4]}
					__a						:  Optional[Any]									=     {'a': {'1': 1}, 'b': 2}
					__a						:  int									=     {'a': 1, 'b': 2, 'c': 3, 'd': 4}
					__a						:  Optional[int]									=     {}
					__a						:  Optional[Any]									=     []
					__a						:  Optional[Any]									=     2
					__a						:  List[str]									=     [2, 3]
					__a						:  Optional[Any]									=     {'a': 2, 'b': 3}
					__a						:  List[str]									=     {'a': [2, 3], 'b': [4, 5]}
					__a						:  List[str]									=     {'a': {'1': 2}, 'b': 3}
					__a						:  Optional[Any]									=     {'a': 2, 'b': 3, 'c': 4, 'd': 5}
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    )    ,       snake_case__    )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    )    ,       snake_case__    )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    )    ,       snake_case__    )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    )    ,       snake_case__    )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    )    ,       snake_case__    )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    )    ,       snake_case__    )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    )    ,       snake_case__    )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    )    ,       snake_case__    )

					__a						:  Tuple									=     2
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    ,       num_proc=snake_case__    )    ,       snake_case__    )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    ,       num_proc=snake_case__    )    ,       snake_case__    )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    ,       num_proc=snake_case__    )    ,       snake_case__    )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    ,       num_proc=snake_case__    )    ,       snake_case__    )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    ,       num_proc=snake_case__    )    ,       snake_case__    )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    ,       num_proc=snake_case__    )    ,       snake_case__    )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    ,       num_proc=snake_case__    )    ,       snake_case__    )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    ,       num_proc=snake_case__    )    ,       snake_case__    )

					__a						:  Union[str, Any]									=     {'a': np.eye(2    ), 'b': np.zeros(3    ), 'c': np.ones(2    )}
					__a						:  Tuple									=     {'a': 2, 'b': 0, 'c': 2}
					__a						:  int									=     {
					    'a': np.eye(2    ).astype(snake_case__    ),
					    'b': np.zeros(3    ).astype(snake_case__    ),
					    'c': np.ones(2    ).astype(snake_case__    ),
					}
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    ,       map_numpy=snake_case__    )    ,       snake_case__    )
					self.assertEqual(
					    {k: v.tolist() for k, v in map_nested(snake_case__    ,       snake_case__    ,       map_numpy=snake_case__    ).items()}    ,       {k: v.tolist() for k, v in expected_map_nested_sna_int.items()}    ,       )
					self.assertEqual(map_nested(snake_case__    ,       snake_case__    ,       map_numpy=snake_case__    ,       num_proc=snake_case__    )    ,       snake_case__    )
					self.assertEqual(
					    {k: v.tolist() for k, v in map_nested(snake_case__    ,       snake_case__    ,       map_numpy=snake_case__    ,       num_proc=snake_case__    ).items()}    ,       {k: v.tolist() for k, v in expected_map_nested_sna_int.items()}    ,       )
					with self.assertRaises(snake_case__    ):  # can't pickle a local lambda
								map_nested(lambda SCREAMING_SNAKE_CASE__    : x + 1    ,       snake_case__    ,       num_proc=snake_case__    )

		def    __lowerCAmelCase  (					self		:    str    ):



					'''simple docstring'''



					__a						:  Optional[int]									=     {'a': 1, 'b': 2}
					__a						:  Optional[int]									=     {'a': 3, 'b': 4}
					__a						:  Optional[int]									=     {'a': 5, 'b': 6}
					__a						:  str									=     sorted([('a', (1, 3, 5)), ('b', (2, 4, 6))]    )
					self.assertEqual(sorted(zip_dict(snake_case__    ,       snake_case__    ,       snake_case__    )    )    ,       snake_case__    )



		def    __lowerCAmelCase  (					self		:    int    ):



					'''simple docstring'''



					class 			_UpperCamelCase:
							__SCREAMING_SNAKE_CASE  :       Tuple							=							"bar"




					__a						:  Tuple									=     Foo()
					self.assertEqual(foo.my_attr    ,       'bar'    )
					with temporary_assignment(snake_case__    ,       'my_attr'    ,       'BAR'    ):
								self.assertEqual(foo.my_attr    ,       'BAR'    )
					self.assertEqual(foo.my_attr    ,       'bar'    )




@pytest.mark.parametrize(
    'iterable_length, num_proc, expected_num_proc'   ,  [
        (1, None, 1),
        (1, 1, 1),
        (2, None, 1),
        (2, 1, 1),
        (2, 2, 1),
        (2, 3, 1),
        (3, 2, 1),
        (1_6, 1_6, 1_6),
        (1_6, 1_7, 1_6),
        (1_7, 1_6, 1_6),
    ]   ,  )
def        UpperCAmelCase__		(						lowerCamelCase_       :       List[Any]   ,  lowerCamelCase_       :       Optional[int]   ,  lowerCamelCase_       :       Optional[Any]				):




			with patch('datasets.utils.py_utils._single_map_nested'				) as mock_single_map_nested, patch(
			    'datasets.parallel.parallel.Pool'				) as mock_multiprocessing_pool:
						__a						:  str									=     {f'''{i}''': i for i in range(lowerCamelCase__				)}
						__a						:  Tuple									=     map_nested(lambda lowerCamelCase_				: x + 1_0   ,  lowerCamelCase__   ,  num_proc=lowerCamelCase__   ,  parallel_min_length=1_6				)
						if expected_num_proc == 1:
									assert mock_single_map_nested.called
									assert not mock_multiprocessing_pool.called
						else:
									assert not mock_single_map_nested.called
									assert mock_multiprocessing_pool.called
									assert mock_multiprocessing_pool.call_args[0][0] == expected_num_proc






class 			_UpperCamelCase(       __UpperCamelCase		):

		@require_tf
		def    __lowerCAmelCase  (					self		:    Optional[int]    ):



					'''simple docstring'''



					import tensorflow as tf
					from tensorflow.keras import layers

					__a						:  Optional[int]									=     layers.Dense(2    )

					def gen_random_output():
								__a						:  Dict									=     tf.random.uniform((1, 3)    )
								return model(snake_case__    ).numpy()

					with temp_seed(4_2    ,       set_tensorflow=snake_case__    ):
								__a						:  Tuple									=     gen_random_output()
					with temp_seed(4_2    ,       set_tensorflow=snake_case__    ):
								__a						:  List[str]									=     gen_random_output()
					__a						:  Optional[Any]									=     gen_random_output()

					np.testing.assert_equal(snake_case__    ,       snake_case__    )
					self.assertGreater(np.abs(outa - outa    ).sum()    ,       0    )

		@require_torch
		def    __lowerCAmelCase  (					self		:    List[str]    ):



					'''simple docstring'''



					import torch

					def gen_random_output():
								__a						:  str									=     torch.nn.Linear(3    ,       2    )
								__a						:  Union[str, Any]									=     torch.rand(1    ,       3    )
								return model(snake_case__    ).detach().numpy()

					with temp_seed(4_2    ,       set_pytorch=snake_case__    ):
								__a						:  Dict									=     gen_random_output()
					with temp_seed(4_2    ,       set_pytorch=snake_case__    ):
								__a						:  Tuple									=     gen_random_output()
					__a						:  List[str]									=     gen_random_output()

					np.testing.assert_equal(snake_case__    ,       snake_case__    )
					self.assertGreater(np.abs(outa - outa    ).sum()    ,       0    )



		def    __lowerCAmelCase  (					self		:    Union[str, Any]    ):



					'''simple docstring'''



					def gen_random_output():
								return np.random.rand(1    ,       3    )

					with temp_seed(4_2    ):
								__a						:  str									=     gen_random_output()
					with temp_seed(4_2    ):
								__a						:  Optional[Any]									=     gen_random_output()
					__a						:  Union[str, Any]									=     gen_random_output()

					np.testing.assert_equal(snake_case__    ,       snake_case__    )
					self.assertGreater(np.abs(outa - outa    ).sum()    ,       0    )




@pytest.mark.parametrize('input_data'   ,  [{}]				)
def        UpperCAmelCase__		(						lowerCamelCase_       :       Optional[Any]				):
			__a						:  Optional[int]									=     NestedDataStructure(lowerCamelCase__				).data
			assert output_data == input_data

@pytest.mark.parametrize(
    'data, expected_output'   ,  [
        ({}, []),
        ([], []),
        ('foo', ['foo']),
        (['foo', 'bar'], ['foo', 'bar']),
        ([['foo', 'bar']], ['foo', 'bar']),
        ([[['foo'], ['bar']]], ['foo', 'bar']),
        ([[['foo'], 'bar']], ['foo', 'bar']),
        ({'a': 1, 'b': 2}, [1, 2]),
        ({'a': [1, 2], 'b': [3, 4]}, [1, 2, 3, 4]),
        ({'a': [[1, 2]], 'b': [[3, 4]]}, [1, 2, 3, 4]),
        ({'a': [[1, 2]], 'b': [3, 4]}, [1, 2, 3, 4]),
        ({'a': [[[1], [2]]], 'b': [[[3], [4]]]}, [1, 2, 3, 4]),
        ({'a': [[[1], [2]]], 'b': [[3, 4]]}, [1, 2, 3, 4]),
        ({'a': [[[1], [2]]], 'b': [3, 4]}, [1, 2, 3, 4]),
        ({'a': [[[1], [2]]], 'b': [3, [4]]}, [1, 2, 3, 4]),
        ({'a': {'1': 1}, 'b': 2}, [1, 2]),
        ({'a': {'1': [1]}, 'b': 2}, [1, 2]),
        ({'a': {'1': [1]}, 'b': [2]}, [1, 2]),
    ]   ,  )
def        UpperCAmelCase__		(						lowerCamelCase_       :       Optional[int]   ,  lowerCamelCase_       :       str				):
			__a						:  Optional[int]									=     NestedDataStructure(lowerCamelCase__				).flatten()
			assert output == expected_output

def        UpperCAmelCase__		(						):
			__a						:  Union[str, Any]									=     A(x=1   ,  y='foobar'				)
			__a						:  int									=     {'x': 1, 'y': 'foobar'}
			assert asdict(lowerCamelCase__				) == expected_output

			__a						:  Optional[int]									=     {'a': {'b': A(x=1_0   ,  y='foo'				)}, 'c': [A(x=2_0   ,  y='bar'				)]}
			__a						:  Tuple									=     {'a': {'b': {'x': 1_0, 'y': 'foo'}}, 'c': [{'x': 2_0, 'y': 'bar'}]}
			assert asdict(lowerCamelCase__				) == expected_output

			with pytest.raises(lowerCamelCase__				):
						asdict([1, A(x=1_0   ,  y='foo'				)]				)

def        UpperCAmelCase__		(						lowerCamelCase_       :       Any				):
			return text.split()

def        UpperCAmelCase__		(						lowerCamelCase_       :       int				):
			yield (time.time(), content)
			time.sleep(2				)
			yield (time.time(), content)

def        UpperCAmelCase__		(						):




			with Pool(2				) as pool:
						__a						:  Optional[Any]									=     list(iflatmap_unordered(lowerCamelCase__   ,  _split_text   ,  kwargs_iterable=[{'text': 'hello there'}] * 1_0				)				)
						assert out.count('hello'				) == 1_0
						assert out.count('there'				) == 1_0
						assert len(lowerCamelCase__				) == 2_0

			# check multiprocess from pathos (uses dill for pickling)
			with multiprocess.Pool(2				) as pool:
						__a						:  Optional[int]									=     list(iflatmap_unordered(lowerCamelCase__   ,  _split_text   ,  kwargs_iterable=[{'text': 'hello there'}] * 1_0				)				)
						assert out.count('hello'				) == 1_0
						assert out.count('there'				) == 1_0
						assert len(lowerCamelCase__				) == 2_0

			# check that we get items as fast as possible
			with Pool(2				) as pool:
						__a						:  List[str]									=     []
						for yield_time, content in iflatmap_unordered(
						    lowerCamelCase__   ,  _aseconds_generator_of_aitems_with_timing   ,  kwargs_iterable=[{'content': 'a'}, {'content': 'b'}]				):
									assert yield_time < time.time() + 0.1, "we should each item directly after it was yielded"
									out.append(lowerCamelCase__				)
						assert out.count('a'				) == 2
						assert out.count('b'				) == 2
						assert len(lowerCamelCase__				) == 4






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




