
# Lint as: python3
import sys
from collections.abc import Mapping
from typing import TYPE_CHECKING, Dict, Optional

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.logging import get_logger
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter


if TYPE_CHECKING:
					import jax
					import jaxlib

UpperCamelCase_   	=	get_logger()

UpperCamelCase_   	=	None

class       a		(   TensorFormatter[Mapping, 'jax.Array', Mapping]					):







							def __init__(						self  :  Any      ,				snake_case__  :  Tuple=None      ,				snake_case__  :  List[str]=None      ,				**snake_case__  :  Optional[Any] ):
								"""simple docstring"""







								super().__init__(features=snake_case__ )
								import jax
								from jaxlib.xla_client import Device

								if isinstance(snake_case__      ,				snake_case__ ):
									raise ValueError(
									    F"Expected {device} to be a `str` not {type(snake_case__ )}, as `jaxlib.xla_extension.Device` "
									    "is not serializable neither with `pickle` nor with `dill`. Instead you can surround "
									    "the device with `str()` to get its string identifier that will be internally mapped "
									    "to the actual `jaxlib.xla_extension.Device`." )
								__lowerCAmelCase         =							device if isinstance(snake_case__      ,				snake_case__ ) else str(jax.devices()[0] )
								# using global variable since `jaxlib.xla_extension.Device` is not serializable neither
								# with `pickle` nor with `dill`, so we need to use a global variable instead
								global DEVICE_MAPPING
								if DEVICE_MAPPING is None:
									__lowerCAmelCase         =							self._map_devices_to_str()
								if self.device not in list(DEVICE_MAPPING.keys() ):
									logger.warning(
									    F"Device with string identifier {self.device} not listed among the available "
									    F"devices: {list(DEVICE_MAPPING.keys() )}, so falling back to the default "
									    F"device: {str(jax.devices()[0] )}." )
									__lowerCAmelCase         =							str(jax.devices()[0] )
								__lowerCAmelCase         =							jnp_array_kwargs







							@staticmethod
							def       UpperCAmelCase__					(						):
								"""simple docstring"""







								import jax

								return {str(snake_case__ ): device for device in jax.devices()}







							def       UpperCAmelCase__					(						self  :  List[Any]      ,				snake_case__  :  Optional[Any] ):
								"""simple docstring"""







								import jax
								import jax.numpy as jnp

								if isinstance(snake_case__      ,				snake_case__ ) and column:
									if all(
									    isinstance(snake_case__      ,				jax.Array ) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column ):
										return jnp.stack(snake_case__      ,				axis=0 )
								return column







							def       UpperCAmelCase__					(						self  :  Any      ,				snake_case__  :  str ):
								"""simple docstring"""







								import jax
								import jax.numpy as jnp

								if isinstance(snake_case__      ,				(str, bytes, type(snake_case__ )) ):
									return value
								elif isinstance(snake_case__      ,				(np.character, np.ndarray) ) and np.issubdtype(value.dtype      ,				np.character ):
									return value.tolist()

								__lowerCAmelCase         =							{}

								if isinstance(snake_case__      ,				(np.number, np.ndarray) ) and np.issubdtype(value.dtype      ,				np.integer ):
									# the default int precision depends on the jax config
									# see https://jax.readthedocs.io/en/latest/notebooks/Common_Gotchas_in_JAX.html#double-64bit-precision
									if jax.config.jax_enable_xaa:
										__lowerCAmelCase         =							{"dtype": jnp.intaa}
									else:
										__lowerCAmelCase         =							{"dtype": jnp.intaa}
								elif isinstance(snake_case__      ,				(np.number, np.ndarray) ) and np.issubdtype(value.dtype      ,				np.floating ):
									__lowerCAmelCase         =							{"dtype": jnp.floataa}
								elif config.PIL_AVAILABLE and "PIL" in sys.modules:
									import PIL.Image

									if isinstance(snake_case__      ,				PIL.Image.Image ):
										__lowerCAmelCase         =							np.asarray(snake_case__ )

        # using global variable since `jaxlib.xla_extension.Device` is not serializable neither
        # with `pickle` nor with `dill`, so we need to use a global variable instead
								global DEVICE_MAPPING
								if DEVICE_MAPPING is None:
									__lowerCAmelCase         =							self._map_devices_to_str()

								with jax.default_device(DEVICE_MAPPING[self.device] ):
									# calling jnp.array on a np.ndarray does copy the data
									# see https://github.com/google/jax/issues/4486
									return jnp.array(snake_case__      ,				**{**default_dtype, **self.jnp_array_kwargs} )







							def       UpperCAmelCase__					(						self  :  Dict      ,				snake_case__  :  Union[str, Any] ):
								"""simple docstring"""







								import jax

								# support for torch, tf, jax etc.
								if config.TORCH_AVAILABLE and "torch" in sys.modules:
									import torch

									if isinstance(snake_case__      ,				torch.Tensor ):
										return self._tensorize(data_struct.detach().cpu().numpy()[()] )
								if hasattr(snake_case__      ,				"__array__" ) and not isinstance(snake_case__      ,				jax.Array ):
									__lowerCAmelCase         =							data_struct.__array__()
								# support for nested types like struct of list of struct
								if isinstance(snake_case__      ,				np.ndarray ):
									if data_struct.dtype == object:  # jax arrays cannot be instantied from an array of objects
										return self._consolidate([self.recursive_tensorize(snake_case__ ) for substruct in data_struct] )
								elif isinstance(snake_case__      ,				(list, tuple) ):
									return self._consolidate([self.recursive_tensorize(snake_case__ ) for substruct in data_struct] )
								return self._tensorize(snake_case__ )







							def       UpperCAmelCase__					(						self  :  List[str]      ,				snake_case__  :  dict ):
								"""simple docstring"""







								return map_nested(self._recursive_tensorize      ,				snake_case__      ,				map_list=snake_case__ )







							def       UpperCAmelCase__					(						self  :  Tuple      ,				snake_case__  :  pa.Table ):
								"""simple docstring"""







								__lowerCAmelCase         =							self.numpy_arrow_extractor().extract_row(snake_case__ )
								__lowerCAmelCase         =							self.python_features_decoder.decode_row(snake_case__ )
								return self.recursive_tensorize(snake_case__ )







							def       UpperCAmelCase__					(						self  :  int      ,				snake_case__  :  pa.Table ):
								"""simple docstring"""







								__lowerCAmelCase         =							self.numpy_arrow_extractor().extract_column(snake_case__ )
								__lowerCAmelCase         =							self.python_features_decoder.decode_column(snake_case__      ,				pa_table.column_names[0] )
								__lowerCAmelCase         =							self.recursive_tensorize(snake_case__ )
								__lowerCAmelCase         =							self._consolidate(snake_case__ )
								return column







							def       UpperCAmelCase__					(						self  :  Optional[int]      ,				snake_case__  :  pa.Table ):
								"""simple docstring"""







								__lowerCAmelCase         =							self.numpy_arrow_extractor().extract_batch(snake_case__ )
								__lowerCAmelCase         =							self.python_features_decoder.decode_batch(snake_case__ )
								__lowerCAmelCase         =							self.recursive_tensorize(snake_case__ )
								for column_name in batch:
									__lowerCAmelCase         =							self._consolidate(batch[column_name] )
								return batch








from copy import deepcopy
from typing import Optional, Union

import numpy as np

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding
from ...utils import TensorType, is_tf_available, is_torch_available


if is_torch_available():
					import torch

if is_tf_available():
					import tensorflow as tf

class       a		(   __UpperCAmelCase					):
							lowercase_		:						Tuple   							=  ['image_processor']
							lowercase_		:						Tuple   							=  'SamImageProcessor'







							def __init__(						self  :  List[str]      ,				snake_case__  :  Optional[Any] ):
								"""simple docstring"""







								super().__init__(snake_case__ )
								__lowerCAmelCase         =							self.image_processor
								__lowerCAmelCase         =							-10
								__lowerCAmelCase         =							self.image_processor.size["longest_edge"]







							def __call__(						self  :  Any      ,				snake_case__  :  Optional[int]=None      ,				snake_case__  :  Optional[int]=None      ,				snake_case__  :  str=None      ,				snake_case__  :  List[str]=None      ,				snake_case__  :  Optional[Union[str, TensorType]] = None      ,				**snake_case__  :  Any      ,				):
								"""simple docstring"""







								__lowerCAmelCase         =							self.image_processor(
								    snake_case__      ,				return_tensors=snake_case__      ,				**snake_case__      ,				)

								# pop arguments that are not used in the foward but used nevertheless
								__lowerCAmelCase         =							encoding_image_processor["original_sizes"]

								if hasattr(snake_case__      ,				"numpy" ):  # Checks if Torch or TF tensor
									__lowerCAmelCase         =							original_sizes.numpy()

								__lowerCAmelCase   ,						__lowerCAmelCase   ,						__lowerCAmelCase         =							self._check_and_preprocess_points(
								    input_points=snake_case__      ,				input_labels=snake_case__      ,				input_boxes=snake_case__      ,				)

								__lowerCAmelCase         =							self._normalize_and_convert(
								    snake_case__      ,				snake_case__      ,				input_points=snake_case__      ,				input_labels=snake_case__      ,				input_boxes=snake_case__      ,				return_tensors=snake_case__      ,				)

								return encoding_image_processor







							def       UpperCAmelCase__					(						self  :  Tuple      ,				snake_case__  :  Any      ,				snake_case__  :  Optional[Any]      ,				snake_case__  :  str=None      ,				snake_case__  :  Any=None      ,				snake_case__  :  Any=None      ,				snake_case__  :  List[str]="pt"      ,				):
								"""simple docstring"""







								if input_points is not None:
									if len(snake_case__ ) != len(snake_case__ ):
										__lowerCAmelCase         =							[
										    self._normalize_coordinates(self.target_size      ,				snake_case__      ,				original_sizes[0] ) for point in input_points
										]
									else:
										__lowerCAmelCase         =							[
										    self._normalize_coordinates(self.target_size      ,				snake_case__      ,				snake_case__ )
										    for point, original_size in zip(snake_case__      ,				snake_case__ )
										]
									# check that all arrays have the same shape
									if not all(point.shape == input_points[0].shape for point in input_points ):
										if input_labels is not None:
											__lowerCAmelCase   ,						__lowerCAmelCase         =							self._pad_points_and_labels(snake_case__      ,				snake_case__ )

									__lowerCAmelCase         =							np.array(snake_case__ )

								if input_labels is not None:
									__lowerCAmelCase         =							np.array(snake_case__ )

								if input_boxes is not None:
									if len(snake_case__ ) != len(snake_case__ ):
										__lowerCAmelCase         =							[
										    self._normalize_coordinates(self.target_size      ,				snake_case__      ,				original_sizes[0]      ,				is_bounding_box=snake_case__ )
										    for box in input_boxes
										]
									else:
										__lowerCAmelCase         =							[
										    self._normalize_coordinates(self.target_size      ,				snake_case__      ,				snake_case__      ,				is_bounding_box=snake_case__ )
										    for box, original_size in zip(snake_case__      ,				snake_case__ )
										]
									__lowerCAmelCase         =							np.array(snake_case__ )

								if input_boxes is not None:
									if return_tensors == "pt":
										__lowerCAmelCase         =							torch.from_numpy(snake_case__ )
										# boxes batch size of 1 by default
										__lowerCAmelCase         =							input_boxes.unsqueeze(1 ) if len(input_boxes.shape ) != 3 else input_boxes
									elif return_tensors == "tf":
										__lowerCAmelCase         =							tf.convert_to_tensor(snake_case__ )
										# boxes batch size of 1 by default
										__lowerCAmelCase         =							tf.expand_dims(snake_case__      ,				1 ) if len(input_boxes.shape ) != 3 else input_boxes
									encoding_image_processor.update({"input_boxes": input_boxes} )
								if input_points is not None:
									if return_tensors == "pt":
										__lowerCAmelCase         =							torch.from_numpy(snake_case__ )
										# point batch size of 1 by default
										__lowerCAmelCase         =							input_points.unsqueeze(1 ) if len(input_points.shape ) != 4 else input_points
									elif return_tensors == "tf":
										__lowerCAmelCase         =							tf.convert_to_tensor(snake_case__ )
										# point batch size of 1 by default
										__lowerCAmelCase         =							tf.expand_dims(snake_case__      ,				1 ) if len(input_points.shape ) != 4 else input_points
									encoding_image_processor.update({"input_points": input_points} )
								if input_labels is not None:
									if return_tensors == "pt":
										__lowerCAmelCase         =							torch.from_numpy(snake_case__ )
										# point batch size of 1 by default
										__lowerCAmelCase         =							input_labels.unsqueeze(1 ) if len(input_labels.shape ) != 3 else input_labels
									elif return_tensors == "tf":
										__lowerCAmelCase         =							tf.convert_to_tensor(snake_case__ )
										# point batch size of 1 by default
										__lowerCAmelCase         =							tf.expand_dims(snake_case__      ,				1 ) if len(input_labels.shape ) != 3 else input_labels
									encoding_image_processor.update({"input_labels": input_labels} )

								return encoding_image_processor







							def       UpperCAmelCase__					(						self  :  str      ,				snake_case__  :  Union[str, Any]      ,				snake_case__  :  Dict ):
								"""simple docstring"""







								__lowerCAmelCase         =							max([point.shape[0] for point in input_points] )
								__lowerCAmelCase         =							[]
								for i, point in enumerate(snake_case__ ):
									if point.shape[0] != expected_nb_points:
										__lowerCAmelCase         =							np.concatenate(
										    [point, np.zeros((expected_nb_points - point.shape[0], 2) ) + self.point_pad_value]      ,				axis=0 )
										__lowerCAmelCase         =							np.append(input_labels[i]      ,				[self.point_pad_value] )
									processed_input_points.append(snake_case__ )
								__lowerCAmelCase         =							processed_input_points
								return input_points, input_labels







							def       UpperCAmelCase__					(						self  :  Dict      ,				snake_case__  :  int      ,				snake_case__  :  np.ndarray      ,				snake_case__  :  List[str]      ,				snake_case__  :  Dict=False ):
								"""simple docstring"""







								__lowerCAmelCase   ,						__lowerCAmelCase         =							original_size
								__lowerCAmelCase   ,						__lowerCAmelCase         =							self.image_processor._get_preprocess_shape(snake_case__      ,				longest_edge=snake_case__ )
								__lowerCAmelCase         =							deepcopy(snake_case__ ).astype(snake_case__ )

								if is_bounding_box:
									__lowerCAmelCase         =							coords.reshape(-1      ,				2      ,				2 )

								__lowerCAmelCase         =							coords[..., 0] * (new_w / old_w)
								__lowerCAmelCase         =							coords[..., 1] * (new_h / old_h)

								if is_bounding_box:
									__lowerCAmelCase         =							coords.reshape(-1      ,				4 )

								return coords







							def       UpperCAmelCase__					(						self  :  List[str]      ,				snake_case__  :  Union[str, Any]=None      ,				snake_case__  :  int=None      ,				snake_case__  :  Any=None      ,				):
								"""simple docstring"""







								if input_points is not None:
									if hasattr(snake_case__      ,				"numpy" ):  # Checks for TF or Torch tensor
										__lowerCAmelCase         =							input_points.numpy().tolist()

									if not isinstance(snake_case__      ,				snake_case__ ) or not isinstance(input_points[0]      ,				snake_case__ ):
										raise ValueError("Input points must be a list of list of floating points." )
									__lowerCAmelCase         =							[np.array(snake_case__ ) for input_point in input_points]
								else:
									__lowerCAmelCase         =							None

								if input_labels is not None:
									if hasattr(snake_case__      ,				"numpy" ):
										__lowerCAmelCase         =							input_labels.numpy().tolist()

									if not isinstance(snake_case__      ,				snake_case__ ) or not isinstance(input_labels[0]      ,				snake_case__ ):
										raise ValueError("Input labels must be a list of list integers." )
									__lowerCAmelCase         =							[np.array(snake_case__ ) for label in input_labels]
								else:
									__lowerCAmelCase         =							None

								if input_boxes is not None:
									if hasattr(snake_case__      ,				"numpy" ):
										__lowerCAmelCase         =							input_boxes.numpy().tolist()

									if (
									    not isinstance(snake_case__      ,				snake_case__ )
									    or not isinstance(input_boxes[0]      ,				snake_case__ )
									    or not isinstance(input_boxes[0][0]      ,				snake_case__ )
									):
										raise ValueError("Input boxes must be a list of list of list of floating points." )
									__lowerCAmelCase         =							[np.array(snake_case__ ).astype(np.floataa ) for box in input_boxes]
								else:
									__lowerCAmelCase         =							None

								return input_points, input_labels, input_boxes







							@property
							def       UpperCAmelCase__					(						self  :  Union[str, Any] ):
								"""simple docstring"""







								__lowerCAmelCase         =							self.image_processor.model_input_names
								return list(dict.fromkeys(snake_case__ ) )







							def       UpperCAmelCase__					(						self  :  Optional[Any]      ,				*snake_case__  :  Dict      ,				**snake_case__  :  Union[str, Any] ):
								"""simple docstring"""







								return self.image_processor.post_process_masks(*snake_case__      ,				**snake_case__ )






