"""simple docstring"""
import os
import sys
import warnings
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, List, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen
from ..table import array_cast
from ..utils.file_utils import is_local_path
from ..utils.py_utils import first_non_null_value, no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
					import PIL.Image

					from .features import FeatureType


SCREAMING_SNAKE_CASE__      						=  None
SCREAMING_SNAKE_CASE__      						=  "<" if sys.byteorder == "little" else ">"
# Origin: https://github.com/python-pillow/Pillow/blob/698951e19e19972aeed56df686868f1329981c12/src/PIL/Image.py#L3126 minus "|i1" which values are not preserved correctly when saving and loading an image
SCREAMING_SNAKE_CASE__      						=  [
    np.dtype("|b1"),
    np.dtype("|u1"),
    np.dtype("<u2"),
    np.dtype(">u2"),
    np.dtype("<i2"),
    np.dtype(">i2"),
    np.dtype("<u4"),
    np.dtype(">u4"),
    np.dtype("<i4"),
    np.dtype(">i4"),
    np.dtype("<f4"),
    np.dtype(">f4"),
    np.dtype("<f8"),
    np.dtype(">f8"),
]



@dataclass
class   lowerCAmelCase_    :



				"""simple docstring"""
				_lowerCAmelCase     :    bool			  =						True
				_lowerCAmelCase     :    Optional[str]			  =						None
				# Automatically constructed
				_lowerCAmelCase     :    ClassVar[str]			  =						"PIL.Image.Image"
				_lowerCAmelCase     :    ClassVar[Any]			  =						pa.struct({"""bytes""": pa.binary(), """path""": pa.string()}					)
				_lowerCAmelCase     :    str			  =						field(default="""Image"""     ,    init=lowerCAmelCase     ,    repr=lowerCAmelCase					)

				def __call__(				self    ):






										"""simple docstring"""

										return self.pa_type

				def 		snake_case (				self		,    lowerCAmelCase    ):






										"""simple docstring"""

										if config.PIL_AVAILABLE:
																import PIL.Image
										else:
																raise ImportError('To support encoding images, please install \'Pillow\'.'    )

										if isinstance(lowerCAmelCase		,    lowerCAmelCase    ):
																snake_case								=			np.array(lowerCAmelCase    )

										if isinstance(lowerCAmelCase		,    lowerCAmelCase    ):
																return {"path": value, "bytes": None}
										elif isinstance(lowerCAmelCase		,    lowerCAmelCase    ):
																return {"path": None, "bytes": value}
										elif isinstance(lowerCAmelCase		,    np.ndarray    ):
																# convert the image array to PNG/TIFF bytes
																return encode_np_array(lowerCAmelCase    )
										elif isinstance(lowerCAmelCase		,    PIL.Image.Image    ):
																# convert the PIL image to bytes (default format is PNG/TIFF)
																return encode_pil_image(lowerCAmelCase    )
										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
																return {"bytes": None, "path": value.get('path'    )}
										elif value.get('bytes'    ) is not None or value.get('path'    ) is not None:
																# store the image bytes, and path is used to infer the image format using the file extension
																return {"bytes": value.get('bytes'    ), "path": value.get('path'    )}
										else:
																raise ValueError(
																    F"""An image sample should have one of 'path' or 'bytes' but they are missing or None in {value}."""    )

				def 		snake_case (				self		,    lowerCAmelCase		,    lowerCAmelCase=None    ):






										"""simple docstring"""

										if not self.decode:
																raise RuntimeError('Decoding is disabled for this feature. Please use Image(decode=True) instead.'    )

										if config.PIL_AVAILABLE:
																import PIL.Image
										else:
																raise ImportError('To support decoding images, please install \'Pillow\'.'    )

										if token_per_repo_id is None:
																snake_case								=			{}

										snake_case					,snake_case								=			value['path'], value['bytes']
										if bytes_ is None:
																if path is None:
																						raise ValueError(F"""An image should have one of 'path' or 'bytes' but both are None in {value}."""    )
																else:
																						if is_local_path(lowerCAmelCase    ):
																												snake_case								=			PIL.Image.open(lowerCAmelCase    )
																						else:
																												snake_case								=			path.split('::'    )[-1]
																												try:
																																		snake_case								=			string_to_dict(lowerCAmelCase		,    config.HUB_DATASETS_URL    )['repo_id']
																																		snake_case								=			token_per_repo_id.get(lowerCAmelCase    )
																												except ValueError:
																																		snake_case								=			None
																												with xopen(lowerCAmelCase		,    'rb'		,    use_auth_token=lowerCAmelCase    ) as f:
																																		snake_case								=			BytesIO(f.read()    )
																												snake_case								=			PIL.Image.open(bytes_    )
										else:
																snake_case								=			PIL.Image.open(BytesIO(bytes_    )    )
										image.load()  # to avoid "Too many open files" errors
										return image

				def 		snake_case (				self    ):






										"""simple docstring"""

										from .features import Value

										return (
										    self
										    if self.decode
										    else {
										        "bytes": Value('binary'    ),
										        "path": Value('string'    ),
										    }
										)

				def 		snake_case (				self		,    lowerCAmelCase    ):






										"""simple docstring"""

										if pa.types.is_string(storage.type    ):
																snake_case								=			pa.array([None] * len(lowerCAmelCase    )		,    type=pa.binary()    )
																snake_case								=			pa.StructArray.from_arrays([bytes_array, storage]		,    ['bytes', 'path']		,    mask=storage.is_null()    )
										elif pa.types.is_binary(storage.type    ):
																snake_case								=			pa.array([None] * len(lowerCAmelCase    )		,    type=pa.string()    )
																snake_case								=			pa.StructArray.from_arrays([storage, path_array]		,    ['bytes', 'path']		,    mask=storage.is_null()    )
										elif pa.types.is_struct(storage.type    ):
																if storage.type.get_field_index('bytes'    ) >= 0:
																						snake_case								=			storage.field('bytes'    )
																else:
																						snake_case								=			pa.array([None] * len(lowerCAmelCase    )		,    type=pa.binary()    )
																if storage.type.get_field_index('path'    ) >= 0:
																						snake_case								=			storage.field('path'    )
																else:
																						snake_case								=			pa.array([None] * len(lowerCAmelCase    )		,    type=pa.string()    )
																snake_case								=			pa.StructArray.from_arrays([bytes_array, path_array]		,    ['bytes', 'path']		,    mask=storage.is_null()    )
										elif pa.types.is_list(storage.type    ):
																snake_case								=			pa.array(
																    [encode_np_array(np.array(lowerCAmelCase    )    )['bytes'] if arr is not None else None for arr in storage.to_pylist()]		,    type=pa.binary()		,    )
																snake_case								=			pa.array([None] * len(lowerCAmelCase    )		,    type=pa.string()    )
																snake_case								=			pa.StructArray.from_arrays(
																    [bytes_array, path_array]		,    ['bytes', 'path']		,    mask=bytes_array.is_null()    )
										return array_cast(lowerCAmelCase		,    self.pa_type    )

				def 		snake_case (				self		,    lowerCAmelCase    ):






										"""simple docstring"""

										@no_op_if_value_is_null
										def path_to_bytes(lowerCAmelCase    ):
																with xopen(lowerCAmelCase		,    'rb'    ) as f:
																						snake_case								=			f.read()
																return bytes_

										snake_case								=			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()		,    )
										snake_case								=			pa.array(
										    [os.path.basename(lowerCAmelCase    ) if path is not None else None for path in storage.field('path'    ).to_pylist()]		,    type=pa.string()		,    )
										snake_case								=			pa.StructArray.from_arrays([bytes_array, path_array]		,    ['bytes', 'path']		,    mask=bytes_array.is_null()    )
										return array_cast(lowerCAmelCase		,    self.pa_type    )





def    lowerCAmelCase__    (			)					->				List[str]:





						"""simple docstring"""






						if config.PIL_AVAILABLE:
												import PIL.Image
						else:
												raise ImportError('To support encoding images, please install \'Pillow\'.'					)

						global _IMAGE_COMPRESSION_FORMATS
						if _IMAGE_COMPRESSION_FORMATS is None:
												PIL.Image.init()
												snake_case								=			list(set(PIL.Image.OPEN.keys()					) & set(PIL.Image.SAVE.keys()					)					)
						return _IMAGE_COMPRESSION_FORMATS





def    lowerCAmelCase__    (			_UpperCamelCase       :				"PIL.Image.Image"					)					->				bytes:





						"""simple docstring"""






						snake_case								=			BytesIO()
						if image.format in list_image_compression_formats():
												snake_case								=			image.format
						else:
												snake_case								=			'PNG' if image.mode in ['1', 'L', 'LA', 'RGB', 'RGBA'] else 'TIFF'
						image.save(_UpperCamelCase ,			format=_UpperCamelCase					)
						return buffer.getvalue()





def    lowerCAmelCase__    (			_UpperCamelCase       :				"PIL.Image.Image"					)					->				dict:





						"""simple docstring"""






						if hasattr(_UpperCamelCase ,			'filename'					) and image.filename != "":
												return {"path": image.filename, "bytes": None}
						else:
												return {"path": None, "bytes": image_to_bytes(_UpperCamelCase					)}





def    lowerCAmelCase__    (			_UpperCamelCase       :				np.ndarray					)					->				dict:





						"""simple docstring"""






						if config.PIL_AVAILABLE:
												import PIL.Image
						else:
												raise ImportError('To support encoding images, please install \'Pillow\'.'					)

						snake_case								=			array.dtype
						snake_case								=			dtype.byteorder if dtype.byteorder != '=' else _NATIVE_BYTEORDER
						snake_case								=			dtype.kind
						snake_case								=			dtype.itemsize

						snake_case								=			None

						# Multi-channel array case (only np.dtype("|u1") is allowed)
						if array.shape[2:]:
												snake_case								=			np.dtype('|u1'					)
												if dtype_kind not in ["u", "i"]:
																		raise TypeError(
																		    f"""Unsupported array dtype {dtype} for image encoding. Only {dest_dtype} is supported for multi-channel arrays."""					)
												if dtype is not dest_dtype:
																		warnings.warn(f"""Downcasting array dtype {dtype} to {dest_dtype} to be compatible with 'Pillow'"""					)
    # Exact match
						elif dtype in _VALID_IMAGE_ARRAY_DTPYES:
												snake_case								=			dtype
						else:  # Downcast the type within the kind (np.can_cast(from_type, to_type, casting="same_kind") doesn't behave as expected, so do it manually)
												while dtype_itemsize >= 1:
																		snake_case								=			dtype_byteorder + dtype_kind + str(_UpperCamelCase					)
																		snake_case								=			np.dtype(_UpperCamelCase					)
																		if dest_dtype in _VALID_IMAGE_ARRAY_DTPYES:
																								warnings.warn(f"""Downcasting array dtype {dtype} to {dest_dtype} to be compatible with 'Pillow'"""					)
																								break
																		else:
																								dtype_itemsize //= 2

						if dest_dtype is None:
												raise TypeError(
												    f"""Cannot convert dtype {dtype} to a valid image dtype. Valid image dtypes: {_VALID_IMAGE_ARRAY_DTPYES}"""					)
						snake_case								=			PIL.Image.fromarray(array.astype(_UpperCamelCase					)					)
						return {"path": None, "bytes": image_to_bytes(_UpperCamelCase					)}





def    lowerCAmelCase__    (			_UpperCamelCase       :				Union[List[str], List[dict], List[np.ndarray], List["PIL.Image.Image"]]					)					->				List[dict]:





						"""simple docstring"""






						if config.PIL_AVAILABLE:
												import PIL.Image
						else:
												raise ImportError('To support encoding images, please install \'Pillow\'.'					)

						if objs:
												snake_case					,snake_case								=			first_non_null_value(_UpperCamelCase					)
												if isinstance(_UpperCamelCase ,			_UpperCamelCase					):
																		return [{"path": obj, "bytes": None} if obj is not None else None for obj in objs]
												if isinstance(_UpperCamelCase ,			np.ndarray					):
																		snake_case								=			no_op_if_value_is_null(_UpperCamelCase					)
																		return [obj_to_image_dict_func(_UpperCamelCase					) for obj in objs]
												elif isinstance(_UpperCamelCase ,			PIL.Image.Image					):
																		snake_case								=			no_op_if_value_is_null(_UpperCamelCase					)
																		return [obj_to_image_dict_func(_UpperCamelCase					) for obj in objs]
												else:
																		return objs
						else:
												return objs



"""simple docstring"""
import subprocess
import sys

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



class   lowerCAmelCase_    (			lowerCAmelCase					):



				"""simple docstring"""

				@require_torch
				def 		snake_case (				self    ):






										"""simple docstring"""

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

										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        '

										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
										snake_case								=			'hf-internal-testing/tiny-random-bert'
										BertConfig.from_pretrained(lowerCAmelCase    )
										BertModel.from_pretrained(lowerCAmelCase    )
										BertTokenizer.from_pretrained(lowerCAmelCase    )
										pipeline(task='fill-mask'		,    model=lowerCAmelCase    )

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

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

				@require_torch
				def 		snake_case (				self    ):






										"""simple docstring"""

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

										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        '

										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
										snake_case								=			'hf-internal-testing/tiny-random-bert'
										BertConfig.from_pretrained(lowerCAmelCase    )
										BertModel.from_pretrained(lowerCAmelCase    )
										BertTokenizer.from_pretrained(lowerCAmelCase    )
										pipeline(task='fill-mask'		,    model=lowerCAmelCase    )

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

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

				@require_torch
				def 		snake_case (				self    ):






										"""simple docstring"""

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

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

										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
										snake_case								=			[sys.executable, '-c', '\n'.join([load, run]    )]

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

										# next emulate no network
										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
										snake_case								=			'1'
										snake_case								=			subprocess.run(lowerCAmelCase		,    env=lowerCAmelCase		,    check=lowerCAmelCase		,    capture_output=lowerCAmelCase    )
										self.assertEqual(result.returncode		,    0		,    result.stderr    )
										self.assertIn('success'		,    result.stdout.decode()    )

				@require_torch
				def 		snake_case (				self    ):






										"""simple docstring"""

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

										snake_case								=			'\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
										snake_case								=			self.get_env()
										snake_case								=			'1'
										snake_case								=			[sys.executable, '-c', '\n'.join([load, mock, run]    )]
										snake_case								=			subprocess.run(lowerCAmelCase		,    env=lowerCAmelCase		,    check=lowerCAmelCase		,    capture_output=lowerCAmelCase    )
										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 		snake_case (				self    ):






										"""simple docstring"""

										snake_case								=			'\nfrom transformers import AutoModel\n        '
										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
										snake_case								=			[sys.executable, '-c', '\n'.join([load, run]    )]

										# should succeed
										snake_case								=			self.get_env()
										snake_case								=			subprocess.run(lowerCAmelCase		,    env=lowerCAmelCase		,    check=lowerCAmelCase		,    capture_output=lowerCAmelCase    )
										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
										snake_case								=			'1'
										snake_case								=			subprocess.run(lowerCAmelCase		,    env=lowerCAmelCase		,    check=lowerCAmelCase		,    capture_output=lowerCAmelCase    )
										self.assertEqual(result.returncode		,    0		,    result.stderr    )
										self.assertIn('success'		,    result.stdout.decode()    )


