

# Copyright 2023 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import torch

from accelerate import PartialState
from accelerate.utils.operations import broadcast, gather, gather_object, pad_across_processes, reduce

def 			lowerCamelCase__			(       _a):
	return (torch.arange(state.num_processes) + 1.0 + (state.num_processes * state.process_index)).to(state.device)

def 			lowerCamelCase__			(       _a):
	SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							create_tensor(a__)
	SCREAMING_SNAKE_CASE :   Any					  =							gather(a__)
	assert gathered_tensor.tolist() == list(range(1				,      state.num_processes**2 + 1))

def 			lowerCamelCase__			(       _a):
	SCREAMING_SNAKE_CASE :   Optional[Any]					  =							[state.process_index]
	SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							gather_object(a__)
	assert len(a__) == state.num_processes, f"{gathered_obj}, {len(a__)} != {state.num_processes}"
	assert gathered_obj == list(range(state.num_processes)), f"{gathered_obj} != {list(range(state.num_processes))}"

def 			lowerCamelCase__			(       _a):
	SCREAMING_SNAKE_CASE :   List[str]					  =							create_tensor(a__)
	SCREAMING_SNAKE_CASE :   Tuple					  =							broadcast(a__)
	assert broadcasted_tensor.shape == torch.Size([state.num_processes])
	assert broadcasted_tensor.tolist() == list(range(1				,      state.num_processes + 1))

def 			lowerCamelCase__			(       _a):
	# We need to pad the tensor with one more element if we are the main process
	# to ensure that we can pad
	if state.is_main_process:
		SCREAMING_SNAKE_CASE :   Any					  =							torch.arange(state.num_processes + 1).to(state.device)
	else:
		SCREAMING_SNAKE_CASE :   str					  =							torch.arange(state.num_processes).to(state.device)
	SCREAMING_SNAKE_CASE :   int					  =							pad_across_processes(a__)
	assert padded_tensor.shape == torch.Size([state.num_processes + 1])
	if not state.is_main_process:
		assert padded_tensor.tolist() == list(range(0				,      state.num_processes)) + [0]

def 			lowerCamelCase__			(       _a):
	# For now runs on only two processes
	if state.num_processes != 2:
		return
	SCREAMING_SNAKE_CASE :   Union[str, Any]					  =							create_tensor(a__)
	SCREAMING_SNAKE_CASE :   str					  =							reduce(a__				,      "sum")
	SCREAMING_SNAKE_CASE :   int					  =							torch.tensor([4.0, 6]).to(state.device)
	assert torch.allclose(a__				,      a__), f"{reduced_tensor} != {truth_tensor}"

def 			lowerCamelCase__			(       _a):
	# For now runs on only two processes
	if state.num_processes != 2:
		return
	SCREAMING_SNAKE_CASE :   Optional[Any]					  =							create_tensor(a__)
	SCREAMING_SNAKE_CASE :   Optional[Any]					  =							reduce(a__				,      "mean")
	SCREAMING_SNAKE_CASE :   List[str]					  =							torch.tensor([2.0, 3]).to(state.device)
	assert torch.allclose(a__				,      a__), f"{reduced_tensor} != {truth_tensor}"

def 			lowerCamelCase__			(       _a):
	# For xla_spawn (TPUs)
	main()

def 			lowerCamelCase__			(       ):
	SCREAMING_SNAKE_CASE :   Any					  =							PartialState()
	state.print(f"State: {state}")
	state.print("testing gather")
	test_gather(a__)
	state.print("testing gather_object")
	test_gather_object(a__)
	state.print("testing broadcast")
	test_broadcast(a__)
	state.print("testing pad_across_processes")
	test_pad_across_processes(a__)
	state.print("testing reduce_sum")
	test_reduce_sum(a__)
	state.print("testing reduce_mean")
	test_reduce_mean(a__)


if __name__ == "__main__":
			main()


import os
from tempfile import TemporaryDirectory
from unittest import TestCase

import pytest
from absl.testing import parameterized

from datasets import config
from datasets.arrow_reader import HF_GCP_BASE_URL
from datasets.builder import DatasetBuilder
from datasets.dataset_dict import IterableDatasetDict
from datasets.iterable_dataset import IterableDataset
from datasets.load import dataset_module_factory, import_main_class
from datasets.utils.file_utils import cached_path


UpperCAmelCase_							:					Optional[Any]		 =   [
    {'''dataset''': '''wikipedia''', '''config_name''': '''20220301.de'''},
    {'''dataset''': '''wikipedia''', '''config_name''': '''20220301.en'''},
    {'''dataset''': '''wikipedia''', '''config_name''': '''20220301.fr'''},
    {'''dataset''': '''wikipedia''', '''config_name''': '''20220301.frr'''},
    {'''dataset''': '''wikipedia''', '''config_name''': '''20220301.it'''},
    {'''dataset''': '''wikipedia''', '''config_name''': '''20220301.simple'''},
    {'''dataset''': '''snli''', '''config_name''': '''plain_text'''},
    {'''dataset''': '''eli5''', '''config_name''': '''LFQA_reddit'''},
    {'''dataset''': '''wiki40b''', '''config_name''': '''en'''},
    {'''dataset''': '''wiki_dpr''', '''config_name''': '''psgs_w100.nq.compressed'''},
    {'''dataset''': '''wiki_dpr''', '''config_name''': '''psgs_w100.nq.no_index'''},
    {'''dataset''': '''wiki_dpr''', '''config_name''': '''psgs_w100.multiset.no_index'''},
    {'''dataset''': '''natural_questions''', '''config_name''': '''default'''},
]
def 							__SCREAMING_SNAKE_CASE     (   a__    :   str=True       )					->				List[Any]:
	if with_config:
		return [
		    {
		        "testcase_name": d["dataset"] + "/" + d["config_name"],
		        "dataset": d["dataset"],
		        "config_name": d["config_name"],
		    }
		    for d in DATASETS_ON_HF_GCP
		]
	else:
		return [
		    {"testcase_name": dataset, "dataset": dataset} for dataset in {d["dataset"] for d in DATASETS_ON_HF_GCP}
		]






@parameterized.named_parameters(list_datasets_on_hf_gcp_parameters(with_config=_lowercase	)	)
class      lowerCamelCase_      (	_lowercase	):
	_lowercase		:   Optional[int]     					=				None
	_lowercase		:   str     					=				None




	def        lowerCAmelCase_       (     self						:     Dict					,      __A						:     Optional[int]					,      __A						:     Optional[Any]							):
		with TemporaryDirectory() as tmp_dir:
			__A :			List[Any]		    =			dataset_module_factory(__A					,      cache_dir=__A							)

			__A :			Tuple		    =			import_main_class(dataset_module.module_path					,      dataset=__A							)

			__A :			DatasetBuilder		    =			builder_cls(
			    cache_dir=__A					,      config_name=__A					,      hash=dataset_module.hash					,      )

			__A :			List[Any]		    =			"""/""".join(
			    [
			        HF_GCP_BASE_URL,
			        builder_instance._relative_data_dir(with_hash=__A							).replace(os.sep					,      """/"""							),
			        config.DATASET_INFO_FILENAME,
			    ]							)
			__A :			Union[str, Any]		    =			cached_path(__A					,      cache_dir=__A							)
			self.assertTrue(os.path.exists(__A							)							)






@pytest.mark.integration
def 							__SCREAMING_SNAKE_CASE     (   a__    :   Dict       )					->				Optional[Any]:
	__A :			Optional[Any]		    =			tmp_path_factory.mktemp("""test_hf_gcp"""       ) / """test_wikipedia_simple"""
	__A :			Union[str, Any]		    =			dataset_module_factory("""wikipedia"""			,cache_dir=a__       )
	__A :			List[Any]		    =			import_main_class(dataset_module.module_path       )
	__A :			DatasetBuilder		    =			builder_cls(
	    cache_dir=a__			,config_name="""20220301.frr"""			,hash=dataset_module.hash			,)
	# use the HF cloud storage, not the original download_and_prepare that uses apache-beam
	__A :			Any		    =			None
	builder_instance.download_and_prepare()
	__A :			Union[str, Any]		    =			builder_instance.as_dataset()
	assert ds







@pytest.mark.integration
def 							__SCREAMING_SNAKE_CASE     (   a__    :   List[str]       )					->				List[str]:
	__A :			Tuple		    =			dataset_module_factory("""wikipedia"""			,cache_dir=a__       )
	__A :			str		    =			import_main_class(dataset_module.module_path			,dataset=a__       )
	__A :			DatasetBuilder		    =			builder_cls(
	    cache_dir=a__			,config_name="""20220301.frr"""			,hash=dataset_module.hash			,)
	__A :			Optional[int]		    =			builder_instance.as_streaming_dataset()
	assert ds
	assert isinstance(a__			,a__       )
	assert "train" in ds
	assert isinstance(ds["""train"""]			,a__       )
	assert next(iter(ds["""train"""]       )       )





