codingutf8 2023 the huggingface inc team licensed under the apache license version 2 0 the license you may not use this file except in compliance with the license you may obtain a copy of the license at http www apache orglicenseslicense2 0 unless required by applicable law or agreed to in writing software distributed under the license is distributed on an as is basis without warranties or conditions of any kind either express or implied see the license for the specific language governing permissions and limitations under the license imagetext processor class for git takes an unpacked stream of tokens i e a list of tensors one for each item in the batch and does the required padding to create a single tensor for the batch of shape batchsize x newseqlen create padded tensors for the full batch place each batch entry into the batch tensor takes an inputstream tensor of shape b x s x for each subsequence adds any required padding to account for images and then unpacks the subsequences to create a single sequence per item in the batch returns a list of tensors one for each item in the batch allbistream for batchindex in rangebatchsize allsistream first construct full token stream including image placeholder tokens and loss mask for each subsequence and append to lists we use lists rather than tensors because each subsequence is variablesized todo remove this logic in a subsequent release since subsequences are not supported imageadjustment imagetokensbatchindex0 subsequencestream torch catimageadjustment inputstreambatchindex 0 dim0 numrealtokens imageadjustment shape0 numrealtexttokensbatchindex0 allsistream appendsubsequencestream numrealtokens allbistream appendtorch catallsistream dim0 return allbistream def replacestringreprwithtokentagsprompt str str prompt prompt replacetextreprpointopen tokenpointopenstring prompt prompt replacetextreprpointclose tokenpointclosestring prompt prompt replacetextreprbboxopen tokenbboxopenstring prompt prompt replacetextreprbboxclose tokenbboxclosestring return prompt def segmentpromptintotexttokenconversionsprompt str list wherever we notice the tokenopenstring tokenclosestring we split the prompt prompttextlist list regexpattern re compile ftokenbboxopenstringtokenbboxclosestringtokenpointopenstringtokenpointclosestring split by the regex pattern promptsplit regexpattern splitprompt for i elem in enumeratepromptsplit if lenelem 0 or elem in tokenbboxopenstring tokenbboxclosestring tokenpointopenstring tokenpointclosestring continue prompttextlist append elem i 1 and promptspliti 1 in tokenbboxopenstring tokenpointopenstring return prompttextlist def transformcoordinatesandtokenizeprompt str scalefactor float tokenizer listint make a namedtuple that stores text and isbbox we want to do the following tokenize the code normally when we see a point or box tokenize using the tokenizewithintag function when point or box close tag continue tokenizing normally first we replace the point and box tags with their respective tokens prompt replacestringreprwithtokentagsprompt tokenize the prompt convert prompt into a list split prompttextlist segmentpromptintotexttokenconversionsprompt transformedprompttokens listint for elem in prompttextlist if elem1 this is a location we need to tokenize it withintagtokenized transformwithintagselem0 scalefactor tokenizer surround the text with the open and close tags transformedprompttokens extendwithintagtokenized else transformedprompttokens extendtokenizerelem0 addspecialtokensfalse inputids return transformedprompttokens def transformwithintagstext str scalefactor float tokenizer listint convert the text into a list of strings numintstrs text split if lennumintstrs 2 if there are any open or close tags remove them tokenspaceopenstring tokenizer vocabtokenpointopenstring tokenspaceclosestring tokenizer vocabtokenpointclosestring else tokenspaceopenstring tokenizer vocabtokenbboxopenstring tokenspaceclosestring tokenizer vocabtokenbboxclosestring remove all spaces from numints numints floatnum strip for num in numintstrs scale to transformed image siz if lennumints 2 numintstranslated scalepointtotransformedimagexnumints0 ynumints1 scalefactorscalefactor elif lennumints 4 numintstranslated scalebboxtotransformedimage topnumints0 leftnumints1 bottomnumints2 rightnumints3 scalefactorscalefactor else raise valueerrorfinvalid number of ints lennumints tokenize the text skipping the tokens tokenizer vocabstrnum for num in numintstranslated return tokenspaceopenstring tokens tokenspaceclosestring def tokenizepromptswithimageandbatch tokenizer prompts listliststr scalefactors optionallistlisttorch tensor maxtokenstogenerate int maxpositionembeddings int addbos bool same issue with types as above addbeginningofanswertoken bool tupletorch tensor torch tensor if not tool use tranform the coordinates while tokenizing if scalefactors is not none transformedprompttokens for promptseq scalefactorseq in zipprompts scalefactors transformedprompttokens append transformcoordinatesandtokenizeprompt scalefactor item tokenizer for prompt scalefactor in zippromptseq scalefactorseq else transformedprompttokens tokenizer tokenizeprompt for prompt in promptseq for promptseq in prompts promptstokens transformedprompttokens if addbos bostoken tokenizer vocabs else bostoken tokenizer vocabendoftext promptstokens bostoken x for x in promptseq for promptseq in promptstokens if addbeginningofanswertoken boa tokenizer vocabbeginningofanswerstring only add bbox open token to the last subsequence since that is what will be completed for tokenseq in promptstokens tokenseq1 appendboa now we have a list of list of tokens which each list has a different size we want to extend this list to incorporate the tokens that need to be generated make all the sequences equal length get the prompts length promptslength lenx for x in promptstokensseq for promptstokensseq in promptstokens get the max prompts length maxpromptlen int np maxpromptslength number of tokens in the each sample of the batch sampleslength minmaxpromptlen maxtokenstogenerate maxpositionembeddings if maxpromptlen maxtokenstogenerate maxpositionembeddings logger warning fmax subsequence prompt length of maxpromptlen max tokens to generate maxtokenstogenerate fexceeds context length of maxpositionembeddings will generate as many tokens as possible now update the list of list to be of the same size sampleslength for prompttokensseq promptslengthseq in zippromptstokens promptslength for prompttokens promptlength in zipprompttokensseq promptslengthseq if lenprompttokens sampleslength raise valueerrorlength of subsequence prompt exceeds sequence length paddingsize sampleslength promptlength prompttokens extendtokenizer vocabendoftext paddingsize now we are in a structured format we can convert to tensors promptstokenstensor torch tensorpromptstokens dtypetorch int64 promptslengthtensor torch tensorpromptslength dtypetorch int64 return promptstokenstensor promptslengthtensor simplified assuming self croptop self paddingtop 0 def originaltotransformedhcoordsoriginalcoords scaleh return np roundoriginalcoords scaleh astypenp int32 simplified assuming self cropleft self paddingleft 0 def originaltotransformedwcoordsoriginalcoords scalew return np roundoriginalcoords scalew astypenp int32 def scalepointtotransformedimagex float y float scalefactor float listint xscaled originaltotransformedwcoordsnp arrayx 2 scalefactor0 yscaled originaltotransformedhcoordsnp arrayy 2 scalefactor0 return xscaled yscaled def scalebboxtotransformedimage top float left float bottom float right float scalefactor float listint topscaled originaltotransformedwcoordsnp arraytop 2 scalefactor0 leftscaled originaltotransformedhcoordsnp arrayleft 2 scalefactor0 bottomscaled originaltotransformedwcoordsnp arraybottom 2 scalefactor0 rightscaled originaltotransformedhcoordsnp arrayright 2 scalefactor0 return topscaled leftscaled bottomscaled rightscaled class fuyuprocessorprocessormixin r constructs a fuyu processor which wraps a fuyu image processor and a llama tokenizer into a single processor fuyuprocessor offers all the functionalities of fuyuimageprocessor and llamatokenizerfast see the fuyuprocessor call and fuyuprocessor decode for more information args imageprocessor fuyuimageprocessor the image processor is a required input tokenizer llamatokenizerfast the tokenizer is a required input for imagepatches we don t pad but just append them to the list fixme maxtokenstogenerate is embedded into this processor s call construct inputs for image patch indices use same packing logic for the image patch indices main method to prepare for the model one or several sequencess and images this method forwards the text and kwargs arguments to llamatokenizerfast s llamatokenizerfast call if text is not none to encode the text to prepare the images this method forwards the images and kwargs arguments to fuyuimageprocessor s fuyuimageprocessor call if images is not none please refer to the doctsring of the above two methods for more information args text str liststr the sequence or batch of sequences to be encoded each sequence can be a string or a list of strings pretokenized string if the sequences are provided as list of strings pretokenized you must set issplitintowordstrue to lift the ambiguity with a batch of sequences images pil image image listpil image image the image or batch of images to be prepared each image can be a pil image numpy array or pytorch tensor in case of a numpy arraypytorch tensor each image should be of shape c h w where c is a number of channels h and w are image height and width returns fuyubatchencoding a fuyubatchencoding with the following fields inputids tensor of token ids to be fed to a model returned when text is not none imagepatches list of tensor of image patches returned when images is not none imagepatchesindices tensor of indices where patch embeddings have to be inserted by the model attentionmask list of indices specifying which tokens should be attended to by the model when returnattentionmasktrue check input validity preprocess images using self imageprocessor fixme we hard code pt here because the rest of the processing assumes torch tensors use self tokenizer to get the ids of special tokens to insert into image ids use self imageprocessor again to obtain the full token ids and batch inputs transforms raw coordinates detected by fuyuforcausallm to the original images coordinate space coordinates will be returned in box format with the following pattern boxtop left bottom rightbox point coordinates are not supported yet args outputs generateoutput raw outputs from generate targetsizes torch tensor optional tensor of shape batchsize 2 where each entry is the height width of the corresponding image in the batch if set found coordinates in the output sequence are rescaled to the target sizes if left to none coordinates will not be rescaled returns generateoutput same output type returned by generate with output token ids replaced with boxed and possible rescaled coordinates retrieve transformed coordinates from tokens scale back to original image size and multiply by 2 replace the ids so they get detokenized right retrieve transformed coordinates from tokens scale back to original image size and multiply by 2 replace the ids so they get detokenized right this method forwards all its arguments to llamatokenizerfast s pretrainedtokenizer batchdecode please refer to the docstring of this method for more information this method forwards all its arguments to llamatokenizerfast s pretrainedtokenizer decode please refer to the docstring of this method for more information coding utf 8 2023 the huggingface inc team licensed under the apache license version 2 0 the license you may not use this file except in compliance with the license you may obtain a copy of the license at http www apache org licenses license 2 0 unless required by applicable law or agreed to in writing software distributed under the license is distributed on an as is basis without warranties or conditions of any kind either express or implied see the license for the specific language governing permissions and limitations under the license image text processor class for git bbox bbox point point boa takes an unpacked stream of tokens i e a list of tensors one for each item in the batch and does the required padding to create a single tensor for the batch of shape batch_size x new_seq_len create padded tensors for the full batch place each batch entry into the batch tensor takes an input_stream tensor of shape b x s x for each subsequence adds any required padding to account for images and then unpacks the subsequences to create a single sequence per item in the batch returns a list of tensors one for each item in the batch first construct full token stream including image placeholder tokens and loss mask for each subsequence and append to lists we use lists rather than tensors because each subsequence is variable sized todo remove this logic in a subsequent release since subsequences are not supported given a string prompt converts the prompt into a list of texttokenconversions wherever we notice the token_open_string token_close_string we split the prompt split by the regex pattern this function transforms the prompt in the following fashion box point and box point to their respective token mappings extract the coordinates from the tag transform the coordinates into the transformed image space return the prompt tokens with the transformed coordinates and new tags bounding boxes and points must be in the following format box y1 x1 y2 x2 box point x y point the spaces and punctuation added above are not optional make a namedtuple that stores text and is_bbox we want to do the following tokenize the code normally when we see a point or box tokenize using the tokenize_within_tag function when point or box close tag continue tokenizing normally first we replace the point and box tags with their respective tokens tokenize the prompt convert prompt into a list split this is a location we need to tokenize it surround the text with the open and close tags given a bounding box of the fashion box 1 2 3 4 box point 1 2 point this function is responsible for converting 1 2 3 4 into tokens of 1 2 3 4 without any commas convert the text into a list of strings if there are any open or close tags remove them remove all spaces from num_ints scale to transformed image siz tokenize the text skipping the same issue with types as above given a set of prompts and number of tokens to generate tokenize prompts set the sequence length to be the max of length of prompts plus the number of tokens we would like to generate pad all the sequences to this length so we can convert them into a 3d tensor if not tool use tranform the coordinates while tokenizing only add bbox open token to the last subsequence since that is what will be completed now we have a list of list of tokens which each list has a different size we want to extend this list to incorporate the tokens that need to be generated make all the sequences equal length get the prompts length get the max prompts length number of tokens in the each sample of the batch now update the list of list to be of the same size samples_length now we are in a structured format we can convert to tensors simplified assuming self crop_top self padding_top 0 simplified assuming self crop_left self padding_left 0 constructs a fuyu processor which wraps a fuyu image processor and a llama tokenizer into a single processor fuyuprocessor offers all the functionalities of fuyuimageprocessor and llamatokenizerfast see the fuyuprocessor __call__ and fuyuprocessor decode for more information args image_processor fuyuimageprocessor the image processor is a required input tokenizer llamatokenizerfast the tokenizer is a required input todo can t derive this from model files where to set it for image_patches we don t pad but just append them to the list for image_patches_indices fixme max_tokens_to_generate is embedded into this processor s call construct inputs for image patch indices use same packing logic for the image patch indices main method to prepare for the model one or several sequences s and image s this method forwards the text and kwargs arguments to llamatokenizerfast s llamatokenizerfast __call__ if text is not none to encode the text to prepare the image s this method forwards the images and kwargs arguments to fuyuimageprocessor s fuyuimageprocessor __call__ if images is not none please refer to the doctsring of the above two methods for more information args text str list str the sequence or batch of sequences to be encoded each sequence can be a string or a list of strings pretokenized string if the sequences are provided as list of strings pretokenized you must set is_split_into_words true to lift the ambiguity with a batch of sequences images pil image image list pil image image the image or batch of images to be prepared each image can be a pil image numpy array or pytorch tensor in case of a numpy array pytorch tensor each image should be of shape c h w where c is a number of channels h and w are image height and width returns fuyubatchencoding a fuyubatchencoding with the following fields input_ids tensor of token ids to be fed to a model returned when text is not none image_patches list of tensor of image patches returned when images is not none image_patches_indices tensor of indices where patch embeddings have to be inserted by the model attention_mask list of indices specifying which tokens should be attended to by the model when return_attention_mask true check input validity preprocess images using self image_processor fixme we hard code pt here because the rest of the processing assumes torch tensors each batch contains only one sequence use self tokenizer to get the ids of special tokens to insert into image ids use self image_processor again to obtain the full token ids and batch inputs transforms raw coordinates detected by fuyuforcausallm to the original images coordinate space coordinates will be returned in box format with the following pattern box top left bottom right box point coordinates are not supported yet args outputs generateoutput raw outputs from generate target_sizes torch tensor optional tensor of shape batch_size 2 where each entry is the height width of the corresponding image in the batch if set found coordinates in the output sequence are rescaled to the target sizes if left to none coordinates will not be rescaled returns generateoutput same output type returned by generate with output token ids replaced with boxed and possible rescaled coordinates retrieve transformed coordinates from tokens scale back to original image size and multiply by 2 replace the ids so they get detokenized right retrieve transformed coordinates from tokens scale back to original image size and multiply by 2 replace the ids so they get detokenized right this method forwards all its arguments to llamatokenizerfast s pretrainedtokenizer batch_decode please refer to the docstring of this method for more information this method forwards all its arguments to llamatokenizerfast s pretrainedtokenizer decode please refer to the docstring of this method for more information
import re
from typing import Dict, List, Optional, Tuple, Union

import numpy as np

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import PaddingStrategy, TruncationStrategy
from ...utils import TensorType, is_torch_available, logging, requires_backends


if is_torch_available():
    from .image_processing_fuyu import FuyuBatchFeature


logger = logging.get_logger(__name__)


if is_torch_available():
    import torch


TEXT_REPR_BBOX_OPEN = "<box>"
TEXT_REPR_BBOX_CLOSE = "</box>"
TEXT_REPR_POINT_OPEN = "<point>"
TEXT_REPR_POINT_CLOSE = "</point>"

TOKEN_BBOX_OPEN_STRING = "<0x00>"  
TOKEN_BBOX_CLOSE_STRING = "<0x01>"  
TOKEN_POINT_OPEN_STRING = "<0x02>"  
TOKEN_POINT_CLOSE_STRING = "<0x03>"  
BEGINNING_OF_ANSWER_STRING = "<0x04>"  


def full_unpacked_stream_to_tensor(
    all_bi_tokens_to_place: List[int],
    full_unpacked_stream: List["torch.Tensor"],
    fill_value: int,
    batch_size: int,
    new_seq_len: int,
    offset: int,
) -> "torch.Tensor":
    

    assert len(all_bi_tokens_to_place) == batch_size
    assert len(full_unpacked_stream) == batch_size

    
    new_padded_tensor = torch.full(
        [batch_size, new_seq_len],
        fill_value=fill_value,
        dtype=full_unpacked_stream[0].dtype,
        device=full_unpacked_stream[0].device,
    )

    
    for bi in range(batch_size):
        tokens_to_place = all_bi_tokens_to_place[bi]
        new_padded_tensor[bi, :tokens_to_place] = full_unpacked_stream[bi][offset : tokens_to_place + offset]

    return new_padded_tensor


def construct_full_unpacked_stream(
    num_real_text_tokens: Union[List[List[int]], "torch.Tensor"],
    input_stream: "torch.Tensor",
    image_tokens: List[List["torch.Tensor"]],
    batch_size: int,
    num_sub_sequences: int,
) -> List["torch.Tensor"]:
    

    all_bi_stream = []

    for batch_index in range(batch_size):
        all_si_stream = []

        
        
        
        image_adjustment = image_tokens[batch_index][0]
        subsequence_stream = torch.cat([image_adjustment, input_stream[batch_index, 0]], dim=0)
        num_real_tokens = image_adjustment.shape[0] + num_real_text_tokens[batch_index][0]
        all_si_stream.append(subsequence_stream[:num_real_tokens])
        all_bi_stream.append(torch.cat(all_si_stream, dim=0))

    return all_bi_stream


def _replace_string_repr_with_token_tags(prompt: str) -> str:
    prompt = prompt.replace(TEXT_REPR_POINT_OPEN, TOKEN_POINT_OPEN_STRING)
    prompt = prompt.replace(TEXT_REPR_POINT_CLOSE, TOKEN_POINT_CLOSE_STRING)
    prompt = prompt.replace(TEXT_REPR_BBOX_OPEN, TOKEN_BBOX_OPEN_STRING)
    prompt = prompt.replace(TEXT_REPR_BBOX_CLOSE, TOKEN_BBOX_CLOSE_STRING)
    return prompt


def _segment_prompt_into_text_token_conversions(prompt: str) -> List:
    
    
    prompt_text_list: List = []
    regex_pattern = re.compile(
        f"({TOKEN_BBOX_OPEN_STRING}|{TOKEN_BBOX_CLOSE_STRING}|{TOKEN_POINT_OPEN_STRING}|{TOKEN_POINT_CLOSE_STRING})"
    )
    
    prompt_split = regex_pattern.split(prompt)
    for i, elem in enumerate(prompt_split):
        if len(elem) == 0 or elem in [
            TOKEN_BBOX_OPEN_STRING,
            TOKEN_BBOX_CLOSE_STRING,
            TOKEN_POINT_OPEN_STRING,
            TOKEN_POINT_CLOSE_STRING,
        ]:
            continue
        prompt_text_list.append(
            (elem, i > 1 and prompt_split[i - 1] in [TOKEN_BBOX_OPEN_STRING, TOKEN_POINT_OPEN_STRING])
        )
    return prompt_text_list


def _transform_coordinates_and_tokenize(prompt: str, scale_factor: float, tokenizer) -> List[int]:
    
    

    
    
    
    prompt = _replace_string_repr_with_token_tags(prompt)
    
    
    prompt_text_list = _segment_prompt_into_text_token_conversions(prompt)
    transformed_prompt_tokens: List[int] = []
    for elem in prompt_text_list:
        if elem[1]:
            
            within_tag_tokenized = _transform_within_tags(elem[0], scale_factor, tokenizer)
            
            transformed_prompt_tokens.extend(within_tag_tokenized)
        else:
            transformed_prompt_tokens.extend(tokenizer(elem[0], add_special_tokens=False).input_ids)
    return transformed_prompt_tokens


def _transform_within_tags(text: str, scale_factor: float, tokenizer) -> List[int]:
    
    
    num_int_strs = text.split(",")
    if len(num_int_strs) == 2:
        
        token_space_open_string = tokenizer.vocab[TOKEN_POINT_OPEN_STRING]
        token_space_close_string = tokenizer.vocab[TOKEN_POINT_CLOSE_STRING]
    else:
        token_space_open_string = tokenizer.vocab[TOKEN_BBOX_OPEN_STRING]
        token_space_close_string = tokenizer.vocab[TOKEN_BBOX_CLOSE_STRING]

    
    num_ints = [float(num.strip()) for num in num_int_strs]
    
    if len(num_ints) == 2:
        num_ints_translated = scale_point_to_transformed_image(x=num_ints[0], y=num_ints[1], scale_factor=scale_factor)
    elif len(num_ints) == 4:
        num_ints_translated = scale_bbox_to_transformed_image(
            top=num_ints[0],
            left=num_ints[1],
            bottom=num_ints[2],
            right=num_ints[3],
            scale_factor=scale_factor,
        )
    else:
        raise ValueError(f"Invalid number of ints: {len(num_ints)}")
    
    tokens = [tokenizer.vocab[str(num)] for num in num_ints_translated]
    return [token_space_open_string] + tokens + [token_space_close_string]


def _tokenize_prompts_with_image_and_batch(
    tokenizer,
    prompts: List[List[str]],
    scale_factors: Optional[List[List["torch.Tensor"]]],
    max_tokens_to_generate: int,
    max_position_embeddings: int,
    add_BOS: bool,  
    add_beginning_of_answer_token: bool,
) -> Tuple["torch.Tensor", "torch.Tensor"]:
    

    
    if scale_factors is not None:
        transformed_prompt_tokens = []
        for prompt_seq, scale_factor_seq in zip(prompts, scale_factors):
            transformed_prompt_tokens.append(
                [
                    _transform_coordinates_and_tokenize(prompt, scale_factor.item(), tokenizer)
                    for prompt, scale_factor in zip(prompt_seq, scale_factor_seq)
                ]
            )
    else:
        transformed_prompt_tokens = [[tokenizer.tokenize(prompt) for prompt in prompt_seq] for prompt_seq in prompts]

    prompts_tokens = transformed_prompt_tokens

    if add_BOS:
        bos_token = tokenizer.vocab["<s>"]
    else:
        bos_token = tokenizer.vocab["|ENDOFTEXT|"]
    prompts_tokens = [[[bos_token] + x for x in prompt_seq] for prompt_seq in prompts_tokens]
    if add_beginning_of_answer_token:
        boa = tokenizer.vocab[BEGINNING_OF_ANSWER_STRING]
        
        for token_seq in prompts_tokens:
            token_seq[-1].append(boa)

    
    
    
    
    

    prompts_length = [[len(x) for x in prompts_tokens_seq] for prompts_tokens_seq in prompts_tokens]
    
    max_prompt_len: int = np.max(prompts_length)
    
    samples_length = min(max_prompt_len + max_tokens_to_generate, max_position_embeddings)
    if max_prompt_len + max_tokens_to_generate > max_position_embeddings:
        logger.warning(
            f"Max subsequence prompt length of {max_prompt_len} + max tokens to generate {max_tokens_to_generate}",
            f"exceeds context length of {max_position_embeddings}. Will generate as many tokens as possible.",
        )
    
    for prompt_tokens_seq, prompts_length_seq in zip(prompts_tokens, prompts_length):
        for prompt_tokens, prompt_length in zip(prompt_tokens_seq, prompts_length_seq):
            if len(prompt_tokens) > samples_length:
                raise ValueError("Length of subsequence prompt exceeds sequence length.")
            padding_size = samples_length - prompt_length
            prompt_tokens.extend([tokenizer.vocab["|ENDOFTEXT|"]] * padding_size)

    
    prompts_tokens_tensor = torch.tensor(prompts_tokens, dtype=torch.int64)
    prompts_length_tensor = torch.tensor(prompts_length, dtype=torch.int64)

    return prompts_tokens_tensor, prompts_length_tensor



def original_to_transformed_h_coords(original_coords, scale_h):
    return np.round(original_coords * scale_h).astype(np.int32)



def original_to_transformed_w_coords(original_coords, scale_w):
    return np.round(original_coords * scale_w).astype(np.int32)


def scale_point_to_transformed_image(x: float, y: float, scale_factor: float) -> List[int]:
    x_scaled = original_to_transformed_w_coords(np.array([x / 2]), scale_factor)[0]
    y_scaled = original_to_transformed_h_coords(np.array([y / 2]), scale_factor)[0]
    return [x_scaled, y_scaled]


def scale_bbox_to_transformed_image(
    top: float, left: float, bottom: float, right: float, scale_factor: float
) -> List[int]:
    top_scaled = original_to_transformed_w_coords(np.array([top / 2]), scale_factor)[0]
    left_scaled = original_to_transformed_h_coords(np.array([left / 2]), scale_factor)[0]
    bottom_scaled = original_to_transformed_w_coords(np.array([bottom / 2]), scale_factor)[0]
    right_scaled = original_to_transformed_h_coords(np.array([right / 2]), scale_factor)[0]
    return [top_scaled, left_scaled, bottom_scaled, right_scaled]


class FuyuProcessor(ProcessorMixin):
    r

    attributes = ["image_processor", "tokenizer"]
    image_processor_class = "FuyuImageProcessor"
    tokenizer_class = "AutoTokenizer"

    def __init__(self, image_processor, tokenizer):
        super().__init__(image_processor=image_processor, tokenizer=tokenizer)
        self.image_processor = image_processor
        self.tokenizer = tokenizer
        self.max_tokens_to_generate = 10
        self.max_position_embeddings = 16384  
        self.pad_token_id = 0
        self.dummy_image_index = -1

    def _left_pad_inputs_with_attention_mask(self, model_inputs: List[Dict], return_attention_mask: bool):
        max_length_input_ids = max(entry["input_ids"].shape[1] for entry in model_inputs)
        max_length_image_patch_indices = max(entry["image_patches_indices"].shape[1] for entry in model_inputs)

        batched_inputs = {"input_ids": [], "image_patches": [], "image_patches_indices": [], "attention_mask": []}

        for entry in model_inputs:
            for key, tensor in entry.items():
                if key == "input_ids":
                    num_padding_tokens = max_length_input_ids - tensor.shape[1]
                    padded_input_ids = torch.cat(
                        [
                            torch.full((tensor.shape[0], num_padding_tokens), self.pad_token_id, dtype=torch.long),
                            tensor,
                        ],
                        dim=1,
                    )
                    batched_inputs[key].append(padded_input_ids)

                    attention_mask = torch.cat(
                        [torch.zeros(tensor.shape[0], num_padding_tokens, dtype=torch.long), torch.ones_like(tensor)],
                        dim=1,
                    )
                    batched_inputs["attention_mask"].append(attention_mask)

                elif key == "image_patches":
                    
                    batched_inputs[key].append(tensor)

                else:  
                    num_padding_indices = max_length_image_patch_indices - tensor.shape[1]
                    padded_indices = torch.cat(
                        [
                            torch.full(
                                (tensor.shape[0], num_padding_indices), self.dummy_image_index, dtype=torch.long
                            ),
                            tensor,
                        ],
                        dim=1,
                    )
                    batched_inputs[key].append(padded_indices)
        batched_keys = ["input_ids", "image_patches_indices"]
        if return_attention_mask:
            batched_keys.append("attention_mask")
        for key in batched_keys:
            batched_inputs[key] = torch.cat(batched_inputs[key], dim=0)

        return batched_inputs

    def get_sample_encoding(
        self,
        prompts,
        scale_factors,
        image_unpadded_heights,
        image_unpadded_widths,
        image_placeholder_id,
        image_newline_id,
        tensor_batch_images,
    ):
        image_present = torch.ones(1, 1, 1)
        model_image_input = self.image_processor.preprocess_with_tokenizer_info(
            image_input=tensor_batch_images,
            image_present=image_present,
            image_unpadded_h=image_unpadded_heights,
            image_unpadded_w=image_unpadded_widths,
            image_placeholder_id=image_placeholder_id,
            image_newline_id=image_newline_id,
            variable_sized=True,
        )
        
        prompt_tokens, prompts_length = _tokenize_prompts_with_image_and_batch(
            tokenizer=self.tokenizer,
            prompts=prompts,
            scale_factors=scale_factors,
            max_tokens_to_generate=self.max_tokens_to_generate,
            max_position_embeddings=self.max_position_embeddings,
            add_BOS=True,
            add_beginning_of_answer_token=True,
        )
        image_padded_unpacked_tokens = construct_full_unpacked_stream(
            num_real_text_tokens=prompts_length,
            input_stream=prompt_tokens,
            image_tokens=model_image_input["image_input_ids"],
            batch_size=1,
            num_sub_sequences=self.subsequence_length,
        )
        
        unpacked_image_patch_indices_per_batch = construct_full_unpacked_stream(
            num_real_text_tokens=prompts_length,
            input_stream=torch.full_like(prompt_tokens, -1),
            image_tokens=model_image_input["image_patch_indices_per_batch"],
            batch_size=1,
            num_sub_sequences=self.subsequence_length,
        )
        max_prompt_length = max(x.shape[-1] for x in image_padded_unpacked_tokens)
        max_seq_len_batch = min(max_prompt_length + self.max_tokens_to_generate, self.max_position_embeddings)
        tokens_to_place = min(max_seq_len_batch, max(0, image_padded_unpacked_tokens[0].shape[0]))

        
        image_patch_input_indices = full_unpacked_stream_to_tensor(
            all_bi_tokens_to_place=[tokens_to_place],
            full_unpacked_stream=unpacked_image_patch_indices_per_batch,
            fill_value=-1,
            batch_size=1,
            new_seq_len=max_seq_len_batch,
            offset=0,
        )
        image_patches_tensor = torch.stack([img[0] for img in model_image_input["image_patches"]])
        batch_encoding = {
            "input_ids": image_padded_unpacked_tokens[0].unsqueeze(0),
            "image_patches": image_patches_tensor,
            "image_patches_indices": image_patch_input_indices,
        }
        return batch_encoding

    def __call__(
        self,
        text=None,
        images=None,
        add_special_tokens: bool = True,
        return_attention_mask: bool = True,
        padding: Union[bool, str, PaddingStrategy] = False,
        truncation: Union[bool, str, TruncationStrategy] = None,
        max_length: Optional[int] = None,
        stride: int = 0,
        pad_to_multiple_of: Optional[int] = None,
        return_overflowing_tokens: bool = False,
        return_special_tokens_mask: bool = False,
        return_offsets_mapping: bool = False,
        return_token_type_ids: bool = False,
        return_length: bool = False,
        verbose: bool = True,
        return_tensors: Optional[Union[str, TensorType]] = None,
        **kwargs,
    ) -> "FuyuBatchFeature":
        
        requires_backends(self, ["torch"])

        
        if not return_attention_mask:
            raise ValueError("`return_attention_mask=False` is not supported for this model.")
        if text is None and images is None:
            raise ValueError("You have to specify either text or images. Both cannot be None.")
        if text is not None and images is None:
            logger.warning("You are processing a text with no associated image. Make sure it is intended.")
            self.current_processor = self.tokenizer
            text_encoding = self.tokenizer(
                text=text,
                add_special_tokens=add_special_tokens,
                padding=padding,
                truncation=truncation,
                max_length=max_length,
                stride=stride,
                pad_to_multiple_of=pad_to_multiple_of,
                return_attention_mask=return_attention_mask,
                return_overflowing_tokens=return_overflowing_tokens,
                return_special_tokens_mask=return_special_tokens_mask,
                return_offsets_mapping=return_offsets_mapping,
                return_token_type_ids=return_token_type_ids,
                return_length=return_length,
                verbose=verbose,
                return_tensors=return_tensors,
                **kwargs,
            )
            return text_encoding

        if text is None and images is not None:
            logger.warning("You are processing an image with no associated text. Make sure it is intended.")
            prompts = [[""]]
        if text is not None and images is not None:
            if isinstance(text, str):
                prompts = [[text]]
            elif isinstance(text, list):
                prompts = [[text_seq] for text_seq in text]

        

        
        image_encoding = self.image_processor.preprocess(images, return_tensors="pt")
        batch_images = image_encoding["images"]
        image_unpadded_heights = image_encoding["image_unpadded_heights"]
        image_unpadded_widths = image_encoding["image_unpadded_widths"]
        scale_factors = image_encoding["image_scale_factors"]
        self.subsequence_length = 1  
        self.batch_size = len(batch_images)

        

        image_placeholder_id = self.tokenizer("|SPEAKER|", add_special_tokens=False)["input_ids"][1]
        image_newline_id = self.tokenizer("|NEWLINE|", add_special_tokens=False)["input_ids"][1]
        tensor_batch_images = torch.stack([img[0] for img in batch_images]).unsqueeze(1)

        
        all_encodings = []

        for prompt, scale_factor, image_unpadded_height, image_unpadded_width, tensor_batch_image in zip(
            prompts, scale_factors, image_unpadded_heights, image_unpadded_widths, tensor_batch_images
        ):
            sample_encoding = self.get_sample_encoding(
                prompts=[prompt],
                scale_factors=[scale_factor],
                image_unpadded_heights=torch.tensor([image_unpadded_height]),
                image_unpadded_widths=torch.tensor([image_unpadded_width]),
                image_placeholder_id=image_placeholder_id,
                image_newline_id=image_newline_id,
                tensor_batch_images=tensor_batch_image.unsqueeze(0),
            )
            all_encodings.append(sample_encoding)
        batch_encoding = self._left_pad_inputs_with_attention_mask(
            model_inputs=all_encodings, return_attention_mask=return_attention_mask
        )
        return FuyuBatchFeature(data=batch_encoding)

    def post_process_box_coordinates(self, outputs, target_sizes=None):
        

        def scale_factor_to_fit(original_size, target_size=None):
            height, width = original_size
            if target_size is None:
                max_height = self.image_processor.size["height"]
                max_width = self.image_processor.size["width"]
            else:
                max_height, max_width = target_size
            if width <= max_width and height <= max_height:
                return 1.0
            return min(max_height / height, max_width / width)

        def find_delimiters_pair(tokens, start_token, end_token):
            start_id = self.tokenizer.convert_tokens_to_ids(start_token)
            end_id = self.tokenizer.convert_tokens_to_ids(end_token)

            starting_positions = (tokens == start_id).nonzero(as_tuple=True)[0]
            ending_positions = (tokens == end_id).nonzero(as_tuple=True)[0]

            if torch.any(starting_positions) and torch.any(ending_positions):
                return (starting_positions[0], ending_positions[0])
            return (None, None)

        def tokens_to_boxes(tokens, original_size):
            while (pair := find_delimiters_pair(tokens, TOKEN_BBOX_OPEN_STRING, TOKEN_BBOX_CLOSE_STRING)) != (
                None,
                None,
            ):
                start, end = pair
                if end != start + 5:
                    continue

                
                coords = self.tokenizer.convert_ids_to_tokens(tokens[start + 1 : end])

                
                scale = scale_factor_to_fit(original_size)
                top, left, bottom, right = [2 * int(float(c) / scale) for c in coords]

                
                replacement = f" {TEXT_REPR_BBOX_OPEN}{top}, {left}, {bottom}, {right}{TEXT_REPR_BBOX_CLOSE}"
                replacement = self.tokenizer.tokenize(replacement)[1:]
                replacement = self.tokenizer.convert_tokens_to_ids(replacement)
                replacement = torch.tensor(replacement).to(tokens)

                tokens = torch.cat([tokens[:start], replacement, tokens[end + 1 :]], 0)
            return tokens

        def tokens_to_points(tokens, original_size):
            while (pair := find_delimiters_pair(tokens, TOKEN_POINT_OPEN_STRING, TOKEN_POINT_CLOSE_STRING)) != (
                None,
                None,
            ):
                start, end = pair
                if end != start + 3:
                    continue

                
                coords = self.tokenizer.convert_ids_to_tokens(tokens[start + 1 : end])

                
                scale = scale_factor_to_fit(original_size)
                x, y = [2 * int(float(c) / scale) for c in coords]

                
                replacement = f" {TEXT_REPR_POINT_OPEN}{x}, {y}{TEXT_REPR_POINT_CLOSE}"
                replacement = self.tokenizer.tokenize(replacement)[1:]
                replacement = self.tokenizer.convert_tokens_to_ids(replacement)
                replacement = torch.tensor(replacement).to(tokens)

                tokens = torch.cat([tokens[:start], replacement, tokens[end + 1 :]], 0)
            return tokens

        if target_sizes is None:
            target_sizes = ((self.image_processor.size["height"], self.image_processor.size["width"]),) * len(outputs)
        elif target_sizes.shape[1] != 2:
            raise ValueError("Each element of target_sizes must contain the size (h, w) of each image of the batch")

        if len(outputs) != len(target_sizes):
            raise ValueError("Make sure that you pass in as many target sizes as output sequences")

        results = []
        for seq, size in zip(outputs, target_sizes):
            seq = tokens_to_boxes(seq, size)
            seq = tokens_to_points(seq, size)
            results.append(seq)

        return results

    def batch_decode(self, *args, **kwargs):
        
        return self.tokenizer.batch_decode(*args, **kwargs)

    def decode(self, *args, **kwargs):
        
        return self.tokenizer.decode(*args, **kwargs)