codingutf8 2022 shi labs and 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 oneformer attributes imageprocessor tokenizer imageprocessorclass oneformerimageprocessor tokenizerclass cliptokenizer cliptokenizerfast def init self imageprocessornone tokenizernone maxseqlength int 77 taskseqlength int 77 kwargs if imageprocessor is none raise valueerroryou need to specify an imageprocessor if tokenizer is none raise valueerroryou need to specify a tokenizer self maxseqlength maxseqlength self taskseqlength taskseqlength super initimageprocessor tokenizer def preprocesstextself textlistnone maxlength77 if textlist is none raise valueerrortokens cannot be none tokens self tokenizertextlist paddingmaxlength maxlengthmaxlength truncationtrue attentionmasks inputids tokensattentionmask tokensinputids tokeninputs for attnmask inputid in zipattentionmasks inputids token torch tensorattnmask torch tensorinputid tokeninputs appendtoken unsqueeze0 tokeninputs torch cattokeninputs dim0 return tokeninputs def callself imagesnone taskinputsnone segmentationmapsnone kwargs if taskinputs is none raise valueerroryou have to specify the taskinput found none elif images is none raise valueerroryou have to specify the image found none if not alltask in semantic instance panoptic for task in taskinputs raise valueerrortaskinputs must be semantic instance or panoptic encodedinputs self imageprocessorimages taskinputs segmentationmaps kwargs if isinstancetaskinputs str taskinputs taskinputs if isinstancetaskinputs list and allisinstancetaskinput str for taskinput in taskinputs tasktokeninputs for task in taskinputs taskinput fthe task is task tasktokeninputs appendtaskinput encodedinputstaskinputs self preprocesstexttasktokeninputs maxlengthself taskseqlength else raise typeerrortask inputs should be a string or a list of strings if hasattrencodedinputs textinputs textslist encodedinputs textinputs textinputs for texts in textslist textinputlist self preprocesstexttexts maxlengthself maxseqlength textinputs appendtextinputlist unsqueeze0 encodedinputstextinputs torch cattextinputs dim0 return encodedinputs def encodeinputsself imagesnone taskinputsnone segmentationmapsnone kwargs if taskinputs is none raise valueerroryou have to specify the taskinput found none elif images is none raise valueerroryou have to specify the image found none if not alltask in semantic instance panoptic for task in taskinputs raise valueerrortaskinputs must be semantic instance or panoptic encodedinputs self imageprocessor encodeinputsimages taskinputs segmentationmaps kwargs if isinstancetaskinputs str taskinputs taskinputs if isinstancetaskinputs list and allisinstancetaskinput str for taskinput in taskinputs tasktokeninputs for task in taskinputs taskinput fthe task is task tasktokeninputs appendtaskinput encodedinputstaskinputs self preprocesstexttasktokeninputs maxlengthself taskseqlength else raise typeerrortask inputs should be a string or a list of strings if hasattrencodedinputs textinputs textslist encodedinputs textinputs textinputs for texts in textslist textinputlist self preprocesstexttexts maxlengthself maxseqlength textinputs appendtextinputlist unsqueeze0 encodedinputstextinputs torch cattextinputs dim0 return encodedinputs def postprocesssemanticsegmentationself args kwargs return self imageprocessor postprocesssemanticsegmentationargs kwargs def postprocessinstancesegmentationself args kwargs return self imageprocessor postprocessinstancesegmentationargs kwargs def postprocesspanopticsegmentationself args kwargs return self imageprocessor postprocesspanopticsegmentationargs kwargs coding utf 8 2022 shi labs and 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 oneformer constructs an oneformer processor which wraps oneformerimageprocessor and cliptokenizer cliptokenizerfast into a single processor that inherits both the image processor and tokenizer functionalities args image_processor oneformerimageprocessor the image processor is a required input tokenizer cliptokenizer cliptokenizerfast the tokenizer is a required input max_seq_len int optional defaults to 77 sequence length for input text list task_seq_len int optional defaults to 77 sequence length for input task token main method to prepare for the model one or several task input s and image s this method forwards the task_inputs and kwargs arguments to cliptokenizer s cliptokenizer __call__ if task_inputs is not none to encode to prepare the image s this method forwards the images and kwargs arguments to oneformerimageprocessor s oneformerimageprocessor __call__ if images is not none please refer to the doctsring of the above two methods for more information args task_inputs str list str the sequence or batch of task_inputs sequences to be encoded each sequence can be a string or a list of strings of the template the task is task images pil image image np ndarray torch tensor list pil image image list np ndarray list torch tensor 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 segmentation_maps imageinput optional the corresponding semantic segmentation maps with the pixel wise annotations bool optional defaults to true whether or not to pad images up to the largest image in a batch and create a pixel mask if left to the default will return a pixel mask that is 1 for pixels that are real i e not masked 0 for pixels that are padding i e masked returns batchfeature a batchfeature with the following fields task_inputs list of token ids to be fed to a model returned when text is not none pixel_values pixel values to be fed to a model returned when images is not none this method forwards all its arguments to oneformerimageprocessor encode_inputs and then tokenizes the task_inputs please refer to the docstring of this method for more information this method forwards all its arguments to oneformerimageprocessor post_process_semantic_segmentation please refer to the docstring of this method for more information this method forwards all its arguments to oneformerimageprocessor post_process_instance_segmentation please refer to the docstring of this method for more information this method forwards all its arguments to oneformerimageprocessor post_process_panoptic_segmentation please refer to the docstring of this method for more information
from typing import List

from ...processing_utils import ProcessorMixin
from ...utils import is_torch_available


if is_torch_available():
    import torch


class OneFormerProcessor(ProcessorMixin):
    r

    attributes = ["image_processor", "tokenizer"]
    image_processor_class = "OneFormerImageProcessor"
    tokenizer_class = ("CLIPTokenizer", "CLIPTokenizerFast")

    def __init__(
        self, image_processor=None, tokenizer=None, max_seq_length: int = 77, task_seq_length: int = 77, **kwargs
    ):
        if image_processor is None:
            raise ValueError("You need to specify an `image_processor`.")
        if tokenizer is None:
            raise ValueError("You need to specify a `tokenizer`.")

        self.max_seq_length = max_seq_length
        self.task_seq_length = task_seq_length

        super().__init__(image_processor, tokenizer)

    def _preprocess_text(self, text_list=None, max_length=77):
        if text_list is None:
            raise ValueError("tokens cannot be None.")

        tokens = self.tokenizer(text_list, padding="max_length", max_length=max_length, truncation=True)

        attention_masks, input_ids = tokens["attention_mask"], tokens["input_ids"]

        token_inputs = []
        for attn_mask, input_id in zip(attention_masks, input_ids):
            token = torch.tensor(attn_mask) * torch.tensor(input_id)
            token_inputs.append(token.unsqueeze(0))

        token_inputs = torch.cat(token_inputs, dim=0)
        return token_inputs

    def __call__(self, images=None, task_inputs=None, segmentation_maps=None, **kwargs):
        

        if task_inputs is None:
            raise ValueError("You have to specify the task_input. Found None.")
        elif images is None:
            raise ValueError("You have to specify the image. Found None.")

        if not all(task in ["semantic", "instance", "panoptic"] for task in task_inputs):
            raise ValueError("task_inputs must be semantic, instance, or panoptic.")

        encoded_inputs = self.image_processor(images, task_inputs, segmentation_maps, **kwargs)

        if isinstance(task_inputs, str):
            task_inputs = [task_inputs]

        if isinstance(task_inputs, List) and all(isinstance(task_input, str) for task_input in task_inputs):
            task_token_inputs = []
            for task in task_inputs:
                task_input = f"the task is {task}"
                task_token_inputs.append(task_input)
            encoded_inputs["task_inputs"] = self._preprocess_text(task_token_inputs, max_length=self.task_seq_length)
        else:
            raise TypeError("Task Inputs should be a string or a list of strings.")

        if hasattr(encoded_inputs, "text_inputs"):
            texts_list = encoded_inputs.text_inputs

            text_inputs = []
            for texts in texts_list:
                text_input_list = self._preprocess_text(texts, max_length=self.max_seq_length)
                text_inputs.append(text_input_list.unsqueeze(0))

            encoded_inputs["text_inputs"] = torch.cat(text_inputs, dim=0)

        return encoded_inputs

    def encode_inputs(self, images=None, task_inputs=None, segmentation_maps=None, **kwargs):
        

        if task_inputs is None:
            raise ValueError("You have to specify the task_input. Found None.")
        elif images is None:
            raise ValueError("You have to specify the image. Found None.")

        if not all(task in ["semantic", "instance", "panoptic"] for task in task_inputs):
            raise ValueError("task_inputs must be semantic, instance, or panoptic.")

        encoded_inputs = self.image_processor.encode_inputs(images, task_inputs, segmentation_maps, **kwargs)

        if isinstance(task_inputs, str):
            task_inputs = [task_inputs]

        if isinstance(task_inputs, List) and all(isinstance(task_input, str) for task_input in task_inputs):
            task_token_inputs = []
            for task in task_inputs:
                task_input = f"the task is {task}"
                task_token_inputs.append(task_input)
            encoded_inputs["task_inputs"] = self._preprocess_text(task_token_inputs, max_length=self.task_seq_length)
        else:
            raise TypeError("Task Inputs should be a string or a list of strings.")

        if hasattr(encoded_inputs, "text_inputs"):
            texts_list = encoded_inputs.text_inputs

            text_inputs = []
            for texts in texts_list:
                text_input_list = self._preprocess_text(texts, max_length=self.max_seq_length)
                text_inputs.append(text_input_list.unsqueeze(0))

            encoded_inputs["text_inputs"] = torch.cat(text_inputs, dim=0)

        return encoded_inputs

    def post_process_semantic_segmentation(self, *args, **kwargs):
        
        return self.image_processor.post_process_semantic_segmentation(*args, **kwargs)

    def post_process_instance_segmentation(self, *args, **kwargs):
        
        return self.image_processor.post_process_instance_segmentation(*args, **kwargs)

    def post_process_panoptic_segmentation(self, *args, **kwargs):
        
        return self.image_processor.post_process_panoptic_segmentation(*args, **kwargs)