# 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      __SCREAMING_SNAKE_CASE  ( __UpperCamelCase   :	Union[str, Any]     )							-> Tuple:







  """simple docstring"""





  return (torch.arange(state.num_processes     ) + 1.0 + (state.num_processes * state.process_index)).to(state.device     )



def      __SCREAMING_SNAKE_CASE  ( __UpperCamelCase   :	Any     )							-> Optional[int]:







  """simple docstring"""





  SCREAMING_SNAKE_CASE__      							=				create_tensor(__UpperCamelCase     )
  SCREAMING_SNAKE_CASE__      							=				gather(__UpperCamelCase     )
  assert gathered_tensor.tolist() == list(range(1			,					state.num_processes**2 + 1     )     )



def      __SCREAMING_SNAKE_CASE  ( __UpperCamelCase   :	Optional[Any]     )							-> Tuple:







  """simple docstring"""





  SCREAMING_SNAKE_CASE__      							=				[state.process_index]
  SCREAMING_SNAKE_CASE__      							=				gather_object(__UpperCamelCase     )
  assert len(__UpperCamelCase     ) == state.num_processes, f"""{gathered_obj}, {len(__UpperCamelCase     )} != {state.num_processes}"""
  assert gathered_obj == list(range(state.num_processes     )     ), f"""{gathered_obj} != {list(range(state.num_processes     )     )}"""



def      __SCREAMING_SNAKE_CASE  ( __UpperCamelCase   :	Optional[Any]     )							-> str:







  """simple docstring"""





  SCREAMING_SNAKE_CASE__      							=				create_tensor(__UpperCamelCase     )
  SCREAMING_SNAKE_CASE__      							=				broadcast(__UpperCamelCase     )
  assert broadcasted_tensor.shape == torch.Size([state.num_processes]     )
  assert broadcasted_tensor.tolist() == list(range(1			,					state.num_processes + 1     )     )



def      __SCREAMING_SNAKE_CASE  ( __UpperCamelCase   :	List[str]     )							-> Dict:







  """simple docstring"""





  if state.is_main_process:
    SCREAMING_SNAKE_CASE__      							=				torch.arange(state.num_processes + 1     ).to(state.device     )
  else:
    SCREAMING_SNAKE_CASE__      							=				torch.arange(state.num_processes     ).to(state.device     )
  SCREAMING_SNAKE_CASE__      							=				pad_across_processes(__UpperCamelCase     )
  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      __SCREAMING_SNAKE_CASE  ( __UpperCamelCase   :	List[str]     )							-> Any:







  """simple docstring"""





  if state.num_processes != 2:
    return
  SCREAMING_SNAKE_CASE__      							=				create_tensor(__UpperCamelCase     )
  SCREAMING_SNAKE_CASE__      							=				reduce(__UpperCamelCase			,					"""sum"""     )
  SCREAMING_SNAKE_CASE__      							=				torch.tensor([4.0, 6]     ).to(state.device     )
  assert torch.allclose(__UpperCamelCase			,					__UpperCamelCase     ), f"""{reduced_tensor} != {truth_tensor}"""



def      __SCREAMING_SNAKE_CASE  ( __UpperCamelCase   :	str     )							-> Dict:







  """simple docstring"""





  if state.num_processes != 2:
    return
  SCREAMING_SNAKE_CASE__      							=				create_tensor(__UpperCamelCase     )
  SCREAMING_SNAKE_CASE__      							=				reduce(__UpperCamelCase			,					"""mean"""     )
  SCREAMING_SNAKE_CASE__      							=				torch.tensor([2.0, 3]     ).to(state.device     )
  assert torch.allclose(__UpperCamelCase			,					__UpperCamelCase     ), f"""{reduced_tensor} != {truth_tensor}"""



def      __SCREAMING_SNAKE_CASE  ( __UpperCamelCase   :	Tuple     )							-> str:







  """simple docstring"""





  main()



def      __SCREAMING_SNAKE_CASE  ( )							-> Tuple:







  """simple docstring"""





  SCREAMING_SNAKE_CASE__      							=				PartialState()
  state.print(f"""State: {state}"""     )
  state.print("""testing gather"""     )
  test_gather(__UpperCamelCase     )
  state.print("""testing gather_object"""     )
  test_gather_object(__UpperCamelCase     )
  state.print("""testing broadcast"""     )
  test_broadcast(__UpperCamelCase     )
  state.print("""testing pad_across_processes"""     )
  test_pad_across_processes(__UpperCamelCase     )
  state.print("""testing reduce_sum"""     )
  test_reduce_sum(__UpperCamelCase     )
  state.print("""testing reduce_mean"""     )
  test_reduce_mean(__UpperCamelCase     )


if __name__ == "__main__":
   main()
import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER, get_tests_dir


sys.path.append(str(Path(__file__).parent.parent.parent.parent / '''utils'''))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


__lowerCamelCase      :			List[Any]							       =		get_tests_dir('''fixtures''')
__lowerCamelCase      :			Optional[int]							       =		get_tests_dir('''fixtures/dummy_feature_extractor_config.json''')
__lowerCamelCase      :			Any							       =		get_tests_dir('''fixtures/dummy-config.json''')







class __snake_case    (						unittest.TestCase		):




   def      __a						(   self      :    Optional[Any]				):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__      							=				0




   def      __a						(   self      :    List[Any]				):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained("""facebook/wav2vec2-base-960h"""				)
     self.assertIsInstance(_lowercase	,     _lowercase				)




   def      __a						(   self      :    Union[str, Any]				):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained(_lowercase				)
     self.assertIsInstance(_lowercase	,     _lowercase				)




   def      __a						(   self      :    Optional[int]				):



     """simple docstring"""






     with tempfile.TemporaryDirectory() as tmpdirname:
       SCREAMING_SNAKE_CASE__      							=				WavaVecaConfig()

       # remove feature_extractor_type to make sure config.json alone is enough to load feature processor locally
       SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained(_lowercase				).to_dict()

       config_dict.pop("""feature_extractor_type"""				)
       SCREAMING_SNAKE_CASE__      							=				WavaVecaFeatureExtractor(**_lowercase				)

       # save in new folder
       model_config.save_pretrained(_lowercase				)
       config.save_pretrained(_lowercase				)

       SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained(_lowercase				)

       # make sure private variable is not incorrectly saved
       SCREAMING_SNAKE_CASE__      							=				json.loads(config.to_json_string()				)
       self.assertTrue("""_processor_class""" not in dict_as_saved				)

     self.assertIsInstance(_lowercase	,     _lowercase				)




   def      __a						(   self      :    Optional[Any]				):



     """simple docstring"""






     SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained(_lowercase				)
     self.assertIsInstance(_lowercase	,     _lowercase				)




   def      __a						(   self      :    Dict				):



     """simple docstring"""






     with self.assertRaisesRegex(
         _lowercase	,     """bert-base is not a local folder and is not a valid model identifier"""				):
       SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained("""bert-base"""				)




   def      __a						(   self      :    Union[str, Any]				):



     """simple docstring"""






     with self.assertRaisesRegex(
         _lowercase	,     R"""aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)"""				):
       SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained(_lowercase	,     revision="""aaaaaa"""				)




   def      __a						(   self      :    List[Any]				):



     """simple docstring"""






     with self.assertRaisesRegex(
         _lowercase	,     """hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json."""	,     ):
       SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained("""hf-internal-testing/config-no-model"""				)




   def      __a						(   self      :    str				):



     """simple docstring"""






     with self.assertRaises(_lowercase				):
       SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained(
           """hf-internal-testing/test_dynamic_feature_extractor"""				)
     # If remote code is disabled, we can't load this config.
     with self.assertRaises(_lowercase				):
       SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained(
           """hf-internal-testing/test_dynamic_feature_extractor"""	,     trust_remote_code=_lowercase				)

     SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained(
         """hf-internal-testing/test_dynamic_feature_extractor"""	,     trust_remote_code=_lowercase				)
     self.assertEqual(feature_extractor.__class__.__name__	,     """NewFeatureExtractor"""				)

     # Test feature extractor can be reloaded.
     with tempfile.TemporaryDirectory() as tmp_dir:
       feature_extractor.save_pretrained(_lowercase				)
       SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained(_lowercase	,     trust_remote_code=_lowercase				)
     self.assertEqual(reloaded_feature_extractor.__class__.__name__	,     """NewFeatureExtractor"""				)




   def      __a						(   self      :    Union[str, Any]				):



     """simple docstring"""






     try:
       AutoConfig.register("""custom"""	,     _lowercase				)
       AutoFeatureExtractor.register(_lowercase	,     _lowercase				)
       # Trying to register something existing in the Transformers library will raise an error
       with self.assertRaises(_lowercase				):
         AutoFeatureExtractor.register(_lowercase	,     _lowercase				)

       # Now that the config is registered, it can be used as any other config with the auto-API
       SCREAMING_SNAKE_CASE__      							=				CustomFeatureExtractor.from_pretrained(_lowercase				)
       with tempfile.TemporaryDirectory() as tmp_dir:
         feature_extractor.save_pretrained(_lowercase				)
         SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained(_lowercase				)
         self.assertIsInstance(_lowercase	,     _lowercase				)

     finally:
       if "custom" in CONFIG_MAPPING._extra_content:
         del CONFIG_MAPPING._extra_content["custom"]
       if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
         del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]




   def      __a						(   self      :    Any				):



     """simple docstring"""






     class __snake_case    (						lowerCamelCase_		):
        lowerCAmelCase_														=							True

     try:
       AutoConfig.register("""custom"""	,     _lowercase				)
       AutoFeatureExtractor.register(_lowercase	,     _lowercase				)
       # If remote code is not set, the default is to use local
       SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained(
           """hf-internal-testing/test_dynamic_feature_extractor"""				)
       self.assertEqual(feature_extractor.__class__.__name__	,     """NewFeatureExtractor"""				)
       self.assertTrue(feature_extractor.is_local				)

       # If remote code is disabled, we load the local one.
       SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained(
           """hf-internal-testing/test_dynamic_feature_extractor"""	,     trust_remote_code=_lowercase				)
       self.assertEqual(feature_extractor.__class__.__name__	,     """NewFeatureExtractor"""				)
       self.assertTrue(feature_extractor.is_local				)

       # If remote is enabled, we load from the Hub
       SCREAMING_SNAKE_CASE__      							=				AutoFeatureExtractor.from_pretrained(
           """hf-internal-testing/test_dynamic_feature_extractor"""	,     trust_remote_code=_lowercase				)
       self.assertEqual(feature_extractor.__class__.__name__	,     """NewFeatureExtractor"""				)
       self.assertTrue(not hasattr(_lowercase	,     """is_local"""				)				)

     finally:
       if "custom" in CONFIG_MAPPING._extra_content:
         del CONFIG_MAPPING._extra_content["custom"]
       if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
         del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]
