


'''simple docstring'''



import json
import os
import shutil
import warnings
from argparse import ArgumentParser, Namespace
from pathlib import Path
from typing import List

from ..utils import logging
from . import BaseTransformersCLICommand


try:
		from cookiecutter.main import cookiecutter

		__A		=True
except ImportError:
		__A		=False

__A		=logging.get_logger(__name__)  # pylint: disable=invalid-name







def       _UpperCamelCase (						UpperCamelCase__   ):
			return AddNewModelCommand(args.testing	, args.testing_file	, path=args.path   )


class 						_snake_case   (       a__		):




					@staticmethod
					def 			snake_case__  ( _lowerCamelCase):
								UpperCAmelCase__					:   int       				=       parser.add_parser("""add-new-model""")
								add_new_model_parser.add_argument("""--testing"""							,							action="""store_true"""							,							help="""If in testing mode.""")
								add_new_model_parser.add_argument("""--testing_file"""							,							type=_lowerCamelCase							,							help="""Configuration file on which to run.""")
								add_new_model_parser.add_argument(
								    """--path"""							,							type=_lowerCamelCase							,							help="""Path to cookiecutter. Should only be used for testing purposes.""")
								add_new_model_parser.set_defaults(func=_lowerCamelCase)




					def __init__( self							,							_lowerCamelCase							,							_lowerCamelCase							,							_lowerCamelCase=None							,							*_lowerCamelCase):
								UpperCAmelCase__					:   Dict       				=       testing
								UpperCAmelCase__					:   Optional[int]       				=       testing_file
								UpperCAmelCase__					:   Union[str, Any]       				=       path




					def 			snake_case__  ( self):
								warnings.warn(
								    """The command `transformers-cli add-new-model` is deprecated and will be removed in v5 of Transformers. """
								    """It is not actively maintained anymore, so might give a result that won't pass all tests and quality """
								    """checks, you should use `transformers-cli add-new-model-like` instead.""")
								if not _has_cookiecutter:
											raise ImportError(
											    """Model creation dependencies are required to use the `add_new_model` command. Install them by running """
											    """the following at the root of your `transformers` clone:\n\n\t$ pip install -e .[modelcreation]\n""")
								# Ensure that there is no other `cookiecutter-template-xxx` directory in the current working directory
								UpperCAmelCase__					:   int       				=       [directory for directory in os.listdir() if """cookiecutter-template-""" == directory[:22]]
								if len(_lowerCamelCase) > 0:
											raise ValueError(
											    """Several directories starting with `cookiecutter-template-` in current working directory. """
											    """Please clean your directory by removing all folders starting with `cookiecutter-template-` or """
											    """change your working directory.""")

								UpperCAmelCase__					:   str       				=       (
								    Path(_lowerCamelCase).parent.parent.parent.parent if self._path is None else Path(self._path).parent.parent
								)
								UpperCAmelCase__					:   List[Any]       				=       path_to_transformer_root / """templates""" / """adding_a_new_model"""

								# Execute cookiecutter
								if not self._testing:
											cookiecutter(str(_lowerCamelCase))
								else:
											with open(self._testing_file							,							"""r""") as configuration_file:
														UpperCAmelCase__					:   Any       				=       json.load(_lowerCamelCase)

											cookiecutter(
											    str(path_to_cookiecutter if self._path is None else self._path)							,							no_input=_lowerCamelCase							,							extra_context=_lowerCamelCase							,							)

								UpperCAmelCase__					:   Optional[Any]       				=       [directory for directory in os.listdir() if """cookiecutter-template-""" in directory[:22]][0]

								# Retrieve configuration
								with open(directory + """/configuration.json"""							,							"""r""") as configuration_file:
											UpperCAmelCase__					:   Dict       				=       json.load(_lowerCamelCase)

								UpperCAmelCase__					:   Dict       				=       configuration["""lowercase_modelname"""]
								UpperCAmelCase__					:   Tuple       				=       configuration["""generate_tensorflow_pytorch_and_flax"""]
								os.remove(f'''{directory}/configuration.json''')

								UpperCAmelCase__					:   Optional[int]       				=       """PyTorch""" in generate_tensorflow_pytorch_and_flax
								UpperCAmelCase__					:   Tuple       				=       """TensorFlow""" in generate_tensorflow_pytorch_and_flax
								UpperCAmelCase__					:   List[Any]       				=       """Flax""" in generate_tensorflow_pytorch_and_flax

								UpperCAmelCase__					:   Optional[int]       				=       f'''{path_to_transformer_root}/src/transformers/models/{lowercase_model_name}'''
								os.makedirs(_lowerCamelCase							,							exist_ok=_lowerCamelCase)
								os.makedirs(f'''{path_to_transformer_root}/tests/models/{lowercase_model_name}'''							,							exist_ok=_lowerCamelCase)

								# Tests require submodules as they have parent imports
								with open(f'''{path_to_transformer_root}/tests/models/{lowercase_model_name}/__init__.py'''							,							"""w"""):
											pass

								shutil.move(
								    f'''{directory}/__init__.py'''							,							f'''{model_dir}/__init__.py'''							,							)
								shutil.move(
								    f'''{directory}/configuration_{lowercase_model_name}.py'''							,							f'''{model_dir}/configuration_{lowercase_model_name}.py'''							,							)

								def remove_copy_lines(_lowerCamelCase):
											with open(_lowerCamelCase							,							"""r""") as f:
														UpperCAmelCase__					:   int       				=       f.readlines()
											with open(_lowerCamelCase							,							"""w""") as f:
														for line in lines:
																	if "# Copied from transformers." not in line:
																				f.write(_lowerCamelCase)

								if output_pytorch:
											if not self._testing:
														remove_copy_lines(f'''{directory}/modeling_{lowercase_model_name}.py''')

											shutil.move(
											    f'''{directory}/modeling_{lowercase_model_name}.py'''							,							f'''{model_dir}/modeling_{lowercase_model_name}.py'''							,							)

											shutil.move(
											    f'''{directory}/test_modeling_{lowercase_model_name}.py'''							,							f'''{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_{lowercase_model_name}.py'''							,							)
								else:
											os.remove(f'''{directory}/modeling_{lowercase_model_name}.py''')
											os.remove(f'''{directory}/test_modeling_{lowercase_model_name}.py''')

								if output_tensorflow:
											if not self._testing:
														remove_copy_lines(f'''{directory}/modeling_tf_{lowercase_model_name}.py''')

											shutil.move(
											    f'''{directory}/modeling_tf_{lowercase_model_name}.py'''							,							f'''{model_dir}/modeling_tf_{lowercase_model_name}.py'''							,							)

											shutil.move(
											    f'''{directory}/test_modeling_tf_{lowercase_model_name}.py'''							,							f'''{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_tf_{lowercase_model_name}.py'''							,							)
								else:
											os.remove(f'''{directory}/modeling_tf_{lowercase_model_name}.py''')
											os.remove(f'''{directory}/test_modeling_tf_{lowercase_model_name}.py''')

								if output_flax:
											if not self._testing:
														remove_copy_lines(f'''{directory}/modeling_flax_{lowercase_model_name}.py''')

											shutil.move(
											    f'''{directory}/modeling_flax_{lowercase_model_name}.py'''							,							f'''{model_dir}/modeling_flax_{lowercase_model_name}.py'''							,							)

											shutil.move(
											    f'''{directory}/test_modeling_flax_{lowercase_model_name}.py'''							,							f'''{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_flax_{lowercase_model_name}.py'''							,							)
								else:
											os.remove(f'''{directory}/modeling_flax_{lowercase_model_name}.py''')
											os.remove(f'''{directory}/test_modeling_flax_{lowercase_model_name}.py''')

								shutil.move(
								    f'''{directory}/{lowercase_model_name}.md'''							,							f'''{path_to_transformer_root}/docs/source/en/model_doc/{lowercase_model_name}.md'''							,							)

								shutil.move(
								    f'''{directory}/tokenization_{lowercase_model_name}.py'''							,							f'''{model_dir}/tokenization_{lowercase_model_name}.py'''							,							)

								shutil.move(
								    f'''{directory}/tokenization_fast_{lowercase_model_name}.py'''							,							f'''{model_dir}/tokenization_{lowercase_model_name}_fast.py'''							,							)

								from os import fdopen, remove
								from shutil import copymode, move
								from tempfile import mkstemp

								def replace(_lowerCamelCase							,							_lowerCamelCase							,							_lowerCamelCase):
											# Create temp file
											UpperCAmelCase__				,    UpperCAmelCase__					:   Any       				=       mkstemp()
											UpperCAmelCase__					:   List[Any]       				=       False
											with fdopen(_lowerCamelCase							,							"""w""") as new_file:
														with open(_lowerCamelCase) as old_file:
																	for line in old_file:
																				new_file.write(_lowerCamelCase)
																				if line_to_copy_below in line:
																							UpperCAmelCase__					:   Tuple       				=       True
																							for line_to_copy in lines_to_copy:
																										new_file.write(_lowerCamelCase)

											if not line_found:
														raise ValueError(f'''Line {line_to_copy_below} was not found in file.''')

											# Copy the file permissions from the old file to the new file
											copymode(_lowerCamelCase							,							_lowerCamelCase)
											# Remove original file
											remove(_lowerCamelCase)
											# Move new file
											move(_lowerCamelCase							,							_lowerCamelCase)

								def skip_units(_lowerCamelCase):
											return (
											    ("generating PyTorch" in line and not output_pytorch)
											    or ("generating TensorFlow" in line and not output_tensorflow)
											    or ("generating Flax" in line and not output_flax)
											)

								def replace_in_files(_lowerCamelCase):
											with open(_lowerCamelCase) as datafile:
														UpperCAmelCase__					:   int       				=       []
														UpperCAmelCase__					:   Any       				=       False
														UpperCAmelCase__					:   Any       				=       False
														for line in datafile:
																	if "# To replace in: " in line and "##" not in line:
																				UpperCAmelCase__					:   Union[str, Any]       				=       line.split("""\"""")[1]
																				UpperCAmelCase__					:   Union[str, Any]       				=       skip_units(_lowerCamelCase)
																	elif "# Below: " in line and "##" not in line:
																				UpperCAmelCase__					:   List[Any]       				=       line.split("""\"""")[1]
																				UpperCAmelCase__					:   Optional[int]       				=       skip_units(_lowerCamelCase)
																	elif "# End." in line and "##" not in line:
																				if not skip_file and not skip_snippet:
																							replace(_lowerCamelCase							,							_lowerCamelCase							,							_lowerCamelCase)

																				UpperCAmelCase__					:   Tuple       				=       []
																	elif "# Replace with" in line and "##" not in line:
																				UpperCAmelCase__					:   Any       				=       []
																	elif "##" not in line:
																				lines_to_copy.append(_lowerCamelCase)

											remove(_lowerCamelCase)

								replace_in_files(f'''{directory}/to_replace_{lowercase_model_name}.py''')
								os.rmdir(_lowerCamelCase)



'''simple docstring'''



import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
		from .features import FeatureType


__A		,	__A		,	__A		=False, False, False


@dataclass
class 						_snake_case   :
					lowerCAmelCase :Optional[int]						=   None
					lowerCAmelCase :bool						=   True
					lowerCAmelCase :bool						=   True
					lowerCAmelCase :Optional[str]						=   None
					# Automatically constructed
					lowerCAmelCase :ClassVar[str]						=   "dict"
					lowerCAmelCase :ClassVar[Any]						=   pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}		)
					lowerCAmelCase :str						=   field(default='''Audio'''    ,				init=a__    ,				repr=a__		)




					def __call__( self):
								return self.pa_type




					def 			snake_case__  ( self							,							_lowerCamelCase):
								try:
											import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
								except ImportError as err:
											raise ImportError("""To support encoding audio data, please install 'soundfile'.""") from err
								if isinstance(_lowerCamelCase							,							_lowerCamelCase):
											return {"bytes": None, "path": value}
								elif isinstance(_lowerCamelCase							,							_lowerCamelCase):
											return {"bytes": value, "path": None}
								elif "array" in value:
											# convert the audio array to wav bytes
											UpperCAmelCase__					:   Optional[int]       				=       BytesIO()
											sf.write(_lowerCamelCase							,							value["""array"""]							,							value["""sampling_rate"""]							,							format="""wav""")
											return {"bytes": buffer.getvalue(), "path": None}
								elif value.get("""path""") is not None and os.path.isfile(value["""path"""]):
											# we set "bytes": None to not duplicate the data if they're already available locally
											if value["path"].endswith("""pcm"""):
														# "PCM" only has raw audio bytes
														if value.get("""sampling_rate""") is None:
																	# At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
																	raise KeyError("""To use PCM files, please specify a 'sampling_rate' in Audio object""")
														if value.get("""bytes"""):
																	# If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
																	UpperCAmelCase__					:   Tuple       				=       np.frombuffer(value["""bytes"""]							,							dtype=np.intaa).astype(np.floataa) / 3_2767
														else:
																	UpperCAmelCase__					:   List[str]       				=       np.memmap(value["""path"""]							,							dtype="""h"""							,							mode="""r""").astype(np.floataa) / 3_2767

														UpperCAmelCase__					:   List[str]       				=       BytesIO(bytes())
														sf.write(_lowerCamelCase							,							_lowerCamelCase							,							value["""sampling_rate"""]							,							format="""wav""")
														return {"bytes": buffer.getvalue(), "path": None}
											else:
														return {"bytes": None, "path": value.get("""path""")}
								elif value.get("""bytes""") is not None or value.get("""path""") is not None:
											# store the audio bytes, and path is used to infer the audio format using the file extension
											return {"bytes": value.get("""bytes"""), "path": value.get("""path""")}
								else:
											raise ValueError(
											    f'''An audio sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}.''')




					def 			snake_case__  ( self							,							_lowerCamelCase							,							_lowerCamelCase = None):
								if not self.decode:
											raise RuntimeError("""Decoding is disabled for this feature. Please use Audio(decode=True) instead.""")

								UpperCAmelCase__				,    UpperCAmelCase__					:   Tuple       				=       (value["""path"""], BytesIO(value["""bytes"""])) if value["""bytes"""] is not None else (value["""path"""], None)
								if path is None and file is None:
											raise ValueError(f'''An audio sample should have one of \'path\' or \'bytes\' but both are None in {value}.''')

								try:
											import librosa
											import soundfile as sf
								except ImportError as err:
											raise ImportError("""To support decoding audio files, please install 'librosa' and 'soundfile'.""") from err

								UpperCAmelCase__					:   Dict       				=       xsplitext(_lowerCamelCase)[1][1:].lower() if path is not None else None
								if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
											raise RuntimeError(
											    """Decoding 'opus' files requires system library 'libsndfile'>=1.0.31, """
											    """You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. """)
								elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
											raise RuntimeError(
											    """Decoding 'mp3' files requires system library 'libsndfile'>=1.1.0, """
											    """You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. """)

								if file is None:
											UpperCAmelCase__					:   int       				=       token_per_repo_id or {}
											UpperCAmelCase__					:   Optional[int]       				=       path.split("""::""")[-1]
											try:
														UpperCAmelCase__					:   Dict       				=       string_to_dict(_lowerCamelCase							,							config.HUB_DATASETS_URL)["""repo_id"""]
														UpperCAmelCase__					:   Dict       				=       token_per_repo_id[repo_id]
											except (ValueError, KeyError):
														UpperCAmelCase__					:   List[Any]       				=       None

											with xopen(_lowerCamelCase							,							"""rb"""							,							use_auth_token=_lowerCamelCase) as f:
														UpperCAmelCase__				,    UpperCAmelCase__					:   List[str]       				=       sf.read(_lowerCamelCase)

								else:
											UpperCAmelCase__				,    UpperCAmelCase__					:   List[Any]       				=       sf.read(_lowerCamelCase)

								UpperCAmelCase__					:   str       				=       array.T
								if self.mono:
											UpperCAmelCase__					:   List[Any]       				=       librosa.to_mono(_lowerCamelCase)
								if self.sampling_rate and self.sampling_rate != sampling_rate:
											UpperCAmelCase__					:   int       				=       librosa.resample(_lowerCamelCase							,							orig_sr=_lowerCamelCase							,							target_sr=self.sampling_rate)
											UpperCAmelCase__					:   Tuple       				=       self.sampling_rate

								return {"path": path, "array": array, "sampling_rate": sampling_rate}




					def 			snake_case__  ( self):
								from .features import Value

								if self.decode:
											raise ValueError("""Cannot flatten a decoded Audio feature.""")
								return {
								    "bytes": Value("""binary"""),
								    "path": Value("""string"""),
								}




					def 			snake_case__  ( self							,							_lowerCamelCase):
								if pa.types.is_string(storage.type):
											UpperCAmelCase__					:   Dict       				=       pa.array([None] * len(_lowerCamelCase)							,							type=pa.binary())
											UpperCAmelCase__					:   Tuple       				=       pa.StructArray.from_arrays([bytes_array, storage]							,							["""bytes""", """path"""]							,							mask=storage.is_null())
								elif pa.types.is_binary(storage.type):
											UpperCAmelCase__					:   Optional[int]       				=       pa.array([None] * len(_lowerCamelCase)							,							type=pa.string())
											UpperCAmelCase__					:   str       				=       pa.StructArray.from_arrays([storage, path_array]							,							["""bytes""", """path"""]							,							mask=storage.is_null())
								elif pa.types.is_struct(storage.type) and storage.type.get_all_field_indices("""array"""):
											UpperCAmelCase__					:   Optional[Any]       				=       pa.array([Audio().encode_example(_lowerCamelCase) if x is not None else None for x in storage.to_pylist()])
								elif pa.types.is_struct(storage.type):
											if storage.type.get_field_index("""bytes""") >= 0:
														UpperCAmelCase__					:   int       				=       storage.field("""bytes""")
											else:
														UpperCAmelCase__					:   List[str]       				=       pa.array([None] * len(_lowerCamelCase)							,							type=pa.binary())
											if storage.type.get_field_index("""path""") >= 0:
														UpperCAmelCase__					:   List[Any]       				=       storage.field("""path""")
											else:
														UpperCAmelCase__					:   Optional[int]       				=       pa.array([None] * len(_lowerCamelCase)							,							type=pa.string())
											UpperCAmelCase__					:   Optional[int]       				=       pa.StructArray.from_arrays([bytes_array, path_array]							,							["""bytes""", """path"""]							,							mask=storage.is_null())
								return array_cast(_lowerCamelCase							,							self.pa_type)




					def 			snake_case__  ( self							,							_lowerCamelCase):

								@no_op_if_value_is_null
								def path_to_bytes(_lowerCamelCase):
											with xopen(_lowerCamelCase							,							"""rb""") as f:
														UpperCAmelCase__					:   int       				=       f.read()
											return bytes_

								UpperCAmelCase__					:   Optional[Any]       				=       pa.array(
								    [
								        (path_to_bytes(x["""path"""]) if x["""bytes"""] is None else x["""bytes"""]) if x is not None else None
								        for x in storage.to_pylist()
								    ]							,							type=pa.binary()							,							)
								UpperCAmelCase__					:   Optional[Any]       				=       pa.array(
								    [os.path.basename(_lowerCamelCase) if path is not None else None for path in storage.field("""path""").to_pylist()]							,							type=pa.string()							,							)
								UpperCAmelCase__					:   List[str]       				=       pa.StructArray.from_arrays([bytes_array, path_array]							,							["""bytes""", """path"""]							,							mask=bytes_array.is_null())
								return array_cast(_lowerCamelCase							,							self.pa_type)