



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


_lowerCAmelCase										=     None
_lowerCAmelCase										=     '''<''' 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										=     [
    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'''




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




  def __call__(   self				)   ->		Tuple:
      return self.pa_type




  def 							UpperCAmelCase_     (   self  ,__UpperCAmelCase				)   ->		dict:
      if config.PIL_AVAILABLE:
          import PIL.Image
      else:
          raise ImportError("""To support encoding images, please install 'Pillow'."""				)

      if isinstance(__UpperCAmelCase  ,__UpperCAmelCase				):
          lowerCAmelCase__			:  int											= 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 							UpperCAmelCase_     (   self  ,__UpperCAmelCase  ,__UpperCAmelCase=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__			:  List[Any]											= {}

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




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

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




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






  def 							UpperCAmelCase_     (   self  ,__UpperCAmelCase				)   ->		pa.StructArray:

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

      lowerCAmelCase__			:  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__			:  Any											= pa.array(
          [os.path.basename(__UpperCAmelCase				) if path is not None else None for path in storage.field("""path"""				).to_pylist()]  ,type=pa.string()  ,)
      lowerCAmelCase__			:  Any											= 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  (	):



    """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__			:  List[str]											= list(set(PIL.Image.OPEN.keys()      ) & set(PIL.Image.SAVE.keys()      )      )
    return _IMAGE_COMPRESSION_FORMATS
def        _SCREAMING_SNAKE_CASE  (	UpperCamelCase      ):



    """simple docstring"""





    lowerCAmelCase__			:  Dict											= BytesIO()
    if image.format in list_image_compression_formats():
        lowerCAmelCase__			:  Dict											= image.format
    else:
        lowerCAmelCase__			:  List[str]											= """PNG""" if image.mode in ["""1""", """L""", """LA""", """RGB""", """RGBA"""] else """TIFF"""
    image.save(UpperCamelCase							,   format=UpperCamelCase      )
    return buffer.getvalue()
def        _SCREAMING_SNAKE_CASE  (	UpperCamelCase      ):



    """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        _SCREAMING_SNAKE_CASE  (	UpperCamelCase      ):



    """simple docstring"""





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

    lowerCAmelCase__			:  Union[str, Any]											= array.dtype
    lowerCAmelCase__			:  Optional[int]											= dtype.byteorder if dtype.byteorder != """=""" else _NATIVE_BYTEORDER
    lowerCAmelCase__			:  Optional[Any]											= dtype.kind
    lowerCAmelCase__			:  str											= dtype.itemsize

    lowerCAmelCase__			:  str											= None

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




def        _SCREAMING_SNAKE_CASE  (	UpperCamelCase      ):



    """simple docstring"""





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

    if objs:
        lowerCAmelCase__			:  Tuple											= 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      ):
            lowerCAmelCase__			:  Dict											= no_op_if_value_is_null(UpperCamelCase      )
            return [obj_to_image_dict_func(UpperCamelCase      ) for obj in objs]
        elif isinstance(UpperCamelCase							,   PIL.Image.Image      ):
            lowerCAmelCase__			:  Optional[Any]											= 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 copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...models.auto.modeling_auto import MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
from ...utils import logging
from ..auto import CONFIG_MAPPING


_lowerCAmelCase										=     logging.get_logger(__name__)

_lowerCAmelCase										=     {
    '''Salesforce/instruct-blip-flan-t5''': '''https://huggingface.co/Salesforce/instruct-blip-flan-t5/resolve/main/config.json''',
}






class 	lowerCAmelCase_(    SCREAMING_SNAKE_CASE_					):
  '''simple docstring'''




  __lowercase							:				Optional[Any]           =				'''instructblip_vision_model'''




  def __init__(   self  ,__UpperCAmelCase=1408  ,__UpperCAmelCase=6144  ,__UpperCAmelCase=39  ,__UpperCAmelCase=16  ,__UpperCAmelCase=224  ,__UpperCAmelCase=14  ,__UpperCAmelCase="gelu"  ,__UpperCAmelCase=1E-6  ,__UpperCAmelCase=0.0  ,__UpperCAmelCase=1E-10  ,__UpperCAmelCase=True  ,**__UpperCAmelCase  ,)   ->		List[Any]:
      super().__init__(**__UpperCAmelCase				)

      lowerCAmelCase__			:  Any											= hidden_size
      lowerCAmelCase__			:  List[str]											= intermediate_size
      lowerCAmelCase__			:  str											= num_hidden_layers
      lowerCAmelCase__			:  Tuple											= num_attention_heads
      lowerCAmelCase__			:  Union[str, Any]											= patch_size
      lowerCAmelCase__			:  int											= image_size
      lowerCAmelCase__			:  Tuple											= initializer_range
      lowerCAmelCase__			:  Optional[int]											= attention_dropout
      lowerCAmelCase__			:  List[str]											= layer_norm_eps
      lowerCAmelCase__			:  List[Any]											= hidden_act
      lowerCAmelCase__			:  int											= qkv_bias






  @classmethod
  def 							UpperCAmelCase_     (   cls  ,__UpperCAmelCase  ,**__UpperCAmelCase				)   ->		"PretrainedConfig":
      cls._set_token_in_kwargs(__UpperCAmelCase				)

      lowerCAmelCase__				,       lowerCAmelCase__			:  Tuple											= cls.get_config_dict(__UpperCAmelCase  ,**__UpperCAmelCase				)

      # get the vision config dict if we are loading from InstructBlipConfig
      if config_dict.get("""model_type"""				) == "instructblip":
          lowerCAmelCase__			:  Optional[int]											= config_dict["""vision_config"""]

      if "model_type" in config_dict and hasattr(cls  ,"""model_type"""				) and config_dict["model_type"] != cls.model_type:
          logger.warning(
              F"""You are using a model of type {config_dict['model_type']} to instantiate a model of type """
              F"""{cls.model_type}. This is not supported for all configurations of models and can yield errors."""				)

      return cls.from_dict(__UpperCAmelCase  ,**__UpperCAmelCase				)







class 	lowerCAmelCase_(    SCREAMING_SNAKE_CASE_					):
  '''simple docstring'''




  __lowercase							:				int           =				'''instructblip_qformer'''




  def __init__(   self  ,__UpperCAmelCase=3_0522  ,__UpperCAmelCase=768  ,__UpperCAmelCase=12  ,__UpperCAmelCase=12  ,__UpperCAmelCase=3072  ,__UpperCAmelCase="gelu"  ,__UpperCAmelCase=0.1  ,__UpperCAmelCase=0.1  ,__UpperCAmelCase=512  ,__UpperCAmelCase=0.0_2  ,__UpperCAmelCase=1E-12  ,__UpperCAmelCase=0  ,__UpperCAmelCase="absolute"  ,__UpperCAmelCase=2  ,__UpperCAmelCase=1408  ,**__UpperCAmelCase  ,)   ->		Tuple:
      super().__init__(pad_token_id=__UpperCAmelCase  ,**__UpperCAmelCase				)

      lowerCAmelCase__			:  List[Any]											= vocab_size
      lowerCAmelCase__			:  Optional[Any]											= hidden_size
      lowerCAmelCase__			:  Any											= num_hidden_layers
      lowerCAmelCase__			:  int											= num_attention_heads
      lowerCAmelCase__			:  Tuple											= hidden_act
      lowerCAmelCase__			:  int											= intermediate_size
      lowerCAmelCase__			:  Tuple											= hidden_dropout_prob
      lowerCAmelCase__			:  Optional[Any]											= attention_probs_dropout_prob
      lowerCAmelCase__			:  Optional[Any]											= max_position_embeddings
      lowerCAmelCase__			:  Any											= initializer_range
      lowerCAmelCase__			:  Tuple											= layer_norm_eps
      lowerCAmelCase__			:  Dict											= position_embedding_type
      lowerCAmelCase__			:  int											= cross_attention_frequency
      lowerCAmelCase__			:  List[Any]											= encoder_hidden_size






  @classmethod
  def 							UpperCAmelCase_     (   cls  ,__UpperCAmelCase  ,**__UpperCAmelCase				)   ->		"PretrainedConfig":
      cls._set_token_in_kwargs(__UpperCAmelCase				)

      lowerCAmelCase__				,       lowerCAmelCase__			:  Any											= cls.get_config_dict(__UpperCAmelCase  ,**__UpperCAmelCase				)

      # get the qformer config dict if we are loading from InstructBlipConfig
      if config_dict.get("""model_type"""				) == "instructblip":
          lowerCAmelCase__			:  Tuple											= config_dict["""qformer_config"""]

      if "model_type" in config_dict and hasattr(cls  ,"""model_type"""				) and config_dict["model_type"] != cls.model_type:
          logger.warning(
              F"""You are using a model of type {config_dict['model_type']} to instantiate a model of type """
              F"""{cls.model_type}. This is not supported for all configurations of models and can yield errors."""				)

      return cls.from_dict(__UpperCAmelCase  ,**__UpperCAmelCase				)







class 	lowerCAmelCase_(    SCREAMING_SNAKE_CASE_					):
  '''simple docstring'''




  __lowercase							:				Any           =				'''instructblip'''
  __lowercase							:				str           =				True




  def __init__(   self  ,__UpperCAmelCase=None  ,__UpperCAmelCase=None  ,__UpperCAmelCase=None  ,__UpperCAmelCase=32  ,**__UpperCAmelCase				)   ->		Any:
      super().__init__(**__UpperCAmelCase				)

      if vision_config is None:
          lowerCAmelCase__			:  Any											= {}
          logger.info("""vision_config is None. initializing the InstructBlipVisionConfig with default values."""				)

      if qformer_config is None:
          lowerCAmelCase__			:  List[str]											= {}
          logger.info("""qformer_config is None. Initializing the InstructBlipQFormerConfig with default values."""				)

      if text_config is None:
          lowerCAmelCase__			:  List[Any]											= {}
          logger.info("""text_config is None. Initializing the text config with default values (`OPTConfig`)."""				)

      lowerCAmelCase__			:  Any											= InstructBlipVisionConfig(**__UpperCAmelCase				)
      lowerCAmelCase__			:  Tuple											= InstructBlipQFormerConfig(**__UpperCAmelCase				)
      lowerCAmelCase__			:  Tuple											= text_config["""model_type"""] if """model_type""" in text_config else """opt"""
      lowerCAmelCase__			:  Any											= CONFIG_MAPPING[text_model_type](**__UpperCAmelCase				)

      lowerCAmelCase__			:  Union[str, Any]											= self.text_config.tie_word_embeddings
      lowerCAmelCase__			:  Any											= self.text_config.is_encoder_decoder

      lowerCAmelCase__			:  int											= num_query_tokens
      lowerCAmelCase__			:  List[str]											= self.vision_config.hidden_size
      lowerCAmelCase__			:  Optional[int]											= self.text_config.model_type in MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
      lowerCAmelCase__			:  Optional[Any]											= 1.0
      lowerCAmelCase__			:  Dict											= 0.0_2




  @classmethod
  def 							UpperCAmelCase_     (   cls  ,__UpperCAmelCase  ,__UpperCAmelCase  ,__UpperCAmelCase  ,**__UpperCAmelCase  ,)   ->		Union[str, Any]:
      return cls(
          vision_config=vision_config.to_dict()  ,qformer_config=qformer_config.to_dict()  ,text_config=text_config.to_dict()  ,**__UpperCAmelCase  ,)






  def 							UpperCAmelCase_     (   self				)   ->		str:
      lowerCAmelCase__			:  List[Any]											= copy.deepcopy(self.__dict__				)
      lowerCAmelCase__			:  str											= self.vision_config.to_dict()
      lowerCAmelCase__			:  Union[str, Any]											= self.qformer_config.to_dict()
      lowerCAmelCase__			:  Union[str, Any]											= self.text_config.to_dict()
      lowerCAmelCase__			:  str											= self.__class__.model_type
      return output


