code: 
  
  
  """simple docstring"""
  import random
  import shutil
  import tempfile
  import unittest
  
  import numpy as np
  
  from transformers import TaTokenizer, TaTokenizerFast
  from transformers.testing_utils import require_sentencepiece, require_torch
  from transformers.utils.import_utils import is_speech_available, is_torch_available
  
  
  if is_torch_available():
       pass
  
  if is_speech_available():
       from transformers import EncodecFeatureExtractor, MusicgenProcessor
  
  
  UpperCAmelCase   =	random.Random()
  
  def 							_snake_case			(      __snake_case       :			Optional[int]	,   __snake_case       :			Dict=1.0	,   __snake_case       :			Any=None	,   __snake_case       :			Optional[Any]=None      ):
  
  
  
  
     """simple docstring"""
  
  
  
  
  
  
  
     if rng is None:
        _lowerCamelCase  :				Union[str, Any]      =		global_rng
  
     _lowerCamelCase  :				Dict      =		[]
     for batch_idx in range(shape[0]      ):
        values.append([]      )
        for _ in range(shape[1]      ):
           values[-1].append(rng.random() * scale      )
  
     return values
  
  
  
  
  
  @require_torch
  @require_sentencepiece
  class   lowercase__							(				unittest.TestCase							):
  
  
  
  
  
     def 		UpperCamelCase_     (       self)     ->     Dict:
        _lowerCamelCase  :				Any      =		"facebook/musicgen-small"
        _lowerCamelCase  :				Any      =		tempfile.mkdtemp()
  
  
  
  
  
     def 		UpperCamelCase_     (       self			,      **SCREAMING_SNAKE_CASE)     ->     Optional[int]:
        return TaTokenizer.from_pretrained(self.checkpoint			,      **lowerCAmelCase__)
  
  
  
  
  
     def 		UpperCamelCase_     (       self			,      **SCREAMING_SNAKE_CASE)     ->     str:
        return EncodecFeatureExtractor.from_pretrained(self.checkpoint			,      **lowerCAmelCase__)
  
  
  
  
  
     def 		UpperCamelCase_     (       self)     ->     List[str]:
        shutil.rmtree(self.tmpdirname)
  
  
  
  
  
     def 		UpperCamelCase_     (       self)     ->     List[str]:
        _lowerCamelCase  :				List[Any]      =		self.get_tokenizer()
        _lowerCamelCase  :				str      =		self.get_feature_extractor()
  
        _lowerCamelCase  :				Optional[Any]      =		MusicgenProcessor(tokenizer=lowerCAmelCase__			,      feature_extractor=lowerCAmelCase__)
  
        processor.save_pretrained(self.tmpdirname)
        _lowerCamelCase  :				List[str]      =		MusicgenProcessor.from_pretrained(self.tmpdirname)
  
        self.assertEqual(processor.tokenizer.get_vocab()			,      tokenizer.get_vocab())
        self.assertIsInstance(processor.tokenizer			,      lowerCAmelCase__)
  
        self.assertEqual(processor.feature_extractor.to_json_string()			,      feature_extractor.to_json_string())
        self.assertIsInstance(processor.feature_extractor			,      lowerCAmelCase__)
  
  
  
  
  
     def 		UpperCamelCase_     (       self)     ->     Optional[int]:
        _lowerCamelCase  :				List[Any]      =		MusicgenProcessor(tokenizer=self.get_tokenizer()			,      feature_extractor=self.get_feature_extractor())
        processor.save_pretrained(self.tmpdirname)
  
        _lowerCamelCase  :				Optional[Any]      =		self.get_tokenizer(bos_token="""(BOS)"""			,      eos_token="""(EOS)""")
        _lowerCamelCase  :				Union[str, Any]      =		self.get_feature_extractor(do_normalize=lowerCAmelCase__			,      padding_value=1.0)
  
        _lowerCamelCase  :				str      =		MusicgenProcessor.from_pretrained(
            self.tmpdirname			,      bos_token="""(BOS)"""			,      eos_token="""(EOS)"""			,      do_normalize=lowerCAmelCase__			,      padding_value=1.0)
  
        self.assertEqual(processor.tokenizer.get_vocab()			,      tokenizer_add_kwargs.get_vocab())
        self.assertIsInstance(processor.tokenizer			,      lowerCAmelCase__)
  
        self.assertEqual(processor.feature_extractor.to_json_string()			,      feature_extractor_add_kwargs.to_json_string())
        self.assertIsInstance(processor.feature_extractor			,      lowerCAmelCase__)
  
  
  
  
  
     def 		UpperCamelCase_     (       self)     ->     Dict:
        _lowerCamelCase  :				Tuple      =		self.get_feature_extractor()
        _lowerCamelCase  :				Optional[int]      =		self.get_tokenizer()
  
        _lowerCamelCase  :				str      =		MusicgenProcessor(tokenizer=lowerCAmelCase__			,      feature_extractor=lowerCAmelCase__)
  
        _lowerCamelCase  :				int      =		floats_list((3, 1000))
  
        _lowerCamelCase  :				List[Any]      =		feature_extractor(lowerCAmelCase__			,      return_tensors="""np""")
        _lowerCamelCase  :				Union[str, Any]      =		processor(lowerCAmelCase__			,      return_tensors="""np""")
  
        for key in input_feat_extract.keys():
           self.assertAlmostEqual(input_feat_extract[key].sum()			,      input_processor[key].sum()			,      delta=1e-2)
  
  
  
  
  
     def 		UpperCamelCase_     (       self)     ->     Dict:
        _lowerCamelCase  :				Any      =		self.get_feature_extractor()
        _lowerCamelCase  :				Dict      =		self.get_tokenizer()
  
        _lowerCamelCase  :				Optional[int]      =		MusicgenProcessor(tokenizer=lowerCAmelCase__			,      feature_extractor=lowerCAmelCase__)
  
        _lowerCamelCase  :				Any      =		"This is a test string"
  
        _lowerCamelCase  :				Optional[int]      =		processor(text=lowerCAmelCase__)
  
        _lowerCamelCase  :				int      =		tokenizer(lowerCAmelCase__)
  
        for key in encoded_tok.keys():
           self.assertListEqual(encoded_tok[key]			,      encoded_processor[key])
  
  
  
  
  
     def 		UpperCamelCase_     (       self)     ->     int:
        _lowerCamelCase  :				Optional[Any]      =		self.get_feature_extractor()
        _lowerCamelCase  :				Any      =		self.get_tokenizer()
  
        _lowerCamelCase  :				Tuple      =		MusicgenProcessor(tokenizer=lowerCAmelCase__			,      feature_extractor=lowerCAmelCase__)
  
        _lowerCamelCase  :				List[Any]      =		[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]
  
        _lowerCamelCase  :				Dict      =		processor.batch_decode(sequences=lowerCAmelCase__)
        _lowerCamelCase  :				Any      =		tokenizer.batch_decode(lowerCAmelCase__)
  
        self.assertListEqual(lowerCAmelCase__			,      lowerCAmelCase__)
  
  
  
  
  
     def 		UpperCamelCase_     (       self)     ->     Union[str, Any]:
        _lowerCamelCase  :				Any      =		self.get_feature_extractor()
        _lowerCamelCase  :				List[str]      =		self.get_tokenizer()
  
        _lowerCamelCase  :				List[str]      =		MusicgenProcessor(tokenizer=lowerCAmelCase__			,      feature_extractor=lowerCAmelCase__)
  
        self.assertListEqual(
            processor.model_input_names			,      feature_extractor.model_input_names			,      msg="""`processor` and `feature_extractor` model input names do not match"""			,      )
  
  
  
  
  
     def 		UpperCamelCase_     (       self)     ->     str:
        _lowerCamelCase  :				Dict      =		self.get_feature_extractor(padding_side="""left""")
        _lowerCamelCase  :				Any      =		self.get_tokenizer()
  
        _lowerCamelCase  :				List[str]      =		MusicgenProcessor(tokenizer=lowerCAmelCase__			,      feature_extractor=lowerCAmelCase__)
  
        _lowerCamelCase  :				List[str]      =		[floats_list((1, x))[0] for x in range(5			,      20			,      5)]
        _lowerCamelCase  :				List[Any]      =		processor(lowerCAmelCase__).padding_mask
  
        _lowerCamelCase  :				Optional[int]      =		np.asarray(floats_list((3, 20)))[:, None, :]
        _lowerCamelCase  :				Union[str, Any]      =		processor.batch_decode(lowerCAmelCase__			,      padding_mask=lowerCAmelCase__)
  
        self.assertIsInstance(lowerCAmelCase__			,      lowerCAmelCase__)
  
        for audio in decoded_audios:
           self.assertIsInstance(lowerCAmelCase__			,      np.ndarray)
  
        self.assertTrue(decoded_audios[0].shape == (1, 10))
        self.assertTrue(decoded_audios[1].shape == (1, 15))
        self.assertTrue(decoded_audios[2].shape == (1, 20))
  
  
style_context: 
  
  
  """simple docstring"""
  import os
  
  from apache_beam.io.filesystems import FileSystems
  from apache_beam.pipeline import Pipeline
  
  from .logging import get_logger
  
  
  UpperCAmelCase   =	2 << 20  # 2mb
  UpperCAmelCase   =	get_logger(__name__)
  
  
  
  
  class   lowercase__							(				A_							):
  
  
  
  
  
     def 		UpperCamelCase_     (       self)     ->     Dict:
        _lowerCamelCase  :				Tuple      =		self._options.get_all_options().get("""runner""")
        return runner in [None, "DirectRunner", "PortableRunner"]
  
  
  
  
  
  
  
  def 							_snake_case			(      __snake_case       :			Union[str, Any]	,   __snake_case       :			List[Any]	,   __snake_case       :			Any=False      ):
  
  
  
  
     """simple docstring"""
  
  
  
  
  
  
  
     _lowerCamelCase  :				Any      =		FileSystems
     if fs.exists(__snake_case      ):
        if force_upload:
           logger.info(F'Remote path already exist: {remote_file_path}. Overwriting it as force_upload=True.'      )
        else:
           logger.info(F'Remote path already exist: {remote_file_path}. Skipping it as force_upload=False.'      )
           return
     with fs.create(__snake_case      ) as remote_file:
        with open(__snake_case	,   """rb"""      ) as local_file:
           _lowerCamelCase  :				Optional[Any]      =		local_file.read(__snake_case      )
           while chunk:
              remote_file.write(__snake_case      )
              _lowerCamelCase  :				Tuple      =		local_file.read(__snake_case      )
  
  
  
  
  
  def 							_snake_case			(      __snake_case       :			Union[str, Any]	,   __snake_case       :			List[Any]	,   __snake_case       :			Union[str, Any]=False      ):
  
  
  
  
     """simple docstring"""
  
  
  
  
  
  
  
     _lowerCamelCase  :				int      =		FileSystems
     if os.path.exists(__snake_case      ):
        if force_download:
           logger.info(F'Local path already exist: {remote_file_path}. Overwriting it as force_upload=True.'      )
        else:
           logger.info(F'Local path already exist: {remote_file_path}. Skipping it as force_upload=False.'      )
           return
     with fs.open(__snake_case      ) as remote_file:
        with open(__snake_case	,   """wb"""      ) as local_file:
           _lowerCamelCase  :				str      =		remote_file.read(__snake_case      )
           while chunk:
              local_file.write(__snake_case      )
              _lowerCamelCase  :				Any      =		remote_file.read(__snake_case      )
  
  