
"""simple docstring"""


# Copyright 2021 The HuggingFace Team. All rights reserved.
#
# 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 pathlib import Path

import torch

from ...utils import is_npu_available, is_xpu_available
from .config_args import ClusterConfig, default_json_config_file
from .config_utils import SubcommandHelpFormatter


SCREAMING_SNAKE_CASE			  =  """Create a default config file for Accelerate with only a few flags set."""



def 						lowerCamelCase__      (				UpperCAmelCase_="no"		,   UpperCAmelCase_ = default_json_config_file		,   UpperCAmelCase_ = False					)->				str:







       """simple docstring"""
       UpperCamelCase       			=		Path(UpperCAmelCase_					)
       path.parent.mkdir(parents=UpperCAmelCase_		,   exist_ok=UpperCAmelCase_					)
       if path.exists():
              print(
                  F"Configuration already exists at {save_location}, will not override. Run `accelerate config` manually or pass a different `save_location`."					)
              return False
       UpperCamelCase       			=		mixed_precision.lower()
       if mixed_precision not in ["no", "fp16", "bf16", "fp8"]:
              raise ValueError(
                  F"`mixed_precision` should be one of 'no', 'fp16', 'bf16', or 'fp8'. Received {mixed_precision}"					)
       UpperCamelCase       			=		{
           "compute_environment": "LOCAL_MACHINE",
           "mixed_precision": mixed_precision,
       }
       if torch.cuda.is_available():
              UpperCamelCase       			=		torch.cuda.device_count()
              UpperCamelCase       			=		num_gpus
              UpperCamelCase       			=		False
              if num_gpus > 1:
                     UpperCamelCase       			=		"MULTI_GPU"
              else:
                     UpperCamelCase       			=		"NO"
       elif is_xpu_available() and use_xpu:
              UpperCamelCase       			=		torch.xpu.device_count()
              UpperCamelCase       			=		num_xpus
              UpperCamelCase       			=		False
              if num_xpus > 1:
                     UpperCamelCase       			=		"MULTI_XPU"
              else:
                     UpperCamelCase       			=		"NO"
       elif is_npu_available():
              UpperCamelCase       			=		torch.npu.device_count()
              UpperCamelCase       			=		num_npus
              UpperCamelCase       			=		False
              if num_npus > 1:
                     UpperCamelCase       			=		"MULTI_NPU"
              else:
                     UpperCamelCase       			=		"NO"
       else:
              UpperCamelCase       			=		0
              UpperCamelCase       			=		True
              UpperCamelCase       			=		1
              UpperCamelCase       			=		"NO"
       UpperCamelCase       			=		ClusterConfig(**UpperCAmelCase_					)
       config.to_json_file(UpperCAmelCase_					)
       return path



def 						lowerCamelCase__      (				UpperCAmelCase_		,   UpperCAmelCase_					)->				Any:







       """simple docstring"""
       UpperCamelCase       			=		parser.add_parser("default"		,   parents=UpperCAmelCase_		,   help=UpperCAmelCase_		,   formatter_class=UpperCAmelCase_					)
       parser.add_argument(
           "--config_file"		,   default=UpperCAmelCase_		,   help=(
               "The path to use to store the config file. Will default to a file named default_config.yaml in the cache "
               "location, which is the content of the environment `HF_HOME` suffixed with 'accelerate', or if you don't have "
               "such an environment variable, your cache directory ('~/.cache' or the content of `XDG_CACHE_HOME`) suffixed "
               "with 'huggingface'."
           )		,   dest="save_location"		,   )

       parser.add_argument(
           "--mixed_precision"		,   choices=["no", "fp16", "bf16"]		,   type=UpperCAmelCase_		,   help="Whether or not to use mixed precision training. "
           "Choose between FP16 and BF16 (bfloat16) training. "
           "BF16 training is only supported on Nvidia Ampere GPUs and PyTorch 1.10 or later."		,   default="no"		,   )
       parser.set_defaults(func=UpperCAmelCase_					)
       return parser



def 						lowerCamelCase__      (				UpperCAmelCase_					)->				Dict:







       """simple docstring"""
       UpperCamelCase       			=		write_basic_config(args.mixed_precision		,   args.save_location					)
       if config_file:
              print(F"accelerate configuration saved at {config_file}"					)







"""simple docstring"""


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

SCREAMING_SNAKE_CASE			  =  logging.get_logger(__name__)



@add_end_docstrings(_lowerCAmelCase						)
class 	__a	(   _lowerCAmelCase						):






       def __init__(	self			:				str							,   **UpperCAmelCase_			:				List[str]      )->	Tuple:







              """simple docstring"""
              super().__init__(**UpperCAmelCase_      )
              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(UpperCAmelCase_      )






       def 						_SCREAMING_SNAKE_CASE							(	self			:				Tuple							,   **UpperCAmelCase_			:				int      )->	List[str]:







              """simple docstring"""
              UpperCamelCase       			=		{}
              UpperCamelCase       			=		{}
              UpperCamelCase       			=		{}
              # preprocess args
              if "points_per_batch" in kwargs:
                     UpperCamelCase       			=		kwargs["points_per_batch"]
              if "points_per_crop" in kwargs:
                     UpperCamelCase       			=		kwargs["points_per_crop"]
              if "crops_n_layers" in kwargs:
                     UpperCamelCase       			=		kwargs["crops_n_layers"]
              if "crop_overlap_ratio" in kwargs:
                     UpperCamelCase       			=		kwargs["crop_overlap_ratio"]
              if "crop_n_points_downscale_factor" in kwargs:
                     UpperCamelCase       			=		kwargs["crop_n_points_downscale_factor"]
              # postprocess args
              if "pred_iou_thresh" in kwargs:
                     UpperCamelCase       			=		kwargs["pred_iou_thresh"]
              if "stability_score_offset" in kwargs:
                     UpperCamelCase       			=		kwargs["stability_score_offset"]
              if "mask_threshold" in kwargs:
                     UpperCamelCase       			=		kwargs["mask_threshold"]
              if "stability_score_thresh" in kwargs:
                     UpperCamelCase       			=		kwargs["stability_score_thresh"]
              if "crops_nms_thresh" in kwargs:
                     UpperCamelCase       			=		kwargs["crops_nms_thresh"]
              if "output_rle_mask" in kwargs:
                     UpperCamelCase       			=		kwargs["output_rle_mask"]
              if "output_bboxes_mask" in kwargs:
                     UpperCamelCase       			=		kwargs["output_bboxes_mask"]
              return preprocess_kwargs, forward_params, postprocess_kwargs






       def __call__(	self			:				Optional[int]							,   UpperCAmelCase_			:				Optional[int]							,   *UpperCAmelCase_			:				Tuple							,   UpperCAmelCase_			:				int=None							,   UpperCAmelCase_			:				Tuple=None							,   **UpperCAmelCase_			:				str      )->	Any:







              """simple docstring"""
              return super().__call__(UpperCAmelCase_							,   *UpperCAmelCase_							,   num_workers=UpperCAmelCase_							,   batch_size=UpperCAmelCase_							,   **UpperCAmelCase_      )






       def 						_SCREAMING_SNAKE_CASE							(	self			:				Optional[int]							,   UpperCAmelCase_			:				Optional[int]							,   UpperCAmelCase_			:				str=64							,   UpperCAmelCase_			:				int = 0							,   UpperCAmelCase_			:				float = 512 / 1_500							,   UpperCAmelCase_			:				Optional[int] = 32							,   UpperCAmelCase_			:				Optional[int] = 1							,   )->	List[Any]:







              """simple docstring"""
              UpperCamelCase       			=		load_image(UpperCAmelCase_      )
              UpperCamelCase       			=		self.image_processor.size["longest_edge"]
              UpperCamelCase			,							UpperCamelCase			,							UpperCamelCase			,							UpperCamelCase       			=		self.image_processor.generate_crop_boxes(
                  UpperCAmelCase_							,   UpperCAmelCase_							,   UpperCAmelCase_							,   UpperCAmelCase_							,   UpperCAmelCase_							,   UpperCAmelCase_      )
              UpperCamelCase       			=		self.image_processor(images=UpperCAmelCase_							,   return_tensors="pt"      )

              with self.device_placement():
                     if self.framework == "pt":
                            UpperCamelCase       			=		self.get_inference_context()
                            with inference_context():
                                   UpperCamelCase       			=		self._ensure_tensor_on_device(UpperCAmelCase_							,   device=self.device      )
                                   UpperCamelCase       			=		self.model.get_image_embeddings(model_inputs.pop("pixel_values"      )      )
                                   UpperCamelCase       			=		image_embeddings

              UpperCamelCase       			=		grid_points.shape[1]
              UpperCamelCase       			=		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							,   UpperCAmelCase_							,   UpperCAmelCase_      ):
                     UpperCamelCase       			=		grid_points[:, i : i + points_per_batch, :, :]
                     UpperCamelCase       			=		input_labels[:, i : i + points_per_batch]
                     UpperCamelCase       			=		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			:				Union[str, Any]							,   UpperCAmelCase_			:				Union[str, Any]							,   UpperCAmelCase_			:				int=0.88							,   UpperCAmelCase_			:				Optional[int]=0.95							,   UpperCAmelCase_			:				Dict=0							,   UpperCAmelCase_			:				int=1							,   )->	Any:







              """simple docstring"""
              UpperCamelCase       			=		model_inputs.pop("input_boxes"      )
              UpperCamelCase       			=		model_inputs.pop("is_last"      )
              UpperCamelCase       			=		model_inputs.pop("original_sizes"      ).tolist()
              UpperCamelCase       			=		model_inputs.pop("reshaped_input_sizes"      ).tolist()

              UpperCamelCase       			=		self.model(**UpperCAmelCase_      )

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






       def 						_SCREAMING_SNAKE_CASE							(	self			:				Any							,   UpperCAmelCase_			:				Tuple							,   UpperCAmelCase_			:				List[str]=False							,   UpperCAmelCase_			:				str=False							,   UpperCAmelCase_			:				Optional[int]=0.7							,   )->	Dict:







              """simple docstring"""
              UpperCamelCase       			=		[]
              UpperCamelCase       			=		[]
              UpperCamelCase       			=		[]
              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"      )      )

              UpperCamelCase       			=		torch.cat(UpperCAmelCase_      )
              UpperCamelCase       			=		torch.cat(UpperCAmelCase_      )
              UpperCamelCase			,							UpperCamelCase			,							UpperCamelCase			,							UpperCamelCase       			=		self.image_processor.post_process_for_mask_generation(
                  UpperCAmelCase_							,   UpperCAmelCase_							,   UpperCAmelCase_							,   UpperCAmelCase_      )

              UpperCamelCase       			=		defaultdict(UpperCAmelCase_      )
              for output in model_outputs:
                     for k, v in output.items():
                            extra[k].append(UpperCAmelCase_      )

              UpperCamelCase       			=		{}
              if output_rle_mask:
                     UpperCamelCase       			=		rle_mask

              if output_bboxes_mask:
                     UpperCamelCase       			=		bounding_boxes

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





