

import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch




class      UpperCamelCase (     SCREAMING_SNAKE_CASE   ):

    @require_torch
    def UpperCamelCase			(			self :      Optional[int]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

     SCREAMING_SNAKE_CASE								=			'\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

     SCREAMING_SNAKE_CASE								=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise RuntimeError("Offline mode is enabled, we shouldn\'t access internet")\nsocket.socket = offline_socket\n        '

     # Force fetching the files so that we can use the cache
     SCREAMING_SNAKE_CASE								=			'hf-internal-testing/tiny-random-bert'
     BertConfig.from_pretrained(snake_case__    )
     BertModel.from_pretrained(snake_case__    )
     BertTokenizer.from_pretrained(snake_case__    )
     pipeline(task='fill-mask'							,    model=snake_case__    )

     # baseline - just load from_pretrained with normal network
     SCREAMING_SNAKE_CASE								=			[sys.executable, '-c', '\n'.join([load, run, mock]    )]

     # should succeed
     SCREAMING_SNAKE_CASE								=			self.get_env()
     # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
     SCREAMING_SNAKE_CASE								=			'1'
     SCREAMING_SNAKE_CASE								=			subprocess.run(snake_case__							,    env=snake_case__							,    check=snake_case__							,    capture_output=snake_case__    )
     self.assertEqual(result.returncode							,    0							,    result.stderr    )
     self.assertIn('success'							,    result.stdout.decode()    )

    @require_torch
    def UpperCamelCase			(			self :      Dict    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

     SCREAMING_SNAKE_CASE								=			'\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

     SCREAMING_SNAKE_CASE								=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Faking flaky internet")\nsocket.socket = offline_socket\n        '

     # Force fetching the files so that we can use the cache
     SCREAMING_SNAKE_CASE								=			'hf-internal-testing/tiny-random-bert'
     BertConfig.from_pretrained(snake_case__    )
     BertModel.from_pretrained(snake_case__    )
     BertTokenizer.from_pretrained(snake_case__    )
     pipeline(task='fill-mask'							,    model=snake_case__    )

     # baseline - just load from_pretrained with normal network
     SCREAMING_SNAKE_CASE								=			[sys.executable, '-c', '\n'.join([load, run, mock]    )]

     # should succeed
     SCREAMING_SNAKE_CASE								=			self.get_env()
     SCREAMING_SNAKE_CASE								=			subprocess.run(snake_case__							,    env=snake_case__							,    check=snake_case__							,    capture_output=snake_case__    )
     self.assertEqual(result.returncode							,    0							,    result.stderr    )
     self.assertIn('success'							,    result.stdout.decode()    )

    @require_torch
    def UpperCamelCase			(			self :      Optional[Any]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			'\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

     SCREAMING_SNAKE_CASE								=			'\nmname = "hf-internal-testing/tiny-random-bert-sharded"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint("success")\n        '

     SCREAMING_SNAKE_CASE								=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise ValueError("Offline mode is enabled")\nsocket.socket = offline_socket\n        '

     # baseline - just load from_pretrained with normal network
     SCREAMING_SNAKE_CASE								=			[sys.executable, '-c', '\n'.join([load, run]    )]

     # should succeed
     SCREAMING_SNAKE_CASE								=			self.get_env()
     SCREAMING_SNAKE_CASE								=			subprocess.run(snake_case__							,    env=snake_case__							,    check=snake_case__							,    capture_output=snake_case__    )
     self.assertEqual(result.returncode							,    0							,    result.stderr    )
     self.assertIn('success'							,    result.stdout.decode()    )

     # next emulate no network
     SCREAMING_SNAKE_CASE								=			[sys.executable, '-c', '\n'.join([load, mock, run]    )]

     # Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
     # env["TRANSFORMERS_OFFLINE"] = "0"
     # result = subprocess.run(cmd, env=env, check=False, capture_output=True)
     # self.assertEqual(result.returncode, 1, result.stderr)

     # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
     SCREAMING_SNAKE_CASE								=			'1'
     SCREAMING_SNAKE_CASE								=			subprocess.run(snake_case__							,    env=snake_case__							,    check=snake_case__							,    capture_output=snake_case__    )
     self.assertEqual(result.returncode							,    0							,    result.stderr    )
     self.assertIn('success'							,    result.stdout.decode()    )

    @require_torch
    def UpperCamelCase			(			self :      Dict    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			'\nfrom transformers import pipeline\n        '
     SCREAMING_SNAKE_CASE								=			'\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

     SCREAMING_SNAKE_CASE								=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
     SCREAMING_SNAKE_CASE								=			self.get_env()
     SCREAMING_SNAKE_CASE								=			'1'
     SCREAMING_SNAKE_CASE								=			[sys.executable, '-c', '\n'.join([load, mock, run]    )]
     SCREAMING_SNAKE_CASE								=			subprocess.run(snake_case__							,    env=snake_case__							,    check=snake_case__							,    capture_output=snake_case__    )
     self.assertEqual(result.returncode							,    1							,    result.stderr    )
     self.assertIn(
         'You cannot infer task automatically within `pipeline` when using offline mode'							,    result.stderr.decode().replace('\n'							,    ''    )							,    )




    @require_torch
    def UpperCamelCase			(			self :      Any    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			'\nfrom transformers import AutoModel\n        '
     SCREAMING_SNAKE_CASE								=			'\nmname = "hf-internal-testing/test_dynamic_model"\nAutoModel.from_pretrained(mname, trust_remote_code=True)\nprint("success")\n        '

     # baseline - just load from_pretrained with normal network
     SCREAMING_SNAKE_CASE								=			[sys.executable, '-c', '\n'.join([load, run]    )]

     # should succeed
     SCREAMING_SNAKE_CASE								=			self.get_env()
     SCREAMING_SNAKE_CASE								=			subprocess.run(snake_case__							,    env=snake_case__							,    check=snake_case__							,    capture_output=snake_case__    )
     self.assertEqual(result.returncode							,    0							,    result.stderr    )
     self.assertIn('success'							,    result.stdout.decode()    )

     # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
     SCREAMING_SNAKE_CASE								=			'1'
     SCREAMING_SNAKE_CASE								=			subprocess.run(snake_case__							,    env=snake_case__							,    check=snake_case__							,    capture_output=snake_case__    )
     self.assertEqual(result.returncode							,    0							,    result.stderr    )
     self.assertIn('success'							,    result.stdout.decode()    )




# 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 ..models.whisper import WhisperForConditionalGeneration, WhisperProcessor
from .base import PipelineTool




class      UpperCamelCase (     SCREAMING_SNAKE_CASE   ):
    __UpperCamelCase         ="openai/whisper-base"
    __UpperCamelCase         =(
        "This is a tool that transcribes an audio into text. It takes an input named `audio` and returns the "
        "transcribed text."
    )
    __UpperCamelCase         ="transcriber"
    __UpperCamelCase         =WhisperProcessor
    __UpperCamelCase         =WhisperForConditionalGeneration

    __UpperCamelCase         =["audio"]
    __UpperCamelCase         =["text"]

    def UpperCamelCase			(			self :      Dict							,    snake_case__ :      Tuple    ):

     """simple docstring"""






     return self.pre_processor(snake_case__							,    return_tensors='pt'    ).input_features

    def UpperCamelCase			(			self :      Optional[int]							,    snake_case__ :      Tuple    ):

     """simple docstring"""






     return self.model.generate(inputs=snake_case__    )




    def UpperCamelCase			(			self :      str							,    snake_case__ :      Union[str, Any]    ):

     """simple docstring"""






     return self.pre_processor.batch_decode(snake_case__							,    skip_special_tokens=snake_case__    )[0]

