added backend

app.py

@@ -1,20 +1,33 @@
 import gradio as gr # needs to be installed
 
 from dataset import Dataset
+from model import Model
 
+from utils import hf_login, compute_wer
 
+hf_login()
 dataset = Dataset()
+models = Model()
 
 
-def eval(data_subset:str, model_1:str, model_2:str, model_3:str)->str:
+def run_tests (dataset_choice:str, model:str):
 
-    result = f"""# HELLO THERE
-    Data Subset: {data_subset}
-    Model 1: {model_1}
-    Model 2: {model_2}
-    Model_3: {model_3}
-    """
-    return result
+    MoDeL = Model()
+    MoDeL.select(model)
+    MoDeL.load()
+    DaTaSeT = Dataset(100)
+    DaTaSeT.load(dataset_choice)
+    references, predictions = MoDeL.process(DaTaSeT)
+    wer = compute_wer(references=references, predictions=predictions)
+    return wer
+
+
+def eval(data_subset:str, model_1:str, model_2:str)->str:
+
+    wer_result_1 = run_tests(data_subset, model_1)
+    wer_result_2 = run_tests(data_subset, model_2)
+
+    return f"WER Score: {wer_result_1}", f"WER Score: {wer_result_2}"
 
 def get_card(selected_model:str)->str:
 
@@ -38,20 +51,18 @@ def is_own(data_subset:str):
 
 with gr.Blocks() as demo:
 
-    gr.Markdown("# VocalVenturer 💬")
+
+    gr.Markdown('# <p style="text-align: center;">VocalVenturer 💬</p>')
     gr.Markdown("-------")
-    gr.Markdown("""
-                Hello there, this is the VocalVenturer, this app is aimed at helping you making more informed model choices for ASR. 
-                Please choose a Data Subset to evalutate the Models on. You also have the opportunity to record and transcibe an own sample. 
-                The Models will be evaluated using the *WER* metric -> here is an amazing Leaderboard for it LINK""")
+    gr.Markdown("""#### Hello there, this is the VocalVenturer, this app is aimed at helping you making more informed model choices for ASR. Please choose a Data Subset to evalutate the Models on. You also have the opportunity to record and transcibe an own sample. The Models will be evaluated using the *WER* metric -> here is an amazing Leaderboard for it LINK""")
 
     with gr.Row():
         with gr.Column(scale=1):
             pass
-        with gr.Column(scale=3):
+        with gr.Column(scale=5):
             data_subset = gr.Radio(
-                value="Libris",
-                choices=["Libris","Common","own"],
+                value="LibriSpeech Clean",
+                choices=dataset.get_options(),
                 label="Data subset / Own Sample",
             )
         with gr.Column(scale=1):
@@ -68,41 +79,51 @@ with gr.Blocks() as demo:
 
         with gr.Column(scale=1):
             model_1 = gr.Dropdown(
-                choices=["None","Model2","Model3"],
-                label="Select Model 1"
+                choices=models.get_options(),
+                label="Select Model"
             )
             model_1_card = gr.Markdown("")
 
         with gr.Column(scale=1):
             model_2 = gr.Dropdown(
-                choices=["None","Model2","Model3"],
-                label="Select Model 2"
+                choices=models.get_options(),
+                label="Select Model"
             )
             model_2_card = gr.Markdown("")
 
-        with gr.Column(scale=1):
-            model_3 = gr.Dropdown(
-                choices=["None","Model2","Model3"],
-                label="Select Model 3"
-            )
-            model_3_card = gr.Markdown("")
+        # with gr.Column(scale=1):
+        #     model_3 = gr.Dropdown(
+        #         choices=models.get_options(),
+        #         label="Select Model"
+        #     )
+        #     model_3_card = gr.Markdown("")
 
         model_1.change(get_card, inputs=model_1, outputs=model_1_card)
         model_2.change(get_card, inputs=model_2, outputs=model_2_card)
-        model_3.change(get_card, inputs=model_3, outputs=model_3_card)
+        # model_3.change(get_card, inputs=model_3, outputs=model_3_card)
 
     eval_btn = gr.Button(
         value="Evaluate",
-        variant="primary"
+        variant="primary",
+        size="sm"
     )
-
     
 
+    
+    gr.Markdown('## <p style="text-align: center;">Results</p>')
     gr.Markdown("-------")
-    gr.Markdown("### Results")
+    
+
+    with gr.Row():
+
+        with gr.Column(scale=1):
+            results_model_1 = gr.Markdown("")
+
+        with gr.Column(scale=1):
+            results_model_2 = gr.Markdown("")
 
-    results = gr.Markdown("")
+    eval_btn.click(eval, [data_subset, model_1, model_2], [results_model_1, results_model_2])
 
-    eval_btn.click(eval, [data_subset, model_1, model_2, model_3], results)
+    
 
 demo.launch(debug=True)

cards.txt

@@ -0,0 +1,5 @@
+#### Whisper Tiny (EN)
+
+- 
+@@
+@@

dataset.py

@@ -1,36 +1,93 @@
-from huggingface_hub import login
 from datasets import load_dataset
 from datasets import Audio
 
-import json
 
-def get_credentials():
-    with open("credentials.json","r") as f:
-        credentials = json.load(f)
-    return credentials['token']
 
+class Dataset:
 
+    def __init__(self, n:int = 100):
 
-class Dataset:
+        self.n = n
+        self.options = ['LibriSpeech Clean', 'LibriSpeech Other', 'Common Voice', 'VoxPopuli', 'TEDLIUM', 'GigaSpeech', 'SPGISpeech', 'AMI', 'OWN']
+        self.selected = None
+        self.dataset = None
+        self.text = None
 
-    def __init__(self, n:int=100):
+    def get_options(self):
+        return self.options
 
-        # # Log the client into HF
-        # login(token=get_credentials())
+    def _check_text(self):
+        sample = next(iter(self.dataset))
+        print(sample)
+        self._get_text(sample)
 
-        # # Load the Dataset in Streaming Mode
-        # self.librispeech_clean = load_dataset("librispeech_asr", "all", split="test.clean", streaming=True)
-        # self.librispeech_other = load_dataset("librispeech_asr", "all", split="test.other", streaming=True)
+    def _get_text(self, sample):
+        if "text" in sample:
+            self.text = "text"
+            return sample["text"]
+        elif "sentence" in sample:
+            self.text = "sentence"
+            return sample["sentence"]
+        elif "normalized_text" in sample:
+            self.text = "normalized_text"
+            return sample["normalized_text"]
+        elif "transcript" in sample:
+            self.text = "transcript"
+            return sample["transcript"]
+        else:
+            raise ValueError(f"Sample: {sample.keys()} has no transcript.")
+        
+    def filter(self, input_column:str = None):
 
-        self.n = n
+        if input_column is None:
+            if self.text is not None:
+                input_column = self.text
+            else:
+                input_column = self._check_text()
 
-        self.options = ['librisspeech_clean','librisspeech_other']
-    
-    def get_option(self):
-        return self.options
+        def is_target_text_in_range(ref):
+            if ref.strip() == "ignore time segment in scoring":
+                return False
+            else:
+                return ref.strip() != ""
+        
+        self.dataset = self.dataset.filter(is_target_text_in_range, input_columns=[input_column])
+        return self.dataset
+        
+    def normalised(self, normalise):
+        self.dataset = self.dataset.map(normalise)
+
+    def _select(self, option:str):
+        if option not in self.options:
+            raise ValueError(f"This value is not an option, please see: {self.options}")
+        self.selected = option
 
+    def _preprocess(self):
 
+        self.dataset = self.dataset.take(self.n)
+        self.dataset = self.dataset.cast_column("audio", Audio(sampling_rate=16000))
 
+    def load(self, option:str = None):
 
+        self._select(option)
+        
+        if option == "OWN":
+            pass
+        elif option == "LibriSpeech Clean":
+            self.dataset = load_dataset("librispeech_asr", "all", split="test.clean", streaming=True)
+        elif option == "LibriSpeech Other":
+            self.dataset = load_dataset("librispeech_asr", "all", split="test.other", streaming=True)
+        elif option == "Common Voice":
+            self.dataset = load_dataset("mozilla-foundation/common_voice_11_0", "en", revision="streaming", split="test", streaming=True, token=True, trust_remote_code=True)
+        elif option == "VoxPopuli":
+            self.dataset = load_dataset("facebook/voxpopuli", "en", split="test", streaming=True, trust_remote_code=True)
+        elif option == "TEDLIUM":
+            self.dataset = load_dataset("LIUM/tedlium", "release3", split="test", streaming=True, trust_remote_code=True)
+        elif option == "GigaSpeech":
+            self.dataset = load_dataset("speechcolab/gigaspeech", "xs", split="test", streaming=True, token=True, trust_remote_code=True)
+        elif option == "SPGISpeech":
+            self.dataset = load_dataset("kensho/spgispeech", "S", split="test", streaming=True, token=True, trust_remote_code=True)
+        elif option == "AMI":
+            self.dataset = load_dataset("edinburghcstr/ami", "ihm", split="test", streaming=True, trust_remote_code=True)
 
-dataset = Dataset()
+        self._preprocess()

eval.py

@@ -0,0 +1,22 @@
+from dataset import Dataset
+from model import Models
+
+
+def data(dataset):
+    for i, item in enumerate(dataset):
+        yield {**item["audio"], "reference": item["norm_text"]}
+
+
+def streamed_infernce(dataset, pipeline):
+    
+
+    # placeholders for predictions and references
+    predictions = []
+    references = []
+
+    # run streamed inference
+    for out in pipeline(data(dataset), batch_size=16):
+        predictions.append(pipeline(out["text"]))
+        references.append(out["reference"][0])
+
+    return predictions, references

model.py

@@ -0,0 +1,104 @@
+from transformers import WhisperProcessor, WhisperForConditionalGeneration
+from transformers import Speech2TextForConditionalGeneration, Speech2TextProcessor
+from transformers import pipeline
+
+from dataset import Dataset
+from utils import data
+
+
+
+class Model:
+
+
+    def __init__(self):
+
+        self.options = [
+            "openai/whisper-tiny.en",
+            "facebook/s2t-medium-librispeech-asr"
+        ]
+        self.selected = None
+        self.pipeline = None
+        self.normalize = None
+
+    def get_options(self):
+        return self.options
+
+    def load(self, option:str = None):
+
+        if option is None:
+            if self.selected is None:
+                raise ValueError("No model selected. Please first select a model")
+            option = self.selected
+
+        if option not in self.options:
+            raise ValueError(f"Selected Option is not a valid value, see: {self.options}")
+
+        if option == "openai/whisper-tiny.en":
+            self.pipeline = pipeline("automatic-speech-recognition", model="openai/whisper-tiny.en", device=0)
+            self.normalize = self.pipeline.tokenizer.normalize
+            
+        elif option == "facebook/s2t-medium-librispeech-asr":
+            self.model = Speech2TextForConditionalGeneration.from_pretrained("facebook/s2t-medium-librispeech-asr")
+            self.processor = Speech2TextProcessor.from_pretrained("facebook/s2t-medium-librispeech-asr", do_upper_case=True)
+
+
+    def select(self, option:str=None):
+        if option not in self.options:
+            raise ValueError(f"This value is not an option, please see: {self.options}")
+        self.selected = option
+
+    def process(self, dataset:Dataset):
+
+        if self.selected is None:
+            raise ValueError("No Model is yet selected. Please select a model first")
+        
+        if self.selected == "openai/whisper-tiny.en":
+            references, predictions = self._process_openai_whisper_tiny_en(dataset)
+        elif self.selected == "facebook/s2t-medium-librispeech-asr":
+            references, predictions = self._process_facebook_s2t_medium(dataset)
+
+        return references, predictions
+
+    def _process_openai_whisper_tiny_en(self, DaTaSeT:Dataset):
+
+        def normalise(batch):
+            batch["norm_text"] = self.normalize(DaTaSeT._get_text(batch))
+            return batch
+        
+        DaTaSeT.normalised(normalise)
+        dataset = DaTaSeT.filter("norm_text")
+
+        predictions = []
+        references = []
+
+        # run streamed inference
+        for out in self.pipeline(data(dataset), batch_size=16):
+            predictions.append(self.normalize(out["text"]))
+            references.append(out["reference"][0])
+
+        return references, predictions
+    
+    def _process_facebook_s2t_medium(self, DaTaSeT:Dataset):
+
+
+
+        def map_to_pred(batch):
+            features = self.processor(batch["audio"]["array"], sampling_rate=16000, padding=True, return_tensors="pt")
+            input_features = features.input_features
+            attention_mask = features.attention_mask
+
+            gen_tokens = self.model.generate(input_features=input_features, attention_mask=attention_mask)
+            batch["transcription"] = self.processor.batch_decode(gen_tokens, skip_special_tokens=True)[0]
+            return batch
+        
+        DaTaSeT.dataset = DaTaSeT.dataset.take(100)
+        result = DaTaSeT.dataset.map(map_to_pred, remove_columns=["audio"])
+
+        predictions = []
+        references = []
+
+        for sample in result:
+            predictions.append(sample['transcription'])
+            references.append(sample['text'])
+
+        return references, predictions

test.v01.py

@@ -0,0 +1,25 @@
+from utils import hf_login, data, compute_wer
+from dataset import Dataset
+from model import Model
+
+hf_login()
+
+
+
+
+
+
+def run_tests (dataset_choice:str, model:str):
+
+    MoDeL = Model()
+    MoDeL.select(model)
+    MoDeL.load()
+    DaTaSeT = Dataset(100)
+    DaTaSeT.load(dataset_choice)
+    references, predictions = MoDeL.process(DaTaSeT)
+    wer = compute_wer(references=references, predictions=predictions)
+    return wer
+
+
+print("WER:", run_tests(dataset_choice="GigaSpeech", model="facebook/s2t-medium-librispeech-asr"))
+

utils.py

@@ -0,0 +1,23 @@
+from huggingface_hub import login
+import json
+import evaluate
+import os
+
+def hf_login():
+    hf_token = os.getenv("HF_TOKEN")
+    # if hf_token is None:
+    #     with open("credentials.json", "r") as f:
+    #         hf_token = json.load(f)["token"]
+    login(token=hf_token)
+
+def data(dataset):
+    for i, item in enumerate(dataset):
+        yield {**item["audio"], "reference": item["norm_text"]}
+
+def compute_wer(references, predictions):
+    wer_metric = evaluate.load("wer")
+    wer = wer_metric.compute(references=references, predictions=predictions)
+    wer = round(100 * wer, 2)
+    return wer
+
+