


'''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

__A					:  Dict											=      logging.get_logger(__name__)






@add_end_docstrings(_SCREAMING_SNAKE_CASE)
class    __snake_case       (   _SCREAMING_SNAKE_CASE):







    """simple docstring"""



    def __init__(   self    :  Union[str, Any]    ,						**lowerCamelCase    :  str		)   ->  Any:
      super().__init__(**lowerCamelCase		)
      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(lowerCamelCase		)

    def __lowercase   (   self    :  Union[str, Any]    ,						**lowerCamelCase    :  Optional[int]		)   ->  Dict:
      lowerCAmelCase_			:      Dict			=   {}
      lowerCAmelCase_			:      Dict			=   {}
      lowerCAmelCase_			:      Optional[int]			=   {}
      # preprocess args
      if "points_per_batch" in kwargs:
        lowerCAmelCase_			:      Optional[int]			=   kwargs["""points_per_batch"""]
      if "points_per_crop" in kwargs:
        lowerCAmelCase_			:      List[Any]			=   kwargs["""points_per_crop"""]
      if "crops_n_layers" in kwargs:
        lowerCAmelCase_			:      Dict			=   kwargs["""crops_n_layers"""]
      if "crop_overlap_ratio" in kwargs:
        lowerCAmelCase_			:      str			=   kwargs["""crop_overlap_ratio"""]
      if "crop_n_points_downscale_factor" in kwargs:
        lowerCAmelCase_			:      Union[str, Any]			=   kwargs["""crop_n_points_downscale_factor"""]
      # postprocess args
      if "pred_iou_thresh" in kwargs:
        lowerCAmelCase_			:      Union[str, Any]			=   kwargs["""pred_iou_thresh"""]
      if "stability_score_offset" in kwargs:
        lowerCAmelCase_			:      List[str]			=   kwargs["""stability_score_offset"""]
      if "mask_threshold" in kwargs:
        lowerCAmelCase_			:      Dict			=   kwargs["""mask_threshold"""]
      if "stability_score_thresh" in kwargs:
        lowerCAmelCase_			:      List[Any]			=   kwargs["""stability_score_thresh"""]
      if "crops_nms_thresh" in kwargs:
        lowerCAmelCase_			:      List[Any]			=   kwargs["""crops_nms_thresh"""]
      if "output_rle_mask" in kwargs:
        lowerCAmelCase_			:      int			=   kwargs["""output_rle_mask"""]
      if "output_bboxes_mask" in kwargs:
        lowerCAmelCase_			:      List[str]			=   kwargs["""output_bboxes_mask"""]
      return preprocess_kwargs, forward_params, postprocess_kwargs

    def __call__(   self    :  List[Any]    ,						lowerCamelCase    :  int    ,						*lowerCamelCase    :  Tuple    ,						lowerCamelCase    :  Optional[Any]=None    ,						lowerCamelCase    :  int=None    ,						**lowerCamelCase    :  Dict		)   ->  Optional[int]:
      return super().__call__(lowerCamelCase    ,						*lowerCamelCase    ,						num_workers=lowerCamelCase    ,						batch_size=lowerCamelCase    ,						**lowerCamelCase		)

    def __lowercase   (   self    :  List[str]    ,						lowerCamelCase    :  Tuple    ,						lowerCamelCase    :  Dict=64    ,						lowerCamelCase    :  int = 0    ,						lowerCamelCase    :  float = 5_12 / 15_00    ,						lowerCamelCase    :  Optional[int] = 32    ,						lowerCamelCase    :  Optional[int] = 1    ,						)   ->  List[Any]:
      lowerCAmelCase_			:      int			=   load_image(lowerCamelCase		)
      lowerCAmelCase_			:      Any			=   self.image_processor.size["""longest_edge"""]
      lowerCAmelCase_,							lowerCAmelCase_,							lowerCAmelCase_,							lowerCAmelCase_			:      Dict			=   self.image_processor.generate_crop_boxes(
          lowerCamelCase    ,						lowerCamelCase    ,						lowerCamelCase    ,						lowerCamelCase    ,						lowerCamelCase    ,						lowerCamelCase		)
      lowerCAmelCase_			:      int			=   self.image_processor(images=lowerCamelCase    ,						return_tensors="""pt"""		)

      with self.device_placement():
        if self.framework == "pt":
          lowerCAmelCase_			:      List[Any]			=   self.get_inference_context()
          with inference_context():
            lowerCAmelCase_			:      List[str]			=   self._ensure_tensor_on_device(lowerCamelCase    ,						device=self.device		)
            lowerCAmelCase_			:      int			=   self.model.get_image_embeddings(model_inputs.pop("""pixel_values"""		)		)
            lowerCAmelCase_			:      Optional[Any]			=   image_embeddings

      lowerCAmelCase_			:      Optional[Any]			=   grid_points.shape[1]
      lowerCAmelCase_			:      Optional[int]			=   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    ,						lowerCamelCase    ,						lowerCamelCase		):
        lowerCAmelCase_			:      List[Any]			=   grid_points[:, i : i + points_per_batch, :, :]
        lowerCAmelCase_			:      List[str]			=   input_labels[:, i : i + points_per_batch]
        lowerCAmelCase_			:      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 __lowercase   (   self    :  Tuple    ,						lowerCamelCase    :  Optional[Any]    ,						lowerCamelCase    :  Optional[int]=0.88    ,						lowerCamelCase    :  int=0.95    ,						lowerCamelCase    :  str=0    ,						lowerCamelCase    :  List[str]=1    ,						)   ->  List[str]:
      lowerCAmelCase_			:      Any			=   model_inputs.pop("""input_boxes"""		)
      lowerCAmelCase_			:      Any			=   model_inputs.pop("""is_last"""		)
      lowerCAmelCase_			:      int			=   model_inputs.pop("""original_sizes"""		).tolist()
      lowerCAmelCase_			:      List[Any]			=   model_inputs.pop("""reshaped_input_sizes"""		).tolist()

      lowerCAmelCase_			:      Any			=   self.model(**lowerCamelCase		)

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





    def __lowercase   (   self    :  Union[str, Any]    ,						lowerCamelCase    :  Union[str, Any]    ,						lowerCamelCase    :  Optional[Any]=False    ,						lowerCamelCase    :  int=False    ,						lowerCamelCase    :  Tuple=0.7    ,						)   ->  Tuple:
      lowerCAmelCase_			:      List[str]			=   []
      lowerCAmelCase_			:      Optional[int]			=   []
      lowerCAmelCase_			:      str			=   []
      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"""		)		)

      lowerCAmelCase_			:      Union[str, Any]			=   torch.cat(lowerCamelCase		)
      lowerCAmelCase_			:      Union[str, Any]			=   torch.cat(lowerCamelCase		)
      lowerCAmelCase_,							lowerCAmelCase_,							lowerCAmelCase_,							lowerCAmelCase_			:      Dict			=   self.image_processor.post_process_for_mask_generation(
          lowerCamelCase    ,						lowerCamelCase    ,						lowerCamelCase    ,						lowerCamelCase		)

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

      lowerCAmelCase_			:      List[Any]			=   {}
      if output_rle_mask:
        lowerCAmelCase_			:      List[Any]			=   rle_mask

      if output_bboxes_mask:
        lowerCAmelCase_			:      Any			=   bounding_boxes

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







'''simple docstring'''






import json
import os
import shutil
import warnings
from argparse import ArgumentParser, Namespace
from pathlib import Path
from typing import List

from ..utils import logging
from . import BaseTransformersCLICommand


try:
  from cookiecutter.main import cookiecutter

  __A					:  List[Any]											=      True
except ImportError:
  __A					:  int											=      False

__A					:  str											=      logging.get_logger(__name__)  # pylint: disable=invalid-name







def 				UpperCamelCase_			(    A__					:   Namespace					):
  '''simple docstring'''





  return AddNewModelCommand(args.testing							,    args.testing_file							,    path=args.path					)






class    __snake_case       (   _SCREAMING_SNAKE_CASE):







    """simple docstring"""



    @staticmethod
    def __lowercase   (   lowerCamelCase    :  ArgumentParser		)   ->  int:
      lowerCAmelCase_			:      Optional[int]			=   parser.add_parser("""add-new-model"""		)
      add_new_model_parser.add_argument("""--testing"""    ,						action="""store_true"""    ,						help="""If in testing mode."""		)
      add_new_model_parser.add_argument("""--testing_file"""    ,						type=lowerCamelCase    ,						help="""Configuration file on which to run."""		)
      add_new_model_parser.add_argument(
          """--path"""    ,						type=lowerCamelCase    ,						help="""Path to cookiecutter. Should only be used for testing purposes."""		)
      add_new_model_parser.set_defaults(func=lowerCamelCase		)

    def __init__(   self    :  List[str]    ,						lowerCamelCase    :  bool    ,						lowerCamelCase    :  str    ,						lowerCamelCase    :  Any=None    ,						*lowerCamelCase    :  List[str]		)   ->  Optional[Any]:
      lowerCAmelCase_			:      int			=   testing
      lowerCAmelCase_			:      Union[str, Any]			=   testing_file
      lowerCAmelCase_			:      Tuple			=   path





    def __lowercase   (   self    :  Tuple		)   ->  int:
      warnings.warn(
          """The command `transformers-cli add-new-model` is deprecated and will be removed in v5 of Transformers. """
          """It is not actively maintained anymore, so might give a result that won't pass all tests and quality """
          """checks, you should use `transformers-cli add-new-model-like` instead."""		)
      if not _has_cookiecutter:
        raise ImportError(
            """Model creation dependencies are required to use the `add_new_model` command. Install them by running """
            """the following at the root of your `transformers` clone:\n\n\t$ pip install -e .[modelcreation]\n"""		)
      # Ensure that there is no other `cookiecutter-template-xxx` directory in the current working directory
      lowerCAmelCase_			:      int			=   [directory for directory in os.listdir() if """cookiecutter-template-""" == directory[:22]]
      if len(lowerCamelCase		) > 0:
        raise ValueError(
            """Several directories starting with `cookiecutter-template-` in current working directory. """
            """Please clean your directory by removing all folders starting with `cookiecutter-template-` or """
            """change your working directory."""		)

      lowerCAmelCase_			:      List[Any]			=   (
          Path(lowerCamelCase		).parent.parent.parent.parent if self._path is None else Path(self._path		).parent.parent
      )
      lowerCAmelCase_			:      Dict			=   path_to_transformer_root / """templates""" / """adding_a_new_model"""

      # Execute cookiecutter
      if not self._testing:
        cookiecutter(str(lowerCamelCase		)		)
      else:
        with open(self._testing_file    ,						"""r"""		) as configuration_file:
          lowerCAmelCase_			:      Tuple			=   json.load(lowerCamelCase		)

        cookiecutter(
            str(path_to_cookiecutter if self._path is None else self._path		)    ,						no_input=lowerCamelCase    ,						extra_context=lowerCamelCase    ,						)

      lowerCAmelCase_			:      List[str]			=   [directory for directory in os.listdir() if """cookiecutter-template-""" in directory[:22]][0]

      # Retrieve configuration
      with open(directory + """/configuration.json"""    ,						"""r"""		) as configuration_file:
        lowerCAmelCase_			:      Tuple			=   json.load(lowerCamelCase		)

      lowerCAmelCase_			:      str			=   configuration["""lowercase_modelname"""]
      lowerCAmelCase_			:      List[str]			=   configuration["""generate_tensorflow_pytorch_and_flax"""]
      os.remove(F'{directory}/configuration.json'		)

      lowerCAmelCase_			:      Dict			=   """PyTorch""" in generate_tensorflow_pytorch_and_flax
      lowerCAmelCase_			:      Optional[int]			=   """TensorFlow""" in generate_tensorflow_pytorch_and_flax
      lowerCAmelCase_			:      List[str]			=   """Flax""" in generate_tensorflow_pytorch_and_flax

      lowerCAmelCase_			:      Union[str, Any]			=   F'{path_to_transformer_root}/src/transformers/models/{lowercase_model_name}'
      os.makedirs(lowerCamelCase    ,						exist_ok=lowerCamelCase		)
      os.makedirs(F'{path_to_transformer_root}/tests/models/{lowercase_model_name}'    ,						exist_ok=lowerCamelCase		)

      # Tests require submodules as they have parent imports
      with open(F'{path_to_transformer_root}/tests/models/{lowercase_model_name}/__init__.py'    ,						"""w"""		):
        pass

      shutil.move(
          F'{directory}/__init__.py'    ,						F'{model_dir}/__init__.py'    ,						)
      shutil.move(
          F'{directory}/configuration_{lowercase_model_name}.py'    ,						F'{model_dir}/configuration_{lowercase_model_name}.py'    ,						)

      def remove_copy_lines(lowerCamelCase    :  Any		):
        with open(lowerCamelCase    ,						"""r"""		) as f:
          lowerCAmelCase_			:      List[str]			=   f.readlines()
        with open(lowerCamelCase    ,						"""w"""		) as f:
          for line in lines:
            if "# Copied from transformers." not in line:
              f.write(lowerCamelCase		)

      if output_pytorch:
        if not self._testing:
          remove_copy_lines(F'{directory}/modeling_{lowercase_model_name}.py'		)

        shutil.move(
            F'{directory}/modeling_{lowercase_model_name}.py'    ,						F'{model_dir}/modeling_{lowercase_model_name}.py'    ,						)

        shutil.move(
            F'{directory}/test_modeling_{lowercase_model_name}.py'    ,						F'{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_{lowercase_model_name}.py'    ,						)
      else:
        os.remove(F'{directory}/modeling_{lowercase_model_name}.py'		)
        os.remove(F'{directory}/test_modeling_{lowercase_model_name}.py'		)

      if output_tensorflow:
        if not self._testing:
          remove_copy_lines(F'{directory}/modeling_tf_{lowercase_model_name}.py'		)

        shutil.move(
            F'{directory}/modeling_tf_{lowercase_model_name}.py'    ,						F'{model_dir}/modeling_tf_{lowercase_model_name}.py'    ,						)

        shutil.move(
            F'{directory}/test_modeling_tf_{lowercase_model_name}.py'    ,						F'{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_tf_{lowercase_model_name}.py'    ,						)
      else:
        os.remove(F'{directory}/modeling_tf_{lowercase_model_name}.py'		)
        os.remove(F'{directory}/test_modeling_tf_{lowercase_model_name}.py'		)

      if output_flax:
        if not self._testing:
          remove_copy_lines(F'{directory}/modeling_flax_{lowercase_model_name}.py'		)

        shutil.move(
            F'{directory}/modeling_flax_{lowercase_model_name}.py'    ,						F'{model_dir}/modeling_flax_{lowercase_model_name}.py'    ,						)

        shutil.move(
            F'{directory}/test_modeling_flax_{lowercase_model_name}.py'    ,						F'{path_to_transformer_root}/tests/models/{lowercase_model_name}/test_modeling_flax_{lowercase_model_name}.py'    ,						)
      else:
        os.remove(F'{directory}/modeling_flax_{lowercase_model_name}.py'		)
        os.remove(F'{directory}/test_modeling_flax_{lowercase_model_name}.py'		)

      shutil.move(
          F'{directory}/{lowercase_model_name}.md'    ,						F'{path_to_transformer_root}/docs/source/en/model_doc/{lowercase_model_name}.md'    ,						)

      shutil.move(
          F'{directory}/tokenization_{lowercase_model_name}.py'    ,						F'{model_dir}/tokenization_{lowercase_model_name}.py'    ,						)

      shutil.move(
          F'{directory}/tokenization_fast_{lowercase_model_name}.py'    ,						F'{model_dir}/tokenization_{lowercase_model_name}_fast.py'    ,						)

      from os import fdopen, remove
      from shutil import copymode, move
      from tempfile import mkstemp

      def replace(lowerCamelCase    :  str    ,						lowerCamelCase    :  str    ,						lowerCamelCase    :  List[str]		):
        # Create temp file
        lowerCAmelCase_,							lowerCAmelCase_			:      int			=   mkstemp()
        lowerCAmelCase_			:      List[Any]			=   False
        with fdopen(lowerCamelCase    ,						"""w"""		) as new_file:
          with open(lowerCamelCase		) as old_file:
            for line in old_file:
              new_file.write(lowerCamelCase		)
              if line_to_copy_below in line:
                lowerCAmelCase_			:      List[str]			=   True
                for line_to_copy in lines_to_copy:
                  new_file.write(lowerCamelCase		)

        if not line_found:
          raise ValueError(F'Line {line_to_copy_below} was not found in file.'		)

        # Copy the file permissions from the old file to the new file
        copymode(lowerCamelCase    ,						lowerCamelCase		)
        # Remove original file
        remove(lowerCamelCase		)
        # Move new file
        move(lowerCamelCase    ,						lowerCamelCase		)

      def skip_units(lowerCamelCase    :  Optional[int]		):
        return (
            ("generating PyTorch" in line and not output_pytorch)
            or ("generating TensorFlow" in line and not output_tensorflow)
            or ("generating Flax" in line and not output_flax)
        )

      def replace_in_files(lowerCamelCase    :  Any		):
        with open(lowerCamelCase		) as datafile:
          lowerCAmelCase_			:      Dict			=   []
          lowerCAmelCase_			:      List[str]			=   False
          lowerCAmelCase_			:      str			=   False
          for line in datafile:
            if "# To replace in: " in line and "##" not in line:
              lowerCAmelCase_			:      Dict			=   line.split("""\""""		)[1]
              lowerCAmelCase_			:      int			=   skip_units(lowerCamelCase		)
            elif "# Below: " in line and "##" not in line:
              lowerCAmelCase_			:      Any			=   line.split("""\""""		)[1]
              lowerCAmelCase_			:      Tuple			=   skip_units(lowerCamelCase		)
            elif "# End." in line and "##" not in line:
              if not skip_file and not skip_snippet:
                replace(lowerCamelCase    ,						lowerCamelCase    ,						lowerCamelCase		)

              lowerCAmelCase_			:      Dict			=   []
            elif "# Replace with" in line and "##" not in line:
              lowerCAmelCase_			:      int			=   []
            elif "##" not in line:
              lines_to_copy.append(lowerCamelCase		)

        remove(lowerCamelCase		)

      replace_in_files(F'{directory}/to_replace_{lowercase_model_name}.py'		)
      os.rmdir(lowerCamelCase		)



