





import inspect
import re
from hashlib import shaaaa
from typing import Dict, List

from .arrow import arrow
from .audiofolder import audiofolder
from .csv import csv
from .imagefolder import imagefolder
from .json import json
from .pandas import pandas
from .parquet import parquet
from .sql import sql  # noqa F401
from .text import text




def      a__			(       lowercase__	):






   '''simple docstring'''






   UpperCAmelCase_           =[]
   for line in lines:
      UpperCAmelCase_           =re.sub(R"#.*"	,     ""	,     lowercase__	)  # remove comments
      if line:
         filtered_lines.append(lowercase__	)
   UpperCAmelCase_           ="\n".join(lowercase__	)

   # Make a hash from all this code
   UpperCAmelCase_           =full_str.encode("utf-8"	)
   return shaaaa(lowercase__	).hexdigest()


# get importable module names and hash for caching
__lowercase	: Tuple       ={
    """csv""": (csv.__name__, _hash_python_lines(inspect.getsource(csv).splitlines())),
    """json""": (json.__name__, _hash_python_lines(inspect.getsource(json).splitlines())),
    """pandas""": (pandas.__name__, _hash_python_lines(inspect.getsource(pandas).splitlines())),
    """parquet""": (parquet.__name__, _hash_python_lines(inspect.getsource(parquet).splitlines())),
    """arrow""": (arrow.__name__, _hash_python_lines(inspect.getsource(arrow).splitlines())),
    """text""": (text.__name__, _hash_python_lines(inspect.getsource(text).splitlines())),
    """imagefolder""": (imagefolder.__name__, _hash_python_lines(inspect.getsource(imagefolder).splitlines())),
    """audiofolder""": (audiofolder.__name__, _hash_python_lines(inspect.getsource(audiofolder).splitlines())),
}

# Used to infer the module to use based on the data files extensions
__lowercase	: Optional[int]       ={
    """.csv""": ("""csv""", {}),
    """.tsv""": ("""csv""", {"""sep""": """\t"""}),
    """.json""": ("""json""", {}),
    """.jsonl""": ("""json""", {}),
    """.parquet""": ("""parquet""", {}),
    """.arrow""": ("""arrow""", {}),
    """.txt""": ("""text""", {}),
}
_EXTENSION_TO_MODULE.update({ext: ("""imagefolder""", {}) for ext in imagefolder.ImageFolder.EXTENSIONS})
_EXTENSION_TO_MODULE.update({ext.upper(): ("""imagefolder""", {}) for ext in imagefolder.ImageFolder.EXTENSIONS})
_EXTENSION_TO_MODULE.update({ext: ("""audiofolder""", {}) for ext in audiofolder.AudioFolder.EXTENSIONS})
_EXTENSION_TO_MODULE.update({ext.upper(): ("""audiofolder""", {}) for ext in audiofolder.AudioFolder.EXTENSIONS})
__lowercase	: Dict       ={"""imagefolder""", """audiofolder"""}

# Used to filter data files based on extensions given a module name
__lowercase	: Dict[str, List[str]]       ={}
for _ext, (_module, _) in _EXTENSION_TO_MODULE.items():
     _MODULE_TO_EXTENSIONS.setdefault(_module, []).append(_ext)

_MODULE_TO_EXTENSIONS["imagefolder"].append(""".zip""")





_MODULE_TO_EXTENSIONS["audiofolder"].append(""".zip""")










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


_lowerCamelCase      :						Optional[List[str]]							=  None
_lowerCamelCase      :						int							=  """<""" 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
_lowerCamelCase      :						Union[str, Any]							=  [
    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 							UpperCamelCase_       :

    '''simple docstring'''






    UpperCAmelCase__				    = True
    UpperCAmelCase__				    = None
    # Automatically constructed
    UpperCAmelCase__				    = "PIL.Image.Image"
    UpperCAmelCase__				    = pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}	)
    UpperCAmelCase__				    = field(default='''Image'''						,   init=UpperCAmelCase__						,   repr=UpperCAmelCase__	)


    def __call__(     self :      List[str])		->List[str]:





           '''simple docstring'''

           return self.pa_type


    def 			SCREAMING_SNAKE_CASE				(     self :      Optional[Any]   ,					UpperCAmelCase__ :      Union[str, bytes, dict, np.ndarray, "PIL.Image.Image"])		->dict:





           '''simple docstring'''

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

           if isinstance(UpperCAmelCase__   ,					UpperCAmelCase__):
                  A__ 					=       np.array(UpperCAmelCase__)

           if isinstance(UpperCAmelCase__   ,					UpperCAmelCase__):
                  return {"path": value, "bytes": None}
           elif isinstance(UpperCAmelCase__   ,					UpperCAmelCase__):
                  return {"path": None, "bytes": value}
           elif isinstance(UpperCAmelCase__   ,					np.ndarray):
                  # convert the image array to PNG/TIFF bytes
                  return encode_np_array(UpperCAmelCase__)
           elif isinstance(UpperCAmelCase__   ,					PIL.Image.Image):
                  # convert the PIL image to bytes (default format is PNG/TIFF)
                  return encode_pil_image(UpperCAmelCase__)
           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 			SCREAMING_SNAKE_CASE				(     self :      Optional[int]   ,					UpperCAmelCase__ :      dict   ,					UpperCAmelCase__ :      str=None)		->"PIL.Image.Image":





           '''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:
                  A__ 					=       {}

           A__					,				A__ 					=       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(UpperCAmelCase__):
                                A__ 					=       PIL.Image.open(UpperCAmelCase__)
                         else:
                                A__ 					=       path.split('''::''')[-1]
                                try:
                                       A__ 					=       string_to_dict(UpperCAmelCase__   ,					config.HUB_DATASETS_URL)['''repo_id''']
                                       A__ 					=       token_per_repo_id.get(UpperCAmelCase__)
                                except ValueError:
                                       A__ 					=       None
                                with xopen(UpperCAmelCase__   ,					'''rb'''   ,					use_auth_token=UpperCAmelCase__) as f:
                                       A__ 					=       BytesIO(f.read())
                                A__ 					=       PIL.Image.open(bytes_)
           else:
                  A__ 					=       PIL.Image.open(BytesIO(bytes_))
           image.load()  # to avoid "Too many open files" errors
           return image


    def 			SCREAMING_SNAKE_CASE				(     self :      Dict)		->Union["FeatureType", Dict[str, "FeatureType"]]:





           '''simple docstring'''

           from .features import Value

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


    def 			SCREAMING_SNAKE_CASE				(     self :      Dict   ,					UpperCAmelCase__ :      Union[pa.StringArray, pa.StructArray, pa.ListArray])		->pa.StructArray:





           '''simple docstring'''

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





    def 			SCREAMING_SNAKE_CASE				(     self :      List[Any]   ,					UpperCAmelCase__ :      pa.StructArray)		->pa.StructArray:





           '''simple docstring'''

           @no_op_if_value_is_null
           def path_to_bytes(UpperCAmelCase__ :      Dict):
                  with xopen(UpperCAmelCase__   ,					'''rb''') as f:
                         A__ 					=       f.read()
                  return bytes_

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







def SCREAMING_SNAKE_CASE							(			)  ->			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()
              A__ 					=       list(set(PIL.Image.OPEN.keys()		) & set(PIL.Image.SAVE.keys()		)		)
       return _IMAGE_COMPRESSION_FORMATS







def SCREAMING_SNAKE_CASE							(			lowercase_		)  ->			bytes:



       """simple docstring"""

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







def SCREAMING_SNAKE_CASE							(			lowercase_		)  ->			dict:



       """simple docstring"""

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







def SCREAMING_SNAKE_CASE							(			lowercase_		)  ->			dict:



       """simple docstring"""

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

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

       A__ 					=       None

       # Multi-channel array case (only np.dtype("|u1") is allowed)
       if array.shape[2:]:
              A__ 					=       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:
              A__ 					=       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:
                     A__ 					=       dtype_byteorder + dtype_kind + str(lowercase_		)
                     A__ 					=       np.dtype(lowercase_		)
                     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}"""		)
       A__ 					=       PIL.Image.fromarray(array.astype(lowercase_		)		)
       return {"path": None, "bytes": image_to_bytes(lowercase_		)}







def SCREAMING_SNAKE_CASE							(			lowercase_		)  ->			List[dict]:



       """simple docstring"""

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

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


