




from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .embeddings import GaussianFourierProjection, TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin
from .unet_ad_blocks import get_down_block, get_mid_block, get_out_block, get_up_block


@dataclass
class     lowercase__  (					__SCREAMING_SNAKE_CASE		):
  A__=  42




class     lowercase__  (					__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE		):






  @register_to_config
  def __init__(					self      :					Dict		,						_lowercase      :					int = 6_55_36		,						_lowercase      :					Optional[int] = None		,						_lowercase      :					int = 2		,						_lowercase      :					int = 2		,						_lowercase      :					int = 0		,						_lowercase      :					str = "fourier"		,						_lowercase      :					bool = True		,						_lowercase      :					bool = False		,						_lowercase      :					float = 0.0		,						_lowercase      :					Tuple[str] = ("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")		,						_lowercase      :					Tuple[str] = ("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")		,						_lowercase      :					Tuple[str] = "UNetMidBlock1D"		,						_lowercase      :					str = None		,						_lowercase      :					Tuple[int] = (32, 32, 64)		,						_lowercase      :					str = None		,						_lowercase      :					int = 8		,						_lowercase      :					int = 1		,						_lowercase      :					bool = False		,						):




         """simple docstring"""







         super().__init__()
         UpperCAmelCase__										=     sample_size

         # time
         if time_embedding_type == "fourier":
                UpperCAmelCase__										=     GaussianFourierProjection(
                    embedding_size=8		,						set_W_to_weight=_lowercase		,						log=_lowercase		,						flip_sin_to_cos=_lowercase )
                UpperCAmelCase__										=     2 * block_out_channels[0]
         elif time_embedding_type == "positional":
                UpperCAmelCase__										=     Timesteps(
                    block_out_channels[0]		,						flip_sin_to_cos=_lowercase		,						downscale_freq_shift=_lowercase )
                UpperCAmelCase__										=     block_out_channels[0]

         if use_timestep_embedding:
                UpperCAmelCase__										=     block_out_channels[0] * 4
                UpperCAmelCase__										=     TimestepEmbedding(
                    in_channels=_lowercase		,						time_embed_dim=_lowercase		,						act_fn=_lowercase		,						out_dim=block_out_channels[0]		,						)

         UpperCAmelCase__										=     nn.ModuleList([] )
         UpperCAmelCase__										=     None
         UpperCAmelCase__										=     nn.ModuleList([] )
         UpperCAmelCase__										=     None

         # down
         UpperCAmelCase__										=     in_channels
         for i, down_block_type in enumerate(_lowercase ):
                UpperCAmelCase__										=     output_channel
                UpperCAmelCase__										=     block_out_channels[i]

                if i == 0:
                       input_channel += extra_in_channels

                UpperCAmelCase__										=     i == len(_lowercase ) - 1

                UpperCAmelCase__										=     get_down_block(
                    _lowercase		,						num_layers=_lowercase		,						in_channels=_lowercase		,						out_channels=_lowercase		,						temb_channels=block_out_channels[0]		,						add_downsample=not is_final_block or downsample_each_block		,						)
                self.down_blocks.append(_lowercase )

         # mid
         UpperCAmelCase__										=     get_mid_block(
             _lowercase		,						in_channels=block_out_channels[-1]		,						mid_channels=block_out_channels[-1]		,						out_channels=block_out_channels[-1]		,						embed_dim=block_out_channels[0]		,						num_layers=_lowercase		,						add_downsample=_lowercase		,						)

         # up
         UpperCAmelCase__										=     list(reversed(_lowercase ) )
         UpperCAmelCase__										=     reversed_block_out_channels[0]
         if out_block_type is None:
                UpperCAmelCase__										=     out_channels
         else:
                UpperCAmelCase__										=     block_out_channels[0]

         for i, up_block_type in enumerate(_lowercase ):
                UpperCAmelCase__										=     output_channel
                UpperCAmelCase__										=     (
                    reversed_block_out_channels[i + 1] if i < len(_lowercase ) - 1 else final_upsample_channels
                )

                UpperCAmelCase__										=     i == len(_lowercase ) - 1

                UpperCAmelCase__										=     get_up_block(
                    _lowercase		,						num_layers=_lowercase		,						in_channels=_lowercase		,						out_channels=_lowercase		,						temb_channels=block_out_channels[0]		,						add_upsample=not is_final_block		,						)
                self.up_blocks.append(_lowercase )
                UpperCAmelCase__										=     output_channel

         # out
         UpperCAmelCase__										=     norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4		,						32 )
         UpperCAmelCase__										=     get_out_block(
             out_block_type=_lowercase		,						num_groups_out=_lowercase		,						embed_dim=block_out_channels[0]		,						out_channels=_lowercase		,						act_fn=_lowercase		,						fc_dim=block_out_channels[-1] // 4		,						)






  def 			_UpperCAmelCase  (					self      :					List[Any]		,						_lowercase      :					torch.FloatTensor		,						_lowercase      :					Union[torch.Tensor, float, int]		,						_lowercase      :					bool = True		,						):




         """simple docstring"""







         UpperCAmelCase__										=     timestep
         if not torch.is_tensor(_lowercase ):
                UpperCAmelCase__										=     torch.tensor([timesteps]		,						dtype=torch.long		,						device=sample.device )
         elif torch.is_tensor(_lowercase ) and len(timesteps.shape ) == 0:
                UpperCAmelCase__										=     timesteps[None].to(sample.device )

         UpperCAmelCase__										=     self.time_proj(_lowercase )
         if self.config.use_timestep_embedding:
                UpperCAmelCase__										=     self.time_mlp(_lowercase )
         else:
                UpperCAmelCase__										=     timestep_embed[..., None]
                UpperCAmelCase__										=     timestep_embed.repeat([1, 1, sample.shape[2]] ).to(sample.dtype )
                UpperCAmelCase__										=     timestep_embed.broadcast_to((sample.shape[:1] + timestep_embed.shape[1:]) )

         # 2. down
         UpperCAmelCase__										=     ()
         for downsample_block in self.down_blocks:
                UpperCAmelCase__						,			UpperCAmelCase__										=     downsample_block(hidden_states=_lowercase		,						temb=_lowercase )
                down_block_res_samples += res_samples

         # 3. mid
         if self.mid_block:
                UpperCAmelCase__										=     self.mid_block(_lowercase		,						_lowercase )

         # 4. up
         for i, upsample_block in enumerate(self.up_blocks ):
                UpperCAmelCase__										=     down_block_res_samples[-1:]
                UpperCAmelCase__										=     down_block_res_samples[:-1]
                UpperCAmelCase__										=     upsample_block(_lowercase		,						res_hidden_states_tuple=_lowercase		,						temb=_lowercase )

         # 5. post-process
         if self.out_block:
                UpperCAmelCase__										=     self.out_block(_lowercase		,						_lowercase )

         if not return_dict:
                return (sample,)

         return UNetaDOutput(sample=_lowercase )









import uuid
from typing import Any, Dict, List, Optional, Union

from ..utils import add_end_docstrings, is_tf_available, is_torch_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_tf_available():
    import tensorflow as tf

if is_torch_available():
    import torch


A      = logging.get_logger(__name__)


class     lowercase__  :






  def __init__(					self      :					List[str]		,						_lowercase      :					str = None		,						_lowercase      :					uuid.UUID = None		,						_lowercase      :					Dict=None		,						_lowercase      :					Any=None ):




         """simple docstring"""







         if not conversation_id:
                UpperCAmelCase__										=     uuid.uuida()
         if past_user_inputs is None:
                UpperCAmelCase__										=     []
         if generated_responses is None:
                UpperCAmelCase__										=     []

         UpperCAmelCase__										=     conversation_id
         UpperCAmelCase__										=     past_user_inputs
         UpperCAmelCase__										=     generated_responses
         UpperCAmelCase__										=     text






  def __eq__(					self      :					Any		,						_lowercase      :					str ):




         """simple docstring"""







         if not isinstance(_lowercase		,						_lowercase ):
                return False
         if self.uuid == other.uuid:
                return True
         return (
             self.new_user_input == other.new_user_input
             and self.past_user_inputs == other.past_user_inputs
             and self.generated_responses == other.generated_responses
         )






  def 			_UpperCAmelCase  (					self      :					Any		,						_lowercase      :					str		,						_lowercase      :					bool = False ):




         """simple docstring"""







         if self.new_user_input:
                if overwrite:
                       logger.warning(
                           F"""User input added while unprocessed input was existing: \"{self.new_user_input}\" was overwritten """
                           F"""with: \"{text}\".""" )
                       UpperCAmelCase__										=     text
                else:
                       logger.warning(
                           F"""User input added while unprocessed input was existing: \"{self.new_user_input}\" new input """
                           F"""ignored: \"{text}\". Set `overwrite` to True to overwrite unprocessed user input""" )
         else:
                UpperCAmelCase__										=     text






  def 			_UpperCAmelCase  (					self      :					int ):




         """simple docstring"""







         if self.new_user_input:
                self.past_user_inputs.append(self.new_user_input )
         UpperCAmelCase__										=     None






  def 			_UpperCAmelCase  (					self      :					List[str]		,						_lowercase      :					str ):




         """simple docstring"""







         self.generated_responses.append(_lowercase )






  def 			_UpperCAmelCase  (					self      :					List[str] ):




         """simple docstring"""







         for user_input, generated_response in zip(self.past_user_inputs		,						self.generated_responses ):
                yield True, user_input
                yield False, generated_response
         if self.new_user_input:
                yield True, self.new_user_input






  def __repr__(					self      :					List[Any] ):




         """simple docstring"""







         UpperCAmelCase__										=     F"""Conversation id: {self.uuid} \n"""
         for is_user, text in self.iter_texts():
                UpperCAmelCase__										=     "user" if is_user else "bot"
                output += F"""{name} >> {text} \n"""
         return output




@add_end_docstrings(
    __SCREAMING_SNAKE_CASE					,							R'\n        min_length_for_response (`int`, *optional*, defaults to 32):\n            The minimum length (in number of tokens) for a response.\n        minimum_tokens (`int`, *optional*, defaults to 10):\n            The minimum length of tokens to leave for a response.\n    '					,							)
class     lowercase__  (					__SCREAMING_SNAKE_CASE		):






  def __init__(					self      :					str		,						*_lowercase      :					Union[str, Any]		,						**_lowercase      :					Any ):




         """simple docstring"""







         super().__init__(*_lowercase		,						**_lowercase )
         if self.tokenizer.pad_token_id is None:
                UpperCAmelCase__										=     self.tokenizer.eos_token






  def 			_UpperCAmelCase  (					self      :					List[str]		,						_lowercase      :					int=None		,						_lowercase      :					Any=None		,						_lowercase      :					List[str]=None		,						**_lowercase      :					Dict ):




         """simple docstring"""







         UpperCAmelCase__										=     {}
         UpperCAmelCase__										=     {}
         UpperCAmelCase__										=     {}

         if min_length_for_response is not None:
                UpperCAmelCase__										=     min_length_for_response
         if minimum_tokens is not None:
                UpperCAmelCase__										=     minimum_tokens

         if "max_length" in generate_kwargs:
                UpperCAmelCase__										=     generate_kwargs["max_length"]
                # self.max_length = generate_kwargs.get("max_length", self.model.config.max_length)
         if clean_up_tokenization_spaces is not None:
                UpperCAmelCase__										=     clean_up_tokenization_spaces

         if generate_kwargs:
                forward_params.update(_lowercase )
         return preprocess_params, forward_params, postprocess_params






  def __call__(					self      :					Dict		,						_lowercase      :					Union[Conversation, List[Conversation]]		,						_lowercase      :					List[str]=0		,						**_lowercase      :					Optional[int] ):




         """simple docstring"""







         UpperCAmelCase__										=     super().__call__(_lowercase		,						num_workers=_lowercase		,						**_lowercase )
         if isinstance(_lowercase		,						_lowercase ) and len(_lowercase ) == 1:
                return outputs[0]
         return outputs






  def 			_UpperCAmelCase  (					self      :					Optional[Any]		,						_lowercase      :					Conversation		,						_lowercase      :					List[Any]=32 ):




         """simple docstring"""







         if not isinstance(_lowercase		,						_lowercase ):
                raise ValueError("ConversationalPipeline, expects Conversation as inputs" )
         if conversation.new_user_input is None:
                raise ValueError(
                    F"""Conversation with UUID {type(conversation.uuid )} does not contain new user input to process. """
                    "Add user inputs with the conversation's `add_user_input` method" )
         if hasattr(self.tokenizer		,						"_build_conversation_input_ids" ):
                UpperCAmelCase__										=     self.tokenizer._build_conversation_input_ids(_lowercase )
         else:
                # If the tokenizer cannot handle conversations, we default to only the old version
                UpperCAmelCase__										=     self._legacy_parse_and_tokenize(_lowercase )

         if self.framework == "pt":
                UpperCAmelCase__										=     torch.LongTensor([input_ids] )
         elif self.framework == "tf":
                UpperCAmelCase__										=     tf.constant([input_ids] )
         return {"input_ids": input_ids, "conversation": conversation}






  def 			_UpperCAmelCase  (					self      :					Optional[Any]		,						_lowercase      :					List[Any]		,						_lowercase      :					List[Any]=10		,						**_lowercase      :					Optional[Any] ):




         """simple docstring"""







         UpperCAmelCase__										=     generate_kwargs.get("max_length"		,						self.model.config.max_length )

         UpperCAmelCase__										=     model_inputs["input_ids"].shape[1]
         if max_length - minimum_tokens < n:
                logger.warning(F"""Conversation input is to long ({n}), trimming it to ({max_length} - {minimum_tokens})""" )
                UpperCAmelCase__										=     max_length - minimum_tokens
                UpperCAmelCase__										=     model_inputs["input_ids"][:, -trim:]
                if "attention_mask" in model_inputs:
                       UpperCAmelCase__										=     model_inputs["attention_mask"][:, -trim:]
         UpperCAmelCase__										=     model_inputs.pop("conversation" )
         UpperCAmelCase__										=     max_length
         UpperCAmelCase__										=     self.model.generate(**_lowercase		,						**_lowercase )
         if self.model.config.is_encoder_decoder:
                UpperCAmelCase__										=     1
         else:
                UpperCAmelCase__										=     n
         return {"output_ids": output_ids[:, start_position:], "conversation": conversation}






  def 			_UpperCAmelCase  (					self      :					Tuple		,						_lowercase      :					Union[str, Any]		,						_lowercase      :					Optional[Any]=True ):




         """simple docstring"""







         UpperCAmelCase__										=     model_outputs["output_ids"]
         UpperCAmelCase__										=     self.tokenizer.decode(
             output_ids[0]		,						skip_special_tokens=_lowercase		,						clean_up_tokenization_spaces=_lowercase		,						)
         UpperCAmelCase__										=     model_outputs["conversation"]
         conversation.mark_processed()
         conversation.append_response(_lowercase )
         return conversation






  def 			_UpperCAmelCase  (					self      :					List[str]		,						_lowercase      :					Conversation ):




         """simple docstring"""







         UpperCAmelCase__										=     self.tokenizer.eos_token_id
         UpperCAmelCase__										=     []
         for is_user, text in conversation.iter_texts():
                if eos_token_id is not None:
                       input_ids.extend(self.tokenizer.encode(_lowercase		,						add_special_tokens=_lowercase ) + [eos_token_id] )
                else:
                       input_ids.extend(self.tokenizer.encode(_lowercase		,						add_special_tokens=_lowercase ) )

         if len(_lowercase ) > self.tokenizer.model_max_length:
                UpperCAmelCase__										=     input_ids[-self.tokenizer.model_max_length :]
         return input_ids





