


'''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


__A :  Optional[List[str]]    							=				None
__A :  List[Any]    							=				"<" 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
__A :  List[str]    							=				[
    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        __snake_case	:







    """simple docstring"""



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

    def __call__(       self   :  Union[str, Any]		)       ->  Dict:
      return self.pa_type

    def __lowercase       (       self   :  Union[str, Any]   ,						lowerCamelCase   :  Union[str, bytes, dict, np.ndarray, "PIL.Image.Image"]		)       ->  dict:
      if config.PIL_AVAILABLE:
        import PIL.Image
      else:
        raise ImportError("""To support encoding images, please install 'Pillow'."""		)

      if isinstance(lowerCamelCase   ,						lowerCamelCase		):
        lowerCAmelCase_      :				Union[str, Any]      =       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 __lowercase       (       self   :  Tuple   ,						lowerCamelCase   :  dict   ,						lowerCamelCase   :  Optional[int]=None		)       ->  "PIL.Image.Image":
      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:
        lowerCAmelCase_      :				Dict      =       {}

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

    def __lowercase       (       self   :  int		)       ->  Union["FeatureType", Dict[str, "FeatureType"]]:
      from .features import Value

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

    def __lowercase       (       self   :  List[str]   ,						lowerCamelCase   :  Union[pa.StringArray, pa.StructArray, pa.ListArray]		)       ->  pa.StructArray:
      if pa.types.is_string(storage.type		):
        lowerCAmelCase_      :				Dict      =       pa.array([None] * len(lowerCamelCase		)   ,						type=pa.binary()		)
        lowerCAmelCase_      :				Optional[Any]      =       pa.StructArray.from_arrays([bytes_array, storage]   ,						["""bytes""", """path"""]   ,						mask=storage.is_null()		)
      elif pa.types.is_binary(storage.type		):
        lowerCAmelCase_      :				Optional[Any]      =       pa.array([None] * len(lowerCamelCase		)   ,						type=pa.string()		)
        lowerCAmelCase_      :				Optional[Any]      =       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:
          lowerCAmelCase_      :				Optional[int]      =       storage.field("""bytes"""		)
        else:
          lowerCAmelCase_      :				List[str]      =       pa.array([None] * len(lowerCamelCase		)   ,						type=pa.binary()		)
        if storage.type.get_field_index("""path"""		) >= 0:
          lowerCAmelCase_      :				str      =       storage.field("""path"""		)
        else:
          lowerCAmelCase_      :				Optional[Any]      =       pa.array([None] * len(lowerCamelCase		)   ,						type=pa.string()		)
        lowerCAmelCase_      :				List[str]      =       pa.StructArray.from_arrays([bytes_array, path_array]   ,						["""bytes""", """path"""]   ,						mask=storage.is_null()		)
      elif pa.types.is_list(storage.type		):
        lowerCAmelCase_      :				Union[str, Any]      =       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()   ,						)
        lowerCAmelCase_      :				List[str]      =       pa.array([None] * len(lowerCamelCase		)   ,						type=pa.string()		)
        lowerCAmelCase_      :				List[Any]      =       pa.StructArray.from_arrays(
            [bytes_array, path_array]   ,						["""bytes""", """path"""]   ,						mask=bytes_array.is_null()		)
      return array_cast(lowerCamelCase   ,						self.pa_type		)





    def __lowercase       (       self   :  Any   ,						lowerCamelCase   :  pa.StructArray		)       ->  pa.StructArray:

      @no_op_if_value_is_null
      def path_to_bytes(lowerCamelCase   :  List[str]		):
        with xopen(lowerCamelCase   ,						"""rb"""		) as f:
          lowerCAmelCase_      :				Union[str, Any]      =       f.read()
        return bytes_

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







def     UpperCamelCase_      (   ):
  '''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()
    lowerCAmelCase_      :				str      =       list(set(PIL.Image.OPEN.keys() ) & set(PIL.Image.SAVE.keys() ) )
  return _IMAGE_COMPRESSION_FORMATS







def     UpperCamelCase_      (   A__ :       "PIL.Image.Image" ):
  '''simple docstring'''





  lowerCAmelCase_      :				Dict      =       BytesIO()
  if image.format in list_image_compression_formats():
    lowerCAmelCase_      :				List[str]      =       image.format
  else:
    lowerCAmelCase_      :				Optional[int]      =       """PNG""" if image.mode in ["""1""", """L""", """LA""", """RGB""", """RGBA"""] else """TIFF"""
  image.save(A__							,   format=A__ )
  return buffer.getvalue()







def     UpperCamelCase_      (   A__ :       "PIL.Image.Image" ):
  '''simple docstring'''





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







def     UpperCamelCase_      (   A__ :       np.ndarray ):
  '''simple docstring'''





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

  lowerCAmelCase_      :				Tuple      =       array.dtype
  lowerCAmelCase_      :				Any      =       dtype.byteorder if dtype.byteorder != """=""" else _NATIVE_BYTEORDER
  lowerCAmelCase_      :				Optional[Any]      =       dtype.kind
  lowerCAmelCase_      :				Any      =       dtype.itemsize

  lowerCAmelCase_      :				Optional[int]      =       None

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







def     UpperCamelCase_      (   A__ :       Union[List[str], List[dict], List[np.ndarray], List["PIL.Image.Image"]] ):
  '''simple docstring'''





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

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







'''simple docstring'''






from ...utils import is_note_seq_available, is_transformers_available, is_torch_available
from ...utils import OptionalDependencyNotAvailable


try:
  if not (is_transformers_available() and is_torch_available()):
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  from ...utils.dummy_torch_and_transformers_objects import *  # noqa F403
else:
  from .notes_encoder import SpectrogramNotesEncoder
  from .continous_encoder import SpectrogramContEncoder
  from .pipeline_spectrogram_diffusion import (
      SpectrogramContEncoder,
      SpectrogramDiffusionPipeline,
      TaFilmDecoder,
  )

try:
  if not (is_transformers_available() and is_torch_available() and is_note_seq_available()):
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  from ...utils.dummy_transformers_and_torch_and_note_seq_objects import *  # noqa F403
else:
  from .midi_utils import MidiProcessor



