


import os
import tempfile
from pathlib import Path
from unittest import TestCase

import pyarrow as pa
import pytest

from datasets.arrow_dataset import Dataset
from datasets.arrow_reader import ArrowReader, BaseReader, FileInstructions, ReadInstruction, make_file_instructions
from datasets.info import DatasetInfo
from datasets.splits import NamedSplit, Split, SplitDict, SplitInfo

from .utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases





class 			lowerCAmelCase__(     snake_case__    ):






    '''simple docstring'''





    def        _lowerCamelCase			(						self					: Union[str, Any]							, __snake_case					: List[str]							, __snake_case					: List[Any]=False						):





       '''simple docstring'''
       UpperCAmelCase_		,   UpperCAmelCase_		,   UpperCAmelCase_				:   Dict							= (
           filename_skip_take['''filename'''],
           filename_skip_take['''skip'''] if '''skip''' in filename_skip_take else None,
           filename_skip_take['''take'''] if '''take''' in filename_skip_take else None,
       )
       open(os.path.join(__snake_case						)							, '''wb'''						).close()
       UpperCAmelCase_				:   Optional[int]							= pa.Table.from_pydict({'''filename''': [Path(__snake_case						).name] * 100}						)
       if take == -1:
          UpperCAmelCase_				:   Any							= len(__snake_case						) - skip
       if skip is not None and take is not None:
          UpperCAmelCase_				:   Optional[Any]							= pa_table.slice(__snake_case							, __snake_case						)
       return pa_table





class 			lowerCAmelCase__(     snake_case__    ):






    '''simple docstring'''





    def        _lowerCamelCase			(						self					: List[str]						):





       '''simple docstring'''
       UpperCAmelCase_				:   int							= '''my_name'''
       UpperCAmelCase_				:   str							= SplitInfo(name='''train'''							, num_examples=100						)
       UpperCAmelCase_				:   Optional[int]							= SplitInfo(name='''test'''							, num_examples=100						)
       UpperCAmelCase_				:   str							= [train_info, test_info]
       UpperCAmelCase_				:   Union[str, Any]							= SplitDict()
       split_dict.add(__snake_case						)
       split_dict.add(__snake_case						)
       UpperCAmelCase_				:   List[str]							= DatasetInfo(splits=__snake_case						)

       with tempfile.TemporaryDirectory() as tmp_dir:
          UpperCAmelCase_				:   List[Any]							= ReaderTest(__snake_case							, __snake_case						)

          UpperCAmelCase_				:   str							= '''test[:33%]'''
          UpperCAmelCase_				:   int							= Dataset(**reader.read(__snake_case							, __snake_case							, __snake_case						)						)
          self.assertEqual(dset['''filename'''][0]							, f'''{name}-test'''						)
          self.assertEqual(dset.num_rows							, 33						)
          self.assertEqual(dset.num_columns							, 1						)

          UpperCAmelCase_				:   Any							= ['''train''', '''test[:33%]''']
          UpperCAmelCase_				:   Any							= [Split.TRAIN, ReadInstruction.from_spec('''test[:33%]'''						)]
          for instructions in [instructionsa, instructionsa]:
             UpperCAmelCase_				:   Union[str, Any]							= [reader.read(__snake_case							, __snake_case							, __snake_case						) for instr in instructions]
             UpperCAmelCase_		,   UpperCAmelCase_				:   Optional[int]							= (Dataset(**__snake_case						) for dataset_kwargs in datasets_kwargs)
             self.assertEqual(train_dset['''filename'''][0]							, f'''{name}-train'''						)
             self.assertEqual(train_dset.num_rows							, 100						)
             self.assertEqual(train_dset.num_columns							, 1						)
             self.assertIsInstance(train_dset.split							, __snake_case						)
             self.assertEqual(str(train_dset.split						)							, '''train'''						)
             self.assertEqual(test_dset['''filename'''][0]							, f'''{name}-test'''						)
             self.assertEqual(test_dset.num_rows							, 33						)
             self.assertEqual(test_dset.num_columns							, 1						)
             self.assertIsInstance(test_dset.split							, __snake_case						)
             self.assertEqual(str(test_dset.split						)							, '''test[:33%]'''						)
             del train_dset, test_dset





    def        _lowerCamelCase			(						self					: List[str]						):





       '''simple docstring'''
       UpperCAmelCase_				:   str							= '''my_name'''
       UpperCAmelCase_				:   Optional[Any]							= SplitInfo(name='''train'''							, num_examples=1_000							, shard_lengths=[100] * 10						)
       UpperCAmelCase_				:   str							= [train_info]
       UpperCAmelCase_				:   Optional[int]							= SplitDict()
       split_dict.add(__snake_case						)
       UpperCAmelCase_				:   Optional[int]							= DatasetInfo(splits=__snake_case						)

       with tempfile.TemporaryDirectory() as tmp_dir:
          UpperCAmelCase_				:   List[Any]							= ReaderTest(__snake_case							, __snake_case						)

          UpperCAmelCase_				:   Union[str, Any]							= '''train[:33%]'''
          UpperCAmelCase_				:   Optional[Any]							= Dataset(**reader.read(__snake_case							, __snake_case							, __snake_case						)						)
          self.assertEqual(dset['''filename'''][0]							, f'''{name}-train-00000-of-00010'''						)
          self.assertEqual(dset['''filename'''][-1]							, f'''{name}-train-00003-of-00010'''						)
          self.assertEqual(dset.num_rows							, 330						)
          self.assertEqual(dset.num_columns							, 1						)





    def        _lowerCamelCase			(						self					: Dict						):





       '''simple docstring'''
       UpperCAmelCase_				:   List[str]							= SplitInfo(name='''train'''							, num_examples=100						)
       UpperCAmelCase_				:   List[Any]							= SplitInfo(name='''test'''							, num_examples=100						)
       UpperCAmelCase_				:   Union[str, Any]							= SplitDict()
       split_dict.add(__snake_case						)
       split_dict.add(__snake_case						)
       UpperCAmelCase_				:   int							= DatasetInfo(splits=__snake_case						)

       with tempfile.TemporaryDirectory() as tmp_dir:
          UpperCAmelCase_				:   Tuple							= ReaderTest(__snake_case							, __snake_case						)

          UpperCAmelCase_				:   List[str]							= [
              {'''filename''': os.path.join(__snake_case							, '''train'''						)},
              {'''filename''': os.path.join(__snake_case							, '''test'''						), '''skip''': 10, '''take''': 10},
          ]
          UpperCAmelCase_				:   str							= Dataset(**reader.read_files(__snake_case							, original_instructions='''train+test[10:20]'''						)						)
          self.assertEqual(dset.num_rows							, 110						)
          self.assertEqual(dset.num_columns							, 1						)
          del dset





@pytest.mark.parametrize('''in_memory'''      ,							[False, True]							)
def   snake_case_    (			__lowercase      ,							__lowercase      ,							__lowercase							):
   UpperCAmelCase_				:   int							= arrow_file
   with assert_arrow_memory_increases() if in_memory else assert_arrow_memory_doesnt_increase():
      UpperCAmelCase_				:   List[Any]							= ArrowReader.read_table(__lowercase      ,							in_memory=__lowercase							)
   assert table.shape == dataset.data.shape
   assert set(table.column_names							) == set(dataset.data.column_names							)
   assert dict(table.to_pydict()							) == dict(dataset.data.to_pydict()							)  # to_pydict returns OrderedDict
@pytest.mark.parametrize('''in_memory'''      ,							[False, True]							)
def   snake_case_    (			__lowercase      ,							__lowercase      ,							__lowercase							):
   UpperCAmelCase_				:   int							= arrow_file
   UpperCAmelCase_				:   Dict							= ArrowReader(''''''      ,							__lowercase							)
   with assert_arrow_memory_increases() if in_memory else assert_arrow_memory_doesnt_increase():
      UpperCAmelCase_				:   Optional[Any]							= reader.read_files([{'''filename''': filename}]      ,							in_memory=__lowercase							)
   assert dataset_kwargs.keys() == {"arrow_table", "info", "split"}
   UpperCAmelCase_				:   List[str]							= dataset_kwargs['''arrow_table''']
   assert table.shape == dataset.data.shape
   assert set(table.column_names							) == set(dataset.data.column_names							)
   assert dict(table.to_pydict()							) == dict(dataset.data.to_pydict()							)  # to_pydict returns OrderedDict
def   snake_case_    (			):
   assert ReadInstruction('''train'''      ,							to=1_0      ,							unit='''abs'''							).to_spec() == "train[:10]"
   assert ReadInstruction('''train'''      ,							from_=-8_0      ,							to=1_0      ,							unit='''%'''							).to_spec() == "train[-80%:10%]"

   UpperCAmelCase_				:   List[str]							= '''train+test'''
   assert ReadInstruction.from_spec(__lowercase							).to_spec() == spec_train_test

   UpperCAmelCase_				:   Any							= '''train[2:10]'''
   assert ReadInstruction.from_spec(__lowercase							).to_spec() == spec_train_abs

   UpperCAmelCase_				:   Any							= '''train[15%:-20%]'''
   assert ReadInstruction.from_spec(__lowercase							).to_spec() == spec_train_pct

   UpperCAmelCase_				:   Tuple							= '''train[:10%](closest)'''
   assert ReadInstruction.from_spec(__lowercase							).to_spec() == "train[:10%]"

   UpperCAmelCase_				:   Optional[int]							= '''train[:10%](pct1_dropremainder)'''
   assert ReadInstruction.from_spec(__lowercase							).to_spec() == spec_train_pct_rounding

   UpperCAmelCase_				:   Tuple							= '''train[:10%](pct1_dropremainder)+test[-10%:](pct1_dropremainder)'''
   assert ReadInstruction.from_spec(__lowercase							).to_spec() == spec_train_test_pct_rounding
def   snake_case_    (			):
   UpperCAmelCase_				:   int							= '''dummy'''
   UpperCAmelCase_				:   List[str]							= [SplitInfo(name='''train'''      ,							num_examples=1_0_0							)]
   UpperCAmelCase_				:   List[Any]							= '''train[:33%]'''
   UpperCAmelCase_				:   Optional[Any]							= '''arrow'''
   UpperCAmelCase_				:   Any							= '''prefix'''

   UpperCAmelCase_				:   int							= make_file_instructions(__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase							)
   assert isinstance(__lowercase      ,							__lowercase							)
   assert file_instructions.num_examples == 3_3
   assert file_instructions.file_instructions == [
       {"filename": os.path.join(__lowercase      ,							F'''{name}-train.arrow'''							), "skip": 0, "take": 3_3}
   ]

   UpperCAmelCase_				:   int							= [SplitInfo(name='''train'''      ,							num_examples=1_0_0      ,							shard_lengths=[1_0] * 1_0							)]
   UpperCAmelCase_				:   Union[str, Any]							= make_file_instructions(__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase							)
   assert isinstance(__lowercase      ,							__lowercase							)
   assert file_instructions.num_examples == 3_3
   assert file_instructions.file_instructions == [
       {"filename": os.path.join(__lowercase      ,							F'''{name}-train-00000-of-00010.arrow'''							), "skip": 0, "take": -1},
       {"filename": os.path.join(__lowercase      ,							F'''{name}-train-00001-of-00010.arrow'''							), "skip": 0, "take": -1},
       {"filename": os.path.join(__lowercase      ,							F'''{name}-train-00002-of-00010.arrow'''							), "skip": 0, "take": -1},
       {"filename": os.path.join(__lowercase      ,							F'''{name}-train-00003-of-00010.arrow'''							), "skip": 0, "take": 3},
   ]


@pytest.mark.parametrize('''name, expected_exception'''      ,							[(None, TypeError), ('''''', ValueError)]							)
def   snake_case_    (			__lowercase      ,							__lowercase							):
   UpperCAmelCase_				:   List[str]							= [SplitInfo(name='''train'''      ,							num_examples=1_0_0							)]
   UpperCAmelCase_				:   Any							= '''train'''
   UpperCAmelCase_				:   Optional[Any]							= '''arrow'''
   UpperCAmelCase_				:   List[Any]							= '''prefix_path'''
   with pytest.raises(__lowercase							):
      UpperCAmelCase_				:   List[str]							= make_file_instructions(__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase							)



import os
import random
from pathlib import Path
from typing import List

import numpy as np
import torch
from torch.cuda.amp import GradScaler

from .utils import (
    MODEL_NAME,
    OPTIMIZER_NAME,
    RNG_STATE_NAME,
    SCALER_NAME,
    SCHEDULER_NAME,
    get_pretty_name,
    is_tpu_available,
    is_xpu_available,
    save,
)


if is_tpu_available(check_device=False):
       import torch_xla.core.xla_model as xm

from .logging import get_logger
from .state import PartialState


__UpperCamelCase					:	int          =				get_logger(__name__)
def   snake_case_    (			__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase = None      ,							):
   # Model states
   for i, state in enumerate(__lowercase							):
      UpperCAmelCase_				:   Dict							= F'''{MODEL_NAME}.bin''' if i == 0 else F'''{MODEL_NAME}_{i}.bin'''
      UpperCAmelCase_				:   List[Any]							= os.path.join(__lowercase      ,							__lowercase							)
      save(__lowercase      ,							__lowercase							)
      logger.info(F'''Model weights saved in {output_model_file}'''							)
   # Optimizer states
   for i, opt in enumerate(__lowercase							):
      UpperCAmelCase_				:   Tuple							= opt.state_dict()
      UpperCAmelCase_				:   Any							= F'''{OPTIMIZER_NAME}.bin''' if i == 0 else F'''{OPTIMIZER_NAME}_{i}.bin'''
      UpperCAmelCase_				:   List[Any]							= os.path.join(__lowercase      ,							__lowercase							)
      save(__lowercase      ,							__lowercase							)
      logger.info(F'''Optimizer state saved in {output_optimizer_file}'''							)
   # Scheduler states
   for i, scheduler in enumerate(__lowercase							):
      UpperCAmelCase_				:   Tuple							= scheduler.state_dict()
      UpperCAmelCase_				:   Tuple							= F'''{SCHEDULER_NAME}.bin''' if i == 0 else F'''{SCHEDULER_NAME}_{i}.bin'''
      UpperCAmelCase_				:   Tuple							= os.path.join(__lowercase      ,							__lowercase							)
      save(__lowercase      ,							__lowercase							)
      logger.info(F'''Scheduler state saved in {output_scheduler_file}'''							)
   # GradScaler state
   if scaler is not None:
      UpperCAmelCase_				:   Union[str, Any]							= scaler.state_dict()
      UpperCAmelCase_				:   Union[str, Any]							= os.path.join(__lowercase      ,							__lowercase							)
      torch.save(__lowercase      ,							__lowercase							)
      logger.info(F'''Gradient scaler state saved in {output_scaler_file}'''							)
   # Random number generator states
   UpperCAmelCase_				:   Union[str, Any]							= {}
   UpperCAmelCase_				:   Tuple							= F'''{RNG_STATE_NAME}_{process_index}.pkl'''
   UpperCAmelCase_				:   List[Any]							= random.getstate()
   UpperCAmelCase_				:   Any							= np.random.get_state()
   UpperCAmelCase_				:   List[Any]							= torch.get_rng_state()
   if is_xpu_available():
      UpperCAmelCase_				:   Optional[int]							= torch.xpu.get_rng_state_all()
   else:
      UpperCAmelCase_				:   Any							= torch.cuda.get_rng_state_all()
   if is_tpu_available():
      UpperCAmelCase_				:   List[str]							= xm.get_rng_state()
   UpperCAmelCase_				:   Any							= os.path.join(__lowercase      ,							__lowercase							)
   torch.save(__lowercase      ,							__lowercase							)
   logger.info(F'''Random states saved in {output_states_file}'''							)
   return output_dir
def   snake_case_    (			__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase=None      ,							__lowercase=None      ,							**__lowercase      ,							):
   if map_location not in [None, "cpu", "on_device"]:
      raise TypeError(
          '''Unsupported optimizer map location passed, please choose one of `None`, `\'cpu\'`, or `\'on_device\'`'''							)
   if map_location is None:
      UpperCAmelCase_				:   Optional[Any]							= '''cpu'''
   elif map_location == "on_device":
      UpperCAmelCase_				:   Optional[int]							= PartialState().device
   # Model states
   for i, model in enumerate(__lowercase							):
      UpperCAmelCase_				:   List[Any]							= F'''{MODEL_NAME}.bin''' if i == 0 else F'''{MODEL_NAME}_{i}.bin'''
      UpperCAmelCase_				:   int							= os.path.join(__lowercase      ,							__lowercase							)
      models[i].load_state_dict(torch.load(__lowercase      ,							map_location=__lowercase							)      ,							**__lowercase							)
   logger.info('''All model weights loaded successfully'''							)

   # Optimizer states
   for i, opt in enumerate(__lowercase							):
      UpperCAmelCase_				:   List[str]							= F'''{OPTIMIZER_NAME}.bin''' if i == 0 else F'''{OPTIMIZER_NAME}_{i}.bin'''
      UpperCAmelCase_				:   List[Any]							= os.path.join(__lowercase      ,							__lowercase							)
      UpperCAmelCase_				:   Tuple							= torch.load(__lowercase      ,							map_location=__lowercase							)
      optimizers[i].load_state_dict(__lowercase							)
   logger.info('''All optimizer states loaded successfully'''							)

   # Scheduler states
   for i, scheduler in enumerate(__lowercase							):
      UpperCAmelCase_				:   Dict							= F'''{SCHEDULER_NAME}.bin''' if i == 0 else F'''{SCHEDULER_NAME}_{i}.bin'''
      UpperCAmelCase_				:   int							= os.path.join(__lowercase      ,							__lowercase							)
      scheduler.load_state_dict(torch.load(__lowercase							)							)
   logger.info('''All scheduler states loaded successfully'''							)

   # GradScaler state
   if scaler is not None:
      UpperCAmelCase_				:   List[str]							= os.path.join(__lowercase      ,							__lowercase							)
      scaler.load_state_dict(torch.load(__lowercase							)							)
      logger.info('''GradScaler state loaded successfully'''							)

   # Random states
   try:
      UpperCAmelCase_				:   Optional[Any]							= torch.load(os.path.join(__lowercase      ,							F'''{RNG_STATE_NAME}_{process_index}.pkl'''							)							)
      random.setstate(states['''random_state''']							)
      np.random.set_state(states['''numpy_random_seed''']							)
      torch.set_rng_state(states['''torch_manual_seed''']							)
      if is_xpu_available():
         torch.xpu.set_rng_state_all(states['''torch_xpu_manual_seed''']							)
      else:
         torch.cuda.set_rng_state_all(states['''torch_cuda_manual_seed''']							)
      if is_tpu_available():
         xm.set_rng_state(states['''xm_seed''']							)
      logger.info('''All random states loaded successfully'''							)
   except Exception:
      logger.info('''Could not load random states'''							)
def   snake_case_    (			__lowercase      ,							__lowercase      ,							__lowercase = 0							):
   UpperCAmelCase_				:   Optional[int]							= Path(__lowercase							) / F'''custom_checkpoint_{index}.pkl'''
   logger.info(F'''Saving the state of {get_pretty_name(__lowercase							)} to {save_location}'''							)
   torch.save(obj.state_dict()      ,							__lowercase							)


def   snake_case_    (			__lowercase      ,							__lowercase      ,							__lowercase = 0							):
   UpperCAmelCase_				:   Dict							= F'''{path}/custom_checkpoint_{index}.pkl'''
   logger.info(F'''Loading the state of {get_pretty_name(__lowercase							)} from {load_location}'''							)
   obj.load_state_dict(torch.load(__lowercase      ,							map_location='''cpu'''							)							)