


import os
import tempfile
import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch


if is_torch_available():
   import torch
   from torch import nn

   from transformers import (
       Adafactor,
       AdamW,
       get_constant_schedule,
       get_constant_schedule_with_warmup,
       get_cosine_schedule_with_warmup,
       get_cosine_with_hard_restarts_schedule_with_warmup,
       get_inverse_sqrt_schedule,
       get_linear_schedule_with_warmup,
       get_polynomial_decay_schedule_with_warmup,
   )






def  __lowerCAmelCase					(SCREAMING_SNAKE_CASE      ,      SCREAMING_SNAKE_CASE=10      )->						List[Any]:




      """simple docstring"""

      snake_case_ =   []
      for _ in range(_snake_case      ):
            lrs.append(scheduler.get_lr()[0]      )
            scheduler.step()
      return lrs






def  __lowerCAmelCase					(SCREAMING_SNAKE_CASE      ,      SCREAMING_SNAKE_CASE=10      )->						str:




      """simple docstring"""

      snake_case_ =   []
      for step in range(_snake_case      ):
            lrs.append(scheduler.get_lr()[0]      )
            scheduler.step()
            if step == num_steps // 2:
                  with tempfile.TemporaryDirectory() as tmpdirname:
                        snake_case_ =   os.path.join(_snake_case      ,      '''schedule.bin'''      )
                        torch.save(scheduler.state_dict()      ,      _snake_case      )

                        snake_case_ =   torch.load(_snake_case      )
                        scheduler.load_state_dict(_snake_case      )
      return lrs







@require_torch
class lowerCAmelCase_		(				unittest.TestCase ):

 '''simple docstring'''

 def 		UpperCamelCase__  (			self	,   _UpperCAmelCase	,   _UpperCAmelCase	,   _UpperCAmelCase      ):
       self.assertEqual(len(UpperCAmelCase_      )	,   len(UpperCAmelCase_      )      )
       for a, b in zip(UpperCAmelCase_	,   UpperCAmelCase_      ):
             self.assertAlmostEqual(UpperCAmelCase_	,   UpperCAmelCase_	,   delta=UpperCAmelCase_      )



 def 		UpperCamelCase__  (			self      ):
       snake_case_ =   torch.tensor([0.1, -0.2, -0.1]	,   requires_grad=UpperCAmelCase_      )
       snake_case_ =   torch.tensor([0.4, 0.2, -0.5]      )
       snake_case_ =   nn.MSELoss()
       # No warmup, constant schedule, no gradient clipping
       snake_case_ =   AdamW(params=[w]	,   lr=2E-1	,   weight_decay=0.0      )
       for _ in range(1_00      ):
             snake_case_ =   criterion(UpperCAmelCase_	,   UpperCAmelCase_      )
             loss.backward()
             optimizer.step()
             w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
             w.grad.zero_()
       self.assertListAlmostEqual(w.tolist()	,   [0.4, 0.2, -0.5]	,   tol=1E-2      )





 def 		UpperCamelCase__  (			self      ):
       snake_case_ =   torch.tensor([0.1, -0.2, -0.1]	,   requires_grad=UpperCAmelCase_      )
       snake_case_ =   torch.tensor([0.4, 0.2, -0.5]      )
       snake_case_ =   nn.MSELoss()
       # No warmup, constant schedule, no gradient clipping
       snake_case_ =   Adafactor(
           params=[w]	,   lr=1E-2	,   eps=(1E-30, 1E-3)	,   clip_threshold=1.0	,   decay_rate=-0.8	,   betaa=UpperCAmelCase_	,   weight_decay=0.0	,   relative_step=UpperCAmelCase_	,   scale_parameter=UpperCAmelCase_	,   warmup_init=UpperCAmelCase_	,   )
       for _ in range(10_00      ):
             snake_case_ =   criterion(UpperCAmelCase_	,   UpperCAmelCase_      )
             loss.backward()
             optimizer.step()
             w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
             w.grad.zero_()
       self.assertListAlmostEqual(w.tolist()	,   [0.4, 0.2, -0.5]	,   tol=1E-2      )







@require_torch
class lowerCAmelCase_		(				unittest.TestCase ):

 '''simple docstring'''

 __snake_case				=						nn.Linear(5_0      ,		5_0 ) if is_torch_available() else None
 __snake_case				=						AdamW(m.parameters()      ,		lr=10.0 ) if is_torch_available() else None
 __snake_case				=						1_0



 def 		UpperCamelCase__  (			self	,   _UpperCAmelCase	,   _UpperCAmelCase	,   _UpperCAmelCase	,   _UpperCAmelCase=None      ):
       self.assertEqual(len(UpperCAmelCase_      )	,   len(UpperCAmelCase_      )      )
       for a, b in zip(UpperCAmelCase_	,   UpperCAmelCase_      ):
             self.assertAlmostEqual(UpperCAmelCase_	,   UpperCAmelCase_	,   delta=UpperCAmelCase_	,   msg=UpperCAmelCase_      )





 def 		UpperCamelCase__  (			self      ):
       snake_case_ =   {'''num_warmup_steps''': 2, '''num_training_steps''': 10}
       # schedulers doct format
       # function: (sched_args_dict, expected_learning_rates)
       snake_case_ =   {
           get_constant_schedule: ({}, [10.0] * self.num_steps),
           get_constant_schedule_with_warmup: (
               {'''num_warmup_steps''': 4},
               [0.0, 2.5, 5.0, 7.5, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0],
           ),
           get_linear_schedule_with_warmup: (
               {**common_kwargs},
               [0.0, 5.0, 10.0, 8.75, 7.5, 6.25, 5.0, 3.75, 2.5, 1.25],
           ),
           get_cosine_schedule_with_warmup: (
               {**common_kwargs},
               [0.0, 5.0, 10.0, 9.61, 8.53, 6.91, 5.0, 3.08, 1.46, 0.38],
           ),
           get_cosine_with_hard_restarts_schedule_with_warmup: (
               {**common_kwargs, '''num_cycles''': 2},
               [0.0, 5.0, 10.0, 8.53, 5.0, 1.46, 10.0, 8.53, 5.0, 1.46],
           ),
           get_polynomial_decay_schedule_with_warmup: (
               {**common_kwargs, '''power''': 2.0, '''lr_end''': 1E-7},
               [0.0, 5.0, 10.0, 7.656, 5.625, 3.906, 2.5, 1.406, 0.625, 0.156],
           ),
           get_inverse_sqrt_schedule: (
               {'''num_warmup_steps''': 2},
               [0.0, 5.0, 10.0, 8.165, 7.071, 6.325, 5.774, 5.345, 5.0, 4.714],
           ),
       }

       for scheduler_func, data in scheds.items():
             snake_case_ ,					snake_case_ =   data

             snake_case_ =   scheduler_func(self.optimizer	,   **UpperCAmelCase_      )
             self.assertEqual(len([scheduler.get_lr()[0]]      )	,   1      )
             snake_case_ =   unwrap_schedule(UpperCAmelCase_	,   self.num_steps      )
             self.assertListAlmostEqual(
                 UpperCAmelCase_	,   UpperCAmelCase_	,   tol=1E-2	,   msg=F'''failed for {scheduler_func} in normal scheduler'''	,   )

             snake_case_ =   scheduler_func(self.optimizer	,   **UpperCAmelCase_      )
             if scheduler_func.__name__ != "get_constant_schedule":
                   LambdaScheduleWrapper.wrap_scheduler(UpperCAmelCase_      )  # wrap to test picklability of the schedule
             snake_case_ =   unwrap_and_save_reload_schedule(UpperCAmelCase_	,   self.num_steps      )
             self.assertListEqual(UpperCAmelCase_	,   UpperCAmelCase_	,   msg=F'''failed for {scheduler_func} in save and reload'''      )







class lowerCAmelCase_		:

 '''simple docstring'''

 def __init__(			self	,   _UpperCAmelCase      ):
       snake_case_ =   fn



 def __call__(			self	,   *_UpperCAmelCase	,   **_UpperCAmelCase      ):
       return self.fn(*UpperCAmelCase_	,   **UpperCAmelCase_      )





 @classmethod
 def 		UpperCamelCase__  (			self	,   _UpperCAmelCase      ):
       snake_case_ =   list(map(self	,   scheduler.lr_lambdas      )      )



import unittest

from huggingface_hub import hf_hub_download

from transformers import MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING, VideoMAEFeatureExtractor
from transformers.pipelines import VideoClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_decord,
    require_tf,
    require_torch,
    require_torch_or_tf,
    require_vision,
)

from .test_pipelines_common import ANY







@is_pipeline_test
@require_torch_or_tf
@require_vision
@require_decord
class lowerCAmelCase_		(				unittest.TestCase ):

 '''simple docstring'''

 __snake_case				=						MODEL_FOR_VIDEO_CLASSIFICATION_MAPPING



 def 		UpperCamelCase__  (			self	,   _UpperCAmelCase	,   _UpperCAmelCase	,   _UpperCAmelCase      ):
       snake_case_ =   hf_hub_download(
           repo_id='''nateraw/video-demo'''	,   filename='''archery.mp4'''	,   repo_type='''dataset'''      )
       snake_case_ =   VideoClassificationPipeline(model=_UpperCAmelCase	,   image_processor=_UpperCAmelCase	,   top_k=2      )
       snake_case_ =   [
           example_video_filepath,
           '''https://huggingface.co/datasets/nateraw/video-demo/resolve/main/archery.mp4''',
       ]
       return video_classifier, examples



 def 		UpperCamelCase__  (			self	,   _UpperCAmelCase	,   _UpperCAmelCase      ):
       for example in examples:
             snake_case_ =   video_classifier(_UpperCAmelCase      )

             self.assertEqual(
                 _UpperCAmelCase	,   [
                     {'''score''': ANY(_UpperCAmelCase      ), '''label''': ANY(_UpperCAmelCase      )},
                     {'''score''': ANY(_UpperCAmelCase      ), '''label''': ANY(_UpperCAmelCase      )},
                 ]	,   )



 @require_torch
 def 		UpperCamelCase__  (			self      ):
       snake_case_ =   '''hf-internal-testing/tiny-random-VideoMAEForVideoClassification'''
       snake_case_ =   VideoMAEFeatureExtractor(
           size={'''shortest_edge''': 10}	,   crop_size={'''height''': 10, '''width''': 10}      )
       snake_case_ =   pipeline(
           '''video-classification'''	,   model=_UpperCAmelCase	,   feature_extractor=_UpperCAmelCase	,   frame_sampling_rate=4      )

       snake_case_ =   hf_hub_download(repo_id='''nateraw/video-demo'''	,   filename='''archery.mp4'''	,   repo_type='''dataset'''      )
       snake_case_ =   video_classifier(_UpperCAmelCase	,   top_k=2      )
       self.assertEqual(
           nested_simplify(_UpperCAmelCase	,   decimals=4      )	,   [{'''score''': 0.5_199, '''label''': '''LABEL_0'''}, {'''score''': 0.4_801, '''label''': '''LABEL_1'''}]	,   )

       snake_case_ =   video_classifier(
           [
               video_file_path,
               video_file_path,
           ]	,   top_k=2	,   )
       self.assertEqual(
           nested_simplify(_UpperCAmelCase	,   decimals=4      )	,   [
               [{'''score''': 0.5_199, '''label''': '''LABEL_0'''}, {'''score''': 0.4_801, '''label''': '''LABEL_1'''}],
               [{'''score''': 0.5_199, '''label''': '''LABEL_0'''}, {'''score''': 0.4_801, '''label''': '''LABEL_1'''}],
           ]	,   )





 @require_tf
 def 		UpperCamelCase__  (			self      ):
       pass