


'''simple docstring'''






import inspect
from typing import Optional, Union

import numpy as np
import PIL
import torch
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPFeatureExtractor, CLIPModel, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    DPMSolverMultistepScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.utils import (
    PIL_INTERPOLATION,
    randn_tensor,
)





def lowerCamelCase__      (						__lowerCamelCase	:							Any	,   __lowerCamelCase	:							List[str]	,   __lowerCamelCase	:							Tuple							):







   '''simple docstring'''




   if isinstance(__lowerCamelCase	,   torch.Tensor							):
      return image
   elif isinstance(__lowerCamelCase	,   PIL.Image.Image							):
      _UpperCAmelCase       :       List[Any]							   =[image]

   if isinstance(image[0]	,   PIL.Image.Image							):
      _UpperCAmelCase       :       List[Any]							   =[np.array(i.resize((w, h)	,   resample=PIL_INTERPOLATION['lanczos']							)							)[None, :] for i in image]
      _UpperCAmelCase       :       List[str]							   =np.concatenate(__lowerCamelCase	,   axis=0							)
      _UpperCAmelCase       :       Optional[Any]							   =np.array(__lowerCamelCase							).astype(np.floataa							) / 2_55.0
      _UpperCAmelCase       :       List[Any]							   =image.transpose(0	,   3	,   1	,   2							)
      _UpperCAmelCase       :       str							   =2.0 * image - 1.0
      _UpperCAmelCase       :       Optional[Any]							   =torch.from_numpy(__lowerCamelCase							)
   elif isinstance(image[0]	,   torch.Tensor							):
      _UpperCAmelCase       :       List[Any]							   =torch.cat(__lowerCamelCase	,   dim=0							)
   return image





def lowerCamelCase__      (						__lowerCamelCase	:							Union[str, Any]	,   __lowerCamelCase	:							Any	,   __lowerCamelCase	:							Tuple	,   __lowerCamelCase	:							Optional[int]=0.99_95							):







   '''simple docstring'''




   if not isinstance(__lowerCamelCase	,   np.ndarray							):
      _UpperCAmelCase       :       Optional[Any]							   =True
      _UpperCAmelCase       :       int							   =va.device
      _UpperCAmelCase       :       List[Any]							   =va.cpu().numpy()
      _UpperCAmelCase       :       Tuple							   =va.cpu().numpy()

   _UpperCAmelCase       :       Any							   =np.sum(va * va / (np.linalg.norm(__lowerCamelCase							) * np.linalg.norm(__lowerCamelCase							))							)
   if np.abs(__lowerCamelCase							) > DOT_THRESHOLD:
      _UpperCAmelCase       :       Union[str, Any]							   =(1 - t) * va + t * va
   else:
      _UpperCAmelCase       :       Optional[int]							   =np.arccos(__lowerCamelCase							)
      _UpperCAmelCase       :       Tuple							   =np.sin(__lowerCamelCase							)
      _UpperCAmelCase       :       str							   =theta_a * t
      _UpperCAmelCase       :       List[Any]							   =np.sin(__lowerCamelCase							)
      _UpperCAmelCase       :       List[str]							   =np.sin(theta_a - theta_t							) / sin_theta_a
      _UpperCAmelCase       :       str							   =sin_theta_t / sin_theta_a
      _UpperCAmelCase       :       int							   =sa * va + sa * va

   if inputs_are_torch:
      _UpperCAmelCase       :       Union[str, Any]							   =torch.from_numpy(__lowerCamelCase							).to(__lowerCamelCase							)

   return va





def lowerCamelCase__      (						__lowerCamelCase	:							Optional[Any]	,   __lowerCamelCase	:							List[str]							):







   '''simple docstring'''




   _UpperCAmelCase       :       Union[str, Any]							   =F.normalize(__lowerCamelCase	,   dim=-1							)
   _UpperCAmelCase       :       List[Any]							   =F.normalize(__lowerCamelCase	,   dim=-1							)
   return (x - y).norm(dim=-1							).div(2							).arcsin().pow(2							).mul(2							)





def lowerCamelCase__      (						__lowerCamelCase	:							List[str]	,   __lowerCamelCase	:							Union[str, Any]							):







   '''simple docstring'''




   for param in model.parameters():
      _UpperCAmelCase       :       Dict							   =value






class  __magic_name__							(		lowerCAmelCase    ):






       def __init__(      self  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case=None  ,							snake_case=None  ,							snake_case=None  ,							)						->  List[Any]:




          '''simple docstring'''
          super().__init__()
          self.register_modules(
              vae=snake_case  ,							text_encoder=snake_case  ,							clip_model=snake_case  ,							tokenizer=snake_case  ,							unet=snake_case  ,							scheduler=snake_case  ,							feature_extractor=snake_case  ,							coca_model=snake_case  ,							coca_tokenizer=snake_case  ,							coca_transform=snake_case  ,							)
          _UpperCAmelCase       :       List[Any]							   =(
              feature_extractor.size
              if isinstance(feature_extractor.size  ,							snake_case)
              else feature_extractor.size['shortest_edge']
          )
          _UpperCAmelCase       :       str							   =transforms.Normalize(mean=feature_extractor.image_mean  ,							std=feature_extractor.image_std)
          set_requires_grad(self.text_encoder  ,							snake_case)
          set_requires_grad(self.clip_model  ,							snake_case)






       def       lowerCAmelCase		(      self  ,							snake_case = "auto")						->  List[Any]:




          '''simple docstring'''
          if slice_size == "auto":
             # half the attention head size is usually a good trade-off between
             # speed and memory
             _UpperCAmelCase       :       Union[str, Any]							   =self.unet.config.attention_head_dim // 2
          self.unet.set_attention_slice(snake_case)






       def       lowerCAmelCase		(      self)						->  int:




          '''simple docstring'''
          self.enable_attention_slicing(snake_case)






       def       lowerCAmelCase		(      self)						->  Tuple:




          '''simple docstring'''
          set_requires_grad(self.vae  ,							snake_case)






       def       lowerCAmelCase		(      self)						->  Union[str, Any]:




          '''simple docstring'''
          set_requires_grad(self.vae  ,							snake_case)






       def       lowerCAmelCase		(      self)						->  List[Any]:




          '''simple docstring'''
          set_requires_grad(self.unet  ,							snake_case)






       def       lowerCAmelCase		(      self)						->  Optional[Any]:




          '''simple docstring'''
          set_requires_grad(self.unet  ,							snake_case)






       def       lowerCAmelCase		(      self  ,							snake_case  ,							snake_case  ,							snake_case)						->  Tuple:




          '''simple docstring'''
          # get the original timestep using init_timestep
          _UpperCAmelCase       :       Union[str, Any]							   =min(int(num_inference_steps * strength)  ,							snake_case)

          _UpperCAmelCase       :       Any							   =max(num_inference_steps - init_timestep  ,							0)
          _UpperCAmelCase       :       int							   =self.scheduler.timesteps[t_start:]

          return timesteps, num_inference_steps - t_start






       def       lowerCAmelCase		(      self  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case=None)						->  Optional[int]:




          '''simple docstring'''
          if not isinstance(snake_case  ,							torch.Tensor):
             raise ValueError(f"`image` has to be of type `torch.Tensor` but is {type(snake_case)}")

          _UpperCAmelCase       :       str							   =image.to(device=snake_case  ,							dtype=snake_case)

          if isinstance(snake_case  ,							snake_case):
             _UpperCAmelCase       :       Optional[Any]							   =[
                 self.vae.encode(image[i : i + 1]).latent_dist.sample(generator[i]) for i in range(snake_case)
             ]
             _UpperCAmelCase       :       Tuple							   =torch.cat(snake_case  ,							dim=0)
          else:
             _UpperCAmelCase       :       List[Any]							   =self.vae.encode(snake_case).latent_dist.sample(snake_case)

          # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
          _UpperCAmelCase       :       Optional[int]							   =0.1_82_15 * init_latents
          _UpperCAmelCase       :       List[str]							   =init_latents.repeat_interleave(snake_case  ,							dim=0)

          _UpperCAmelCase       :       Union[str, Any]							   =randn_tensor(init_latents.shape  ,							generator=snake_case  ,							device=snake_case  ,							dtype=snake_case)

          # get latents
          _UpperCAmelCase       :       Optional[int]							   =self.scheduler.add_noise(snake_case  ,							snake_case  ,							snake_case)
          _UpperCAmelCase       :       List[Any]							   =init_latents

          return latents






       def       lowerCAmelCase		(      self  ,							snake_case)						->  List[Any]:




          '''simple docstring'''
          _UpperCAmelCase       :       Union[str, Any]							   =self.coca_transform(snake_case).unsqueeze(0)
          with torch.no_grad(), torch.cuda.amp.autocast():
             _UpperCAmelCase       :       str							   =self.coca_model.generate(transformed_image.to(device=self.device  ,							dtype=self.coca_model.dtype))
          _UpperCAmelCase       :       Tuple							   =self.coca_tokenizer.decode(generated[0].cpu().numpy())
          return generated.split('<end_of_text>')[0].replace('<start_of_text>'  ,							'').rstrip(' .,')






       def       lowerCAmelCase		(      self  ,							snake_case  ,							snake_case)						->  Optional[int]:




          '''simple docstring'''
          _UpperCAmelCase       :       Any							   =self.feature_extractor.preprocess(snake_case)
          _UpperCAmelCase       :       Optional[Any]							   =torch.from_numpy(clip_image_input['pixel_values'][0]).unsqueeze(0).to(self.device).half()
          _UpperCAmelCase       :       Dict							   =self.clip_model.get_image_features(snake_case)
          _UpperCAmelCase       :       int							   =image_embeddings_clip / image_embeddings_clip.norm(p=2  ,							dim=-1  ,							keepdim=snake_case)
          _UpperCAmelCase       :       List[str]							   =image_embeddings_clip.repeat_interleave(snake_case  ,							dim=0)
          return image_embeddings_clip






       @torch.enable_grad()
       def       lowerCAmelCase		(      self  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case  ,							)						->  Optional[Any]:




          '''simple docstring'''
          _UpperCAmelCase       :       Dict							   =latents.detach().requires_grad_()

          _UpperCAmelCase       :       str							   =self.scheduler.scale_model_input(snake_case  ,							snake_case)

          # predict the noise residual
          _UpperCAmelCase       :       int							   =self.unet(snake_case  ,							snake_case  ,							encoder_hidden_states=snake_case).sample

          if isinstance(self.scheduler  ,							(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)):
             _UpperCAmelCase       :       Optional[int]							   =self.scheduler.alphas_cumprod[timestep]
             _UpperCAmelCase       :       Any							   =1 - alpha_prod_t
             # compute predicted original sample from predicted noise also called
             # "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
             _UpperCAmelCase       :       str							   =(latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

             _UpperCAmelCase       :       Union[str, Any]							   =torch.sqrt(snake_case)
             _UpperCAmelCase       :       List[str]							   =pred_original_sample * (fac) + latents * (1 - fac)
          elif isinstance(self.scheduler  ,							snake_case):
             _UpperCAmelCase       :       Optional[int]							   =self.scheduler.sigmas[index]
             _UpperCAmelCase       :       Tuple							   =latents - sigma * noise_pred
          else:
             raise ValueError(f"scheduler type {type(self.scheduler)} not supported")

          # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
          _UpperCAmelCase       :       Tuple							   =1 / 0.1_82_15 * sample
          _UpperCAmelCase       :       Optional[Any]							   =self.vae.decode(snake_case).sample
          _UpperCAmelCase       :       Tuple							   =(image / 2 + 0.5).clamp(0  ,							1)

          _UpperCAmelCase       :       int							   =transforms.Resize(self.feature_extractor_size)(snake_case)
          _UpperCAmelCase       :       Optional[int]							   =self.normalize(snake_case).to(latents.dtype)

          _UpperCAmelCase       :       str							   =self.clip_model.get_image_features(snake_case)
          _UpperCAmelCase       :       str							   =image_embeddings_clip / image_embeddings_clip.norm(p=2  ,							dim=-1  ,							keepdim=snake_case)

          _UpperCAmelCase       :       Optional[int]							   =spherical_dist_loss(snake_case  ,							snake_case).mean() * clip_guidance_scale

          _UpperCAmelCase       :       List[str]							   =-torch.autograd.grad(snake_case  ,							snake_case)[0]

          if isinstance(self.scheduler  ,							snake_case):
             _UpperCAmelCase       :       Optional[Any]							   =latents.detach() + grads * (sigma**2)
             _UpperCAmelCase       :       str							   =noise_pred_original
          else:
             _UpperCAmelCase       :       str							   =noise_pred_original - torch.sqrt(snake_case) * grads
          return noise_pred, latents






       @torch.no_grad()
       def __call__(      self  ,							snake_case  ,							snake_case  ,							snake_case = None  ,							snake_case = None  ,							snake_case = 5_1_2  ,							snake_case = 5_1_2  ,							snake_case = 0.6  ,							snake_case = 5_0  ,							snake_case = 7.5  ,							snake_case = 1  ,							snake_case = 0.0  ,							snake_case = 1_0_0  ,							snake_case = None  ,							snake_case = "pil"  ,							snake_case = True  ,							snake_case = 0.8  ,							snake_case = 0.1  ,							snake_case = 0.1  ,							)						->  List[str]:




          '''simple docstring'''
          if isinstance(snake_case  ,							snake_case) and len(snake_case) != batch_size:
             raise ValueError(f"You have passed {batch_size} batch_size, but only {len(snake_case)} generators.")

          if height % 8 != 0 or width % 8 != 0:
             raise ValueError(f"`height` and `width` have to be divisible by 8 but are {height} and {width}.")

          if isinstance(snake_case  ,							torch.Generator) and batch_size > 1:
             _UpperCAmelCase       :       List[str]							   =[generator] + [None] * (batch_size - 1)

          _UpperCAmelCase       :       Tuple							   =[
              ('model', self.coca_model is None),
              ('tokenizer', self.coca_tokenizer is None),
              ('transform', self.coca_transform is None),
          ]
          _UpperCAmelCase       :       Tuple							   =[x[0] for x in coca_is_none if x[1]]
          _UpperCAmelCase       :       Union[str, Any]							   =', '.join(snake_case)
          # generate prompts with coca model if prompt is None
          if content_prompt is None:
             if len(snake_case):
                raise ValueError(
                    f"Content prompt is None and CoCa [{coca_is_none_str}] is None."
                    f"Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline.")
             _UpperCAmelCase       :       Optional[int]							   =self.get_image_description(snake_case)
          if style_prompt is None:
             if len(snake_case):
                raise ValueError(
                    f"Style prompt is None and CoCa [{coca_is_none_str}] is None."
                    f" Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline.")
             _UpperCAmelCase       :       List[str]							   =self.get_image_description(snake_case)

          # get prompt text embeddings for content and style
          _UpperCAmelCase       :       Optional[Any]							   =self.tokenizer(
              snake_case  ,							padding='max_length'  ,							max_length=self.tokenizer.model_max_length  ,							truncation=snake_case  ,							return_tensors='pt'  ,							)
          _UpperCAmelCase       :       Dict							   =self.text_encoder(content_text_input.input_ids.to(self.device))[0]

          _UpperCAmelCase       :       Optional[int]							   =self.tokenizer(
              snake_case  ,							padding='max_length'  ,							max_length=self.tokenizer.model_max_length  ,							truncation=snake_case  ,							return_tensors='pt'  ,							)
          _UpperCAmelCase       :       Tuple							   =self.text_encoder(style_text_input.input_ids.to(self.device))[0]

          _UpperCAmelCase       :       List[Any]							   =slerp(snake_case  ,							snake_case  ,							snake_case)

          # duplicate text embeddings for each generation per prompt
          _UpperCAmelCase       :       Optional[Any]							   =text_embeddings.repeat_interleave(snake_case  ,							dim=0)

          # set timesteps
          _UpperCAmelCase       :       Any							   ='offset' in set(inspect.signature(self.scheduler.set_timesteps).parameters.keys())
          _UpperCAmelCase       :       int							   ={}
          if accepts_offset:
             _UpperCAmelCase       :       Union[str, Any]							   =1

          self.scheduler.set_timesteps(snake_case  ,							**snake_case)
          # Some schedulers like PNDM have timesteps as arrays
          # It's more optimized to move all timesteps to correct device beforehand
          self.scheduler.timesteps.to(self.device)

          _UpperCAmelCase   ,	_UpperCAmelCase       :       int							   =self.get_timesteps(snake_case  ,							snake_case  ,							self.device)
          _UpperCAmelCase       :       Dict							   =timesteps[:1].repeat(snake_case)

          # Preprocess image
          _UpperCAmelCase       :       int							   =preprocess(snake_case  ,							snake_case  ,							snake_case)
          _UpperCAmelCase       :       Tuple							   =self.prepare_latents(
              snake_case  ,							snake_case  ,							snake_case  ,							text_embeddings.dtype  ,							self.device  ,							snake_case)

          _UpperCAmelCase       :       Optional[Any]							   =preprocess(snake_case  ,							snake_case  ,							snake_case)
          _UpperCAmelCase       :       List[Any]							   =self.prepare_latents(
              snake_case  ,							snake_case  ,							snake_case  ,							text_embeddings.dtype  ,							self.device  ,							snake_case)

          _UpperCAmelCase       :       List[Any]							   =slerp(snake_case  ,							snake_case  ,							snake_case)

          if clip_guidance_scale > 0:
             _UpperCAmelCase       :       Optional[int]							   =self.get_clip_image_embeddings(snake_case  ,							snake_case)
             _UpperCAmelCase       :       int							   =self.get_clip_image_embeddings(snake_case  ,							snake_case)
             _UpperCAmelCase       :       Dict							   =slerp(
                 snake_case  ,							snake_case  ,							snake_case)

          # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
          # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
          # corresponds to doing no classifier free guidance.
          _UpperCAmelCase       :       int							   =guidance_scale > 1.0
          # get unconditional embeddings for classifier free guidance
          if do_classifier_free_guidance:
             _UpperCAmelCase       :       Union[str, Any]							   =content_text_input.input_ids.shape[-1]
             _UpperCAmelCase       :       List[str]							   =self.tokenizer(['']  ,							padding='max_length'  ,							max_length=snake_case  ,							return_tensors='pt')
             _UpperCAmelCase       :       Union[str, Any]							   =self.text_encoder(uncond_input.input_ids.to(self.device))[0]
             # duplicate unconditional embeddings for each generation per prompt
             _UpperCAmelCase       :       List[Any]							   =uncond_embeddings.repeat_interleave(snake_case  ,							dim=0)

             # For classifier free guidance, we need to do two forward passes.
             # Here we concatenate the unconditional and text embeddings into a single batch
             # to avoid doing two forward passes
             _UpperCAmelCase       :       Any							   =torch.cat([uncond_embeddings, text_embeddings])

          # get the initial random noise unless the user supplied it

          # Unlike in other pipelines, latents need to be generated in the target device
          # for 1-to-1 results reproducibility with the CompVis implementation.
          # However this currently doesn't work in `mps`.
          _UpperCAmelCase       :       str							   =(batch_size, self.unet.config.in_channels, height // 8, width // 8)
          _UpperCAmelCase       :       Union[str, Any]							   =text_embeddings.dtype
          if latents is None:
             if self.device.type == "mps":
                # randn does not work reproducibly on mps
                _UpperCAmelCase       :       int							   =torch.randn(snake_case  ,							generator=snake_case  ,							device='cpu'  ,							dtype=snake_case).to(
                    self.device)
             else:
                _UpperCAmelCase       :       Optional[int]							   =torch.randn(snake_case  ,							generator=snake_case  ,							device=self.device  ,							dtype=snake_case)
          else:
             if latents.shape != latents_shape:
                raise ValueError(f"Unexpected latents shape, got {latents.shape}, expected {latents_shape}")
             _UpperCAmelCase       :       List[str]							   =latents.to(self.device)

          # scale the initial noise by the standard deviation required by the scheduler
          _UpperCAmelCase       :       str							   =latents * self.scheduler.init_noise_sigma

          # prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
          # eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
          # eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
          # and should be between [0, 1]
          _UpperCAmelCase       :       List[str]							   ='eta' in set(inspect.signature(self.scheduler.step).parameters.keys())
          _UpperCAmelCase       :       Union[str, Any]							   ={}
          if accepts_eta:
             _UpperCAmelCase       :       Optional[int]							   =eta

          # check if the scheduler accepts generator
          _UpperCAmelCase       :       Union[str, Any]							   ='generator' in set(inspect.signature(self.scheduler.step).parameters.keys())
          if accepts_generator:
             _UpperCAmelCase       :       Dict							   =generator

          with self.progress_bar(total=snake_case):
             for i, t in enumerate(snake_case):
                # expand the latents if we are doing classifier free guidance
                _UpperCAmelCase       :       Dict							   =torch.cat([latents] * 2) if do_classifier_free_guidance else latents
                _UpperCAmelCase       :       Optional[int]							   =self.scheduler.scale_model_input(snake_case  ,							snake_case)

                # predict the noise residual
                _UpperCAmelCase       :       Optional[int]							   =self.unet(snake_case  ,							snake_case  ,							encoder_hidden_states=snake_case).sample

                # perform classifier free guidance
                if do_classifier_free_guidance:
                   _UpperCAmelCase   ,	_UpperCAmelCase       :       int							   =noise_pred.chunk(2)
                   _UpperCAmelCase       :       Dict							   =noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

                # perform clip guidance
                if clip_guidance_scale > 0:
                   _UpperCAmelCase       :       Tuple							   =(
                       text_embeddings.chunk(2)[1] if do_classifier_free_guidance else text_embeddings
                   )
                   _UpperCAmelCase   ,	_UpperCAmelCase       :       Optional[int]							   =self.cond_fn(
                       snake_case  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case  ,							snake_case  ,							)

                # compute the previous noisy sample x_t -> x_t-1
                _UpperCAmelCase       :       List[str]							   =self.scheduler.step(snake_case  ,							snake_case  ,							snake_case  ,							**snake_case).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
          _UpperCAmelCase       :       Optional[Any]							   =1 / 0.1_82_15 * latents
          _UpperCAmelCase       :       Optional[int]							   =self.vae.decode(snake_case).sample

          _UpperCAmelCase       :       str							   =(image / 2 + 0.5).clamp(0  ,							1)
          _UpperCAmelCase       :       Union[str, Any]							   =image.cpu().permute(0  ,							2  ,							3  ,							1).numpy()

          if output_type == "pil":
             _UpperCAmelCase       :       List[str]							   =self.numpy_to_pil(snake_case)

          if not return_dict:
             return (image, None)

          return StableDiffusionPipelineOutput(images=snake_case  ,							nsfw_content_detected=snake_case)





'''simple docstring'''






from typing import Any





def lowerCamelCase__      (						__lowerCamelCase	:							list	,   __lowerCamelCase	:							list	,   __lowerCamelCase	:							dict	,   __lowerCamelCase	:							dict	,   __lowerCamelCase	:							dict	,   ):







   '''simple docstring'''




   _validation(
       __lowerCamelCase	,   __lowerCamelCase	,   __lowerCamelCase	,   __lowerCamelCase	,   __lowerCamelCase	,   )
   # Creates data structures and fill initial step
   _UpperCAmelCase       :       dict							   ={}
   _UpperCAmelCase       :       dict							   ={}
   for state in states_space:
      _UpperCAmelCase       :       int							   =observations_space[0]
      _UpperCAmelCase       :       int							   =(
          initial_probabilities[state] * emission_probabilities[state][observation]
      )
      _UpperCAmelCase       :       int							   =None

   # Fills the data structure with the probabilities of
   # different transitions and pointers to previous states
   for o in range(1	,   len(__lowerCamelCase							)							):
      _UpperCAmelCase       :       List[Any]							   =observations_space[o]
      _UpperCAmelCase       :       Optional[int]							   =observations_space[o - 1]
      for state in states_space:
         # Calculates the argmax for probability function
         _UpperCAmelCase       :       List[str]							   =''
         _UpperCAmelCase       :       Dict							   =-1
         for k_state in states_space:
            _UpperCAmelCase       :       List[str]							   =(
                probabilities[(k_state, prior_observation)]
                * transition_probabilities[k_state][state]
                * emission_probabilities[state][observation]
            )
            if probability > max_probability:
               _UpperCAmelCase       :       int							   =probability
               _UpperCAmelCase       :       List[Any]							   =k_state

            # Update probabilities and pointers dicts
         _UpperCAmelCase       :       str							   =(
             probabilities[(arg_max, prior_observation)]
             * transition_probabilities[arg_max][state]
             * emission_probabilities[state][observation]
         )

         _UpperCAmelCase       :       List[Any]							   =arg_max

    # The final observation
   _UpperCAmelCase       :       int							   =observations_space[len(__lowerCamelCase							) - 1]

   # argmax for given final observation
   _UpperCAmelCase       :       Any							   =''
   _UpperCAmelCase       :       Union[str, Any]							   =-1
   for k_state in states_space:
      _UpperCAmelCase       :       Optional[int]							   =probabilities[(k_state, final_observation)]
      if probability > max_probability:
         _UpperCAmelCase       :       Union[str, Any]							   =probability
         _UpperCAmelCase       :       int							   =k_state
   _UpperCAmelCase       :       int							   =arg_max

   # Process pointers backwards
   _UpperCAmelCase       :       List[str]							   =last_state
   _UpperCAmelCase       :       Optional[int]							   =[]
   for o in range(len(__lowerCamelCase							) - 1	,   -1	,   -1							):
      result.append(__lowerCamelCase							)
      _UpperCAmelCase       :       Optional[Any]							   =pointers[previous, observations_space[o]]
   result.reverse()

   return result





def lowerCamelCase__      (						__lowerCamelCase	:							Any	,   __lowerCamelCase	:							Any	,   __lowerCamelCase	:							Any	,   __lowerCamelCase	:							Any	,   __lowerCamelCase	:							Any	,   ):







   '''simple docstring'''




   _validate_not_empty(
       __lowerCamelCase	,   __lowerCamelCase	,   __lowerCamelCase	,   __lowerCamelCase	,   __lowerCamelCase	,   )
   _validate_lists(__lowerCamelCase	,   __lowerCamelCase							)
   _validate_dicts(
       __lowerCamelCase	,   __lowerCamelCase	,   __lowerCamelCase							)





def lowerCamelCase__      (						__lowerCamelCase	:							Any	,   __lowerCamelCase	:							Any	,   __lowerCamelCase	:							Any	,   __lowerCamelCase	:							Any	,   __lowerCamelCase	:							Any	,   ):







   '''simple docstring'''




   if not all(
       [
           observations_space,
           states_space,
           initial_probabilities,
           transition_probabilities,
           emission_probabilities,
       ]							):
      raise ValueError('There\'s an empty parameter'							)





def lowerCamelCase__      (						__lowerCamelCase	:							Any	,   __lowerCamelCase	:							Any							):







   '''simple docstring'''




   _validate_list(__lowerCamelCase	,   'observations_space'							)
   _validate_list(__lowerCamelCase	,   'states_space'							)





def lowerCamelCase__      (						__lowerCamelCase	:							Any	,   __lowerCamelCase	:							str							):







   '''simple docstring'''




   if not isinstance(_object	,   __lowerCamelCase							):
      _UpperCAmelCase       :       Any							   =f"{var_name} must be a list"
      raise ValueError(__lowerCamelCase							)
   else:
      for x in _object:
         if not isinstance(__lowerCamelCase	,   __lowerCamelCase							):
            _UpperCAmelCase       :       Optional[int]							   =f"{var_name} must be a list of strings"
            raise ValueError(__lowerCamelCase							)





def lowerCamelCase__      (						__lowerCamelCase	:							Any	,   __lowerCamelCase	:							Any	,   __lowerCamelCase	:							Any	,   ):







   '''simple docstring'''




   _validate_dict(__lowerCamelCase	,   'initial_probabilities'	,   __lowerCamelCase							)
   _validate_nested_dict(__lowerCamelCase	,   'transition_probabilities'							)
   _validate_nested_dict(__lowerCamelCase	,   'emission_probabilities'							)





def lowerCamelCase__      (						__lowerCamelCase	:							Any	,   __lowerCamelCase	:							str							):







   '''simple docstring'''




   _validate_dict(_object	,   __lowerCamelCase	,   __lowerCamelCase							)
   for x in _object.values():
      _validate_dict(__lowerCamelCase	,   __lowerCamelCase	,   __lowerCamelCase	,   __lowerCamelCase							)





def lowerCamelCase__      (						__lowerCamelCase	:							Any	,   __lowerCamelCase	:							str	,   __lowerCamelCase	:							type	,   __lowerCamelCase	:							bool = False							):







   '''simple docstring'''




   if not isinstance(_object	,   __lowerCamelCase							):
      _UpperCAmelCase       :       List[str]							   =f"{var_name} must be a dict"
      raise ValueError(__lowerCamelCase							)
   if not all(isinstance(__lowerCamelCase	,   __lowerCamelCase							) for x in _object							):
      _UpperCAmelCase       :       str							   =f"{var_name} all keys must be strings"
      raise ValueError(__lowerCamelCase							)
   if not all(isinstance(__lowerCamelCase	,   __lowerCamelCase							) for x in _object.values()							):
      _UpperCAmelCase       :       int							   ='nested dictionary ' if nested else ''
      _UpperCAmelCase       :       Optional[int]							   =f"{var_name} {nested_text}all values must be {value_type.__name__}"
      raise ValueError(__lowerCamelCase							)


if __name__ == "__main__":
 from doctest import testmod

 testmod()

