




'''simple docstring'''

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				(   _lowerCamelCase     ):
				snake_case_      =		["image_processor"]
				snake_case_      =		"SamImageProcessor"


				def __init__(		self    :					List[Any]       ,				lowercase_    :					List[Any]					):
									super().__init__(lowercase_					)
									snake_case_									=				self.image_processor
									snake_case_									=				-10
									snake_case_									=				self.image_processor.size['''longest_edge''']


				def __call__(		self    :					List[str]       ,				lowercase_    :					Tuple=None       ,				lowercase_    :					Tuple=None       ,				lowercase_    :					Optional[int]=None       ,				lowercase_    :					Dict=None       ,				lowercase_    :					Optional[Union[str, TensorType]] = None       ,				**lowercase_    :					Any       ,				):
									snake_case_									=				self.image_processor(
									    lowercase_       ,				return_tensors=lowercase_       ,				**lowercase_       ,				)

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

									if hasattr(lowercase_       ,				'''numpy'''					):  # Checks if Torch or TF tensor
														snake_case_									=				original_sizes.numpy()

									snake_case_			,snake_case_			,snake_case_									=				self._check_and_preprocess_points(
									    input_points=lowercase_       ,				input_labels=lowercase_       ,				input_boxes=lowercase_       ,				)

									snake_case_									=				self._normalize_and_convert(
									    lowercase_       ,				lowercase_       ,				input_points=lowercase_       ,				input_labels=lowercase_       ,				input_boxes=lowercase_       ,				return_tensors=lowercase_       ,				)

									return encoding_image_processor


				def   A_ (		self    :					Dict       ,				lowercase_    :					Union[str, Any]       ,				lowercase_    :					str       ,				lowercase_    :					Optional[Any]=None       ,				lowercase_    :					Tuple=None       ,				lowercase_    :					Optional[Any]=None       ,				lowercase_    :					Optional[Any]="pt"       ,				):
									if input_points is not None:
														if len(lowercase_					) != len(lowercase_					):
																			snake_case_									=				[
																			    self._normalize_coordinates(self.target_size       ,				lowercase_       ,				original_sizes[0]					) for point in input_points
																			]
														else:
																			snake_case_									=				[
																			    self._normalize_coordinates(self.target_size       ,				lowercase_       ,				lowercase_					)
																			    for point, original_size in zip(lowercase_       ,				lowercase_					)
																			]
														# 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:
																								snake_case_			,snake_case_									=				self._pad_points_and_labels(lowercase_       ,				lowercase_					)

														snake_case_									=				np.array(lowercase_					)

									if input_labels is not None:
														snake_case_									=				np.array(lowercase_					)

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

									if input_boxes is not None:
														if return_tensors == "pt":
																			snake_case_									=				torch.from_numpy(lowercase_					)
																			# boxes batch size of 1 by default
																			snake_case_									=				input_boxes.unsqueeze(1					) if len(input_boxes.shape					) != 3 else input_boxes
														elif return_tensors == "tf":
																			snake_case_									=				tf.convert_to_tensor(lowercase_					)
																			# boxes batch size of 1 by default
																			snake_case_									=				tf.expand_dims(lowercase_       ,				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":
																			snake_case_									=				torch.from_numpy(lowercase_					)
																			# point batch size of 1 by default
																			snake_case_									=				input_points.unsqueeze(1					) if len(input_points.shape					) != 4 else input_points
														elif return_tensors == "tf":
																			snake_case_									=				tf.convert_to_tensor(lowercase_					)
																			# point batch size of 1 by default
																			snake_case_									=				tf.expand_dims(lowercase_       ,				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":
																			snake_case_									=				torch.from_numpy(lowercase_					)
																			# point batch size of 1 by default
																			snake_case_									=				input_labels.unsqueeze(1					) if len(input_labels.shape					) != 3 else input_labels
														elif return_tensors == "tf":
																			snake_case_									=				tf.convert_to_tensor(lowercase_					)
																			# point batch size of 1 by default
																			snake_case_									=				tf.expand_dims(lowercase_       ,				1					) if len(input_labels.shape					) != 3 else input_labels
														encoding_image_processor.update({'''input_labels''': input_labels}					)

									return encoding_image_processor


				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					Any       ,				lowercase_    :					Dict					):
									snake_case_									=				max([point.shape[0] for point in input_points]					)
									snake_case_									=				[]
									for i, point in enumerate(lowercase_					):
														if point.shape[0] != expected_nb_points:
																			snake_case_									=				np.concatenate(
																			    [point, np.zeros((expected_nb_points - point.shape[0], 2)					) + self.point_pad_value]       ,				axis=0					)
																			snake_case_									=				np.append(input_labels[i]       ,				[self.point_pad_value]					)
														processed_input_points.append(lowercase_					)
									snake_case_									=				processed_input_points
									return input_points, input_labels


				def   A_ (		self    :					int       ,				lowercase_    :					int       ,				lowercase_    :					np.ndarray       ,				lowercase_    :					List[str]       ,				lowercase_    :					Tuple=False					):
									snake_case_			,snake_case_									=				original_size
									snake_case_			,snake_case_									=				self.image_processor._get_preprocess_shape(lowercase_       ,				longest_edge=lowercase_					)
									snake_case_									=				deepcopy(lowercase_					).astype(lowercase_					)

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

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

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

									return coords


				def   A_ (		self    :					Any       ,				lowercase_    :					int=None       ,				lowercase_    :					int=None       ,				lowercase_    :					Optional[Any]=None       ,				):
									if input_points is not None:
														if hasattr(lowercase_       ,				'''numpy'''					):  # Checks for TF or Torch tensor
																			snake_case_									=				input_points.numpy().tolist()

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

									if input_labels is not None:
														if hasattr(lowercase_       ,				'''numpy'''					):
																			snake_case_									=				input_labels.numpy().tolist()

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

									if input_boxes is not None:
														if hasattr(lowercase_       ,				'''numpy'''					):
																			snake_case_									=				input_boxes.numpy().tolist()

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

									return input_points, input_labels, input_boxes


				@property
				def   A_ (		self    :					Optional[int]					):
									snake_case_									=				self.image_processor.model_input_names
									return list(dict.fromkeys(lowercase_					)					)


				def   A_ (		self    :					Optional[Any]       ,				*lowercase_    :					Optional[Any]       ,				**lowercase_    :					List[Any]					):
									return self.image_processor.post_process_masks(*lowercase_       ,				**lowercase_					)








import math
from typing import Any, Callable, List, Optional, Tuple, Union

import numpy as np
import torch

from ...models import TaFilmDecoder
from ...schedulers import DDPMScheduler
from ...utils import is_onnx_available, logging, randn_tensor


if is_onnx_available():
  from ..onnx_utils import OnnxRuntimeModel

from ..pipeline_utils import AudioPipelineOutput, DiffusionPipeline
from .continous_encoder import SpectrogramContEncoder
from .notes_encoder import SpectrogramNotesEncoder


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

UpperCAmelCase_     							= 256






class     lowercase__       (	__lowerCamelCase				):

   '''simple docstring'''


   a						:							List[str]   		=		["melgan"]




   def __init__(    self,		__magic_name__,		__magic_name__,		__magic_name__,		__magic_name__,		__magic_name__,		) ->      None:






       """simple docstring"""
       super().__init__()

       # From MELGAN
       UpperCamelCase__							:				Optional[int]          =      math.log(1E-5							)  # Matches MelGAN training.
       UpperCamelCase__							:				int          =      4.0  # Largest value for most examples
       UpperCamelCase__							:				Optional[int]          =      128

       self.register_modules(
           notes_encoder=__magic_name__,		continuous_encoder=__magic_name__,		decoder=__magic_name__,		scheduler=__magic_name__,		melgan=__magic_name__,		)




   def       UpperCamelCase__  (    self,		__magic_name__,		__magic_name__=(-1.0, 1.0),		__magic_name__=False							) ->      Any:






       """simple docstring"""
       UpperCamelCase__   ,UpperCamelCase__							:				str          =      output_range
       if clip:
           UpperCamelCase__							:				Union[str, Any]          =      torch.clip(__magic_name__,		self.min_value,		self.max_value							)
       # Scale to [0, 1].
       UpperCamelCase__							:				List[str]          =      (features - self.min_value) / (self.max_value - self.min_value)
       # Scale to [min_out, max_out].
       return zero_one * (max_out - min_out) + min_out




   def       UpperCamelCase__  (    self,		__magic_name__,		__magic_name__=(-1.0, 1.0),		__magic_name__=False							) ->      Optional[int]:






       """simple docstring"""
       UpperCamelCase__   ,UpperCamelCase__							:				List[str]          =      input_range
       UpperCamelCase__							:				Any          =      torch.clip(__magic_name__,		__magic_name__,		__magic_name__							) if clip else outputs
       # Scale to [0, 1].
       UpperCamelCase__							:				Any          =      (outputs - min_out) / (max_out - min_out)
       # Scale to [self.min_value, self.max_value].
       return zero_one * (self.max_value - self.min_value) + self.min_value




   def       UpperCamelCase__  (    self,		__magic_name__,		__magic_name__,		__magic_name__							) ->      List[str]:






       """simple docstring"""
       UpperCamelCase__							:				Union[str, Any]          =      input_tokens > 0
       UpperCamelCase__   ,UpperCamelCase__							:				Any          =      self.notes_encoder(
           encoder_input_tokens=__magic_name__,		encoder_inputs_mask=__magic_name__							)

       UpperCamelCase__   ,UpperCamelCase__							:				Optional[Any]          =      self.continuous_encoder(
           encoder_inputs=__magic_name__,		encoder_inputs_mask=__magic_name__							)

       return [(tokens_encoded, tokens_mask), (continuous_encoded, continuous_mask)]




   def       UpperCamelCase__  (    self,		__magic_name__,		__magic_name__,		__magic_name__							) ->      str:






       """simple docstring"""
       UpperCamelCase__							:				Any          =      noise_time
       if not torch.is_tensor(__magic_name__							):
           UpperCamelCase__							:				Tuple          =      torch.tensor([timesteps],		dtype=torch.long,		device=input_tokens.device							)
       elif torch.is_tensor(__magic_name__							) and len(timesteps.shape							) == 0:
           UpperCamelCase__							:				Union[str, Any]          =      timesteps[None].to(input_tokens.device							)

       # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
       UpperCamelCase__							:				Dict          =      timesteps * torch.ones(input_tokens.shape[0],		dtype=timesteps.dtype,		device=timesteps.device							)

       UpperCamelCase__							:				List[str]          =      self.decoder(
           encodings_and_masks=__magic_name__,		decoder_input_tokens=__magic_name__,		decoder_noise_time=__magic_name__							)
       return logits




   @torch.no_grad()
   def __call__(    self,		__magic_name__,		__magic_name__ = None,		__magic_name__ = 100,		__magic_name__ = True,		__magic_name__ = "numpy",		__magic_name__ = None,		__magic_name__ = 1,		) ->      Union[AudioPipelineOutput, Tuple]:






       """simple docstring"""
       if (callback_steps is None) or (
           callback_steps is not None and (not isinstance(__magic_name__,		__magic_name__							) or callback_steps <= 0)
       ):
           raise ValueError(
               f"`callback_steps` has to be a positive integer but is {callback_steps} of type"
               f" {type(__magic_name__							)}."							)

       UpperCamelCase__							:				Dict          =      np.zeros([1, TARGET_FEATURE_LENGTH, self.n_dims],		dtype=np.floataa							)
       UpperCamelCase__							:				Tuple          =      np.zeros([1, 0, self.n_dims],		np.floataa							)
       UpperCamelCase__							:				List[Any]          =      torch.ones((1, TARGET_FEATURE_LENGTH),		dtype=__magic_name__,		device=self.device							)

       for i, encoder_input_tokens in enumerate(__magic_name__							):
           if i == 0:
               UpperCamelCase__							:				str          =      torch.from_numpy(pred_mel[:1].copy()							).to(
                   device=self.device,		dtype=self.decoder.dtype							)
               # The first chunk has no previous context.
               UpperCamelCase__							:				Any          =      torch.zeros((1, TARGET_FEATURE_LENGTH),		dtype=__magic_name__,		device=self.device							)
           else:
               # The full song pipeline does not feed in a context feature, so the mask
               # will be all 0s after the feature converter. Because we know we're
               # feeding in a full context chunk from the previous prediction, set it
               # to all 1s.
               UpperCamelCase__							:				List[str]          =      ones

           UpperCamelCase__							:				int          =      self.scale_features(
               __magic_name__,		output_range=[-1.0, 1.0],		clip=__magic_name__							)

           UpperCamelCase__							:				Union[str, Any]          =      self.encode(
               input_tokens=torch.IntTensor([encoder_input_tokens]							).to(device=self.device							),		continuous_inputs=__magic_name__,		continuous_mask=__magic_name__,		)

           # Sample encoder_continuous_inputs shaped gaussian noise to begin loop
           UpperCamelCase__							:				Optional[int]          =      randn_tensor(
               shape=encoder_continuous_inputs.shape,		generator=__magic_name__,		device=self.device,		dtype=self.decoder.dtype,		)

           # set step values
           self.scheduler.set_timesteps(__magic_name__							)

           # Denoising diffusion loop
           for j, t in enumerate(self.progress_bar(self.scheduler.timesteps							)							):
               UpperCamelCase__							:				Union[str, Any]          =      self.decode(
                   encodings_and_masks=__magic_name__,		input_tokens=__magic_name__,		noise_time=t / self.scheduler.config.num_train_timesteps,		)

               # Compute previous output: x_t -> x_t-1
               UpperCamelCase__							:				List[Any]          =      self.scheduler.step(__magic_name__,		__magic_name__,		__magic_name__,		generator=__magic_name__							).prev_sample

           UpperCamelCase__							:				List[Any]          =      self.scale_to_features(__magic_name__,		input_range=[-1.0, 1.0]							)
           UpperCamelCase__							:				List[Any]          =      mel[:1]
           UpperCamelCase__							:				int          =      mel.cpu().float().numpy()

           UpperCamelCase__							:				Union[str, Any]          =      np.concatenate([full_pred_mel, pred_mel[:1]],		axis=1							)

           # call the callback, if provided
           if callback is not None and i % callback_steps == 0:
               callback(__magic_name__,		__magic_name__							)

           logger.info('''Generated segment''',		__magic_name__							)

       if output_type == "numpy" and not is_onnx_available():
           raise ValueError(
               '''Cannot return output in \'np\' format if ONNX is not available. Make sure to have ONNX installed or set \'output_type\' to \'mel\'.'''							)
       elif output_type == "numpy" and self.melgan is None:
           raise ValueError(
               '''Cannot return output in \'np\' format if melgan component is not defined. Make sure to define `self.melgan` or set \'output_type\' to \'mel\'.'''							)

       if output_type == "numpy":
           UpperCamelCase__							:				Optional[int]          =      self.melgan(input_features=full_pred_mel.astype(np.floataa							)							)
       else:
           UpperCamelCase__							:				Any          =      full_pred_mel

       if not return_dict:
           return (output,)

       return AudioPipelineOutput(audios=__magic_name__							)

