




import copy
from typing import Dict, Optional

from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ..auto import CONFIG_MAPPING
from ..detr import DetrConfig
from ..swin import SwinConfig


lowerCamelCase	:	Optional[Any]				=   {
    'facebook/maskformer-swin-base-ade': (
        'https://huggingface.co/facebook/maskformer-swin-base-ade/blob/main/config.json'
    )
    # See all MaskFormer models at https://huggingface.co/models?filter=maskformer
}

lowerCamelCase	:	Union[str, Any]				=   logging.get_logger(__name__)




class 						__lowercase		(UpperCamelCase__							):




      """simple docstring"""





      _snake_case												=						"""maskformer"""
      _snake_case												=						{"""hidden_size""": """mask_feature_size"""}
      _snake_case												=						["""resnet""", """swin"""]
      _snake_case												=						["""detr"""]

      def __init__(				self    ,				A = 2_5_6    ,				A = 2_5_6    ,				A = 0.1    ,				A = False    ,				A = None    ,				A = None    ,				A = 0.02    ,				A = 1.0    ,				A = 1.0    ,				A = 1.0    ,				A = 20.0    ,				A = None    ,				**A    ,				)		->    int:
         if backbone_config is None:
            # fall back to https://huggingface.co/microsoft/swin-base-patch4-window12-384-in22k
            snake_case				:    Any           =	SwinConfig(
                image_size=3_8_4    ,				in_channels=3    ,				patch_size=4    ,				embed_dim=1_2_8    ,				depths=[2, 2, 1_8, 2]    ,				num_heads=[4, 8, 1_6, 3_2]    ,				window_size=1_2    ,				drop_path_rate=0.3    ,				out_features=["""stage1""", """stage2""", """stage3""", """stage4"""]    ,				)

         if isinstance(A    ,				A     ):
            snake_case				:    List[Any]           =	backbone_config.pop("""model_type"""     )
            snake_case				:    Optional[int]           =	CONFIG_MAPPING[backbone_model_type]
            snake_case				:    Any           =	config_class.from_dict(A     )

         # verify that the backbone is supported
         if backbone_config.model_type not in self.backbones_supported:
            logger.warning_once(
                f"""Backbone {backbone_config.model_type} is not a supported model and may not be compatible with MaskFormer. """
                f"""Supported model types: {",".join(self.backbones_supported     )}"""     )

         if decoder_config is None:
            # fall back to https://huggingface.co/facebook/detr-resnet-50
            snake_case				:    Tuple           =	DetrConfig()
         else:
            # verify that the decoder is supported
            snake_case				:    Union[str, Any]           =	(
                decoder_config.pop("""model_type"""     ) if isinstance(A    ,				A     ) else decoder_config.model_type
            )
            if decoder_type not in self.decoders_supported:
               raise ValueError(
                   f"""Transformer Decoder {decoder_type} not supported, please use one of"""
                   f""" {",".join(self.decoders_supported     )}"""     )
            if isinstance(A    ,				A     ):
               snake_case				:    List[str]           =	CONFIG_MAPPING[decoder_type]
               snake_case				:    Optional[Any]           =	config_class.from_dict(A     )

         snake_case				:    List[Any]           =	backbone_config
         snake_case				:    Tuple           =	decoder_config
         # main feature dimension for the model
         snake_case				:    str           =	fpn_feature_size
         snake_case				:    Optional[Any]           =	mask_feature_size
         # initializer
         snake_case				:    Union[str, Any]           =	init_std
         snake_case				:    Optional[int]           =	init_xavier_std
         # Hungarian matcher && loss
         snake_case				:    int           =	cross_entropy_weight
         snake_case				:    Union[str, Any]           =	dice_weight
         snake_case				:    int           =	mask_weight
         snake_case				:    Dict           =	use_auxiliary_loss
         snake_case				:    List[Any]           =	no_object_weight
         snake_case				:    str           =	output_auxiliary_logits

         snake_case				:    List[Any]           =	self.decoder_config.encoder_attention_heads
         snake_case				:    Tuple           =	self.decoder_config.num_hidden_layers
         super().__init__(**A     )

      @classmethod
      def      UpperCAmelCase				(				cls    ,				A    ,				A    ,				**A     )		->    Union[str, Any]:
         return cls(
             backbone_config=A    ,				decoder_config=A    ,				**A    ,				)

      def      UpperCAmelCase				(				self     )		->    Dict[str, any]:
         snake_case				:    Union[str, Any]           =	copy.deepcopy(self.__dict__     )
         snake_case				:    int           =	self.backbone_config.to_dict()
         snake_case				:    Optional[Any]           =	self.decoder_config.to_dict()
         snake_case				:    List[str]           =	self.__class__.model_type
         return output











from io import BytesIO
from typing import List, Union

import requests

from ..utils import add_end_docstrings, is_decord_available, is_torch_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_decord_available():
       import numpy as np
       from decord import VideoReader


if is_torch_available():
       from ..models.auto.modeling_auto import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING

lowerCamelCase	:	int				=   logging.get_logger(__name__)




@add_end_docstrings(UpperCamelCase__							)
class 						__lowercase		(UpperCamelCase__							):




      """simple docstring"""





      def __init__(				self    ,				*A    ,				**A     )		->    Optional[int]:
         super().__init__(*A    ,				**A     )
         requires_backends(self    ,				"""decord"""     )
         self.check_model_type(A     )

      def      UpperCAmelCase				(				self    ,				A=None    ,				A=None    ,				A=None     )		->    int:
         snake_case				:    Any           =	{}
         if frame_sampling_rate is not None:
            snake_case				:    int           =	frame_sampling_rate
         if num_frames is not None:
            snake_case				:    Union[str, Any]           =	num_frames

         snake_case				:    str           =	{}
         if top_k is not None:
            snake_case				:    Optional[int]           =	top_k
         return preprocess_params, {}, postprocess_params

      def __call__(				self    ,				A    ,				**A     )		->    Dict:
         return super().__call__(A    ,				**A     )

      def      UpperCAmelCase				(				self    ,				A    ,				A=None    ,				A=1     )		->    Tuple:
         if num_frames is None:
            snake_case				:    Tuple           =	self.model.config.num_frames

         if video.startswith("""http://"""     ) or video.startswith("""https://"""     ):
            snake_case				:    Optional[int]           =	BytesIO(requests.get(A     ).content     )

         snake_case				:    Optional[Any]           =	VideoReader(A     )
         videoreader.seek(0     )

         snake_case				:    Optional[Any]           =	0
         snake_case				:    Optional[Any]           =	num_frames * frame_sampling_rate - 1
         snake_case				:    Any           =	np.linspace(A    ,				A    ,				num=A    ,				dtype=np.intaa     )

         snake_case				:    int           =	videoreader.get_batch(A     ).asnumpy()
         snake_case				:    List[Any]           =	list(A     )

         snake_case				:    Optional[Any]           =	self.image_processor(A    ,				return_tensors=self.framework     )
         return model_inputs

      def      UpperCAmelCase				(				self    ,				A     )		->    List[str]:
         snake_case				:    Dict           =	self.model(**A     )
         return model_outputs

      def      UpperCAmelCase				(				self    ,				A    ,				A=5     )		->    int:
         if top_k > self.model.config.num_labels:
            snake_case				:    str           =	self.model.config.num_labels

         if self.framework == "pt":
            snake_case				:    List[Any]           =	model_outputs.logits.softmax(-1     )[0]
            snake_case     , snake_case				:    Tuple           =	probs.topk(A     )
         else:
            raise ValueError(f"""Unsupported framework: {self.framework}"""     )

         snake_case				:    List[Any]           =	scores.tolist()
         snake_case				:    str           =	ids.tolist()
         return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(A    ,				A     )]





