# coding=utf-8 # Copyright 2025 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. from typing import Optional from transformers.models.maskformer.image_processing_maskformer_fast import MaskFormerImageProcessorFast from ...utils import ( TensorType, is_torch_available, logging, ) from .image_processing_mask2former import ( compute_segments, convert_segmentation_to_rle, remove_low_and_no_objects, ) if is_torch_available(): import torch from torch import nn logger = logging.get_logger(__name__) class Mask2FormerImageProcessorFast(MaskFormerImageProcessorFast): def post_process_semantic_segmentation( self, outputs, target_sizes: Optional[list[tuple[int, int]]] = None ) -> "torch.Tensor": """ Converts the output of [`Mask2FormerForUniversalSegmentation`] into semantic segmentation maps. Only supports PyTorch. Args: outputs ([`Mask2FormerForUniversalSegmentation`]): Raw outputs of the model. target_sizes (`List[Tuple[int, int]]`, *optional*): List of length (batch_size), where each list item (`Tuple[int, int]]`) corresponds to the requested final size (height, width) of each prediction. If left to None, predictions will not be resized. Returns: `List[torch.Tensor]`: A list of length `batch_size`, where each item is a semantic segmentation map of shape (height, width) corresponding to the target_sizes entry (if `target_sizes` is specified). Each entry of each `torch.Tensor` correspond to a semantic class id. """ class_queries_logits = outputs.class_queries_logits # [batch_size, num_queries, num_classes+1] masks_queries_logits = outputs.masks_queries_logits # [batch_size, num_queries, height, width] # Scale back to preprocessed image size - (384, 384) for all models masks_queries_logits = torch.nn.functional.interpolate( masks_queries_logits, size=(384, 384), mode="bilinear", align_corners=False ) # Remove the null class `[..., :-1]` masks_classes = class_queries_logits.softmax(dim=-1)[..., :-1] masks_probs = masks_queries_logits.sigmoid() # [batch_size, num_queries, height, width] # Semantic segmentation logits of shape (batch_size, num_classes, height, width) segmentation = torch.einsum("bqc, bqhw -> bchw", masks_classes, masks_probs) batch_size = class_queries_logits.shape[0] # Resize logits and compute semantic segmentation maps if target_sizes is not None: if batch_size != len(target_sizes): raise ValueError( "Make sure that you pass in as many target sizes as the batch dimension of the logits" ) semantic_segmentation = [] for idx in range(batch_size): resized_logits = torch.nn.functional.interpolate( segmentation[idx].unsqueeze(dim=0), size=target_sizes[idx], mode="bilinear", align_corners=False ) semantic_map = resized_logits[0].argmax(dim=0) semantic_segmentation.append(semantic_map) else: semantic_segmentation = segmentation.argmax(dim=1) semantic_segmentation = [semantic_segmentation[i] for i in range(semantic_segmentation.shape[0])] return semantic_segmentation def post_process_instance_segmentation( self, outputs, threshold: float = 0.5, mask_threshold: float = 0.5, overlap_mask_area_threshold: float = 0.8, target_sizes: Optional[list[tuple[int, int]]] = None, return_coco_annotation: Optional[bool] = False, return_binary_maps: Optional[bool] = False, ) -> list[dict]: """ Converts the output of [`Mask2FormerForUniversalSegmentationOutput`] into instance segmentation predictions. Only supports PyTorch. If instances could overlap, set either return_coco_annotation or return_binary_maps to `True` to get the correct segmentation result. Args: outputs ([`Mask2FormerForUniversalSegmentation`]): Raw outputs of the model. threshold (`float`, *optional*, defaults to 0.5): The probability score threshold to keep predicted instance masks. mask_threshold (`float`, *optional*, defaults to 0.5): Threshold to use when turning the predicted masks into binary values. overlap_mask_area_threshold (`float`, *optional*, defaults to 0.8): The overlap mask area threshold to merge or discard small disconnected parts within each binary instance mask. target_sizes (`List[Tuple]`, *optional*): List of length (batch_size), where each list item (`Tuple[int, int]]`) corresponds to the requested final size (height, width) of each prediction. If left to None, predictions will not be resized. return_coco_annotation (`bool`, *optional*, defaults to `False`): If set to `True`, segmentation maps are returned in COCO run-length encoding (RLE) format. return_binary_maps (`bool`, *optional*, defaults to `False`): If set to `True`, segmentation maps are returned as a concatenated tensor of binary segmentation maps (one per detected instance). Returns: `List[Dict]`: A list of dictionaries, one per image, each dictionary containing two keys: - **segmentation** -- A tensor of shape `(height, width)` where each pixel represents a `segment_id`, or `List[List]` run-length encoding (RLE) of the segmentation map if return_coco_annotation is set to `True`, or a tensor of shape `(num_instances, height, width)` if return_binary_maps is set to `True`. Set to `None` if no mask if found above `threshold`. - **segments_info** -- A dictionary that contains additional information on each segment. - **id** -- An integer representing the `segment_id`. - **label_id** -- An integer representing the label / semantic class id corresponding to `segment_id`. - **score** -- Prediction score of segment with `segment_id`. """ if return_coco_annotation and return_binary_maps: raise ValueError("return_coco_annotation and return_binary_maps can not be both set to True.") # [batch_size, num_queries, num_classes+1] class_queries_logits = outputs.class_queries_logits # [batch_size, num_queries, height, width] masks_queries_logits = outputs.masks_queries_logits # Scale back to preprocessed image size - (384, 384) for all models masks_queries_logits = torch.nn.functional.interpolate( masks_queries_logits, size=(384, 384), mode="bilinear", align_corners=False ) device = masks_queries_logits.device num_classes = class_queries_logits.shape[-1] - 1 num_queries = class_queries_logits.shape[-2] # Loop over items in batch size results: list[dict[str, TensorType]] = [] for i in range(class_queries_logits.shape[0]): mask_pred = masks_queries_logits[i] mask_cls = class_queries_logits[i] scores = torch.nn.functional.softmax(mask_cls, dim=-1)[:, :-1] labels = torch.arange(num_classes, device=device).unsqueeze(0).repeat(num_queries, 1).flatten(0, 1) scores_per_image, topk_indices = scores.flatten(0, 1).topk(num_queries, sorted=False) labels_per_image = labels[topk_indices] topk_indices = torch.div(topk_indices, num_classes, rounding_mode="floor") mask_pred = mask_pred[topk_indices] pred_masks = (mask_pred > 0).float() # Calculate average mask prob mask_scores_per_image = (mask_pred.sigmoid().flatten(1) * pred_masks.flatten(1)).sum(1) / ( pred_masks.flatten(1).sum(1) + 1e-6 ) pred_scores = scores_per_image * mask_scores_per_image pred_classes = labels_per_image segmentation = torch.zeros((384, 384)) - 1 if target_sizes is not None: segmentation = torch.zeros(target_sizes[i]) - 1 pred_masks = torch.nn.functional.interpolate( pred_masks.unsqueeze(0), size=target_sizes[i], mode="nearest" )[0] instance_maps, segments = [], [] current_segment_id = 0 for j in range(num_queries): score = pred_scores[j].item() if not torch.all(pred_masks[j] == 0) and score >= threshold: segmentation[pred_masks[j] == 1] = current_segment_id segments.append( { "id": current_segment_id, "label_id": pred_classes[j].item(), "was_fused": False, "score": round(score, 6), } ) current_segment_id += 1 instance_maps.append(pred_masks[j]) # Return segmentation map in run-length encoding (RLE) format if return_coco_annotation: segmentation = convert_segmentation_to_rle(segmentation) # Return a concatenated tensor of binary instance maps if return_binary_maps and len(instance_maps) != 0: segmentation = torch.stack(instance_maps, dim=0) results.append({"segmentation": segmentation, "segments_info": segments}) return results def post_process_panoptic_segmentation( self, outputs, threshold: float = 0.5, mask_threshold: float = 0.5, overlap_mask_area_threshold: float = 0.8, label_ids_to_fuse: Optional[set[int]] = None, target_sizes: Optional[list[tuple[int, int]]] = None, ) -> list[dict]: """ Converts the output of [`Mask2FormerForUniversalSegmentationOutput`] into image panoptic segmentation predictions. Only supports PyTorch. Args: outputs ([`Mask2FormerForUniversalSegmentationOutput`]): The outputs from [`Mask2FormerForUniversalSegmentation`]. threshold (`float`, *optional*, defaults to 0.5): The probability score threshold to keep predicted instance masks. mask_threshold (`float`, *optional*, defaults to 0.5): Threshold to use when turning the predicted masks into binary values. overlap_mask_area_threshold (`float`, *optional*, defaults to 0.8): The overlap mask area threshold to merge or discard small disconnected parts within each binary instance mask. label_ids_to_fuse (`Set[int]`, *optional*): The labels in this state will have all their instances be fused together. For instance we could say there can only be one sky in an image, but several persons, so the label ID for sky would be in that set, but not the one for person. target_sizes (`List[Tuple]`, *optional*): List of length (batch_size), where each list item (`Tuple[int, int]]`) corresponds to the requested final size (height, width) of each prediction in batch. If left to None, predictions will not be resized. Returns: `List[Dict]`: A list of dictionaries, one per image, each dictionary containing two keys: - **segmentation** -- a tensor of shape `(height, width)` where each pixel represents a `segment_id`, set to `None` if no mask if found above `threshold`. If `target_sizes` is specified, segmentation is resized to the corresponding `target_sizes` entry. - **segments_info** -- A dictionary that contains additional information on each segment. - **id** -- an integer representing the `segment_id`. - **label_id** -- An integer representing the label / semantic class id corresponding to `segment_id`. - **was_fused** -- a boolean, `True` if `label_id` was in `label_ids_to_fuse`, `False` otherwise. Multiple instances of the same class / label were fused and assigned a single `segment_id`. - **score** -- Prediction score of segment with `segment_id`. """ if label_ids_to_fuse is None: logger.warning("`label_ids_to_fuse` unset. No instance will be fused.") label_ids_to_fuse = set() class_queries_logits = outputs.class_queries_logits # [batch_size, num_queries, num_classes+1] masks_queries_logits = outputs.masks_queries_logits # [batch_size, num_queries, height, width] # Scale back to preprocessed image size - (384, 384) for all models masks_queries_logits = torch.nn.functional.interpolate( masks_queries_logits, size=(384, 384), mode="bilinear", align_corners=False ) batch_size = class_queries_logits.shape[0] num_labels = class_queries_logits.shape[-1] - 1 mask_probs = masks_queries_logits.sigmoid() # [batch_size, num_queries, height, width] # Predicted label and score of each query (batch_size, num_queries) pred_scores, pred_labels = nn.functional.softmax(class_queries_logits, dim=-1).max(-1) # Loop over items in batch size results: list[dict[str, TensorType]] = [] for i in range(batch_size): mask_probs_item, pred_scores_item, pred_labels_item = remove_low_and_no_objects( mask_probs[i], pred_scores[i], pred_labels[i], threshold, num_labels ) # No mask found if mask_probs_item.shape[0] <= 0: height, width = target_sizes[i] if target_sizes is not None else mask_probs_item.shape[1:] segmentation = torch.zeros((height, width)) - 1 results.append({"segmentation": segmentation, "segments_info": []}) continue # Get segmentation map and segment information of batch item target_size = target_sizes[i] if target_sizes is not None else None segmentation, segments = compute_segments( mask_probs=mask_probs_item, pred_scores=pred_scores_item, pred_labels=pred_labels_item, mask_threshold=mask_threshold, overlap_mask_area_threshold=overlap_mask_area_threshold, label_ids_to_fuse=label_ids_to_fuse, target_size=target_size, ) results.append({"segmentation": segmentation, "segments_info": segments}) return results def post_process_segmentation(): raise NotImplementedError("Segmentation post-processing is not implemented for Mask2Former yet.") __all__ = ["Mask2FormerImageProcessorFast"]