"""simple docstring"""





import math
from dataclasses import dataclass
from typing import List, Optional, Union

import numpy as np
import PIL
import torch
from transformers import CLIPTextModelWithProjection, CLIPTokenizer

from ...models import PriorTransformer
from ...pipelines import DiffusionPipeline
from ...schedulers import HeunDiscreteScheduler
from ...utils import (
    BaseOutput,
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)
from .renderer import ShapERenderer


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

_UpperCamelCase       :						Union[str, Any]     =   "\n    Examples:\n        ```py\n        >>> import torch\n        >>> from diffusers import DiffusionPipeline\n        >>> from diffusers.utils import export_to_gif\n\n        >>> device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n\n        >>> repo = \"openai/shap-e\"\n        >>> pipe = DiffusionPipeline.from_pretrained(repo, torch_dtype=torch.float16)\n        >>> pipe = pipe.to(device)\n\n        >>> guidance_scale = 15.0\n        >>> prompt = \"a shark\"\n\n        >>> images = pipe(\n        ...     prompt,\n        ...     guidance_scale=guidance_scale,\n        ...     num_inference_steps=64,\n        ...     frame_size=256,\n        ... ).images\n\n        >>> gif_path = export_to_gif(images[0], \"shark_3d.gif\")\n        ```\n"


@dataclass
class 					UpperCAmelCase_     (   _a):
      lowerCamelCase__		:   Union[List[List[PIL.Image.Image]], List[List[np.ndarray]]]





class 					UpperCAmelCase_     (   _a):


      def __init__(       self		,      a		,      a		,      a		,      a		,      a		,      )	-> int:
             super().__init__()

             self.register_modules(
                 prior=a		,      text_encoder=a		,      tokenizer=a		,      scheduler=a		,      renderer=a		,      )


      def 		_UpperCAmelCase	(       self		,      a		,      a		,      a		,      a		,      a		,      a							)	-> Any:
             if latents is None:
                    lowercase__      :		str          =     randn_tensor(a		,      generator=a		,      device=a		,      dtype=a							)
             else:
                    if latents.shape != shape:
                           raise ValueError(f"""Unexpected latents shape, got {latents.shape}, expected {shape}"""							)
                    lowercase__      :		Any          =     latents.to(a							)

             lowercase__      :		int          =     latents * scheduler.init_noise_sigma
             return latents


      def 		_UpperCAmelCase	(       self		,      a=0							)	-> Union[str, Any]:
             if is_accelerate_available():
                    from accelerate import cpu_offload
             else:
                    raise ImportError('Please install accelerate via `pip install accelerate`'							)

             lowercase__      :		str          =     torch.device(f"""cuda:{gpu_id}"""							)

             lowercase__      :		List[Any]          =     [self.text_encoder, self.prior]
             for cpu_offloaded_model in models:
                    if cpu_offloaded_model is not None:
                           cpu_offload(a		,      a							)


      def 		_UpperCAmelCase	(       self		,      a=0							)	-> Optional[Any]:
             if is_accelerate_available() and is_accelerate_version('>='		,      '0.17.0.dev0'							):
                    from accelerate import cpu_offload_with_hook
             else:
                    raise ImportError('`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher.'							)

             lowercase__      :		str          =     torch.device(f"""cuda:{gpu_id}"""							)

             if self.device.type != "cpu":
                    self.to('cpu'		,      silence_dtype_warnings=a							)
                    torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

             lowercase__      :		int          =     None
             for cpu_offloaded_model in [self.text_encoder, self.prior, self.renderer]:
                    lowercase__   ,				lowercase__      :		Any          =     cpu_offload_with_hook(a		,      a		,      prev_module_hook=a							)

             if self.safety_checker is not None:
                    lowercase__   ,				lowercase__      :		Optional[Any]          =     cpu_offload_with_hook(self.safety_checker		,      a		,      prev_module_hook=a							)

             # We'll offload the last model manually.
             lowercase__      :		Optional[int]          =     hook


      @property
      def 		_UpperCAmelCase	(       self							)	-> Optional[Any]:
             if self.device != torch.device('meta'							) or not hasattr(self.text_encoder		,      '_hf_hook'							):
                    return self.device
             for module in self.text_encoder.modules():
                    if (
                        hasattr(a		,      '_hf_hook'							)
                        and hasattr(module._hf_hook		,      'execution_device'							)
                        and module._hf_hook.execution_device is not None
                    ):
                           return torch.device(module._hf_hook.execution_device							)
             return self.device


      def 		_UpperCAmelCase	(       self		,      a		,      a		,      a		,      a		,      )	-> Any:
             len(a							) if isinstance(a		,      a							) else 1

             # YiYi Notes: set pad_token_id to be 0, not sure why I can't set in the config file
             lowercase__      :		Optional[int]          =     0
             # get prompt text embeddings
             lowercase__      :		List[Any]          =     self.tokenizer(
                 a		,      padding='max_length'		,      max_length=self.tokenizer.model_max_length		,      truncation=a		,      return_tensors='pt'		,      )
             lowercase__      :		Optional[int]          =     text_inputs.input_ids
             lowercase__      :		Optional[int]          =     self.tokenizer(a		,      padding='longest'		,      return_tensors='pt'							).input_ids

             if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(a		,      a							):
                    lowercase__      :		int          =     self.tokenizer.batch_decode(untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1]							)
                    logger.warning(
                        'The following part of your input was truncated because CLIP can only handle sequences up to'
                        f""" {self.tokenizer.model_max_length} tokens: {removed_text}"""							)

             lowercase__      :		str          =     self.text_encoder(text_input_ids.to(a							)							)
             lowercase__      :		Union[str, Any]          =     text_encoder_output.text_embeds

             lowercase__      :		Dict          =     prompt_embeds.repeat_interleave(a		,      dim=0							)
             # in Shap-E it normalize the prompt_embeds and then later rescale it
             lowercase__      :		List[Any]          =     prompt_embeds / torch.linalg.norm(a		,      dim=-1		,      keepdim=a							)

             if do_classifier_free_guidance:
                    lowercase__      :		Dict          =     torch.zeros_like(a							)

                    # 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
                    lowercase__      :		Optional[int]          =     torch.cat([negative_prompt_embeds, prompt_embeds]							)

             # Rescale the features to have unit variance
             lowercase__      :		Optional[int]          =     math.sqrt(prompt_embeds.shape[1]							) * prompt_embeds

             return prompt_embeds


      @torch.no_grad()
      @replace_example_docstring(a							)
      def __call__(       self		,      a		,      a = 1		,      a = 2_5		,      a = None		,      a = None		,      a = 4.0		,      a = 6_4		,      a = "pil"		,      a = True		,      )	-> List[str]:

             if isinstance(a		,      a							):
                    lowercase__      :		List[str]          =     1
             elif isinstance(a		,      a							):
                    lowercase__      :		str          =     len(a							)
             else:
                    raise ValueError(f"""`prompt` has to be of type `str` or `list` but is {type(a							)}"""							)

             lowercase__      :		Union[str, Any]          =     self._execution_device

             lowercase__      :		Union[str, Any]          =     batch_size * num_images_per_prompt

             lowercase__      :		Optional[int]          =     guidance_scale > 1.0
             lowercase__      :		List[str]          =     self._encode_prompt(a		,      a		,      a		,      a							)

             # prior

             self.scheduler.set_timesteps(a		,      device=a							)
             lowercase__      :		List[Any]          =     self.scheduler.timesteps

             lowercase__      :		Dict          =     self.prior.config.num_embeddings
             lowercase__      :		Any          =     self.prior.config.embedding_dim

             lowercase__      :		Union[str, Any]          =     self.prepare_latents(
                 (batch_size, num_embeddings * embedding_dim)		,      prompt_embeds.dtype		,      a		,      a		,      a		,      self.scheduler		,      )

             # YiYi notes: for testing only to match ldm, we can directly create a latents with desired shape: batch_size, num_embeddings, embedding_dim
             lowercase__      :		List[Any]          =     latents.reshape(latents.shape[0]		,      a		,      a							)

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

                    lowercase__      :		Tuple          =     self.prior(
                        a		,      timestep=a		,      proj_embedding=a		,      ).predicted_image_embedding

                    # remove the variance
                    lowercase__   ,				lowercase__      :		Tuple          =     noise_pred.split(
                        scaled_model_input.shape[2]		,      dim=2							)  # batch_size, num_embeddings, embedding_dim

                    if do_classifier_free_guidance is not None:
                           lowercase__   ,				lowercase__      :		Optional[Any]          =     noise_pred.chunk(2							)
                           lowercase__      :		Tuple          =     noise_pred_uncond + guidance_scale * (noise_pred - noise_pred_uncond)

                    lowercase__      :		str          =     self.scheduler.step(
                        a		,      timestep=a		,      sample=a		,      ).prev_sample

             if output_type == "latent":
                    return ShapEPipelineOutput(images=a							)

             lowercase__      :		List[Any]          =     []
             for i, latent in enumerate(a							):
                    lowercase__      :		Optional[int]          =     self.renderer.decode(
                        latent[None, :]		,      a		,      size=a		,      ray_batch_size=4_0_9_6		,      n_coarse_samples=6_4		,      n_fine_samples=1_2_8		,      )
                    images.append(a							)

             lowercase__      :		List[str]          =     torch.stack(a							)

             if output_type not in ["np", "pil"]:
                    raise ValueError(f"""Only the output types `pil` and `np` are supported not output_type={output_type}"""							)

             lowercase__      :		Optional[Any]          =     images.cpu().numpy()

             if output_type == "pil":
                    lowercase__      :		List[str]          =     [self.numpy_to_pil(a							) for image in images]

             # Offload last model to CPU
             if hasattr(self		,      'final_offload_hook'							) and self.final_offload_hook is not None:
                    self.final_offload_hook.offload()

             if not return_dict:
                    return (images,)

             return ShapEPipelineOutput(images=a							)


"""simple docstring"""





import logging
from pathlib import Path
from typing import Any, List, Tuple

import yaml


# loading package files: https://stackoverflow.com/a/20885799
try:
       import importlib.resources as pkg_resources
except ImportError:
       # Try backported to PY<37 `importlib_resources`.
       import importlib_resources as pkg_resources

from . import resources


_UpperCamelCase       :						Optional[int]     =   "https://github.com/huggingface/datasets/tree/main/src/datasets/utils"
_UpperCamelCase       :						Union[str, Any]     =   f'''{BASE_REF_URL}/{__file__}'''
_UpperCamelCase       :						Any     =   logging.getLogger(__name__)





def     a_					(				_lowerCAmelCase  :			str					):





       '''simple docstring'''

       lowercase__      :		Optional[Any]          =     pkg_resources.read_text(_lowerCAmelCase       ,			_lowerCAmelCase					)
       return yaml.safe_load(_lowerCAmelCase					), f"""{BASE_REF_URL}/resources/{resource}"""


_UpperCamelCase   ,				_UpperCamelCase       :						int     =   load_yaml_resource("readme_structure.yaml")

_UpperCamelCase       :						Optional[Any]     =   [
    "[Needs More Information]",
    "[More Information Needed]",
    "(https://github.com/huggingface/datasets/blob/main/CONTRIBUTING.md#how-to-contribute-to-the-dataset-cards)",
]

# Dictionary representation of section/readme, error_list, warning_list
_UpperCamelCase       :						List[Any]     =   Tuple[dict, List[str], List[str]]


class 					UpperCAmelCase_     :


      def __init__(       self		,      a		,      a		,      a = None		,      a = False							)	-> str:
             lowercase__      :		List[str]          =     name
             lowercase__      :		int          =     level
             lowercase__      :		Dict          =     lines
             lowercase__      :		Optional[int]          =     ''
             lowercase__      :		List[str]          =     True
             lowercase__      :		Union[str, Any]          =     {}
             lowercase__      :		Any          =     []
             lowercase__      :		Optional[int]          =     []
             if self.lines is not None:
                    self.parse(suppress_parsing_errors=a							)


      def 		_UpperCAmelCase	(       self		,      a = False							)	-> Any:
             lowercase__      :		List[str]          =     ''
             lowercase__      :		int          =     []
             lowercase__      :		Any          =     False
             for line in self.lines:
                    if line.strip(' \n'							) == "":
                           continue
                    elif line.strip(' \n'							)[:3] == "```":
                           lowercase__      :		Optional[Any]          =     not code_start
                    elif line.split()[0] == self.level + "#" and not code_start:
                           if current_sub_level != "":
                                  lowercase__      :		Tuple          =     Section(a		,      self.level + '#'		,      a							)
                                  lowercase__      :		str          =     []
                           else:
                                  if current_lines != []:
                                         self.text += "".join(a							).strip()
                                         if self.text != "" and self.text not in FILLER_TEXT:
                                                lowercase__      :		List[Any]          =     False
                                         lowercase__      :		List[str]          =     []

                           lowercase__      :		Dict          =     ' '.join(line.split()[1:]							).strip(' \n'							)
                    else:
                           current_lines.append(a							)
             else:
                    if current_sub_level != "":
                           if current_sub_level in self.content:
                                  self.parsing_error_list.append(
                                      f"""Multiple sections with the same heading `{current_sub_level}` have been found. Please keep only one of these sections."""							)
                           lowercase__      :		Tuple          =     Section(a		,      self.level + '#'		,      a							)
                    else:
                           if current_lines != []:
                                  self.text += "".join(a							).strip()
                                  if self.text != "" and self.text not in FILLER_TEXT:
                                         lowercase__      :		Dict          =     False

             if self.level == "" and not suppress_parsing_errors:
                    if self.parsing_error_list != [] or self.parsing_warning_list != []:
                           lowercase__      :		Optional[Any]          =     '\n'.join('-\t' + x for x in self.parsing_error_list + self.parsing_warning_list							)
                           lowercase__      :		int          =     f"""The following issues were found while parsing the README at `{self.name}`:\n""" + errors
                           raise ValueError(a							)


      def 		_UpperCAmelCase	(       self		,      a							)	-> ReadmeValidatorOutput:
             lowercase__      :		Tuple          =     []
             lowercase__      :		int          =     []
             if structure["allow_empty"] is False:
                    # If content is expected
                    if self.is_empty_text and self.content == {}:
                           # If no content is found, mention it in the error_list
                           error_list.append(f"""Expected some content in section `{self.name}` but it is empty."""							)

             if structure["allow_empty_text"] is False:
                    # If some text is expected
                    if self.is_empty_text:
                           # If no text is found, mention it in the error_list
                           error_list.append(
                               f"""Expected some text in section `{self.name}` but it is empty (text in subsections are ignored)."""							)
        # Subsections Validation
             if structure["subsections"] is not None:
                    # If subsections are expected
                    if self.content == {}:
                           # If no subsections are present
                           lowercase__      :		str          =     [subsection['name'] for subsection in structure['subsections']]
                           # Mention the expected values in the error_list
                           error_list.append(
                               f"""Section `{self.name}` expected the following subsections: {", ".join(["`"+x+"`" for x in values]							)}. Found 'None'."""							)
                    else:
                           # If some subsections are present
                           lowercase__      :		Union[str, Any]          =     [subsection['name'] for subsection in structure['subsections']]
                           lowercase__      :		Any          =     False
                           for idx, name in enumerate(a							):
                                  if name not in self.content:
                                         # If the expected subsection is not present
                                         error_list.append(f"""Section `{self.name}` is missing subsection: `{name}`."""							)
                                         lowercase__      :		str          =     True
                                  else:
                                         # If the subsection is present, validate subsection, return the result
                                         # and concat the errors from subsection to section error_list

                                         # Skip sublevel validation if current level is `###`
                                         if self.level == "###":
                                                continue
                                         else:
                                                lowercase__   ,				lowercase__   ,				lowercase__      :		Optional[Any]          =     self.content[name].validate(
                                                    structure['subsections'][idx]							)
                                         error_list += subsec_error_list
                                         warning_list += subsec_warning_list

                           if has_missing_subsections:  # we only allow to have extra subsections if all the other ones are here
                                  for name in self.content:
                                         if name not in structure_names:
                                                # If an extra subsection is present
                                                warning_list.append(
                                                    f"""`{self.name}` has an extra subsection: `{name}`. Skipping further validation checks for this subsection as expected structure is unknown."""							)
             if error_list:
                    # If there are errors, do not return the dictionary as it is invalid
                    return {}, error_list, warning_list
             else:
                    return self.to_dict(), error_list, warning_list


      def 		_UpperCAmelCase	(       self							)	-> dict:
             return {
                 "name": self.name,
                 "text": self.text,
                 "is_empty_text": self.is_empty_text,
                 "subsections": [value.to_dict() for value in self.content.values()],
             }





class 					UpperCAmelCase_     (   _a):  # Level 0


      def __init__(       self		,      a		,      a		,      a = None		,      a = False							)	-> Any:
             super().__init__(name=a		,      level=''							)  # Not using lines here as we need to use a child class parse
             lowercase__      :		List[Any]          =     structure
             lowercase__      :		Dict          =     -2
             lowercase__      :		Union[str, Any]          =     0
             lowercase__      :		Dict          =     lines
             if self.lines is not None:
                    self.parse(suppress_parsing_errors=a							)


      def 		_UpperCAmelCase	(       self							)	-> Tuple:
             if self.structure is None:
                    lowercase__   ,				lowercase__   ,				lowercase__      :		Any          =     self._validate(a							)
             else:
                    lowercase__   ,				lowercase__   ,				lowercase__      :		Dict          =     self._validate(self.structure							)
             if error_list != [] or warning_list != []:
                    lowercase__      :		Optional[Any]          =     '\n'.join(['-\t' + x for x in error_list + warning_list]							)
                    lowercase__      :		Union[str, Any]          =     f"""The following issues were found for the README at `{self.name}`:\n""" + errors
                    raise ValueError(a							)


      @classmethod
      def 		_UpperCAmelCase	(       cls		,      a		,      a = None		,      a = False							)	-> Tuple:
             with open(a		,      encoding='utf-8'							) as f:
                    lowercase__      :		List[Any]          =     f.readlines()
             return cls(a		,      a		,      a		,      suppress_parsing_errors=a							)


      @classmethod
      def 		_UpperCAmelCase	(       cls		,      a		,      a = None		,      a = "root"		,      a = False							)	-> int:
             lowercase__      :		Optional[Any]          =     string.split('\n'							)
             return cls(a		,      a		,      a		,      suppress_parsing_errors=a							)


      def 		_UpperCAmelCase	(       self		,      a = False							)	-> List[Any]:
             # Skip Tags
             lowercase__      :		Tuple          =     0

             for line in self.lines:
                    self.yaml_tags_line_count += 1
                    if line.strip(' \n'							) == "---":
                           self.tag_count += 1
                           if self.tag_count == 2:
                                  break
                    line_count += 1
             if self.tag_count == 2:
                    lowercase__      :		Optional[int]          =     self.lines[line_count + 1 :]  # Get the last + 1 th item.
             else:
                    lowercase__      :		List[str]          =     self.lines[self.tag_count :]
             super().parse(suppress_parsing_errors=a							)


      def __str__(       self							)	-> Any:
             return str(self.to_dict()							)


      def 		_UpperCAmelCase	(       self		,      a							)	-> str:
             lowercase__      :		Any          =     []
             lowercase__      :		Optional[int]          =     []
             if self.yaml_tags_line_count == 0:
                    warning_list.append('Empty YAML markers are present in the README.'							)
             elif self.tag_count == 0:
                    warning_list.append('No YAML markers are present in the README.'							)
             elif self.tag_count == 1:
                    warning_list.append('Only the start of YAML tags present in the README.'							)
             # Check how many first level sections are present.
             lowercase__      :		int          =     len(self.content.keys()							)
             if num_first_level_keys > 1:
                    # If more than one, add to the error list, continue
                    error_list.append(
                        f"""The README has several first-level headings: {", ".join(["`"+x+"`" for x in list(self.content.keys()							)]							)}. Only one heading is expected. Skipping further validation for this README."""							)
             elif num_first_level_keys < 1:
                    # If less than one, append error.
                    error_list.append(
                        'The README has no first-level headings. One heading is expected. Skipping further validation for this README.'							)

             else:
                    # If one exactly
                    lowercase__      :		Dict          =     list(self.content.keys()							)[0]  # Get the key
                    if start_key.startswith('Dataset Card for'							):  # Check correct start
                           # If the starting is correct, validate all the sections
                           lowercase__   ,				lowercase__   ,				lowercase__      :		Any          =     self.content[start_key].validate(
                               readme_structure['subsections'][0]							)
                           error_list += sec_error_list
                           warning_list += sec_warning_list
                    else:
                           # If not found, append error
                           error_list.append(
                               'No first-level heading starting with `Dataset Card for` found in README. Skipping further validation for this README.'							)
             if error_list:
                    # If there are errors, do not return the dictionary as it is invalid
                    return {}, error_list, warning_list
             else:
                    return self.to_dict(), error_list, warning_list


if __name__ == "__main__":
       from argparse import ArgumentParser

       _UpperCamelCase       :						Tuple     =   ArgumentParser(usage="Validate the content (excluding YAML tags) of a README.md file.")
       ap.add_argument("readme_filepath")
       _UpperCamelCase       :						str     =   ap.parse_args()
       _UpperCamelCase       :						Optional[int]     =   Path(args.readme_filepath)
       _UpperCamelCase       :						Dict     =   ReadMe.from_readme(readme_filepath)

