

import argparse
import os
from dataclasses import dataclass

import torch
import torchaudio
from tqdm import tqdm

from transformers import AutoConfig, AutoModelForCTC, AutoProcessor



class     lowercase						:



 '''simple docstring'''


 def __init__(self       ,       __a       ,       __a       ,       __a ) ->					Union[str, Any]:






  """simple docstring"""





  UpperCAmelCase__							=				cuda
  UpperCAmelCase__							=				AutoConfig.from_pretrained(__a )
  UpperCAmelCase__							=				AutoModelForCTC.from_pretrained(__a )
  self.model.eval()
  if self.cuda:
   self.model.to(device='cuda' )
  UpperCAmelCase__							=				AutoProcessor.from_pretrained(__a )
  UpperCAmelCase__							=				torchaudio.transforms.Resample(__a       ,       16000 )
  UpperCAmelCase__							=				0
  UpperCAmelCase__							=				list(self.processor.tokenizer.get_vocab().keys() )
  for i in range(len(__a ) ):
   if vocab[i] == "[PAD]" or vocab[i] == "<pad>":
    UpperCAmelCase__							=				i
  print('Blank Token id [PAD]/<pad>'       ,       __a )
  UpperCAmelCase__							=				blank_id




 def   UpperCamelCase__							(self       ,       __a ) ->					List[str]:






  """simple docstring"""





  UpperCAmelCase__				,    UpperCAmelCase__							=				torchaudio.load(__a )
  UpperCAmelCase__							=				self.resampler(__a ).squeeze().numpy()
  return speech




 def   UpperCamelCase__							(self       ,       __a ) ->					Union[str, Any]:






  """simple docstring"""





  UpperCAmelCase__							=				self.blank_id
  UpperCAmelCase__							=				'|'.join(item['sent'].split(' ' ) )
  if not os.path.isfile(item['wav_path'] ):
   print(item['wav_path']       ,       'not found in wavs directory' )

  UpperCAmelCase__							=				self.speech_file_to_array_fn(item['wav_path'] )
  UpperCAmelCase__							=				self.processor(__a       ,       sampling_rate=16000       ,       return_tensors='pt'       ,       padding=__a )
  if self.cuda:
   UpperCAmelCase__							=				inputs.to(device='cuda' )

  with torch.no_grad():
   UpperCAmelCase__							=				self.model(inputs.input_values ).logits

  # get the emission probability at frame level
  UpperCAmelCase__							=				torch.log_softmax(__a       ,       dim=-1 )
  UpperCAmelCase__							=				emissions[0].cpu().detach()

  # get labels from vocab
  UpperCAmelCase__							=				([''] + list(self.processor.tokenizer.get_vocab().keys() ))[
      :-1
  ]  # logits don't align with the tokenizer's vocab

  UpperCAmelCase__							=				{c: i for i, c in enumerate(__a )}
  UpperCAmelCase__							=				[]
  for c in transcript:
   if c in dictionary:
    tokens.append(dictionary[c] )

  def get_trellis(__a       ,       __a       ,       __a=0 ):
   UpperCAmelCase__							=				emission.size(0 )
   UpperCAmelCase__							=				len(__a )

   # Trellis has extra diemsions for both time axis and tokens.
   # The extra dim for tokens represents <SoS> (start-of-sentence)
   # The extra dim for time axis is for simplification of the code.
   UpperCAmelCase__							=				torch.full((num_frames + 1, num_tokens + 1)       ,       -float('inf' ) )
   UpperCAmelCase__							=				0
   for t in range(__a ):
    UpperCAmelCase__							=				torch.maximum(
        # Score for staying at the same token
        trellis[t, 1:] + emission[t, blank_id]       ,       trellis[t, :-1] + emission[t, tokens]       ,       )
   return trellis

  UpperCAmelCase__							=				get_trellis(__a       ,       __a       ,       __a )



  @dataclass
  class     lowercase						:



   '''simple docstring'''


   __SCREAMING_SNAKE_CASE												=				42
   __SCREAMING_SNAKE_CASE												=				42
   __SCREAMING_SNAKE_CASE												=				42

  def backtrack(__a       ,       __a       ,       __a       ,       __a=0 ):
   UpperCAmelCase__							=				trellis.size(1 ) - 1
   UpperCAmelCase__							=				torch.argmax(trellis[:, j] ).item()

   UpperCAmelCase__							=				[]
   for t in range(__a       ,       0       ,       -1 ):
    # 1. Figure out if the current position was stay or change
    # Note (again):
    # `emission[J-1]` is the emission at time frame `J` of trellis dimension.
    # Score for token staying the same from time frame J-1 to T.
    UpperCAmelCase__							=				trellis[t - 1, j] + emission[t - 1, blank_id]
    # Score for token changing from C-1 at T-1 to J at T.
    UpperCAmelCase__							=				trellis[t - 1, j - 1] + emission[t - 1, tokens[j - 1]]

    # 2. Store the path with frame-wise probability.
    UpperCAmelCase__							=				emission[t - 1, tokens[j - 1] if changed > stayed else 0].exp().item()
    # Return token index and time index in non-trellis coordinate.
    path.append(Point(j - 1       ,       t - 1       ,       __a ) )

    # 3. Update the token
    if changed > stayed:
     j -= 1
     if j == 0:
      break
   else:
    raise ValueError('Failed to align' )
   return path[::-1]

  UpperCAmelCase__							=				backtrack(__a       ,       __a       ,       __a       ,       __a )



  @dataclass
  class     lowercase						:



   '''simple docstring'''


   __SCREAMING_SNAKE_CASE												=				42
   __SCREAMING_SNAKE_CASE												=				42
   __SCREAMING_SNAKE_CASE												=				42
   __SCREAMING_SNAKE_CASE												=				42




   def __repr__(self ) ->					List[Any]:






    """simple docstring"""





    return F"{self.label}\t{self.score:4.2f}\t{self.start*20:5d}\t{self.end*20:5d}"




   @property
   def   UpperCamelCase__							(self ) ->					Optional[Any]:






    """simple docstring"""





    return self.end - self.start

  def merge_repeats(__a ):
   UpperCAmelCase__				,    UpperCAmelCase__							=				0, 0
   UpperCAmelCase__							=				[]
   while ia < len(__a ):
    while ia < len(__a ) and path[ia].token_index == path[ia].token_index:
     ia += 1
    UpperCAmelCase__							=				sum(path[k].score for k in range(__a       ,       __a ) ) / (ia - ia)
    segments.append(
        Segment(
            transcript[path[ia].token_index]       ,       path[ia].time_index       ,       path[ia - 1].time_index + 1       ,       __a       ,       ) )
    UpperCAmelCase__							=				ia
   return segments

  UpperCAmelCase__							=				merge_repeats(__a )
  with open(item['out_path']       ,       'w' ) as out_align:
   for seg in segments:
    out_align.write(str(__a ) + '\n' )




 def   UpperCamelCase__							(self       ,       __a       ,       __a       ,       __a ) ->					Optional[Any]:






  """simple docstring"""





  if not os.path.exists(__a ):
   os.makedirs(__a )

  # load text file
  UpperCAmelCase__							=				open(__a       ,       encoding='utf8' ).readlines()

  UpperCAmelCase__							=				[]
  for line in lines:
   if len(line.strip().split('\t' ) ) != 2:
    print('Script must be in format: 00001  this is my sentence' )
    exit()

   UpperCAmelCase__				,    UpperCAmelCase__							=				line.strip().split('\t' )
   UpperCAmelCase__							=				os.path.join(__a       ,       wav_name + '.wav' )
   UpperCAmelCase__							=				os.path.join(__a       ,       wav_name + '.txt' )

   items.append({'sent': sentence, 'wav_path': wav_path, 'out_path': out_path} )
  print('Number of samples found in script file'       ,       len(__a ) )

  for item in tqdm(__a ):
   self.align_single_sample(__a )







def 						UpperCamelCase_(			)       -> List[str]:
 UpperCAmelCase__							=				argparse.ArgumentParser()

 parser.add_argument(
     '--model_name'  ,    type=snake_case__  ,    default='arijitx/wav2vec2-xls-r-300m-bengali'  ,    help='wav2vec model name'   )
 parser.add_argument('--wav_dir'  ,    type=snake_case__  ,    default='./wavs'  ,    help='directory containing wavs'   )
 parser.add_argument('--text_file'  ,    type=snake_case__  ,    default='script.txt'  ,    help='file containing text'   )
 parser.add_argument('--input_wavs_sr'  ,    type=snake_case__  ,    default=1_60_00  ,    help='sampling rate of input audios'   )
 parser.add_argument(
     '--output_dir'  ,    type=snake_case__  ,    default='./out_alignment'  ,    help='output directory containing the alignment files'   )
 parser.add_argument('--cuda'  ,    action='store_true'   )

 UpperCAmelCase__							=				parser.parse_args()

 UpperCAmelCase__							=				WavaVecaAligner(args.model_name  ,    args.input_wavs_sr  ,    args.cuda   )
 aligner.align_data(args.wav_dir  ,    args.text_file  ,    args.output_dir   )


if __name__ == "__main__":
 main()






import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


_UpperCamelCase			       =			namedtuple('''ExampleNamedTuple''', '''a b c''')



class     lowercase						(       unittest.TestCase					):



 '''simple docstring'''


 def   UpperCamelCase__							(self ) ->					Tuple:






  """simple docstring"""





  UpperCAmelCase__							=				torch.randn(5       ,       2 )
  UpperCAmelCase__							=				torch.device('cuda' ) if torch.cuda.is_available() else torch.device('cpu' )

  UpperCAmelCase__							=				send_to_device(__a       ,       __a )
  self.assertTrue(torch.equal(resulta.cpu()       ,       __a ) )

  UpperCAmelCase__							=				send_to_device((tensor, [tensor, tensor], 1)       ,       __a )
  self.assertIsInstance(__a       ,       __a )
  self.assertTrue(torch.equal(resulta[0].cpu()       ,       __a ) )
  self.assertIsInstance(resulta[1]       ,       __a )
  self.assertTrue(torch.equal(resulta[1][0].cpu()       ,       __a ) )
  self.assertTrue(torch.equal(resulta[1][1].cpu()       ,       __a ) )
  self.assertEqual(resulta[2]       ,       1 )

  UpperCAmelCase__							=				send_to_device({'a': tensor, 'b': [tensor, tensor], 'c': 1}       ,       __a )
  self.assertIsInstance(__a       ,       __a )
  self.assertTrue(torch.equal(resulta['a'].cpu()       ,       __a ) )
  self.assertIsInstance(resulta['b']       ,       __a )
  self.assertTrue(torch.equal(resulta['b'][0].cpu()       ,       __a ) )
  self.assertTrue(torch.equal(resulta['b'][1].cpu()       ,       __a ) )
  self.assertEqual(resulta['c']       ,       1 )

  UpperCAmelCase__							=				send_to_device(ExampleNamedTuple(a=__a       ,       b=[tensor, tensor]       ,       c=1 )       ,       __a )
  self.assertIsInstance(__a       ,       __a )
  self.assertTrue(torch.equal(resulta.a.cpu()       ,       __a ) )
  self.assertIsInstance(resulta.b       ,       __a )
  self.assertTrue(torch.equal(resulta.b[0].cpu()       ,       __a ) )
  self.assertTrue(torch.equal(resulta.b[1].cpu()       ,       __a ) )
  self.assertEqual(resulta.c       ,       1 )

  UpperCAmelCase__							=				send_to_device(UserDict({'a': tensor, 'b': [tensor, tensor], 'c': 1} )       ,       __a )
  self.assertIsInstance(__a       ,       __a )
  self.assertTrue(torch.equal(resulta['a'].cpu()       ,       __a ) )
  self.assertIsInstance(resulta['b']       ,       __a )
  self.assertTrue(torch.equal(resulta['b'][0].cpu()       ,       __a ) )
  self.assertTrue(torch.equal(resulta['b'][1].cpu()       ,       __a ) )
  self.assertEqual(resulta['c']       ,       1 )




 def   UpperCamelCase__							(self ) ->					List[str]:






  """simple docstring"""





  with self.assertRaises(__a ) as cm:
   UpperCAmelCase__							=				recursively_apply(torch.tensor       ,       (torch.tensor(1 ), 1)       ,       error_on_other_type=__a )
  self.assertEqual(
      str(cm.exception )       ,       'Unsupported types (<class \'int\'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed.'       ,       )




 def   UpperCamelCase__							(self ) ->					List[str]:






  """simple docstring"""





  UpperCAmelCase__							=				torch.tensor([1, 2, 3, 4, 5] )
  self.assertEqual(listify(__a )       ,       [1, 2, 3, 4, 5] )

  UpperCAmelCase__							=				torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]] )
  self.assertEqual(listify(__a )       ,       [[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]] )

  UpperCAmelCase__							=				torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]] )
  self.assertEqual(
      listify(__a )       ,       [[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]] )




 def   UpperCamelCase__							(self ) ->					int:






  """simple docstring"""





  with patch_environment(aa=1       ,       BB=2 ):
   self.assertEqual(os.environ.get('AA' )       ,       '1' )
   self.assertEqual(os.environ.get('BB' )       ,       '2' )

  self.assertNotIn('AA'       ,       os.environ )
  self.assertNotIn('BB'       ,       os.environ )




 def   UpperCamelCase__							(self ) ->					Dict:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  UpperCAmelCase__							=				model.forward
  UpperCAmelCase__							=				convert_outputs_to_fpaa(model.forward )
  UpperCAmelCase__							=				extract_model_from_parallel(__a       ,       keep_fpaa_wrapper=__a )
  UpperCAmelCase__							=				pickle.dumps(__a )




 @require_cuda
 def   UpperCamelCase__							(self ) ->					Tuple:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  UpperCAmelCase__							=				model.forward
  UpperCAmelCase__							=				torch.cuda.amp.autocast(dtype=torch.floataa )(model.forward )
  UpperCAmelCase__							=				convert_outputs_to_fpaa(model.forward )
  UpperCAmelCase__							=				extract_model_from_parallel(__a       ,       keep_fpaa_wrapper=__a )
  UpperCAmelCase__							=				pickle.dumps(__a )




 @require_cuda
 @require_torch_min_version(version='2.0' )
 def   UpperCamelCase__							(self ) ->					List[str]:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  UpperCAmelCase__							=				model.forward
  UpperCAmelCase__							=				torch.cuda.amp.autocast(dtype=torch.floataa )(model.forward )
  UpperCAmelCase__							=				convert_outputs_to_fpaa(model.forward )
  UpperCAmelCase__							=				torch.compile(model.forward       ,       backend='inductor' )
  UpperCAmelCase__							=				torch.randn(4       ,       10 ).cuda()
  UpperCAmelCase__							=				model(__a )




 def   UpperCamelCase__							(self ) ->					Optional[Any]:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
  UpperCAmelCase__							=				torch.nn.parallel.DataParallel(__a )
  UpperCAmelCase__							=				extract_model_from_parallel(__a )

  self.assertEqual(__a       ,       __a )




 @require_torch_min_version(version='2.0' )
 def   UpperCamelCase__							(self ) ->					Optional[Any]:






  """simple docstring"""





  UpperCAmelCase__							=				RegressionModel()
  UpperCAmelCase__							=				torch.compile(__a )

  # could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
  UpperCAmelCase__							=				torch.nn.parallel.DataParallel(__a )
  UpperCAmelCase__							=				torch.compile(__a )
  UpperCAmelCase__							=				extract_model_from_parallel(__a )

  self.assertEqual(compiled_model._orig_mod       ,       compiled_model_unwrapped._orig_mod )




 def   UpperCamelCase__							(self ) ->					str:






  """simple docstring"""





  self.assertEqual(find_device([1, 'a', torch.tensor([1, 2, 3] )] )       ,       torch.device('cpu' ) )
  self.assertEqual(find_device({'a': 1, 'b': torch.tensor([1, 2, 3] )} )       ,       torch.device('cpu' ) )
  self.assertIsNone(find_device([1, 'a'] ) )



