


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


UpperCAmelCase_				:			Optional[List[str]]   =	None
UpperCAmelCase_				:			Optional[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
UpperCAmelCase_				:			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 			SCREAMING_SNAKE_CASE__  :
      snake_case__    :					bool					       =						True
      snake_case__    :					Optional[str]					       =						None
      # Automatically constructed
      snake_case__    :					ClassVar[str]					       =						"PIL.Image.Image"
      snake_case__    :					ClassVar[Any]					       =						pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}      )
      snake_case__    :					str					       =						field(default='''Image'''						,   init=lowercase__						,   repr=lowercase__      )
      def __call__(	self    :	str					)		->      List[str]:
             return self.pa_type
      def 	SCREAMING_SNAKE_CASE							(	self    :	Optional[Any]					,							SCREAMING_SNAKE_CASE__    :	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(SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					):
                    a_ :		Optional[int]							=  np.array(SCREAMING_SNAKE_CASE__					)

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

             a_							,		a_ :		List[str]							=  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(SCREAMING_SNAKE_CASE__					):
                                  a_ :		List[Any]							=  PIL.Image.open(SCREAMING_SNAKE_CASE__					)
                           else:
                                  a_ :		int							=  path.split('::'					)[-1]
                                  try:
                                         a_ :		List[Any]							=  string_to_dict(SCREAMING_SNAKE_CASE__					,							config.HUB_DATASETS_URL					)['repo_id']
                                         a_ :		Any							=  token_per_repo_id.get(SCREAMING_SNAKE_CASE__					)
                                  except ValueError:
                                         a_ :		Tuple							=  None
                                  with xopen(SCREAMING_SNAKE_CASE__					,							'rb'					,							use_auth_token=SCREAMING_SNAKE_CASE__					) as f:
                                         a_ :		Any							=  BytesIO(f.read()					)
                                  a_ :		str							=  PIL.Image.open(bytes_					)
             else:
                    a_ :		int							=  PIL.Image.open(BytesIO(bytes_					)					)
             image.load()  # to avoid "Too many open files" errors
             return image
      def 	SCREAMING_SNAKE_CASE							(	self    :	List[Any]					)		->      Union["FeatureType", Dict[str, "FeatureType"]]:
             from .features import Value

             return (
                 self
                 if self.decode
                 else {
                     "bytes": Value('binary'					),
                     "path": Value('string'					),
                 }
             )
      def 	SCREAMING_SNAKE_CASE							(	self    :	Union[str, Any]					,							SCREAMING_SNAKE_CASE__    :	Union[pa.StringArray, pa.StructArray, pa.ListArray]					)		->      pa.StructArray:
             if pa.types.is_string(storage.type					):
                    a_ :		Optional[int]							=  pa.array([None] * len(SCREAMING_SNAKE_CASE__					)					,							type=pa.binary()					)
                    a_ :		Union[str, Any]							=  pa.StructArray.from_arrays([bytes_array, storage]					,							['bytes', 'path']					,							mask=storage.is_null()					)
             elif pa.types.is_binary(storage.type					):
                    a_ :		Any							=  pa.array([None] * len(SCREAMING_SNAKE_CASE__					)					,							type=pa.string()					)
                    a_ :		str							=  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_ :		Optional[int]							=  storage.field('bytes'					)
                    else:
                           a_ :		Optional[Any]							=  pa.array([None] * len(SCREAMING_SNAKE_CASE__					)					,							type=pa.binary()					)
                    if storage.type.get_field_index('path'					) >= 0:
                           a_ :		Union[str, Any]							=  storage.field('path'					)
                    else:
                           a_ :		Optional[Any]							=  pa.array([None] * len(SCREAMING_SNAKE_CASE__					)					,							type=pa.string()					)
                    a_ :		int							=  pa.StructArray.from_arrays([bytes_array, path_array]					,							['bytes', 'path']					,							mask=storage.is_null()					)
             elif pa.types.is_list(storage.type					):
                    a_ :		str							=  pa.array(
                        [encode_np_array(np.array(SCREAMING_SNAKE_CASE__					)					)['bytes'] if arr is not None else None for arr in storage.to_pylist()]					,							type=pa.binary()					,							)
                    a_ :		Optional[Any]							=  pa.array([None] * len(SCREAMING_SNAKE_CASE__					)					,							type=pa.string()					)
                    a_ :		Tuple							=  pa.StructArray.from_arrays(
                        [bytes_array, path_array]					,							['bytes', 'path']					,							mask=bytes_array.is_null()					)
             return array_cast(SCREAMING_SNAKE_CASE__					,							self.pa_type					)

      def 	SCREAMING_SNAKE_CASE							(	self    :	Tuple					,							SCREAMING_SNAKE_CASE__    :	pa.StructArray					)		->      pa.StructArray:

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

             a_ :		int							=  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_ :		List[str]							=  pa.array(
                 [os.path.basename(SCREAMING_SNAKE_CASE__					) if path is not None else None for path in storage.field('path'					).to_pylist()]					,							type=pa.string()					,							)
             a_ :		Any							=  pa.StructArray.from_arrays([bytes_array, path_array]					,							['bytes', 'path']					,							mask=bytes_array.is_null()					)
             return array_cast(SCREAMING_SNAKE_CASE__					,							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[Any]							=  list(set(PIL.Image.OPEN.keys()    ) & set(PIL.Image.SAVE.keys()    )    )
       return _IMAGE_COMPRESSION_FORMATS







def 	SCREAMING_SNAKE_CASE_						(      __A			:      "PIL.Image.Image"    )     ->		bytes:

       """simple docstring"""

       a_ :		int							=  BytesIO()
       if image.format in list_image_compression_formats():
              a_ :		Union[str, Any]							=  image.format
       else:
              a_ :		Dict							=  'PNG' if image.mode in ['1', 'L', 'LA', 'RGB', 'RGBA'] else 'TIFF'
       image.save(__A  ,   format=__A    )
       return buffer.getvalue()







def 	SCREAMING_SNAKE_CASE_						(      __A			:      "PIL.Image.Image"    )     ->		dict:

       """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 	SCREAMING_SNAKE_CASE_						(      __A			:      np.ndarray    )     ->		dict:

       """simple docstring"""

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

       a_ :		int							=  array.dtype
       a_ :		Any							=  dtype.byteorder if dtype.byteorder != '=' else _NATIVE_BYTEORDER
       a_ :		Dict							=  dtype.kind
       a_ :		Optional[int]							=  dtype.itemsize

       a_ :		Tuple							=  None

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







def 	SCREAMING_SNAKE_CASE_						(      __A			:      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:
              a_							,		a_ :		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    ):
                     a_ :		Optional[Any]							=  no_op_if_value_is_null(__A    )
                     return [obj_to_image_dict_func(__A    ) for obj in objs]
              elif isinstance(__A  ,   PIL.Image.Image    ):
                     a_ :		int							=  no_op_if_value_is_null(__A    )
                     return [obj_to_image_dict_func(__A    ) for obj in objs]
              else:
                     return objs
       else:
              return objs






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


UpperCAmelCase_				:			str   =	logging.get_logger(__name__)

UpperCAmelCase_				:			str   =	{
    'Salesforce/instruct-blip-flan-t5': 'https://huggingface.co/Salesforce/instruct-blip-flan-t5/resolve/main/config.json',
}
class 			SCREAMING_SNAKE_CASE__  (    lowercase__      ):
      snake_case__    :					int					       =						'''instructblip_vision_model'''
      def __init__(	self    :	Optional[int]					,							SCREAMING_SNAKE_CASE__    :	List[Any]=1_4_0_8					,							SCREAMING_SNAKE_CASE__    :	int=6_1_4_4					,							SCREAMING_SNAKE_CASE__    :	Dict=3_9					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=1_6					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=2_2_4					,							SCREAMING_SNAKE_CASE__    :	Optional[int]=1_4					,							SCREAMING_SNAKE_CASE__    :	List[Any]="gelu"					,							SCREAMING_SNAKE_CASE__    :	List[str]=1E-6					,							SCREAMING_SNAKE_CASE__    :	Tuple=0.0					,							SCREAMING_SNAKE_CASE__    :	List[str]=1E-10					,							SCREAMING_SNAKE_CASE__    :	str=True					,							**SCREAMING_SNAKE_CASE__    :	Dict					,							)		->      int:
             super().__init__(**SCREAMING_SNAKE_CASE__					)

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

      @classmethod
      def 	SCREAMING_SNAKE_CASE							(	cls    :	Any					,							SCREAMING_SNAKE_CASE__    :	Union[str, os.PathLike]					,							**SCREAMING_SNAKE_CASE__    :	List[str]					)		->      "PretrainedConfig":
             cls._set_token_in_kwargs(SCREAMING_SNAKE_CASE__					)

             a_							,		a_ :		Dict							=  cls.get_config_dict(SCREAMING_SNAKE_CASE__					,							**SCREAMING_SNAKE_CASE__					)

             # get the vision config dict if we are loading from InstructBlipConfig
             if config_dict.get('model_type'					) == "instructblip":
                    a_ :		List[Any]							=  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(SCREAMING_SNAKE_CASE__					,							**SCREAMING_SNAKE_CASE__					)
class 			SCREAMING_SNAKE_CASE__  (    lowercase__      ):
      snake_case__    :					List[Any]					       =						'''instructblip_qformer'''
      def __init__(	self    :	Optional[int]					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=3_0_5_2_2					,							SCREAMING_SNAKE_CASE__    :	str=7_6_8					,							SCREAMING_SNAKE_CASE__    :	List[str]=1_2					,							SCREAMING_SNAKE_CASE__    :	Any=1_2					,							SCREAMING_SNAKE_CASE__    :	Optional[int]=3_0_7_2					,							SCREAMING_SNAKE_CASE__    :	List[str]="gelu"					,							SCREAMING_SNAKE_CASE__    :	Dict=0.1					,							SCREAMING_SNAKE_CASE__    :	int=0.1					,							SCREAMING_SNAKE_CASE__    :	Optional[int]=5_1_2					,							SCREAMING_SNAKE_CASE__    :	Optional[Any]=0.02					,							SCREAMING_SNAKE_CASE__    :	int=1E-12					,							SCREAMING_SNAKE_CASE__    :	Dict=0					,							SCREAMING_SNAKE_CASE__    :	Optional[Any]="absolute"					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=2					,							SCREAMING_SNAKE_CASE__    :	str=1_4_0_8					,							**SCREAMING_SNAKE_CASE__    :	Dict					,							)		->      Optional[Any]:
             super().__init__(pad_token_id=SCREAMING_SNAKE_CASE__					,							**SCREAMING_SNAKE_CASE__					)

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

      @classmethod
      def 	SCREAMING_SNAKE_CASE							(	cls    :	str					,							SCREAMING_SNAKE_CASE__    :	Union[str, os.PathLike]					,							**SCREAMING_SNAKE_CASE__    :	Union[str, Any]					)		->      "PretrainedConfig":
             cls._set_token_in_kwargs(SCREAMING_SNAKE_CASE__					)

             a_							,		a_ :		Union[str, Any]							=  cls.get_config_dict(SCREAMING_SNAKE_CASE__					,							**SCREAMING_SNAKE_CASE__					)

             # get the qformer config dict if we are loading from InstructBlipConfig
             if config_dict.get('model_type'					) == "instructblip":
                    a_ :		Dict							=  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(SCREAMING_SNAKE_CASE__					,							**SCREAMING_SNAKE_CASE__					)
class 			SCREAMING_SNAKE_CASE__  (    lowercase__      ):
      snake_case__    :					Union[str, Any]					       =						'''instructblip'''
      snake_case__    :					List[str]					       =						True
      def __init__(	self    :	Tuple					,							SCREAMING_SNAKE_CASE__    :	Optional[Any]=None					,							SCREAMING_SNAKE_CASE__    :	Dict=None					,							SCREAMING_SNAKE_CASE__    :	List[Any]=None					,							SCREAMING_SNAKE_CASE__    :	int=3_2					,							**SCREAMING_SNAKE_CASE__    :	Union[str, Any]					)		->      Tuple:
             super().__init__(**SCREAMING_SNAKE_CASE__					)

             if vision_config is None:
                    a_ :		Dict							=  {}
                    logger.info('vision_config is None. initializing the InstructBlipVisionConfig with default values.'					)

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

             if text_config is None:
                    a_ :		Any							=  {}
                    logger.info('text_config is None. Initializing the text config with default values (`OPTConfig`).'					)

             a_ :		int							=  InstructBlipVisionConfig(**SCREAMING_SNAKE_CASE__					)
             a_ :		int							=  InstructBlipQFormerConfig(**SCREAMING_SNAKE_CASE__					)
             a_ :		Optional[Any]							=  text_config['model_type'] if 'model_type' in text_config else 'opt'
             a_ :		List[Any]							=  CONFIG_MAPPING[text_model_type](**SCREAMING_SNAKE_CASE__					)

             a_ :		Tuple							=  self.text_config.tie_word_embeddings
             a_ :		List[Any]							=  self.text_config.is_encoder_decoder

             a_ :		Optional[Any]							=  num_query_tokens
             a_ :		int							=  self.vision_config.hidden_size
             a_ :		List[str]							=  self.text_config.model_type in MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
             a_ :		Optional[int]							=  1.0
             a_ :		Any							=  0.02
      @classmethod
      def 	SCREAMING_SNAKE_CASE							(	cls    :	List[Any]					,							SCREAMING_SNAKE_CASE__    :	InstructBlipVisionConfig					,							SCREAMING_SNAKE_CASE__    :	InstructBlipQFormerConfig					,							SCREAMING_SNAKE_CASE__    :	PretrainedConfig					,							**SCREAMING_SNAKE_CASE__    :	List[str]					,							)		->      Any:
             return cls(
                 vision_config=vision_config.to_dict()					,							qformer_config=qformer_config.to_dict()					,							text_config=text_config.to_dict()					,							**SCREAMING_SNAKE_CASE__					,							)

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


