





import json
import logging
import os
import socket

import git
import numpy as np
import torch


logging.basicConfig(
    format="%(asctime)s - %(levelname)s - %(name)s - PID: %(process)d -  %(message)s",
    datefmt="%m/%d/%Y %H:%M:%S",
    level=logging.INFO,
)
A_												=     logging.getLogger(__name__)





def    __UpperCamelCase      (    a) ->Optional[int]:
					lowerCamelCase__       =       git.Repo(search_parent_directories=a)
					lowerCamelCase__       =       {
					    "repo_id": str(a),
					    "repo_sha": str(repo.head.object.hexsha),
					    "repo_branch": str(repo.active_branch),
					}

					with open(os.path.join(a,     "git_log.json"),     "w") as f:
										json.dump(a,     a,     indent=4)





def    __UpperCamelCase      (    a) ->Union[str, Any]:
					if params.n_gpu <= 0:
										lowerCamelCase__       =       0
										lowerCamelCase__       =       -1
										lowerCamelCase__       =       True
										lowerCamelCase__       =       False
										return

					assert torch.cuda.is_available()

					logger.info("Initializing GPUs")
					if params.n_gpu > 1:
										assert params.local_rank != -1

										lowerCamelCase__       =       int(os.environ["WORLD_SIZE"])
										lowerCamelCase__       =       int(os.environ["N_GPU_NODE"])
										lowerCamelCase__       =       int(os.environ["RANK"])

										# number of nodes / node ID
										lowerCamelCase__       =       params.world_size // params.n_gpu_per_node
										lowerCamelCase__       =       params.global_rank // params.n_gpu_per_node
										lowerCamelCase__       =       True

										assert params.n_nodes == int(os.environ["N_NODES"])
										assert params.node_id == int(os.environ["NODE_RANK"])

					# local job (single GPU)
					else:
										assert params.local_rank == -1

										lowerCamelCase__       =       1
										lowerCamelCase__       =       0
										lowerCamelCase__       =       0
										lowerCamelCase__       =       0
										lowerCamelCase__       =       1
										lowerCamelCase__       =       1
										lowerCamelCase__       =       False

					# sanity checks
					assert params.n_nodes >= 1
					assert 0 <= params.node_id < params.n_nodes
					assert 0 <= params.local_rank <= params.global_rank < params.world_size
					assert params.world_size == params.n_nodes * params.n_gpu_per_node

					# define whether this is the master process / if we are in multi-node distributed mode
					lowerCamelCase__       =       params.node_id == 0 and params.local_rank == 0
					lowerCamelCase__       =       params.n_nodes > 1

					# summary
					lowerCamelCase__       =       f"--- Global rank: {params.global_rank} - "
					logger.info(PREFIX + "Number of nodes: %i" % params.n_nodes)
					logger.info(PREFIX + "Node ID        : %i" % params.node_id)
					logger.info(PREFIX + "Local rank     : %i" % params.local_rank)
					logger.info(PREFIX + "World size     : %i" % params.world_size)
					logger.info(PREFIX + "GPUs per node  : %i" % params.n_gpu_per_node)
					logger.info(PREFIX + "Master         : %s" % str(params.is_master))
					logger.info(PREFIX + "Multi-node     : %s" % str(params.multi_node))
					logger.info(PREFIX + "Multi-GPU      : %s" % str(params.multi_gpu))
					logger.info(PREFIX + "Hostname       : %s" % socket.gethostname())

					# set GPU device
					torch.cuda.set_device(params.local_rank)

					# initialize multi-GPU
					if params.multi_gpu:
										logger.info("Initializing PyTorch distributed")
										torch.distributed.init_process_group(
										    init_method="env://",     backend="nccl",     )







def    __UpperCamelCase      (    a) ->Optional[int]:
					np.random.seed(args.seed)
					torch.manual_seed(args.seed)
					if args.n_gpu > 0:
										torch.cuda.manual_seed_all(args.seed)








import os
import tempfile
from functools import partial
from unittest import TestCase
from unittest.mock import patch

import datasets
import datasets.config

from .utils import require_beam

class SCREAMING_SNAKE_CASE_ (					datasets.BeamBasedBuilder   ):





					"""simple docstring"""


					def 					__magic_name__ ( self	):
										return datasets.DatasetInfo(
										    features=datasets.Features({"content": datasets.Value("string"	)}	)	,  supervised_keys=_lowerCAmelCase	,  )





					def 					__magic_name__ ( self	,  _lowerCAmelCase	,  _lowerCAmelCase	):
										return [datasets.SplitGenerator(name=datasets.Split.TRAIN	,  gen_kwargs={"examples": get_test_dummy_examples()}	)]






					def 					__magic_name__ ( self	,  _lowerCAmelCase	,  _lowerCAmelCase	):
										import apache_beam as beam

										return pipeline | "Load Examples" >> beam.Create(_lowerCAmelCase	)

class SCREAMING_SNAKE_CASE_ (					datasets.BeamBasedBuilder   ):





					"""simple docstring"""


					def 					__magic_name__ ( self	):
										return datasets.DatasetInfo(
										    features=datasets.Features({"a": datasets.Sequence({"b": datasets.Value("string"	)}	)}	)	,  supervised_keys=_lowerCAmelCase	,  )





					def 					__magic_name__ ( self	,  _lowerCAmelCase	,  _lowerCAmelCase	):
										return [
										    datasets.SplitGenerator(name=datasets.Split.TRAIN	,  gen_kwargs={"examples": get_test_nested_examples()}	)
										]






					def 					__magic_name__ ( self	,  _lowerCAmelCase	,  _lowerCAmelCase	):
										import apache_beam as beam

										return pipeline | "Load Examples" >> beam.Create(_lowerCAmelCase	)





def    __UpperCamelCase      (    ) ->Optional[Any]:
					return [(i, {"content": content}) for i, content in enumerate(["foo", "bar", "foobar"])]





def    __UpperCamelCase      (    ) ->Optional[Any]:
					return [(i, {"a": {"b": [content]}}) for i, content in enumerate(["foo", "bar", "foobar"])]







class SCREAMING_SNAKE_CASE_ (					lowercase_   ):





					"""simple docstring"""


					@require_beam
					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       len(get_test_dummy_examples()	)
										with tempfile.TemporaryDirectory() as tmp_cache_dir:
															lowerCamelCase__       =       DummyBeamDataset(cache_dir=_lowerCAmelCase	,  beam_runner="DirectRunner"	)
															builder.download_and_prepare()
															self.assertTrue(
															    os.path.exists(
															        os.path.join(_lowerCAmelCase	,  builder.name	,  "default"	,  "0.0.0"	,  F"{builder.name}-train.arrow"	)	)	)
															self.assertDictEqual(builder.info.features	,  datasets.Features({"content": datasets.Value("string"	)}	)	)
															lowerCamelCase__       =       builder.as_dataset()
															self.assertEqual(dset["train"].num_rows	,  _lowerCAmelCase	)
															self.assertEqual(dset["train"].info.splits["train"].num_examples	,  _lowerCAmelCase	)
															self.assertDictEqual(dset["train"][0]	,  get_test_dummy_examples()[0][1]	)
															self.assertDictEqual(
															    dset["train"][expected_num_examples - 1]	,  get_test_dummy_examples()[expected_num_examples - 1][1]	)
															self.assertTrue(
															    os.path.exists(os.path.join(_lowerCAmelCase	,  builder.name	,  "default"	,  "0.0.0"	,  "dataset_info.json"	)	)	)
															del dset





					@require_beam
					def 					__magic_name__ ( self	):
										import apache_beam as beam

										lowerCamelCase__       =       beam.io.parquetio.WriteToParquet

										lowerCamelCase__       =       len(get_test_dummy_examples()	)
										with tempfile.TemporaryDirectory() as tmp_cache_dir:
															lowerCamelCase__       =       DummyBeamDataset(cache_dir=_lowerCAmelCase	,  beam_runner="DirectRunner"	)
															with patch("apache_beam.io.parquetio.WriteToParquet"	) as write_parquet_mock:
																				lowerCamelCase__       =       partial(_lowerCAmelCase	,  num_shards=2	)
																				builder.download_and_prepare()
															self.assertTrue(
															    os.path.exists(
															        os.path.join(
															            _lowerCAmelCase	,  builder.name	,  "default"	,  "0.0.0"	,  F"{builder.name}-train-00000-of-00002.arrow"	)	)	)
															self.assertTrue(
															    os.path.exists(
															        os.path.join(
															            _lowerCAmelCase	,  builder.name	,  "default"	,  "0.0.0"	,  F"{builder.name}-train-00000-of-00002.arrow"	)	)	)
															self.assertDictEqual(builder.info.features	,  datasets.Features({"content": datasets.Value("string"	)}	)	)
															lowerCamelCase__       =       builder.as_dataset()
															self.assertEqual(dset["train"].num_rows	,  _lowerCAmelCase	)
															self.assertEqual(dset["train"].info.splits["train"].num_examples	,  _lowerCAmelCase	)
															# Order is not preserved when sharding, so we just check that all the elements are there
															self.assertListEqual(sorted(dset["train"]["content"]	)	,  sorted(["foo", "bar", "foobar"]	)	)
															self.assertTrue(
															    os.path.exists(os.path.join(_lowerCAmelCase	,  builder.name	,  "default"	,  "0.0.0"	,  "dataset_info.json"	)	)	)
															del dset





					@require_beam
					def 					__magic_name__ ( self	):
										with tempfile.TemporaryDirectory() as tmp_cache_dir:
															lowerCamelCase__       =       DummyBeamDataset(cache_dir=_lowerCAmelCase	)
															self.assertRaises(datasets.builder.MissingBeamOptions	,  builder.download_and_prepare	)






					@require_beam
					def 					__magic_name__ ( self	):
										lowerCamelCase__       =       len(get_test_nested_examples()	)
										with tempfile.TemporaryDirectory() as tmp_cache_dir:
															lowerCamelCase__       =       NestedBeamDataset(cache_dir=_lowerCAmelCase	,  beam_runner="DirectRunner"	)
															builder.download_and_prepare()
															self.assertTrue(
															    os.path.exists(
															        os.path.join(_lowerCAmelCase	,  builder.name	,  "default"	,  "0.0.0"	,  F"{builder.name}-train.arrow"	)	)	)
															self.assertDictEqual(
															    builder.info.features	,  datasets.Features({"a": datasets.Sequence({"b": datasets.Value("string"	)}	)}	)	)
															lowerCamelCase__       =       builder.as_dataset()
															self.assertEqual(dset["train"].num_rows	,  _lowerCAmelCase	)
															self.assertEqual(dset["train"].info.splits["train"].num_examples	,  _lowerCAmelCase	)
															self.assertDictEqual(dset["train"][0]	,  get_test_nested_examples()[0][1]	)
															self.assertDictEqual(
															    dset["train"][expected_num_examples - 1]	,  get_test_nested_examples()[expected_num_examples - 1][1]	)
															self.assertTrue(
															    os.path.exists(os.path.join(_lowerCAmelCase	,  builder.name	,  "default"	,  "0.0.0"	,  "dataset_info.json"	)	)	)
															del dset

