



'''simple docstring'''







import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


SCREAMING_SNAKE_CASE__     : List[str]				=   namedtuple("""ExampleNamedTuple""", """a b c""")




class 				__lowerCAmelCase(							unittest.TestCase      ):






			def 				_lowercase	(   self  :							Optional[Any]	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:Tuple										=      torch.randn(5      ,       2	)
					SCREAMING_SNAKE_CASE_						:int										=      torch.device('cuda'	) if torch.cuda.is_available() else torch.device('cpu'	)

					SCREAMING_SNAKE_CASE_						:Union[str, Any]										=      send_to_device(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta.cpu()      ,       SCREAMING_SNAKE_CASE	)	)

					SCREAMING_SNAKE_CASE_						:Optional[Any]										=      send_to_device((tensor, [tensor, tensor], 1)      ,       SCREAMING_SNAKE_CASE	)
					self.assertIsInstance(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta[0].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertIsInstance(resulta[1]      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta[1][0].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertTrue(torch.equal(resulta[1][1].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertEqual(resulta[2]      ,       1	)

					SCREAMING_SNAKE_CASE_						:Tuple										=      send_to_device({'a': tensor, 'b': [tensor, tensor], 'c': 1}      ,       SCREAMING_SNAKE_CASE	)
					self.assertIsInstance(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta['a'].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertIsInstance(resulta['b']      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta['b'][0].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertTrue(torch.equal(resulta['b'][1].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertEqual(resulta['c']      ,       1	)

					SCREAMING_SNAKE_CASE_						:List[Any]										=      send_to_device(ExampleNamedTuple(a=SCREAMING_SNAKE_CASE      ,       b=[tensor, tensor]      ,       c=1	)      ,       SCREAMING_SNAKE_CASE	)
					self.assertIsInstance(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta.a.cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertIsInstance(resulta.b      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta.b[0].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertTrue(torch.equal(resulta.b[1].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertEqual(resulta.c      ,       1	)

					SCREAMING_SNAKE_CASE_						:Tuple										=      send_to_device(UserDict({'a': tensor, 'b': [tensor, tensor], 'c': 1}	)      ,       SCREAMING_SNAKE_CASE	)
					self.assertIsInstance(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta['a'].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertIsInstance(resulta['b']      ,       SCREAMING_SNAKE_CASE	)
					self.assertTrue(torch.equal(resulta['b'][0].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertTrue(torch.equal(resulta['b'][1].cpu()      ,       SCREAMING_SNAKE_CASE	)	)
					self.assertEqual(resulta['c']      ,       1	)






			def 				_lowercase	(   self  :							List[str]	):




					"""simple docstring"""







					with self.assertRaises(SCREAMING_SNAKE_CASE	) as cm:
							SCREAMING_SNAKE_CASE_						:Tuple										=      recursively_apply(torch.tensor      ,       (torch.tensor(1	), 1)      ,       error_on_other_type=SCREAMING_SNAKE_CASE	)
					self.assertEqual(
					    str(cm.exception	)      ,       'Unsupported types (<class \'int\'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed.'      ,       )






			def 				_lowercase	(   self  :							str	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:List[str]										=      torch.tensor([1, 2, 3, 4, 5]	)
					self.assertEqual(listify(SCREAMING_SNAKE_CASE	)      ,       [1, 2, 3, 4, 5]	)

					SCREAMING_SNAKE_CASE_						:Any										=      torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]	)
					self.assertEqual(listify(SCREAMING_SNAKE_CASE	)      ,       [[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]]	)

					SCREAMING_SNAKE_CASE_						:List[Any]										=      torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]	)
					self.assertEqual(
					    listify(SCREAMING_SNAKE_CASE	)      ,       [[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]]	)






			def 				_lowercase	(   self  :							Tuple	):




					"""simple docstring"""







					with patch_environment(aa=1      ,       BB=2	):
							self.assertEqual(os.environ.get('AA'	)      ,       '1'	)
							self.assertEqual(os.environ.get('BB'	)      ,       '2'	)

					self.assertNotIn('AA'      ,       os.environ	)
					self.assertNotIn('BB'      ,       os.environ	)






			def 				_lowercase	(   self  :							Optional[int]	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:str										=      RegressionModel()
					SCREAMING_SNAKE_CASE_						:Tuple										=      model.forward
					SCREAMING_SNAKE_CASE_						:Any										=      convert_outputs_to_fpaa(model.forward	)
					SCREAMING_SNAKE_CASE_						:Tuple										=      extract_model_from_parallel(SCREAMING_SNAKE_CASE      ,       keep_fpaa_wrapper=SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:List[str]										=      pickle.dumps(SCREAMING_SNAKE_CASE	)






			@require_cuda
			def 				_lowercase	(   self  :							List[Any]	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:List[str]										=      RegressionModel()
					SCREAMING_SNAKE_CASE_						:Tuple										=      model.forward
					SCREAMING_SNAKE_CASE_						:Tuple										=      torch.cuda.amp.autocast(dtype=torch.floataa	)(model.forward	)
					SCREAMING_SNAKE_CASE_						:Any										=      convert_outputs_to_fpaa(model.forward	)
					SCREAMING_SNAKE_CASE_						:Optional[Any]										=      extract_model_from_parallel(SCREAMING_SNAKE_CASE      ,       keep_fpaa_wrapper=SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:List[Any]										=      pickle.dumps(SCREAMING_SNAKE_CASE	)






			@require_cuda
			@require_torch_min_version(version='2.0'	)
			def 				_lowercase	(   self  :							Dict	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:Dict										=      RegressionModel()
					SCREAMING_SNAKE_CASE_						:Tuple										=      model.forward
					SCREAMING_SNAKE_CASE_						:int										=      torch.cuda.amp.autocast(dtype=torch.floataa	)(model.forward	)
					SCREAMING_SNAKE_CASE_						:int										=      convert_outputs_to_fpaa(model.forward	)
					SCREAMING_SNAKE_CASE_						:List[Any]										=      torch.compile(model.forward      ,       backend='inductor'	)
					SCREAMING_SNAKE_CASE_						:str										=      torch.randn(4      ,       10	).cuda()
					SCREAMING_SNAKE_CASE_						:int										=      model(SCREAMING_SNAKE_CASE	)






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




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:List[str]										=      RegressionModel()
					# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
					SCREAMING_SNAKE_CASE_						:Optional[int]										=      torch.nn.parallel.DataParallel(SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:int										=      extract_model_from_parallel(SCREAMING_SNAKE_CASE	)

					self.assertEqual(SCREAMING_SNAKE_CASE      ,       SCREAMING_SNAKE_CASE	)






			@require_torch_min_version(version='2.0'	)
			def 				_lowercase	(   self  :							Optional[int]	):




					"""simple docstring"""







					SCREAMING_SNAKE_CASE_						:List[Any]										=      RegressionModel()
					SCREAMING_SNAKE_CASE_						:int										=      torch.compile(SCREAMING_SNAKE_CASE	)

					# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
					SCREAMING_SNAKE_CASE_						:Tuple										=      torch.nn.parallel.DataParallel(SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:Tuple										=      torch.compile(SCREAMING_SNAKE_CASE	)
					SCREAMING_SNAKE_CASE_						:Optional[int]										=      extract_model_from_parallel(SCREAMING_SNAKE_CASE	)

					self.assertEqual(compiled_model._orig_mod      ,       compiled_model_unwrapped._orig_mod	)






			def 				_lowercase	(   self  :							Optional[Any]	):




					"""simple docstring"""







					self.assertEqual(find_device([1, 'a', torch.tensor([1, 2, 3]	)]	)      ,       torch.device('cpu'	)	)
					self.assertEqual(find_device({'a': 1, 'b': torch.tensor([1, 2, 3]	)}	)      ,       torch.device('cpu'	)	)
					self.assertIsNone(find_device([1, 'a']	)	)





'''simple docstring'''




# Copyright 2023 The HuggingFace Inc. 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.
from typing import TYPE_CHECKING

from ..utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_torch_available,
)


a  :     Optional[Any]       =       {
    """agents""": ["""Agent""", """AzureOpenAiAgent""", """HfAgent""", """LocalAgent""", """OpenAiAgent"""],
    """base""": ["""PipelineTool""", """RemoteTool""", """Tool""", """launch_gradio_demo""", """load_tool"""],
}

try:
							if not is_torch_available():
														raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
							pass
else:
							a  :     Dict       =       ["""DocumentQuestionAnsweringTool"""]
							a  :     List[str]       =       ["""ImageCaptioningTool"""]
							a  :     Tuple       =       ["""ImageQuestionAnsweringTool"""]
							a  :     Tuple       =       ["""ImageSegmentationTool"""]
							a  :     List[str]       =       ["""SpeechToTextTool"""]
							a  :     Dict       =       ["""TextClassificationTool"""]
							a  :     Union[str, Any]       =       ["""TextQuestionAnsweringTool"""]
							a  :     List[str]       =       ["""TextSummarizationTool"""]
							a  :     Optional[Any]       =       ["""TextToSpeechTool"""]
							a  :     str       =       ["""TranslationTool"""]

if TYPE_CHECKING:
							from .agents import Agent, AzureOpenAiAgent, HfAgent, LocalAgent, OpenAiAgent
							from .base import PipelineTool, RemoteTool, Tool, launch_gradio_demo, load_tool

							try:
														if not is_torch_available():
																					raise OptionalDependencyNotAvailable()
							except OptionalDependencyNotAvailable:
														pass
							else:
														from .document_question_answering import DocumentQuestionAnsweringTool
														from .image_captioning import ImageCaptioningTool
														from .image_question_answering import ImageQuestionAnsweringTool
														from .image_segmentation import ImageSegmentationTool
														from .speech_to_text import SpeechToTextTool
														from .text_classification import TextClassificationTool
														from .text_question_answering import TextQuestionAnsweringTool
														from .text_summarization import TextSummarizationTool
														from .text_to_speech import TextToSpeechTool
														from .translation import TranslationTool
else:
							import sys

							a  :     Dict       =       _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)

