



'''simple docstring'''




import contextlib
from multiprocessing import Pool, RLock

from tqdm.auto import tqdm

from ..utils import experimental, logging


lowerCamelCase							:int				    =					logging.get_logger(__name__)







class 	_lowerCAmelCase							:
	__SCREAMING_SNAKE_CASE   :		int					=					None






@experimental
def 	a			( lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__    ):
			'''simple docstring'''





			if ParallelBackendConfig.backend_name is None:
						return _map_with_multiprocessing_pool(
						    lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__    )

			return _map_with_joblib(lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__    )



def 	a			( lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__    ):
			'''simple docstring'''





			A_	:						Dict	=     num_proc if num_proc <= len(lowerCamelCase__    ) else len(lowerCamelCase__    )
			A_	:						int	=     []  # We organize the splits ourselve (contiguous splits)
			for index in range(lowerCamelCase__    ):
						A_	:						Optional[int]	=     len(lowerCamelCase__    ) // num_proc
						A_	:						str	=     len(lowerCamelCase__    ) % num_proc
						A_	:						Union[str, Any]	=     div * index + min(lowerCamelCase__						,      lowerCamelCase__    )
						A_	:						Union[str, Any]	=     start + div + (1 if index < mod else 0)
						split_kwds.append((function, iterable[start:end], types, index, disable_tqdm, desc)    )

			if len(lowerCamelCase__    ) != sum(len(i[1]    ) for i in split_kwds    ):
						raise ValueError(
						    f'Error dividing inputs iterable among processes. '
						    f'Total number of objects {len(lowerCamelCase__    )}, '
						    f'length: {sum(len(i[1]    ) for i in split_kwds    )}'    )

			logger.info(
			    f'Spawning {num_proc} processes for {len(lowerCamelCase__    )} objects in slices of {[len(i[1]    ) for i in split_kwds]}'    )
			A_,    A_	:						Optional[int]	=     None, None
			if not disable_tqdm:
						A_,    A_	:						Optional[int]	=     (RLock(),), tqdm.set_lock
			with Pool(lowerCamelCase__						,      initargs=lowerCamelCase__						,      initializer=lowerCamelCase__    ) as pool:
						A_	:						Any	=     pool.map(lowerCamelCase__						,      lowerCamelCase__    )
			logger.info(f'Finished {num_proc} processes'    )
			A_	:						Any	=     [obj for proc_res in mapped for obj in proc_res]
			logger.info(f'Unpacked {len(lowerCamelCase__    )} objects'    )

			return mapped



def 	a			( lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__						,      lowerCamelCase__    ):
			'''simple docstring'''





			import joblib

			with joblib.parallel_backend(ParallelBackendConfig.backend_name						,      n_jobs=lowerCamelCase__    ):
						return joblib.Parallel()(
						    joblib.delayed(lowerCamelCase__    )((function, obj, types, None, True, None)    ) for obj in iterable    )



@experimental
@contextlib.contextmanager
def 	a			( lowerCamelCase__    ):
			'''simple docstring'''





			A_	:						List[Any]	=     backend_name

			if backend_name == "spark":
						from joblibspark import register_spark

						register_spark()

						# TODO: call create_cache_and_write_probe if "download" in steps
						# TODO: raise NotImplementedError when Dataset.map etc is called

			try:
						yield
			finally:
						A_	:						str	=     None




'''simple docstring'''




import unittest

import numpy as np
import torch

from .utils_summarization import build_mask, compute_token_type_ids, process_story, truncate_or_pad







class 	_lowerCAmelCase							(			unittest.TestCase ):


	def     _a	(self							):
				A_	:						Optional[Any]	=     10


	def     _a	(self							):
				A_	:						Dict	=     [1, 2, 3, 4]
				A_	:						List[Any]	=     [1, 2, 3, 4, 0, 0, 0, 0, 0, 0]
				self.assertEqual(truncate_or_pad(lowercase       ,      self.block_size       ,      0							)       ,      lowercase							)


	def     _a	(self							):
				A_	:						List[Any]	=     [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
				A_	:						Tuple	=     [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
				self.assertEqual(truncate_or_pad(lowercase       ,      self.block_size       ,      0							)       ,      lowercase							)


	def     _a	(self							):
				A_	:						Optional[Any]	=     [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13]
				A_	:						Any	=     [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
				self.assertEqual(truncate_or_pad(lowercase       ,      self.block_size       ,      0							)       ,      lowercase							)


	def     _a	(self							):
				A_	:						List[str]	=     """It was the year of Our Lord one thousand seven hundred and
        seventy-five.\n\nSpiritual revelations were conceded to England at that
        favoured period, as at this."""
				A_,    A_	:						Dict	=     process_story(lowercase							)
				self.assertEqual(lowercase       ,      []							)


	def     _a	(self							):
				A_	:						Optional[int]	=     """"""
				A_,    A_	:						List[str]	=     process_story(lowercase							)
				self.assertEqual(lowercase       ,      []							)
				self.assertEqual(lowercase       ,      []							)


	def     _a	(self							):
				A_	:						Optional[Any]	=     (
				    """It was the year of Our Lord one thousand seven hundred and """
				    """seventy-five\n\nSpiritual revelations were conceded to England """
				    """at that favoured period, as at this.\n@highlight\n\nIt was the best of times"""
				)
				A_,    A_	:						int	=     process_story(lowercase							)

				A_	:						Optional[Any]	=     [
				    """It was the year of Our Lord one thousand seven hundred and seventy-five.""",
				    """Spiritual revelations were conceded to England at that favoured period, as at this.""",
				]
				self.assertEqual(lowercase       ,      lowercase							)

				A_	:						Dict	=     ["""It was the best of times."""]
				self.assertEqual(lowercase       ,      lowercase							)


	def     _a	(self							):
				A_	:						Optional[int]	=     torch.tensor([1, 2, 3, 4]							)
				A_	:						Dict	=     torch.tensor([1, 1, 1, 1]							)
				np.testing.assert_array_equal(build_mask(lowercase       ,      0							).numpy()       ,      expected.numpy()							)


	def     _a	(self							):
				A_	:						str	=     torch.tensor([1, 2, 3, 4, 23, 23, 23]							)
				A_	:						str	=     torch.tensor([1, 1, 1, 1, 0, 0, 0]							)
				np.testing.assert_array_equal(build_mask(lowercase       ,      23							).numpy()       ,      expected.numpy()							)


	def     _a	(self							):
				A_	:						Any	=     torch.tensor([8, 2, 3, 4, 1, 1, 1]							)
				A_	:						List[str]	=     torch.tensor([1, 1, 1, 1, 0, 0, 0]							)
				np.testing.assert_array_equal(build_mask(lowercase       ,      1							).numpy()       ,      expected.numpy()							)


	def     _a	(self							):
				A_	:						List[Any]	=     101
				A_	:						List[Any]	=     torch.tensor([[1, 2, 3, 4, 5, 6], [1, 2, 3, 101, 5, 6], [1, 101, 3, 4, 101, 6]]							)
				A_	:						List[str]	=     torch.tensor([[1, 1, 1, 1, 1, 1], [1, 1, 1, 0, 0, 0], [1, 0, 0, 0, 1, 1]]							)

				A_	:						Dict	=     compute_token_type_ids(lowercase       ,      lowercase							)
				np.testing.assert_array_equal(lowercase       ,      lowercase							)