




import argparse
import os
from dataclasses import dataclass

import torch
import torchaudio
from tqdm import tqdm

from transformers import AutoConfig, AutoModelForCTC, AutoProcessor
class       _UpperCamelCase   :
      def __init__(      self :Optional[int]			,   lowerCamelCase :Optional[int]			,   lowerCamelCase :Any			,   lowerCamelCase :List[Any]   )  ->      int:
       UpperCAmelCase__							=    cuda
       UpperCAmelCase__							=    AutoConfig.from_pretrained(lowerCamelCase   )
       UpperCAmelCase__							=    AutoModelForCTC.from_pretrained(lowerCamelCase   )
       self.model.eval()
       if self.cuda:
        self.model.to(device="cuda"   )
       UpperCAmelCase__							=    AutoProcessor.from_pretrained(lowerCamelCase   )
       UpperCAmelCase__							=    torchaudio.transforms.Resample(lowerCamelCase			,   1_6000   )
       UpperCAmelCase__							=    0
       UpperCAmelCase__							=    list(self.processor.tokenizer.get_vocab().keys()   )
       for i in range(len(lowerCamelCase   )   ):
        if vocab[i] == "[PAD]" or vocab[i] == "<pad>":
         UpperCAmelCase__							=    i
       print("Blank Token id [PAD]/<pad>"			,   lowerCamelCase   )
       UpperCAmelCase__							=    blank_id
      def 						UpperCAmelCase_    (      self :List[Any]			,   lowerCamelCase :List[str]   )  ->      List[Any]:
       UpperCAmelCase__			,			UpperCAmelCase__							=    torchaudio.load(lowerCamelCase   )
       UpperCAmelCase__							=    self.resampler(lowerCamelCase   ).squeeze().numpy()
       return speech
      def 						UpperCAmelCase_    (      self :List[str]			,   lowerCamelCase :Any   )  ->      Union[str, Any]:
       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(lowerCamelCase			,   sampling_rate=1_6000			,   return_tensors="pt"			,   padding=lowerCamelCase   )
       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(lowerCamelCase			,   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(lowerCamelCase   )}
       UpperCAmelCase__							=    []
       for c in transcript:
        if c in dictionary:
         tokens.append(dictionary[c]   )

       def get_trellis(lowerCamelCase :Optional[Any]			,   lowerCamelCase :Any			,   lowerCamelCase :List[Any]=0   ):
        UpperCAmelCase__							=    emission.size(0   )
        UpperCAmelCase__							=    len(lowerCamelCase   )

        # 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(lowerCamelCase   ):
         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(lowerCamelCase			,   lowerCamelCase			,   lowerCamelCase   )
       @dataclass
       class       _UpperCamelCase   :
             UpperCAmelCase_          =		42
             UpperCAmelCase_          =		42
             UpperCAmelCase_          =		42

       def backtrack(lowerCamelCase :Union[str, Any]			,   lowerCamelCase :Union[str, Any]			,   lowerCamelCase :Union[str, Any]			,   lowerCamelCase :Tuple=0   ):
        UpperCAmelCase__							=    trellis.size(1   ) - 1
        UpperCAmelCase__							=    torch.argmax(trellis[:, j]   ).item()

        UpperCAmelCase__							=    []
        for t in range(lowerCamelCase			,   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			,   lowerCamelCase   )   )

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

       UpperCAmelCase__							=    backtrack(lowerCamelCase			,   lowerCamelCase			,   lowerCamelCase			,   lowerCamelCase   )
       @dataclass
       class       _UpperCamelCase   :
             UpperCAmelCase_          =		42
             UpperCAmelCase_          =		42
             UpperCAmelCase_          =		42
             UpperCAmelCase_          =		42
             def __repr__(      self :List[str]   )  ->      Optional[Any]:
              return f'''{self.label}\t{self.score:4.2f}\t{self.start*20:5d}\t{self.end*20:5d}'''







             @property
             def 						UpperCAmelCase_    (      self :str   )  ->      Any:
              return self.end - self.start

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

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







      def 						UpperCAmelCase_    (      self :Union[str, Any]			,   lowerCamelCase :Tuple			,   lowerCamelCase :Union[str, Any]			,   lowerCamelCase :str   )  ->      Any:
       if not os.path.exists(lowerCamelCase   ):
        os.makedirs(lowerCamelCase   )

       # load text file
       UpperCAmelCase__							=    open(lowerCamelCase			,   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(lowerCamelCase			,   wav_name + ".wav"   )
        UpperCAmelCase__							=    os.path.join(lowerCamelCase			,   wav_name + ".txt"   )

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

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



def     lowerCAmelCase		(		):




 """simple docstring"""







 UpperCAmelCase__							=    argparse.ArgumentParser()

 parser.add_argument(
     "--model_name"    ,				type=_lowerCAmelCase    ,				default="arijitx/wav2vec2-xls-r-300m-bengali"    ,				help="wav2vec model name"						)
 parser.add_argument("--wav_dir"    ,				type=_lowerCAmelCase    ,				default="./wavs"    ,				help="directory containing wavs"						)
 parser.add_argument("--text_file"    ,				type=_lowerCAmelCase    ,				default="script.txt"    ,				help="file containing text"						)
 parser.add_argument("--input_wavs_sr"    ,				type=_lowerCAmelCase    ,				default=1_6000    ,				help="sampling rate of input audios"						)
 parser.add_argument(
     "--output_dir"    ,				type=_lowerCAmelCase    ,				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 subprocess
from typing import Union

import numpy as np
import requests

from ..utils import add_end_docstrings, is_torch_available, is_torchaudio_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_torch_available():
  from ..models.auto.modeling_auto import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING

_lowerCAmelCase    : List[str]		= logging.get_logger(__name__)



def     lowerCAmelCase		(		_lowerCAmelCase				: bytes    ,				_lowerCAmelCase				: int						):




 """simple docstring"""







 UpperCAmelCase__							=    F'''{sampling_rate}'''
 UpperCAmelCase__							=    "1"
 UpperCAmelCase__							=    "f32le"
 UpperCAmelCase__							=    [
     "ffmpeg",
     "-i",
     "pipe:0",
     "-ac",
     ac,
     "-ar",
     ar,
     "-f",
     format_for_conversion,
     "-hide_banner",
     "-loglevel",
     "quiet",
     "pipe:1",
 ]

 try:
  UpperCAmelCase__							=    subprocess.Popen(_lowerCAmelCase    ,				stdin=subprocess.PIPE    ,				stdout=subprocess.PIPE						)
 except FileNotFoundError:
  raise ValueError("ffmpeg was not found but is required to load audio files from filename"						)
 UpperCAmelCase__							=    ffmpeg_process.communicate(_lowerCAmelCase						)
 UpperCAmelCase__							=    output_stream[0]

 UpperCAmelCase__							=    np.frombuffer(_lowerCAmelCase    ,				np.floataa						)
 if audio.shape[0] == 0:
  raise ValueError("Malformed soundfile"						)
 return audio


@add_end_docstrings(lowerCAmelCase		)
class       _UpperCamelCase   (					lowerCAmelCase		):
      def __init__(      self :Any			,   *lowerCamelCase :Any			,   **lowerCamelCase :Optional[Any]   )  ->      Optional[int]:
       # Default, might be overriden by the model.config.
       UpperCAmelCase__							=    5
       super().__init__(*lowerCamelCase			,   **lowerCamelCase   )

       if self.framework != "pt":
        raise ValueError(f'''The {self.__class__} is only available in PyTorch.'''   )

       self.check_model_type(lowerCamelCase   )
      def __call__(      self :Optional[Any]			,   lowerCamelCase :Union[np.ndarray, bytes, str]			,   **lowerCamelCase :List[Any]			,   )  ->      Tuple:
       return super().__call__(lowerCamelCase			,   **lowerCamelCase   )
      def 						UpperCAmelCase_    (      self :str			,   lowerCamelCase :Dict=None			,   **lowerCamelCase :str   )  ->      int:
       # No parameters on this pipeline right now
       UpperCAmelCase__							=    {}
       if top_k is not None:
        if top_k > self.model.config.num_labels:
         UpperCAmelCase__							=    self.model.config.num_labels
        UpperCAmelCase__							=    top_k
       return {}, {}, postprocess_params
      def 						UpperCAmelCase_    (      self :Optional[Any]			,   lowerCamelCase :Dict   )  ->      Optional[int]:
       if isinstance(lowerCamelCase			,   lowerCamelCase   ):
        if inputs.startswith("http://"   ) or inputs.startswith("https://"   ):
         # We need to actually check for a real protocol, otherwise it's impossible to use a local file
         # like http_huggingface_co.png
         UpperCAmelCase__							=    requests.get(lowerCamelCase   ).content
        else:
         with open(lowerCamelCase			,   "rb"   ) as f:
          UpperCAmelCase__							=    f.read()

       if isinstance(lowerCamelCase			,   lowerCamelCase   ):
        UpperCAmelCase__							=    ffmpeg_read(lowerCamelCase			,   self.feature_extractor.sampling_rate   )

       if isinstance(lowerCamelCase			,   lowerCamelCase   ):
        # Accepting `"array"` which is the key defined in `datasets` for
        # better integration
        if not ("sampling_rate" in inputs and ("raw" in inputs or "array" in inputs)):
         raise ValueError(
             "When passing a dictionary to AudioClassificationPipeline, the dict needs to contain a "
             "\"raw\" key containing the numpy array representing the audio and a \"sampling_rate\" key, "
             "containing the sampling_rate associated with that array"   )

        UpperCAmelCase__							=    inputs.pop("raw"			,   lowerCamelCase   )
        if _inputs is None:
         # Remove path which will not be used from `datasets`.
         inputs.pop("path"			,   lowerCamelCase   )
         UpperCAmelCase__							=    inputs.pop("array"			,   lowerCamelCase   )
        UpperCAmelCase__							=    inputs.pop("sampling_rate"   )
        UpperCAmelCase__							=    _inputs
        if in_sampling_rate != self.feature_extractor.sampling_rate:
         import torch

         if is_torchaudio_available():
          from torchaudio import functional as F
         else:
          raise ImportError(
              "torchaudio is required to resample audio samples in AudioClassificationPipeline. "
              "The torchaudio package can be installed through: `pip install torchaudio`."   )

         UpperCAmelCase__							=    F.resample(
             torch.from_numpy(lowerCamelCase   )			,   lowerCamelCase			,   self.feature_extractor.sampling_rate   ).numpy()

       if not isinstance(lowerCamelCase			,   np.ndarray   ):
        raise ValueError("We expect a numpy ndarray as input"   )
       if len(inputs.shape   ) != 1:
        raise ValueError("We expect a single channel audio input for AudioClassificationPipeline"   )

       UpperCAmelCase__							=    self.feature_extractor(
           lowerCamelCase			,   sampling_rate=self.feature_extractor.sampling_rate			,   return_tensors="pt"   )
       return processed
      def 						UpperCAmelCase_    (      self :Tuple			,   lowerCamelCase :Union[str, Any]   )  ->      List[Any]:
       UpperCAmelCase__							=    self.model(**lowerCamelCase   )
       return model_outputs







      def 						UpperCAmelCase_    (      self :List[str]			,   lowerCamelCase :Optional[Any]			,   lowerCamelCase :List[Any]=5   )  ->      Tuple:
       UpperCAmelCase__							=    model_outputs.logits[0].softmax(-1   )
       UpperCAmelCase__			,			UpperCAmelCase__							=    probs.topk(lowerCamelCase   )

       UpperCAmelCase__							=    scores.tolist()
       UpperCAmelCase__							=    ids.tolist()

       UpperCAmelCase__							=    [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(lowerCamelCase			,   lowerCamelCase   )]

       return labels





