


from collections import defaultdict
from typing import Optional

from ..image_utils import load_image
from ..utils import (
    add_end_docstrings,
    is_torch_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, ChunkPipeline


if is_torch_available():
   import torch

   from ..models.auto.modeling_auto import MODEL_FOR_MASK_GENERATION_MAPPING

UpperCAmelCase_				:			List[Any]   =	logging.get_logger(__name__)
@add_end_docstrings(lowercase__      )
class 			SCREAMING_SNAKE_CASE__  (    lowercase__      ):
      def __init__(	self    :	List[str]					,							**SCREAMING_SNAKE_CASE__    :	Any					)		->      Union[str, Any]:
             super().__init__(**SCREAMING_SNAKE_CASE__					)
             requires_backends(self					,							'vision'					)
             requires_backends(self					,							'torch'					)

             if self.framework != "pt":
                    raise ValueError(F"""The {self.__class__} is only available in PyTorch."""					)

             self.check_model_type(SCREAMING_SNAKE_CASE__					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	str					,							**SCREAMING_SNAKE_CASE__    :	Any					)		->      int:
             a_ :		Optional[Any]							=  {}
             a_ :		Optional[int]							=  {}
             a_ :		Optional[int]							=  {}
             # preprocess args
             if "points_per_batch" in kwargs:
                    a_ :		List[Any]							=  kwargs['points_per_batch']
             if "points_per_crop" in kwargs:
                    a_ :		str							=  kwargs['points_per_crop']
             if "crops_n_layers" in kwargs:
                    a_ :		Optional[int]							=  kwargs['crops_n_layers']
             if "crop_overlap_ratio" in kwargs:
                    a_ :		Optional[Any]							=  kwargs['crop_overlap_ratio']
             if "crop_n_points_downscale_factor" in kwargs:
                    a_ :		Optional[Any]							=  kwargs['crop_n_points_downscale_factor']
             # postprocess args
             if "pred_iou_thresh" in kwargs:
                    a_ :		str							=  kwargs['pred_iou_thresh']
             if "stability_score_offset" in kwargs:
                    a_ :		int							=  kwargs['stability_score_offset']
             if "mask_threshold" in kwargs:
                    a_ :		Dict							=  kwargs['mask_threshold']
             if "stability_score_thresh" in kwargs:
                    a_ :		Dict							=  kwargs['stability_score_thresh']
             if "crops_nms_thresh" in kwargs:
                    a_ :		Any							=  kwargs['crops_nms_thresh']
             if "output_rle_mask" in kwargs:
                    a_ :		Optional[Any]							=  kwargs['output_rle_mask']
             if "output_bboxes_mask" in kwargs:
                    a_ :		str							=  kwargs['output_bboxes_mask']
             return preprocess_kwargs, forward_params, postprocess_kwargs
      def __call__(	self    :	Optional[Any]					,							SCREAMING_SNAKE_CASE__    :	int					,							*SCREAMING_SNAKE_CASE__    :	Union[str, Any]					,							SCREAMING_SNAKE_CASE__    :	List[Any]=None					,							SCREAMING_SNAKE_CASE__    :	List[Any]=None					,							**SCREAMING_SNAKE_CASE__    :	str					)		->      List[str]:
             return super().__call__(SCREAMING_SNAKE_CASE__					,							*SCREAMING_SNAKE_CASE__					,							num_workers=SCREAMING_SNAKE_CASE__					,							batch_size=SCREAMING_SNAKE_CASE__					,							**SCREAMING_SNAKE_CASE__					)
      def 	SCREAMING_SNAKE_CASE							(	self    :	Union[str, Any]					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]					,							SCREAMING_SNAKE_CASE__    :	Optional[Any]=6_4					,							SCREAMING_SNAKE_CASE__    :	int = 0					,							SCREAMING_SNAKE_CASE__    :	float = 5_1_2 / 1_5_0_0					,							SCREAMING_SNAKE_CASE__    :	Optional[int] = 3_2					,							SCREAMING_SNAKE_CASE__    :	Optional[int] = 1					,							)		->      Union[str, Any]:
             a_ :		Union[str, Any]							=  load_image(SCREAMING_SNAKE_CASE__					)
             a_ :		Tuple							=  self.image_processor.size['longest_edge']
             a_							,		a_							,		a_							,		a_ :		Dict							=  self.image_processor.generate_crop_boxes(
                 SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					)
             a_ :		str							=  self.image_processor(images=SCREAMING_SNAKE_CASE__					,							return_tensors='pt'					)

             with self.device_placement():
                    if self.framework == "pt":
                           a_ :		str							=  self.get_inference_context()
                           with inference_context():
                                  a_ :		Tuple							=  self._ensure_tensor_on_device(SCREAMING_SNAKE_CASE__					,							device=self.device					)
                                  a_ :		List[Any]							=  self.model.get_image_embeddings(model_inputs.pop('pixel_values'					)					)
                                  a_ :		Union[str, Any]							=  image_embeddings

             a_ :		List[str]							=  grid_points.shape[1]
             a_ :		Dict							=  points_per_batch if points_per_batch is not None else n_points

             if points_per_batch <= 0:
                    raise ValueError(
                        'Cannot have points_per_batch<=0. Must be >=1 to returned batched outputs. '
                        'To return all points at once, set points_per_batch to None'					)

             for i in range(0					,							SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					):
                    a_ :		Union[str, Any]							=  grid_points[:, i : i + points_per_batch, :, :]
                    a_ :		Optional[Any]							=  input_labels[:, i : i + points_per_batch]
                    a_ :		Union[str, Any]							=  i == n_points - points_per_batch
                    yield {
                        "input_points": batched_points,
                        "input_labels": labels,
                        "input_boxes": crop_boxes,
                        "is_last": is_last,
                        **model_inputs,
                    }
      def 	SCREAMING_SNAKE_CASE							(	self    :	Tuple					,							SCREAMING_SNAKE_CASE__    :	Dict					,							SCREAMING_SNAKE_CASE__    :	Optional[int]=0.88					,							SCREAMING_SNAKE_CASE__    :	Union[str, Any]=0.95					,							SCREAMING_SNAKE_CASE__    :	Optional[int]=0					,							SCREAMING_SNAKE_CASE__    :	Tuple=1					,							)		->      List[Any]:
             a_ :		int							=  model_inputs.pop('input_boxes'					)
             a_ :		List[str]							=  model_inputs.pop('is_last'					)
             a_ :		Optional[int]							=  model_inputs.pop('original_sizes'					).tolist()
             a_ :		Optional[int]							=  model_inputs.pop('reshaped_input_sizes'					).tolist()

             a_ :		int							=  self.model(**SCREAMING_SNAKE_CASE__					)

             # post processing happens here in order to avoid CPU GPU copies of ALL the masks
             a_ :		List[str]							=  model_outputs['pred_masks']
             a_ :		Union[str, Any]							=  self.image_processor.post_process_masks(
                 SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					,							binarize=SCREAMING_SNAKE_CASE__					)
             a_ :		List[Any]							=  model_outputs['iou_scores']
             a_							,		a_							,		a_ :		Any							=  self.image_processor.filter_masks(
                 masks[0]					,							iou_scores[0]					,							original_sizes[0]					,							input_boxes[0]					,							SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					,							)
             return {
                 "masks": masks,
                 "is_last": is_last,
                 "boxes": boxes,
                 "iou_scores": iou_scores,
             }

      def 	SCREAMING_SNAKE_CASE							(	self    :	Tuple					,							SCREAMING_SNAKE_CASE__    :	List[str]					,							SCREAMING_SNAKE_CASE__    :	Optional[Any]=False					,							SCREAMING_SNAKE_CASE__    :	str=False					,							SCREAMING_SNAKE_CASE__    :	Tuple=0.7					,							)		->      Optional[Any]:
             a_ :		List[str]							=  []
             a_ :		Optional[int]							=  []
             a_ :		Any							=  []
             for model_output in model_outputs:
                    all_scores.append(model_output.pop('iou_scores'					)					)
                    all_masks.extend(model_output.pop('masks'					)					)
                    all_boxes.append(model_output.pop('boxes'					)					)

             a_ :		int							=  torch.cat(SCREAMING_SNAKE_CASE__					)
             a_ :		Optional[int]							=  torch.cat(SCREAMING_SNAKE_CASE__					)
             a_							,		a_							,		a_							,		a_ :		List[Any]							=  self.image_processor.post_process_for_mask_generation(
                 SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					,							SCREAMING_SNAKE_CASE__					)

             a_ :		Optional[Any]							=  defaultdict(SCREAMING_SNAKE_CASE__					)
             for output in model_outputs:
                    for k, v in output.items():
                           extra[k].append(SCREAMING_SNAKE_CASE__					)

             a_ :		Any							=  {}
             if output_rle_mask:
                    a_ :		Tuple							=  rle_mask

             if output_bboxes_mask:
                    a_ :		List[str]							=  bounding_boxes

             return {"masks": output_masks, "scores": iou_scores, **optional, **extra}






import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch
class 			SCREAMING_SNAKE_CASE__  (    lowercase__      ):
      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	Optional[Any]					)		->      List[Any]:
             # this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
             # `transformers` is loaded, and it's too late for inside pytest - so we are changing it
             # while running an external program

             # python one-liner segments

             # this must be loaded before socket.socket is monkey-patched
             a_ :		str							=  '\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

             a_ :		Any							=  '\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

             a_ :		Dict							=  '\nimport socket\ndef offline_socket(*args, **kwargs): raise RuntimeError("Offline mode is enabled, we shouldn\'t access internet")\nsocket.socket = offline_socket\n        '

             # Force fetching the files so that we can use the cache
             a_ :		Dict							=  'hf-internal-testing/tiny-random-bert'
             BertConfig.from_pretrained(SCREAMING_SNAKE_CASE__					)
             BertModel.from_pretrained(SCREAMING_SNAKE_CASE__					)
             BertTokenizer.from_pretrained(SCREAMING_SNAKE_CASE__					)
             pipeline(task='fill-mask'					,							model=SCREAMING_SNAKE_CASE__					)

             # baseline - just load from_pretrained with normal network
             a_ :		Union[str, Any]							=  [sys.executable, '-c', '\n'.join([load, run, mock]					)]

             # should succeed
             a_ :		List[str]							=  self.get_env()
             # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
             a_ :		Any							=  '1'
             a_ :		Optional[Any]							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)
      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	str					)		->      List[Any]:
             # python one-liner segments
             # this must be loaded before socket.socket is monkey-patched
             a_ :		Tuple							=  '\nfrom transformers import BertConfig, BertModel, BertTokenizer, pipeline\n        '

             a_ :		str							=  '\nmname = "hf-internal-testing/tiny-random-bert"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nBertTokenizer.from_pretrained(mname)\npipe = pipeline(task="fill-mask", model=mname)\nprint("success")\n        '

             a_ :		Any							=  '\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Faking flaky internet")\nsocket.socket = offline_socket\n        '

             # Force fetching the files so that we can use the cache
             a_ :		Tuple							=  'hf-internal-testing/tiny-random-bert'
             BertConfig.from_pretrained(SCREAMING_SNAKE_CASE__					)
             BertModel.from_pretrained(SCREAMING_SNAKE_CASE__					)
             BertTokenizer.from_pretrained(SCREAMING_SNAKE_CASE__					)
             pipeline(task='fill-mask'					,							model=SCREAMING_SNAKE_CASE__					)

             # baseline - just load from_pretrained with normal network
             a_ :		Optional[Any]							=  [sys.executable, '-c', '\n'.join([load, run, mock]					)]

             # should succeed
             a_ :		Dict							=  self.get_env()
             a_ :		Dict							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)
      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	int					)		->      Union[str, Any]:
             # this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
             # `transformers` is loaded, and it's too late for inside pytest - so we are changing it
             # while running an external program

             # python one-liner segments

             # this must be loaded before socket.socket is monkey-patched
             a_ :		Any							=  '\nfrom transformers import BertConfig, BertModel, BertTokenizer\n        '

             a_ :		Dict							=  '\nmname = "hf-internal-testing/tiny-random-bert-sharded"\nBertConfig.from_pretrained(mname)\nBertModel.from_pretrained(mname)\nprint("success")\n        '

             a_ :		Any							=  '\nimport socket\ndef offline_socket(*args, **kwargs): raise ValueError("Offline mode is enabled")\nsocket.socket = offline_socket\n        '

             # baseline - just load from_pretrained with normal network
             a_ :		Any							=  [sys.executable, '-c', '\n'.join([load, run]					)]

             # should succeed
             a_ :		Tuple							=  self.get_env()
             a_ :		int							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)

             # next emulate no network
             a_ :		str							=  [sys.executable, '-c', '\n'.join([load, mock, run]					)]

             # Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
             # env["TRANSFORMERS_OFFLINE"] = "0"
             # result = subprocess.run(cmd, env=env, check=False, capture_output=True)
             # self.assertEqual(result.returncode, 1, result.stderr)

             # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
             a_ :		Dict							=  '1'
             a_ :		int							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)
      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	Optional[int]					)		->      Union[str, Any]:
             a_ :		Union[str, Any]							=  '\nfrom transformers import pipeline\n        '
             a_ :		Optional[Any]							=  '\nmname = "hf-internal-testing/tiny-random-bert"\npipe = pipeline(model=mname)\n        '

             a_ :		Any							=  '\nimport socket\ndef offline_socket(*args, **kwargs): raise socket.error("Offline mode is enabled")\nsocket.socket = offline_socket\n        '
             a_ :		Union[str, Any]							=  self.get_env()
             a_ :		Optional[Any]							=  '1'
             a_ :		int							=  [sys.executable, '-c', '\n'.join([load, mock, run]					)]
             a_ :		Union[str, Any]							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							1					,							result.stderr					)
             self.assertIn(
                 'You cannot infer task automatically within `pipeline` when using offline mode'					,							result.stderr.decode().replace('\n'					,							''					)					,							)

      @require_torch
      def 	SCREAMING_SNAKE_CASE							(	self    :	int					)		->      Dict:
             a_ :		Optional[int]							=  '\nfrom transformers import AutoModel\n        '
             a_ :		Dict							=  '\nmname = "hf-internal-testing/test_dynamic_model"\nAutoModel.from_pretrained(mname, trust_remote_code=True)\nprint("success")\n        '

             # baseline - just load from_pretrained with normal network
             a_ :		Tuple							=  [sys.executable, '-c', '\n'.join([load, run]					)]

             # should succeed
             a_ :		Optional[Any]							=  self.get_env()
             a_ :		Dict							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)

             # should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
             a_ :		Optional[int]							=  '1'
             a_ :		Optional[Any]							=  subprocess.run(SCREAMING_SNAKE_CASE__					,							env=SCREAMING_SNAKE_CASE__					,							check=SCREAMING_SNAKE_CASE__					,							capture_output=SCREAMING_SNAKE_CASE__					)
             self.assertEqual(result.returncode					,							0					,							result.stderr					)
             self.assertIn('success'					,							result.stdout.decode()					)


