





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


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

__A							   =							256



class   _SCREAMING_SNAKE_CASE		(      __SCREAMING_SNAKE_CASE       ):







 '''simple docstring'''
 lowercase_						=     ["melgan"]

 def __init__(self							:  Tuple			,					UpperCAmelCase_							:  SpectrogramNotesEncoder			,					UpperCAmelCase_							:  SpectrogramContEncoder			,					UpperCAmelCase_							:  TaFilmDecoder			,					UpperCAmelCase_							:  DDPMScheduler			,					UpperCAmelCase_							:  OnnxRuntimeModel if is_onnx_available() else Any			,					)			->None:



   '''simple docstring'''



   super().__init__()

   # From MELGAN
   lowerCamelCase__: List[Any]         =math.log(1E-5)  # Matches MelGAN training.
   lowerCamelCase__: str         =4.0  # Largest value for most examples
   lowerCamelCase__: Dict         =128

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

 def 				SCREAMING_SNAKE_CASE_   (self							:  Union[str, Any]			,					UpperCAmelCase_							:  int			,					UpperCAmelCase_							:  Any=(-1.0, 1.0)			,					UpperCAmelCase_							:  Dict=False)			->Union[str, Any]:



   '''simple docstring'''



   lowerCamelCase__    ,   lowerCamelCase__: str         =output_range
   if clip:
     lowerCamelCase__: Any         =torch.clip(UpperCAmelCase_			,					self.min_value			,					self.max_value)
   # Scale to [0, 1].
   lowerCamelCase__: Tuple         =(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 				SCREAMING_SNAKE_CASE_   (self							:  Optional[Any]			,					UpperCAmelCase_							:  Optional[int]			,					UpperCAmelCase_							:  List[Any]=(-1.0, 1.0)			,					UpperCAmelCase_							:  List[str]=False)			->Dict:



   '''simple docstring'''



   lowerCamelCase__    ,   lowerCamelCase__: List[Any]         =input_range
   lowerCamelCase__: List[str]         =torch.clip(UpperCAmelCase_			,					UpperCAmelCase_			,					UpperCAmelCase_) if clip else outputs
   # Scale to [0, 1].
   lowerCamelCase__: List[str]         =(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 				SCREAMING_SNAKE_CASE_   (self							:  List[Any]			,					UpperCAmelCase_							:  int			,					UpperCAmelCase_							:  Optional[int]			,					UpperCAmelCase_							:  Optional[Any])			->int:



   '''simple docstring'''



   lowerCamelCase__: Dict         =input_tokens > 0
   lowerCamelCase__    ,   lowerCamelCase__: int         =self.notes_encoder(
       encoder_input_tokens=UpperCAmelCase_			,					encoder_inputs_mask=UpperCAmelCase_)

   lowerCamelCase__    ,   lowerCamelCase__: Dict         =self.continuous_encoder(
       encoder_inputs=UpperCAmelCase_			,					encoder_inputs_mask=UpperCAmelCase_)

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

 def 				SCREAMING_SNAKE_CASE_   (self							:  Tuple			,					UpperCAmelCase_							:  Tuple			,					UpperCAmelCase_							:  str			,					UpperCAmelCase_							:  Tuple)			->int:



   '''simple docstring'''



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

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

   lowerCamelCase__: Tuple         =self.decoder(
       encodings_and_masks=UpperCAmelCase_			,					decoder_input_tokens=UpperCAmelCase_			,					decoder_noise_time=UpperCAmelCase_)
   return logits







 @torch.no_grad()
 def __call__(self							:  Union[str, Any]			,					UpperCAmelCase_							:  List[List[int]]			,					UpperCAmelCase_							:  Optional[torch.Generator] = None			,					UpperCAmelCase_							:  int = 100			,					UpperCAmelCase_							:  bool = True			,					UpperCAmelCase_							:  str = "numpy"			,					UpperCAmelCase_							:  Optional[Callable[[int, int, torch.FloatTensor], None]] = None			,					UpperCAmelCase_							:  int = 1			,					)			->Union[AudioPipelineOutput, Tuple]:



   '''simple docstring'''



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

   lowerCamelCase__: List[str]         =np.zeros([1, TARGET_FEATURE_LENGTH, self.n_dims]			,					dtype=np.floataa)
   lowerCamelCase__: Union[str, Any]         =np.zeros([1, 0, self.n_dims]			,					np.floataa)
   lowerCamelCase__: Optional[int]         =torch.ones((1, TARGET_FEATURE_LENGTH)			,					dtype=UpperCAmelCase_			,					device=self.device)

   for i, encoder_input_tokens in enumerate(UpperCAmelCase_):
     if i == 0:
       lowerCamelCase__: Optional[Any]         =torch.from_numpy(pred_mel[:1].copy()).to(
           device=self.device			,					dtype=self.decoder.dtype)
       # The first chunk has no previous context.
       lowerCamelCase__: int         =torch.zeros((1, TARGET_FEATURE_LENGTH)			,					dtype=UpperCAmelCase_			,					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.
       lowerCamelCase__: Tuple         =ones

     lowerCamelCase__: Optional[Any]         =self.scale_features(
         UpperCAmelCase_			,					output_range=[-1.0, 1.0]			,					clip=UpperCAmelCase_)

     lowerCamelCase__: Tuple         =self.encode(
         input_tokens=torch.IntTensor([encoder_input_tokens]).to(device=self.device)			,					continuous_inputs=UpperCAmelCase_			,					continuous_mask=UpperCAmelCase_			,					)

     # Sample encoder_continuous_inputs shaped gaussian noise to begin loop
     lowerCamelCase__: List[Any]         =randn_tensor(
         shape=encoder_continuous_inputs.shape			,					generator=UpperCAmelCase_			,					device=self.device			,					dtype=self.decoder.dtype			,					)

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

     # Denoising diffusion loop
     for j, t in enumerate(self.progress_bar(self.scheduler.timesteps)):
       lowerCamelCase__: Optional[int]         =self.decode(
           encodings_and_masks=UpperCAmelCase_			,					input_tokens=UpperCAmelCase_			,					noise_time=t / self.scheduler.config.num_train_timesteps			,					)

       # Compute previous output: x_t -> x_t-1
       lowerCamelCase__: str         =self.scheduler.step(UpperCAmelCase_			,					UpperCAmelCase_			,					UpperCAmelCase_			,					generator=UpperCAmelCase_).prev_sample

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

     lowerCamelCase__: int         =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(UpperCAmelCase_			,					UpperCAmelCase_)

     logger.info("Generated segment"			,					UpperCAmelCase_)

   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":
     lowerCamelCase__: Any         =self.melgan(input_features=full_pred_mel.astype(np.floataa))
   else:
     lowerCamelCase__: Union[str, Any]         =full_pred_mel

   if not return_dict:
     return (output,)

   return AudioPipelineOutput(audios=UpperCAmelCase_)






import requests
from bsa import BeautifulSoup

def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:str = "AAPL"				)  ->     str:
   _A      =       F'''https://in.finance.yahoo.com/quote/{symbol}?s={symbol}'''
   _A      =       BeautifulSoup(requests.get(_snake_case				).text	,   '''html.parser'''				)
   _A      =       '''My(6px) Pos(r) smartphone_Mt(6px)'''
   return soup.find('''div'''	,   class_=class_				).find('''span'''				).text


if __name__ == "__main__":
   for symbol in "AAPL AMZN IBM GOOG MSFT ORCL".split():
      print(f'Current {symbol:<4} stock price is {stock_price(symbol):>8}')



