

import json
import os
from dataclasses import dataclass
from functools import partial
from typing import Callable

import flax.linen as nn
import jax
import jax.numpy as jnp
import joblib
import optax
import wandb
from flax import jax_utils, struct, traverse_util
from flax.serialization import from_bytes, to_bytes
from flax.training import train_state
from flax.training.common_utils import shard
from tqdm.auto import tqdm

from transformers import BigBirdConfig, FlaxBigBirdForQuestionAnswering
from transformers.models.big_bird.modeling_flax_big_bird import FlaxBigBirdForQuestionAnsweringModule



class SCREAMING_SNAKE_CASE     (  a_			):
  """simple docstring"""






  lowerCamelCase   :      BigBirdConfig
  lowerCamelCase   :      jnp.dtype										=jnp.floataa
  lowerCamelCase   :      bool										=True







  def 	SCREAMING_SNAKE_CASE   (      self			:   List[str]   )							->     Optional[int]:






     """simple docstring"""







     super().setup()
     __lowerCAmelCase  :      List[Any]    =						nn.Dense(5	,							dtype=self.dtype   )







  def __call__(      self			:   Optional[Any]	,							*lowerCAmelCase			:   int	,							**lowerCAmelCase			:   Tuple   )							->     List[str]:






     """simple docstring"""







     __lowerCAmelCase  :      Optional[Any]    =						super().__call__(*lowerCAmelCase	,							**lowerCAmelCase   )
     __lowerCAmelCase  :      str    =						self.cls(outputs[2]   )
     return outputs[:2] + (cls_out,)






class SCREAMING_SNAKE_CASE     (  a_			):
  """simple docstring"""






  lowerCamelCase   :      Optional[Any]										=FlaxBigBirdForNaturalQuestionsModule






def      snake_case_				(__A					:						int ,       __A					:						Union[str, Any] ,       __A					:						int ,       __A					:						str ,       __A					:						int ,       __A					:						List[Any]					)      ->     Dict:
   def cross_entropy(__A					:						Union[str, Any] ,       __A					:						List[str] ,       __A					:						Any=None					):
      __lowerCAmelCase  :      List[Any]    =						logits.shape[-1]
      __lowerCAmelCase  :      Dict    =						(labels[..., None] == jnp.arange(__A					)[None]).astype("""f4"""					)
      __lowerCAmelCase  :      Dict    =						jax.nn.log_softmax(__A ,       axis=-1					)
      __lowerCAmelCase  :      Optional[int]    =						-jnp.sum(labels * logits ,       axis=-1					)
      if reduction is not None:
         __lowerCAmelCase  :      Tuple    =						reduction(__A					)
      return loss

   __lowerCAmelCase  :      int    =						partial(__A ,       reduction=jnp.mean					)
   __lowerCAmelCase  :      Optional[int]    =						cross_entropy(__A ,       __A					)
   __lowerCAmelCase  :      Union[str, Any]    =						cross_entropy(__A ,       __A					)
   __lowerCAmelCase  :      List[str]    =						cross_entropy(__A ,       __A					)
   return (start_loss + end_loss + pooled_loss) / 3



@dataclass
class SCREAMING_SNAKE_CASE     :
  """simple docstring"""






  lowerCamelCase   :      str										="google/bigbird-roberta-base"
  lowerCamelCase   :      int										=3000
  lowerCamelCase   :      int										=1_0500

  lowerCamelCase   :      int										=128
  lowerCamelCase   :      int										=3

  lowerCamelCase   :      int										=1
  lowerCamelCase   :      int										=5

  # tx_args
  lowerCamelCase   :      float										=3e-5
  lowerCamelCase   :      float										=0.0
  lowerCamelCase   :      int										=2_0000
  lowerCamelCase   :      float										=0.0095

  lowerCamelCase   :      str										="bigbird-roberta-natural-questions"
  lowerCamelCase   :      str										="training-expt"
  lowerCamelCase   :      str										="data/nq-training.jsonl"
  lowerCamelCase   :      str										="data/nq-validation.jsonl"







  def 	SCREAMING_SNAKE_CASE   (      self			:   List[str]   )							->     int:






     """simple docstring"""







     os.makedirs(self.base_dir	,							exist_ok=lowerCAmelCase   )
     __lowerCAmelCase  :      Union[str, Any]    =						os.path.join(self.base_dir	,							self.save_dir   )
     __lowerCAmelCase  :      Optional[Any]    =						self.batch_size_per_device * jax.device_count()






@dataclass
class SCREAMING_SNAKE_CASE     :
  """simple docstring"""






  lowerCamelCase   :      int
  lowerCamelCase   :      int										=4096  # no dynamic padding on TPUs







  def __call__(      self			:   str	,							lowerCAmelCase			:   Dict   )							->     Optional[Any]:






     """simple docstring"""







     __lowerCAmelCase  :      Optional[int]    =						self.collate_fn(lowerCAmelCase   )
     __lowerCAmelCase  :      List[str]    =						jax.tree_util.tree_map(lowerCAmelCase	,							lowerCAmelCase   )
     return batch







  def 	SCREAMING_SNAKE_CASE   (      self			:   Any	,							lowerCAmelCase			:   Optional[int]   )							->     Dict:






     """simple docstring"""







     __lowerCAmelCase   ,__lowerCAmelCase  :      Union[str, Any]    =						self.fetch_inputs(features["""input_ids"""]   )
     __lowerCAmelCase  :      int    =						{
         """input_ids""": jnp.array(lowerCAmelCase	,							dtype=jnp.intaa   ),
         """attention_mask""": jnp.array(lowerCAmelCase	,							dtype=jnp.intaa   ),
         """start_labels""": jnp.array(features["""start_token"""]	,							dtype=jnp.intaa   ),
         """end_labels""": jnp.array(features["""end_token"""]	,							dtype=jnp.intaa   ),
         """pooled_labels""": jnp.array(features["""category"""]	,							dtype=jnp.intaa   ),
     }
     return batch







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






     """simple docstring"""







     __lowerCAmelCase  :      Union[str, Any]    =						[self._fetch_inputs(lowerCAmelCase   ) for ids in input_ids]
     return zip(*lowerCAmelCase   )







  def 	SCREAMING_SNAKE_CASE   (      self			:   List[Any]	,							lowerCAmelCase			:   list   )							->     int:






     """simple docstring"""







     __lowerCAmelCase  :      Union[str, Any]    =						[1 for _ in range(len(lowerCAmelCase   )   )]
     while len(lowerCAmelCase   ) < self.max_length:
        input_ids.append(self.pad_id   )
        attention_mask.append(0   )
     return input_ids, attention_mask






def      snake_case_				(__A					:						int ,       __A					:						List[Any] ,       __A					:						str=None					)      ->     Dict:
   if seed is not None:
      __lowerCAmelCase  :      Optional[Any]    =						dataset.shuffle(seed=__A					)
   for i in range(len(__A					) // batch_size					):
      __lowerCAmelCase  :      Optional[int]    =						dataset[i * batch_size : (i + 1) * batch_size]
      yield dict(__A					)





@partial(jax.pmap ,       axis_name="""batch"""					)
def      snake_case_				(__A					:						Optional[Any] ,       __A					:						Dict ,       **__A					:						str					)      ->     List[str]:
   def loss_fn(__A					:						Any					):
      __lowerCAmelCase  :      Dict    =						model_inputs.pop("""start_labels"""					)
      __lowerCAmelCase  :      str    =						model_inputs.pop("""end_labels"""					)
      __lowerCAmelCase  :      int    =						model_inputs.pop("""pooled_labels"""					)

      __lowerCAmelCase  :      List[str]    =						state.apply_fn(**__A ,       params=__A ,       dropout_rng=__A ,       train=__A					)
      __lowerCAmelCase   ,__lowerCAmelCase   ,__lowerCAmelCase  :      Tuple    =						outputs

      return state.loss_fn(
          __A ,       __A ,       __A ,       __A ,       __A ,       __A ,       )

   __lowerCAmelCase   ,__lowerCAmelCase  :      Any    =						jax.random.split(__A					)
   __lowerCAmelCase  :      Optional[int]    =						jax.value_and_grad(__A					)
   __lowerCAmelCase   ,__lowerCAmelCase  :      List[Any]    =						grad_fn(state.params					)
   __lowerCAmelCase  :      Tuple    =						jax.lax.pmean({"""loss""": loss} ,       axis_name="""batch"""					)
   __lowerCAmelCase  :      Optional[Any]    =						jax.lax.pmean(__A ,       """batch"""					)

   __lowerCAmelCase  :      Dict    =						state.apply_gradients(grads=__A					)
   return state, metrics, new_drp_rng





@partial(jax.pmap ,       axis_name="""batch"""					)
def      snake_case_				(__A					:						int ,       **__A					:						Optional[Any]					)      ->     Any:
   __lowerCAmelCase  :      List[Any]    =						model_inputs.pop("""start_labels"""					)
   __lowerCAmelCase  :      Tuple    =						model_inputs.pop("""end_labels"""					)
   __lowerCAmelCase  :      Union[str, Any]    =						model_inputs.pop("""pooled_labels"""					)

   __lowerCAmelCase  :      List[Any]    =						state.apply_fn(**__A ,       params=state.params ,       train=__A					)
   __lowerCAmelCase   ,__lowerCAmelCase   ,__lowerCAmelCase  :      List[Any]    =						outputs

   __lowerCAmelCase  :      Optional[Any]    =						state.loss_fn(__A ,       __A ,       __A ,       __A ,       __A ,       __A					)
   __lowerCAmelCase  :      int    =						jax.lax.pmean({"""loss""": loss} ,       axis_name="""batch"""					)
   return metrics



class SCREAMING_SNAKE_CASE     (  train_state.TrainState			):
  """simple docstring"""






  lowerCamelCase   :      Callable										=struct.field(pytree_node=a_			)






@dataclass
class SCREAMING_SNAKE_CASE     :
  """simple docstring"""






  lowerCamelCase   :      Args
  lowerCamelCase   :      Callable
  lowerCamelCase   :      Callable
  lowerCamelCase   :      Callable
  lowerCamelCase   :      Callable
  lowerCamelCase   :      wandb
  lowerCamelCase   :      Callable										=None







  def 	SCREAMING_SNAKE_CASE   (      self			:   Optional[Any]	,							lowerCAmelCase			:   Optional[Any]	,							lowerCAmelCase			:   Union[str, Any]	,							lowerCAmelCase			:   str	,							lowerCAmelCase			:   Any=None   )							->     str:






     """simple docstring"""







     __lowerCAmelCase  :      Optional[Any]    =						model.params
     __lowerCAmelCase  :      Optional[Any]    =						TrainState.create(
         apply_fn=model.__call__	,							params=lowerCAmelCase	,							tx=lowerCAmelCase	,							loss_fn=lowerCAmelCase	,							)
     if ckpt_dir is not None:
        __lowerCAmelCase   ,__lowerCAmelCase   ,__lowerCAmelCase   ,__lowerCAmelCase   ,__lowerCAmelCase  :      Optional[int]    =						restore_checkpoint(lowerCAmelCase	,							lowerCAmelCase   )
        __lowerCAmelCase  :      Tuple    =						{
            """lr""": args.lr,
            """init_lr""": args.init_lr,
            """warmup_steps""": args.warmup_steps,
            """num_train_steps""": num_train_steps,
            """weight_decay""": args.weight_decay,
        }
        __lowerCAmelCase   ,__lowerCAmelCase  :      List[str]    =						build_tx(**lowerCAmelCase   )
        __lowerCAmelCase  :      str    =						train_state.TrainState(
            step=lowerCAmelCase	,							apply_fn=model.__call__	,							params=lowerCAmelCase	,							tx=lowerCAmelCase	,							opt_state=lowerCAmelCase	,							)
        __lowerCAmelCase  :      Union[str, Any]    =						args
        __lowerCAmelCase  :      Tuple    =						data_collator
        __lowerCAmelCase  :      Optional[Any]    =						lr
        __lowerCAmelCase  :      int    =						params
     __lowerCAmelCase  :      Any    =						jax_utils.replicate(lowerCAmelCase   )
     return state







  def 	SCREAMING_SNAKE_CASE   (      self			:   Any	,							lowerCAmelCase			:   Any	,							lowerCAmelCase			:   Optional[Any]	,							lowerCAmelCase			:   Tuple   )							->     int:






     """simple docstring"""







     __lowerCAmelCase  :      int    =						self.args
     __lowerCAmelCase  :      Optional[int]    =						len(lowerCAmelCase   ) // args.batch_size

     __lowerCAmelCase  :      Tuple    =						jax.random.PRNGKey(0   )
     __lowerCAmelCase  :      Tuple    =						jax.random.split(lowerCAmelCase	,							jax.device_count()   )
     for epoch in range(args.max_epochs   ):
        __lowerCAmelCase  :      Tuple    =						jnp.array(0	,							dtype=jnp.floataa   )
        __lowerCAmelCase  :      List[Any]    =						get_batched_dataset(lowerCAmelCase	,							args.batch_size	,							seed=lowerCAmelCase   )
        __lowerCAmelCase  :      int    =						0
        for batch in tqdm(lowerCAmelCase	,							total=lowerCAmelCase	,							desc=f'''Running EPOCH-{epoch}'''   ):
           __lowerCAmelCase  :      Tuple    =						self.data_collator(lowerCAmelCase   )
           __lowerCAmelCase   ,__lowerCAmelCase   ,__lowerCAmelCase  :      str    =						self.train_step_fn(lowerCAmelCase	,							lowerCAmelCase	,							**lowerCAmelCase   )
           running_loss += jax_utils.unreplicate(metrics["""loss"""]   )
           i += 1
           if i % args.logging_steps == 0:
              __lowerCAmelCase  :      Union[str, Any]    =						jax_utils.unreplicate(state.step   )
              __lowerCAmelCase  :      Optional[int]    =						running_loss.item() / i
              __lowerCAmelCase  :      Tuple    =						self.scheduler_fn(state_step - 1   )

              __lowerCAmelCase  :      List[Any]    =						self.evaluate(lowerCAmelCase	,							lowerCAmelCase   )
              __lowerCAmelCase  :      int    =						{
                  """step""": state_step.item(),
                  """eval_loss""": eval_loss.item(),
                  """tr_loss""": tr_loss,
                  """lr""": lr.item(),
              }
              tqdm.write(str(lowerCAmelCase   )   )
              self.logger.log(lowerCAmelCase	,							commit=lowerCAmelCase   )

           if i % args.save_steps == 0:
              self.save_checkpoint(args.save_dir + f'''-e{epoch}-s{i}'''	,							state=lowerCAmelCase   )







  def 	SCREAMING_SNAKE_CASE   (      self			:   List[Any]	,							lowerCAmelCase			:   Union[str, Any]	,							lowerCAmelCase			:   int   )							->     Optional[Any]:






     """simple docstring"""







     __lowerCAmelCase  :      Optional[int]    =						get_batched_dataset(lowerCAmelCase	,							self.args.batch_size   )
     __lowerCAmelCase  :      Optional[int]    =						len(lowerCAmelCase   ) // self.args.batch_size
     __lowerCAmelCase  :      str    =						jnp.array(0	,							dtype=jnp.floataa   )
     __lowerCAmelCase  :      Optional[int]    =						0
     for batch in tqdm(lowerCAmelCase	,							total=lowerCAmelCase	,							desc="""Evaluating ... """   ):
        __lowerCAmelCase  :      List[Any]    =						self.data_collator(lowerCAmelCase   )
        __lowerCAmelCase  :      Dict    =						self.val_step_fn(lowerCAmelCase	,							**lowerCAmelCase   )
        running_loss += jax_utils.unreplicate(metrics["""loss"""]   )
        i += 1
     return running_loss / i







  def 	SCREAMING_SNAKE_CASE   (      self			:   List[Any]	,							lowerCAmelCase			:   List[Any]	,							lowerCAmelCase			:   Optional[int]   )							->     List[Any]:






     """simple docstring"""







     __lowerCAmelCase  :      Union[str, Any]    =						jax_utils.unreplicate(lowerCAmelCase   )
     print(f'''SAVING CHECKPOINT IN {save_dir}'''	,							end=""" ... """   )
     self.model_save_fn(lowerCAmelCase	,							params=state.params   )
     with open(os.path.join(lowerCAmelCase	,							"""opt_state.msgpack"""   )	,							"""wb"""   ) as f:
        f.write(to_bytes(state.opt_state   )   )
     joblib.dump(self.args	,							os.path.join(lowerCAmelCase	,							"""args.joblib"""   )   )
     joblib.dump(self.data_collator	,							os.path.join(lowerCAmelCase	,							"""data_collator.joblib"""   )   )
     with open(os.path.join(lowerCAmelCase	,							"""training_state.json"""   )	,							"""w"""   ) as f:
        json.dump({"""step""": state.step.item()}	,							lowerCAmelCase   )
     print("""DONE"""   )






def      snake_case_				(__A					:						int ,       __A					:						List[str]					)      ->     int:
   print(f'''RESTORING CHECKPOINT FROM {save_dir}''' ,       end=""" ... """					)
   with open(os.path.join(__A ,       """flax_model.msgpack"""					) ,       """rb"""					) as f:
      __lowerCAmelCase  :      Optional[Any]    =						from_bytes(state.params ,       f.read()					)

   with open(os.path.join(__A ,       """opt_state.msgpack"""					) ,       """rb"""					) as f:
      __lowerCAmelCase  :      Union[str, Any]    =						from_bytes(state.opt_state ,       f.read()					)

   __lowerCAmelCase  :      Optional[int]    =						joblib.load(os.path.join(__A ,       """args.joblib"""					)					)
   __lowerCAmelCase  :      Optional[Any]    =						joblib.load(os.path.join(__A ,       """data_collator.joblib"""					)					)

   with open(os.path.join(__A ,       """training_state.json"""					) ,       """r"""					) as f:
      __lowerCAmelCase  :      List[str]    =						json.load(__A					)
   __lowerCAmelCase  :      Any    =						training_state["""step"""]

   print("""DONE"""					)
   return params, opt_state, step, args, data_collator





def      snake_case_				(__A					:						Any ,       __A					:						List[Any] ,       __A					:						str ,       __A					:						List[str]					)      ->     Dict:
   __lowerCAmelCase  :      int    =						num_train_steps - warmup_steps
   __lowerCAmelCase  :      Any    =						optax.linear_schedule(init_value=__A ,       end_value=__A ,       transition_steps=__A					)
   __lowerCAmelCase  :      Optional[int]    =						optax.linear_schedule(init_value=__A ,       end_value=1e-7 ,       transition_steps=__A					)
   __lowerCAmelCase  :      Any    =						optax.join_schedules(schedules=[warmup_fn, decay_fn] ,       boundaries=[warmup_steps]					)
   return lr





def      snake_case_				(__A					:						Dict ,       __A					:						Tuple ,       __A					:						str ,       __A					:						Tuple ,       __A					:						Optional[int]					)      ->     List[Any]:
   def weight_decay_mask(__A					:						Union[str, Any]					):
      __lowerCAmelCase  :      Optional[Any]    =						traverse_util.flatten_dict(__A					)
      __lowerCAmelCase  :      Any    =						{k: (v[-1] != """bias""" and v[-2:] != ("""LayerNorm""", """scale""")) for k, v in params.items()}
      return traverse_util.unflatten_dict(__A					)

   __lowerCAmelCase  :      int    =						scheduler_fn(__A ,       __A ,       __A ,       __A					)

   __lowerCAmelCase  :      Any    =						optax.adamw(learning_rate=__A ,       weight_decay=__A ,       mask=__A					)
   return tx, lr





import copy
from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import Audio, ClassLabel, Features
from .base import TaskTemplate



@dataclass(frozen=a_			)
class SCREAMING_SNAKE_CASE     (  a_			):
  """simple docstring"""






  lowerCamelCase   :      str										=field(default="audio-classification"       ,							metadata={"include_in_asdict_even_if_is_default": True}			)
  lowerCamelCase   :      ClassVar[Features]										=Features({"audio": Audio()}			)
  lowerCamelCase   :      ClassVar[Features]										=Features({"labels": ClassLabel}			)
  lowerCamelCase   :      str										="audio"
  lowerCamelCase   :      str										="labels"







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






     """simple docstring"""







     if self.label_column not in features:
        raise ValueError(f'''Column {self.label_column} is not present in features.'''   )
     if not isinstance(features[self.label_column]	,							lowerCAmelCase   ):
        raise ValueError(f'''Column {self.label_column} is not a ClassLabel.'''   )
     __lowerCAmelCase  :      Tuple    =						copy.deepcopy(self   )
     __lowerCAmelCase  :      Tuple    =						self.label_schema.copy()
     __lowerCAmelCase  :      Union[str, Any]    =						features[self.label_column]
     __lowerCAmelCase  :      Any    =						label_schema
     return task_template







  @property
  def 	SCREAMING_SNAKE_CASE   (      self			:   str   )							->     Dict[str, str]:






     """simple docstring"""







     return {
         self.audio_column: "audio",
         self.label_column: "labels",
     }


