






"""simple docstring"""



from copy import deepcopy
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple, Union

import torch
from torch import nn
from torch.utils.data import Dataset

from .deepspeed import is_deepspeed_zeroa_enabled
from .generation.configuration_utils import GenerationConfig
from .trainer import Trainer
from .utils import logging


if TYPE_CHECKING:
 from .data.data_collator import DataCollator
 from .modeling_utils import PreTrainedModel
 from .tokenization_utils_base import PreTrainedTokenizerBase
 from .trainer_callback import TrainerCallback
 from .trainer_utils import EvalPrediction, PredictionOutput
 from .training_args import TrainingArguments


_lowerCAmelCase     :     Optional[Any]            =	logging.get_logger(__name__)




class        lowerCAmelCase__       (  __magic_name__					):
   def __init__(						self	:     Dict				,       snake_case__	:     Union["PreTrainedModel", nn.Module] = None				,       snake_case__	:     "TrainingArguments" = None				,       snake_case__	:     Optional["DataCollator"] = None				,       snake_case__	:     Optional[Dataset] = None				,       snake_case__	:     Optional[Union[Dataset, Dict[str, Dataset]]] = None				,       snake_case__	:     Optional["PreTrainedTokenizerBase"] = None				,       snake_case__	:     Optional[Callable[[], "PreTrainedModel"]] = None				,       snake_case__	:     Optional[Callable[["EvalPrediction"], Dict]] = None				,       snake_case__	:     Optional[List["TrainerCallback"]] = None				,       snake_case__	:     Tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LambdaLR] = (None, None)				,       snake_case__	:     Optional[Callable[[torch.Tensor, torch.Tensor], torch.Tensor]] = None				,       ):


        '''simple docstring'''
        super().__init__(
            model=snake_case__				,       args=snake_case__				,       data_collator=snake_case__				,       train_dataset=snake_case__				,       eval_dataset=snake_case__				,       tokenizer=snake_case__				,       model_init=snake_case__				,       compute_metrics=snake_case__				,       callbacks=snake_case__				,       optimizers=snake_case__				,       preprocess_logits_for_metrics=snake_case__				,       )

        # Override self.model.generation_config if a GenerationConfig is specified in args.
        # Priority: args.generation_config > model.generation_config > default GenerationConfig.
        if self.args.generation_config is not None:
             UpperCAmelCase__		:  List[Any]				=				self.load_generation_config(self.args.generation_config				)
             UpperCAmelCase__		:  Tuple				=				gen_config
   @staticmethod
   def 	__a				(						snake_case__	:     Union[str, GenerationConfig]				):


        '''simple docstring'''

        # GenerationConfig provided, nothing to do
        if isinstance(snake_case__				,       snake_case__				):
             return deepcopy(snake_case__				)

        # str or Path
        UpperCAmelCase__		:  int				=				Path(snake_case__				) if isinstance(snake_case__				,       snake_case__				) else gen_config_arg
        UpperCAmelCase__		:  Optional[int]				=				None

        # Figuring if it is path pointing to a file, pointing to a directory or else a model id or URL
        # This step is required in order to determine config_file_name
        if pretrained_model_name.is_file():
             UpperCAmelCase__		:  int				=				pretrained_model_name.name
             UpperCAmelCase__		:  Dict				=				pretrained_model_name.parent
        # dir path
        elif pretrained_model_name.is_dir():
             pass
        # model id or URL
        else:
             UpperCAmelCase__		:  str				=				gen_config_arg

        UpperCAmelCase__		:  Dict				=				GenerationConfig.from_pretrained(snake_case__				,       snake_case__				)
        return gen_config
   def 	__a				(						self	:     str				,       snake_case__	:     Optional[Dataset] = None				,       snake_case__	:     Optional[List[str]] = None				,       snake_case__	:     str = "eval"				,       **snake_case__	:     List[str]				,       ):


        '''simple docstring'''
        UpperCAmelCase__		:  Optional[Any]				=				gen_kwargs.copy()
        if gen_kwargs.get("max_length"				) is None and gen_kwargs.get("max_new_tokens"				) is None:
             UpperCAmelCase__		:  Tuple				=				self.args.generation_max_length
        UpperCAmelCase__		:  List[str]				=				(
            gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"				) is not None else self.args.generation_num_beams
        )
        UpperCAmelCase__		:  Union[str, Any]				=				gen_kwargs

        return super().evaluate(snake_case__				,       ignore_keys=snake_case__				,       metric_key_prefix=snake_case__				)
   def 	__a				(						self	:     Union[str, Any]				,       snake_case__	:     Dataset				,       snake_case__	:     Optional[List[str]] = None				,       snake_case__	:     str = "test"				,       **snake_case__	:     str				,       ):


        '''simple docstring'''
        UpperCAmelCase__		:  int				=				gen_kwargs.copy()
        if gen_kwargs.get("max_length"				) is None and gen_kwargs.get("max_new_tokens"				) is None:
             UpperCAmelCase__		:  Optional[Any]				=				self.args.generation_max_length
        UpperCAmelCase__		:  Optional[Any]				=				(
            gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"				) is not None else self.args.generation_num_beams
        )
        UpperCAmelCase__		:  Tuple				=				gen_kwargs

        return super().predict(snake_case__				,       ignore_keys=snake_case__				,       metric_key_prefix=snake_case__				)
   def 	__a				(						self	:     int				,       snake_case__	:     nn.Module				,       snake_case__	:     Dict[str, Union[torch.Tensor, Any]]				,       snake_case__	:     bool				,       snake_case__	:     Optional[List[str]] = None				,       **snake_case__	:     List[Any]				,       ):


        '''simple docstring'''

        if not self.args.predict_with_generate or prediction_loss_only:
             return super().prediction_step(
                 snake_case__				,       snake_case__				,       prediction_loss_only=snake_case__				,       ignore_keys=snake_case__				)

        UpperCAmelCase__		:  int				=				"labels" in inputs
        UpperCAmelCase__		:  Optional[Any]				=				self._prepare_inputs(snake_case__				)

        # XXX: adapt synced_gpus for fairscale as well
        # Priority (handled in generate):
        # gen_kwargs > model.generation_config > default GenerationConfig()

        if len(snake_case__				) == 0 and hasattr(self				,       "_gen_kwargs"				):
             UpperCAmelCase__		:  str				=				self._gen_kwargs.copy()

        if gen_kwargs.get("max_length"				) is None and gen_kwargs.get("max_new_tokens"				) is None:
             UpperCAmelCase__		:  Any				=				self.model.config.max_length
        UpperCAmelCase__		:  Dict				=				(
            gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"				) is not None else self.model.config.num_beams
        )
        UpperCAmelCase__		:  List[str]				=				True if is_deepspeed_zeroa_enabled() else False
        UpperCAmelCase__		:  List[str]				=				(
            gen_kwargs["synced_gpus"] if gen_kwargs.get("synced_gpus"				) is not None else default_synced_gpus
        )

        # If the `decoder_input_ids` was created from `labels`, evict the former, so that the model can freely generate
        # (otherwise, it would continue generating from the padded `decoder_input_ids`)
        if (
            "labels" in inputs
            and "decoder_input_ids" in inputs
            and inputs["labels"].shape == inputs["decoder_input_ids"].shape
        ):
             UpperCAmelCase__		:  Dict				=				{k: v for k, v in inputs.items() if k != "decoder_input_ids"}
        UpperCAmelCase__		:  Union[str, Any]				=				self.model.generate(**snake_case__				,       **snake_case__				)

        # Temporary hack to ensure the generation config is not initialized for each iteration of the evaluation loop
        # TODO: remove this hack when the legacy code that initializes generation_config from a model config is
        # removed in https://github.com/huggingface/transformers/blob/98d88b23f54e5a23e741833f1e973fdf600cc2c5/src/transformers/generation/utils.py#L1183
        if self.model.generation_config._from_model_config:
             UpperCAmelCase__		:  str				=				False

        # Retrieves GenerationConfig from model.generation_config
        UpperCAmelCase__		:  List[str]				=				self.model.generation_config
        # in case the batch is shorter than max length, the output should be padded
        if generated_tokens.shape[-1] < gen_config.max_length:
             UpperCAmelCase__		:  Tuple				=				self._pad_tensors_to_max_len(snake_case__				,       gen_config.max_length				)
        elif gen_config.max_new_tokens is not None and generated_tokens.shape[-1] < gen_config.max_new_tokens + 1:
             UpperCAmelCase__		:  Optional[int]				=				self._pad_tensors_to_max_len(snake_case__				,       gen_config.max_new_tokens + 1				)

        with torch.no_grad():
             if has_labels:
                  with self.compute_loss_context_manager():
                       UpperCAmelCase__		:  Tuple				=				model(**snake_case__				)
                  if self.label_smoother is not None:
                       UpperCAmelCase__		:  Optional[Any]				=				self.label_smoother(snake_case__				,       inputs["labels"]				).mean().detach()
                  else:
                       UpperCAmelCase__		:  Union[str, Any]				=				(outputs["loss"] if isinstance(snake_case__				,       snake_case__				) else outputs[0]).mean().detach()
             else:
                  UpperCAmelCase__		:  Union[str, Any]				=				None

        if self.args.prediction_loss_only:
             return loss, None, None

        if has_labels:
             UpperCAmelCase__		:  Tuple				=				inputs["labels"]
             if labels.shape[-1] < gen_config.max_length:
                  UpperCAmelCase__		:  Any				=				self._pad_tensors_to_max_len(snake_case__				,       gen_config.max_length				)
             elif gen_config.max_new_tokens is not None and labels.shape[-1] < gen_config.max_new_tokens + 1:
                  UpperCAmelCase__		:  Tuple				=				self._pad_tensors_to_max_len(snake_case__				,       gen_config.max_new_tokens + 1				)
        else:
             UpperCAmelCase__		:  List[Any]				=				None

        return loss, generated_tokens, labels




   def 	__a				(						self	:     Dict				,       snake_case__	:     int				,       snake_case__	:     Optional[Any]				):


        '''simple docstring'''
        if self.tokenizer is not None and hasattr(self.tokenizer				,       "pad_token_id"				):
             # If PAD token is not defined at least EOS token has to be defined
             UpperCAmelCase__		:  Optional[int]				=				(
                 self.tokenizer.pad_token_id if self.tokenizer.pad_token_id is not None else self.tokenizer.eos_token_id
             )
        else:
             if self.model.config.pad_token_id is not None:
                  UpperCAmelCase__		:  int				=				self.model.config.pad_token_id
             else:
                  raise ValueError("Pad_token_id must be set in the configuration of the model, in order to pad tensors"				)

        UpperCAmelCase__		:  List[Any]				=				pad_token_id * torch.ones(
            (tensor.shape[0], max_length)				,       dtype=tensor.dtype				,       device=tensor.device				)
        UpperCAmelCase__		:  str				=				tensor
        return padded_tensor










"""simple docstring"""



import shutil

import requests


def      SCREAMING_SNAKE_CASE__    (   snake_case    :      str       , snake_case    :      bool = False       , snake_case    :      str = "."							)->						dict:






     '''simple docstring'''



     UpperCAmelCase__		:  Union[str, Any]				=				"https://api.nasa.gov/planetary/apod"
     return requests.get(snake_case       , params={"api_key": api_key}							).json()


def      SCREAMING_SNAKE_CASE__    (   snake_case    :      str       , snake_case    :      str = "."							)->						dict:






     '''simple docstring'''



     UpperCAmelCase__		:  List[Any]				=				get_apod_data(snake_case							)
     UpperCAmelCase__		:  str				=				apod_data["url"]
     UpperCAmelCase__		:  Tuple				=				img_url.split("/"							)[-1]
     UpperCAmelCase__		:  Dict				=				requests.get(snake_case       , stream=snake_case							)

     with open(f'{path}/{img_name}'       , "wb+"							) as img_file:
          shutil.copyfileobj(response.raw       , snake_case							)
     del response
     return apod_data


def      SCREAMING_SNAKE_CASE__    (   snake_case    :      str							)->						dict:






     '''simple docstring'''



     UpperCAmelCase__		:  List[str]				=				"https://images-api.nasa.gov/search"
     return requests.get(snake_case       , params={"q": query}							).json()


if __name__ == "__main__":
 print(save_apod("""YOUR API KEY"""))
 _lowerCAmelCase     :     List[Any]            =	get_archive_data("""apollo 2011""")["""collection"""]["""items"""]
 print(apollo_aaaa_items[0]["""data"""][0]["""description"""])


