


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,
)
UpperCamelCase								=	logging.getLogger(__name__)


def   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE       ):
  A_      :				Optional[int]		     =					git.Repo(search_parent_directories=SCREAMING_SNAKE_CASE       )
  A_      :				List[str]		     =					{
      '''repo_id''': str(SCREAMING_SNAKE_CASE       ),
      '''repo_sha''': str(repo.head.object.hexsha       ),
      '''repo_branch''': str(repo.active_branch       ),
  }

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


def   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE       ):
  if params.n_gpu <= 0:
    A_      :				List[str]		     =					0
    A_      :				List[str]		     =					-1
    A_      :				Tuple		     =					True
    A_      :				Optional[Any]		     =					False
    return

  assert torch.cuda.is_available()

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

    A_      :				str		     =					int(os.environ['''WORLD_SIZE''']       )
    A_      :				str		     =					int(os.environ['''N_GPU_NODE''']       )
    A_      :				List[Any]		     =					int(os.environ['''RANK''']       )

    # number of nodes / node ID
    A_      :				List[str]		     =					params.world_size // params.n_gpu_per_node
    A_      :				Optional[Any]		     =					params.global_rank // params.n_gpu_per_node
    A_      :				int		     =					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

    A_      :				Optional[int]		     =					1
    A_      :				List[str]		     =					0
    A_      :				Optional[Any]		     =					0
    A_      :				str		     =					0
    A_      :				Optional[Any]		     =					1
    A_      :				Union[str, Any]		     =					1
    A_      :				List[str]		     =					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
  A_      :				List[Any]		     =					params.node_id == 0 and params.local_rank == 0
  A_      :				str		     =					params.n_nodes > 1

  # summary
  A_      :				str		     =					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   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE       ):
  np.random.seed(args.seed       )
  torch.manual_seed(args.seed       )
  if args.n_gpu > 0:
    torch.cuda.manual_seed_all(args.seed       )








import os
from typing import BinaryIO, Optional, Union

import numpy as np
import pyarrow.parquet as pq

from .. import Audio, Dataset, Features, Image, NamedSplit, Value, config
from ..features.features import FeatureType, _visit
from ..formatting import query_table
from ..packaged_modules import _PACKAGED_DATASETS_MODULES
from ..packaged_modules.parquet.parquet import Parquet
from ..utils import logging
from ..utils.typing import NestedDataStructureLike, PathLike
from .abc import AbstractDatasetReader


def   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE       ):
  A_      :				Any		     =					np.inf

  def set_batch_size(SCREAMING_SNAKE_CASE       ) -> None:
    nonlocal batch_size
    if isinstance(SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE       ):
      A_      :				Tuple		     =					min(SCREAMING_SNAKE_CASE ,			config.PARQUET_ROW_GROUP_SIZE_FOR_IMAGE_DATASETS       )
    elif isinstance(SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE       ):
      A_      :				str		     =					min(SCREAMING_SNAKE_CASE ,			config.PARQUET_ROW_GROUP_SIZE_FOR_AUDIO_DATASETS       )
    elif isinstance(SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE       ) and feature.dtype == "binary":
      A_      :				Union[str, Any]		     =					min(SCREAMING_SNAKE_CASE ,			config.PARQUET_ROW_GROUP_SIZE_FOR_BINARY_DATASETS       )

  _visit(SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE       )

  return None if batch_size is np.inf else batch_size






class   _lowerCamelCase (					UpperCamelCase			):




       """simple docstring"""



       def __init__(      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = False				,       _SCREAMING_SNAKE_CASE = False				,       _SCREAMING_SNAKE_CASE = None				,       **_SCREAMING_SNAKE_CASE				,       )->str:




         '''simple docstring'''





         super().__init__(
             _SCREAMING_SNAKE_CASE				,       split=_SCREAMING_SNAKE_CASE				,       features=_SCREAMING_SNAKE_CASE				,       cache_dir=_SCREAMING_SNAKE_CASE				,       keep_in_memory=_SCREAMING_SNAKE_CASE				,       streaming=_SCREAMING_SNAKE_CASE				,       num_proc=_SCREAMING_SNAKE_CASE				,       **_SCREAMING_SNAKE_CASE				,       )
         A_      :				str		     =					path_or_paths if isinstance(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		) else {self.split: path_or_paths}
         A_      :				Optional[int]		     =					_PACKAGED_DATASETS_MODULES['''parquet'''][1]
         A_      :				Union[str, Any]		     =					Parquet(
             cache_dir=_SCREAMING_SNAKE_CASE				,       data_files=_SCREAMING_SNAKE_CASE				,       features=_SCREAMING_SNAKE_CASE				,       hash=_SCREAMING_SNAKE_CASE				,       **_SCREAMING_SNAKE_CASE				,       )







       def        _snake_case       (      self		)->Optional[int]:




         '''simple docstring'''





         if self.streaming:
           A_      :				List[str]		     =					self.builder.as_streaming_dataset(split=self.split		)
         # Build regular (map-style) dataset
         else:
           A_      :				List[str]		     =					None
           A_      :				List[str]		     =					None
           A_      :				List[Any]		     =					None
           A_      :				Dict		     =					None

           self.builder.download_and_prepare(
               download_config=_SCREAMING_SNAKE_CASE				,       download_mode=_SCREAMING_SNAKE_CASE				,       verification_mode=_SCREAMING_SNAKE_CASE				,       base_path=_SCREAMING_SNAKE_CASE				,       num_proc=self.num_proc				,       )
           A_      :				Union[str, Any]		     =					self.builder.as_dataset(
               split=self.split				,       verification_mode=_SCREAMING_SNAKE_CASE				,       in_memory=self.keep_in_memory		)
         return dataset






class   _lowerCamelCase :




       """simple docstring"""



       def __init__(      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE = None				,       **_SCREAMING_SNAKE_CASE				,       )->List[Any]:




         '''simple docstring'''





         A_      :				Union[str, Any]		     =					dataset
         A_      :				Union[str, Any]		     =					path_or_buf
         A_      :				Any		     =					batch_size or get_writer_batch_size(dataset.features		)
         A_      :				Optional[int]		     =					parquet_writer_kwargs



       def        _snake_case       (      self		)->int:




         '''simple docstring'''





         A_      :				Union[str, Any]		     =					self.batch_size if self.batch_size else config.DEFAULT_MAX_BATCH_SIZE

         if isinstance(self.path_or_buf				,       (str, bytes, os.PathLike)		):
           with open(self.path_or_buf				,       '''wb+'''		) as buffer:
             A_      :				str		     =					self._write(file_obj=_SCREAMING_SNAKE_CASE				,       batch_size=_SCREAMING_SNAKE_CASE				,       **self.parquet_writer_kwargs		)
         else:
           A_      :				Tuple		     =					self._write(file_obj=self.path_or_buf				,       batch_size=_SCREAMING_SNAKE_CASE				,       **self.parquet_writer_kwargs		)
         return written







       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       **_SCREAMING_SNAKE_CASE		)->int:




         '''simple docstring'''





         A_      :				List[Any]		     =					0
         A_      :				int		     =					parquet_writer_kwargs.pop('''path_or_buf'''				,       _SCREAMING_SNAKE_CASE		)
         A_      :				List[Any]		     =					self.dataset.features.arrow_schema

         A_      :				List[str]		     =					pq.ParquetWriter(_SCREAMING_SNAKE_CASE				,       schema=_SCREAMING_SNAKE_CASE				,       **_SCREAMING_SNAKE_CASE		)

         for offset in logging.tqdm(
             range(0				,       len(self.dataset		)				,       _SCREAMING_SNAKE_CASE		)				,       unit='''ba'''				,       disable=not logging.is_progress_bar_enabled()				,       desc='''Creating parquet from Arrow format'''				,       ):
           A_      :				List[Any]		     =					query_table(
               table=self.dataset._data				,       key=slice(_SCREAMING_SNAKE_CASE				,       offset + batch_size		)				,       indices=self.dataset._indices if self.dataset._indices is not None else None				,       )
           writer.write_table(_SCREAMING_SNAKE_CASE		)
           written += batch.nbytes
         writer.close()
         return written




