

'''simple docstring'''

import io
import math
from typing import Dict, Optional, Union

import numpy as np
from huggingface_hub import hf_hub_download

from ...image_processing_utils import BaseImageProcessor, BatchFeature
from ...image_transforms import convert_to_rgb, normalize, to_channel_dimension_format, to_pil_image
from ...image_utils import (
    ChannelDimension,
    ImageInput,
    get_image_size,
    infer_channel_dimension_format,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_torch_available, is_vision_available, logging
from ...utils.import_utils import requires_backends


if is_vision_available():
    import textwrap

    from PIL import Image, ImageDraw, ImageFont

if is_torch_available():
    import torch

    from transformers.pytorch_utils import is_torch_greater_or_equal_than_1_11
else:
    A_       						=							False


A_       						=							logging.get_logger(__name__)
A_       						=							"ybelkada/fonts"





def    A_  ( ):
  if is_torch_available() and not is_torch_greater_or_equal_than_1_11:
    raise ImportError(
        F'''You are using torch=={torch.__version__}, but torch>=1.11.0 is required to use '''
        "Pix2StructImageProcessor. Please upgrade torch." )





def    A_  ( snake_case      ,       snake_case      ,       snake_case ):
  requires_backends(snake_case      ,       ["torch"] )
  _check_torch_version()

  SCREAMING_SNAKE_CASE:List[str]						      =					image_tensor.unsqueeze(0 )
  SCREAMING_SNAKE_CASE:Any						      =					torch.nn.functional.unfold(snake_case      ,       (patch_height, patch_width)      ,       stride=(patch_height, patch_width) )
  SCREAMING_SNAKE_CASE:Any						      =					patches.reshape(image_tensor.size(0 )      ,       image_tensor.size(1 )      ,       snake_case      ,       snake_case      ,       -1 )
  SCREAMING_SNAKE_CASE:Optional[Any]						      =					patches.permute(0      ,       4      ,       2      ,       3      ,       1 ).reshape(
      image_tensor.size(2 ) // patch_height      ,       image_tensor.size(3 ) // patch_width      ,       image_tensor.size(1 ) * patch_height * patch_width      ,       )
  return patches.unsqueeze(0 )





def    A_  ( snake_case      ,       snake_case = 36      ,       snake_case = "black"      ,       snake_case = "white"      ,       snake_case = 5      ,       snake_case = 5      ,       snake_case = 5      ,       snake_case = 5      ,       snake_case = None      ,       snake_case = None      ,       ):
  requires_backends(snake_case      ,       "vision" )
  # Add new lines so that each line is no more than 80 characters.

  SCREAMING_SNAKE_CASE:List[Any]						      =					textwrap.TextWrapper(width=80 )
  SCREAMING_SNAKE_CASE:str						      =					wrapper.wrap(text=snake_case )
  SCREAMING_SNAKE_CASE:Tuple						      =					"\n".join(snake_case )

  if font_bytes is not None and font_path is None:
    SCREAMING_SNAKE_CASE:Dict						      =					io.BytesIO(snake_case )
  elif font_path is not None:
    SCREAMING_SNAKE_CASE:List[str]						      =					font_path
  else:
    SCREAMING_SNAKE_CASE:str						      =					hf_hub_download(snake_case      ,       "Arial.TTF" )
  SCREAMING_SNAKE_CASE:Union[str, Any]						      =					ImageFont.truetype(snake_case      ,       encoding="UTF-8"      ,       size=snake_case )

  # Use a temporary canvas to determine the width and height in pixels when
  # rendering the text.
  SCREAMING_SNAKE_CASE:str						      =					ImageDraw.Draw(Image.new("RGB"      ,       (1, 1)      ,       snake_case ) )
  SCREAMING_SNAKE_CASE      ,    SCREAMING_SNAKE_CASE      ,    SCREAMING_SNAKE_CASE      ,    SCREAMING_SNAKE_CASE:Any						      =					temp_draw.textbbox((0, 0)      ,       snake_case      ,       snake_case )

  # Create the actual image with a bit of padding around the text.
  SCREAMING_SNAKE_CASE:str						      =					text_width + left_padding + right_padding
  SCREAMING_SNAKE_CASE:List[str]						      =					text_height + top_padding + bottom_padding
  SCREAMING_SNAKE_CASE:Union[str, Any]						      =					Image.new("RGB"      ,       (image_width, image_height)      ,       snake_case )
  SCREAMING_SNAKE_CASE:str						      =					ImageDraw.Draw(snake_case )
  draw.text(xy=(left_padding, top_padding)      ,       text=snake_case      ,       fill=snake_case      ,       font=snake_case )
  return image





def    A_  ( snake_case      ,       snake_case      ,       **snake_case ):
  requires_backends(snake_case      ,       "vision" )

  # Convert to PIL image if necessary
  SCREAMING_SNAKE_CASE:Optional[int]						      =					to_pil_image(snake_case )

  SCREAMING_SNAKE_CASE:List[Any]						      =					render_text(snake_case      ,       **snake_case )
  SCREAMING_SNAKE_CASE:Optional[int]						      =					max(header_image.width      ,       image.width )

  SCREAMING_SNAKE_CASE:Any						      =					int(image.height * (new_width / image.width) )
  SCREAMING_SNAKE_CASE:int						      =					int(header_image.height * (new_width / header_image.width) )

  SCREAMING_SNAKE_CASE:Any						      =					Image.new("RGB"      ,       (new_width, new_height + new_header_height)      ,       "white" )
  new_image.paste(header_image.resize((new_width, new_header_height) )      ,       (0, 0) )
  new_image.paste(image.resize((new_width, new_height) )      ,       (0, new_header_height) )

  # Convert back to the original framework if necessary
  SCREAMING_SNAKE_CASE:Union[str, Any]						      =					to_numpy_array(snake_case )

  if infer_channel_dimension_format(snake_case ) == ChannelDimension.LAST:
    SCREAMING_SNAKE_CASE:List[Any]						      =					to_channel_dimension_format(snake_case      ,       ChannelDimension.LAST )

  return new_image





class 							_snake_case (       _a						):
   _A       :     Any       =					['''flattened_patches''']
   def __init__( self			:      List[Any]     ,SCREAMING_SNAKE_CASE__			:      bool = True     ,SCREAMING_SNAKE_CASE__			:      bool = True     ,SCREAMING_SNAKE_CASE__			:      Dict[str, int] = None     ,SCREAMING_SNAKE_CASE__			:      int = 2_048     ,SCREAMING_SNAKE_CASE__			:      bool = False     ,**SCREAMING_SNAKE_CASE__			:      Dict     ,):
     super().__init__(**SCREAMING_SNAKE_CASE__							)
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					patch_size if patch_size is not None else {"height": 16, "width": 16}
     SCREAMING_SNAKE_CASE:Dict						      =					do_normalize
     SCREAMING_SNAKE_CASE:List[Any]						      =					do_convert_rgb
     SCREAMING_SNAKE_CASE:str						      =					max_patches
     SCREAMING_SNAKE_CASE:Tuple						      =					is_vqa
   def     __UpperCamelCase						( self			:      Optional[Any]     ,SCREAMING_SNAKE_CASE__			:      np.ndarray     ,SCREAMING_SNAKE_CASE__			:      int     ,SCREAMING_SNAKE_CASE__			:      dict     ,**SCREAMING_SNAKE_CASE__			:      Any							):
     requires_backends(self.extract_flattened_patches     ,"torch"							)
     _check_torch_version()

     # convert to torch
     SCREAMING_SNAKE_CASE:Tuple						      =					to_channel_dimension_format(SCREAMING_SNAKE_CASE__     ,ChannelDimension.FIRST							)
     SCREAMING_SNAKE_CASE:Any						      =					torch.from_numpy(SCREAMING_SNAKE_CASE__							)

     SCREAMING_SNAKE_CASE      ,    SCREAMING_SNAKE_CASE:Union[str, Any]						      =					patch_size["height"], patch_size["width"]
     SCREAMING_SNAKE_CASE      ,    SCREAMING_SNAKE_CASE:Optional[int]						      =					get_image_size(SCREAMING_SNAKE_CASE__							)

     # maximize scale s.t.
     SCREAMING_SNAKE_CASE:Optional[Any]						      =					math.sqrt(max_patches * (patch_height / image_height) * (patch_width / image_width)							)
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					max(min(math.floor(scale * image_height / patch_height							)     ,SCREAMING_SNAKE_CASE__							)     ,1							)
     SCREAMING_SNAKE_CASE:Dict						      =					max(min(math.floor(scale * image_width / patch_width							)     ,SCREAMING_SNAKE_CASE__							)     ,1							)
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					max(num_feasible_rows * patch_height     ,1							)
     SCREAMING_SNAKE_CASE:Any						      =					max(num_feasible_cols * patch_width     ,1							)

     SCREAMING_SNAKE_CASE:Optional[Any]						      =					torch.nn.functional.interpolate(
         image.unsqueeze(0							)     ,size=(resized_height, resized_width)     ,mode="bilinear"     ,align_corners=SCREAMING_SNAKE_CASE__     ,antialias=SCREAMING_SNAKE_CASE__     ,).squeeze(0							)

     # [1, rows, columns, patch_height * patch_width * image_channels]
     SCREAMING_SNAKE_CASE:Dict						      =					torch_extract_patches(SCREAMING_SNAKE_CASE__     ,SCREAMING_SNAKE_CASE__     ,SCREAMING_SNAKE_CASE__							)

     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					patches.shape
     SCREAMING_SNAKE_CASE:Any						      =					patches_shape[1]
     SCREAMING_SNAKE_CASE:Optional[Any]						      =					patches_shape[2]
     SCREAMING_SNAKE_CASE:List[Any]						      =					patches_shape[3]

     # [rows * columns, patch_height * patch_width * image_channels]
     SCREAMING_SNAKE_CASE:Optional[Any]						      =					patches.reshape([rows * columns, depth]							)

     # [rows * columns, 1]
     SCREAMING_SNAKE_CASE:int						      =					torch.arange(SCREAMING_SNAKE_CASE__							).reshape([rows, 1]							).repeat(1     ,SCREAMING_SNAKE_CASE__							).reshape([rows * columns, 1]							)
     SCREAMING_SNAKE_CASE:List[Any]						      =					torch.arange(SCREAMING_SNAKE_CASE__							).reshape([1, columns]							).repeat(SCREAMING_SNAKE_CASE__     ,1							).reshape([rows * columns, 1]							)

     # Offset by 1 so the ids do not contain zeros, which represent padding.
     row_ids += 1
     col_ids += 1

     # Prepare additional patch features.
     # [rows * columns, 1]
     SCREAMING_SNAKE_CASE:List[Any]						      =					row_ids.to(torch.floataa							)
     SCREAMING_SNAKE_CASE:List[str]						      =					col_ids.to(torch.floataa							)

     # [rows * columns, 2 + patch_height * patch_width * image_channels]
     SCREAMING_SNAKE_CASE:int						      =					torch.cat([row_ids, col_ids, patches]     ,-1							)

     # [max_patches, 2 + patch_height * patch_width * image_channels]
     SCREAMING_SNAKE_CASE:str						      =					torch.nn.functional.pad(SCREAMING_SNAKE_CASE__     ,[0, 0, 0, max_patches - (rows * columns)]							).float()

     SCREAMING_SNAKE_CASE:List[str]						      =					to_numpy_array(SCREAMING_SNAKE_CASE__							)

     return result
   def     __UpperCamelCase						( self			:      Optional[Any]     ,SCREAMING_SNAKE_CASE__			:      np.ndarray     ,SCREAMING_SNAKE_CASE__			:      Optional[Union[str, ChannelDimension]] = None     ,**SCREAMING_SNAKE_CASE__			:      List[Any]							):
     if image.dtype == np.uinta:
       SCREAMING_SNAKE_CASE:Optional[int]						      =					image.astype(np.floataa							)

     # take mean across the whole `image`
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					np.mean(SCREAMING_SNAKE_CASE__							)
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					np.std(SCREAMING_SNAKE_CASE__							)
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					max(SCREAMING_SNAKE_CASE__     ,1.0 / math.sqrt(np.prod(image.shape							)							)							)

     return normalize(SCREAMING_SNAKE_CASE__     ,mean=SCREAMING_SNAKE_CASE__     ,std=SCREAMING_SNAKE_CASE__     ,**SCREAMING_SNAKE_CASE__							)







   def     __UpperCamelCase						( self			:      List[str]     ,SCREAMING_SNAKE_CASE__			:      ImageInput     ,SCREAMING_SNAKE_CASE__			:      Optional[str] = None     ,SCREAMING_SNAKE_CASE__			:      bool = None     ,SCREAMING_SNAKE_CASE__			:      Optional[bool] = None     ,SCREAMING_SNAKE_CASE__			:      Optional[int] = None     ,SCREAMING_SNAKE_CASE__			:      Optional[Dict[str, int]] = None     ,SCREAMING_SNAKE_CASE__			:      Optional[Union[str, TensorType]] = None     ,SCREAMING_SNAKE_CASE__			:      ChannelDimension = ChannelDimension.FIRST     ,**SCREAMING_SNAKE_CASE__			:      Optional[int]     ,):
     SCREAMING_SNAKE_CASE:Any						      =					do_normalize if do_normalize is not None else self.do_normalize
     SCREAMING_SNAKE_CASE:Tuple						      =					do_convert_rgb if do_convert_rgb is not None else self.do_convert_rgb
     SCREAMING_SNAKE_CASE:List[Any]						      =					patch_size if patch_size is not None else self.patch_size
     SCREAMING_SNAKE_CASE:List[Any]						      =					max_patches if max_patches is not None else self.max_patches
     SCREAMING_SNAKE_CASE:Optional[int]						      =					self.is_vqa

     if kwargs.get("data_format"     ,SCREAMING_SNAKE_CASE__							) is not None:
       raise ValueError("data_format is not an accepted input as the outputs are "							)

     SCREAMING_SNAKE_CASE:List[Any]						      =					make_list_of_images(SCREAMING_SNAKE_CASE__							)

     if not valid_images(SCREAMING_SNAKE_CASE__							):
       raise ValueError(
           "Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, "
           "torch.Tensor, tf.Tensor or jax.ndarray."							)

     # PIL RGBA images are converted to RGB
     if do_convert_rgb:
       SCREAMING_SNAKE_CASE:Optional[int]						      =					[convert_to_rgb(SCREAMING_SNAKE_CASE__							) for image in images]

     # All transformations expect numpy arrays.
     SCREAMING_SNAKE_CASE:List[str]						      =					[to_numpy_array(SCREAMING_SNAKE_CASE__							) for image in images]

     if is_vqa:
       if header_text is None:
         raise ValueError("A header text must be provided for VQA models."							)
       SCREAMING_SNAKE_CASE:Optional[int]						      =					kwargs.pop("font_bytes"     ,SCREAMING_SNAKE_CASE__							)
       SCREAMING_SNAKE_CASE:Tuple						      =					kwargs.pop("font_path"     ,SCREAMING_SNAKE_CASE__							)

       if isinstance(SCREAMING_SNAKE_CASE__     ,SCREAMING_SNAKE_CASE__							):
         SCREAMING_SNAKE_CASE:Any						      =					[header_text] * len(SCREAMING_SNAKE_CASE__							)

       SCREAMING_SNAKE_CASE:Tuple						      =					[
           render_header(SCREAMING_SNAKE_CASE__     ,header_text[i]     ,font_bytes=SCREAMING_SNAKE_CASE__     ,font_path=SCREAMING_SNAKE_CASE__							)
           for i, image in enumerate(SCREAMING_SNAKE_CASE__							)
       ]

     if do_normalize:
       SCREAMING_SNAKE_CASE:Union[str, Any]						      =					[self.normalize(image=SCREAMING_SNAKE_CASE__							) for image in images]

     # convert to torch tensor and permute
     SCREAMING_SNAKE_CASE:Union[str, Any]						      =					[
         self.extract_flattened_patches(image=SCREAMING_SNAKE_CASE__     ,max_patches=SCREAMING_SNAKE_CASE__     ,patch_size=SCREAMING_SNAKE_CASE__							)
         for image in images
     ]

     # create attention mask in numpy
     SCREAMING_SNAKE_CASE:List[Any]						      =					[(image.sum(axis=-1							) != 0).astype(np.floataa							) for image in images]

     SCREAMING_SNAKE_CASE:int						      =					BatchFeature(
         data={"flattened_patches": images, "attention_mask": attention_masks}     ,tensor_type=SCREAMING_SNAKE_CASE__							)

     return encoded_outputs









'''simple docstring'''

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_  ( snake_case ):
  SCREAMING_SNAKE_CASE:List[Any]						      =					[]
  for line in lines:
    SCREAMING_SNAKE_CASE:Any						      =					re.sub(r"#.*"      ,       ""      ,       snake_case )  # remove comments
    if line:
      filtered_lines.append(snake_case )
  SCREAMING_SNAKE_CASE:int						      =					"\n".join(snake_case )

  # Make a hash from all this code
  SCREAMING_SNAKE_CASE:str						      =					full_str.encode("utf-8" )
  return shaaaa(snake_case ).hexdigest()


# get importable module names and hash for caching
A_       						=							{
    "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
A_       						=							{
    ".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})
A_       						=							{"imagefolder", "audiofolder"}

# Used to filter data files based on extensions given a module name
A_       						=							{}
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")






