tdd + diarization

app.py

@@ -4,18 +4,14 @@
 from fastapi import FastAPI, UploadFile
 from fastapi.staticfiles import StaticFiles
 from fastapi.responses import FileResponse
-from pyannote.audio import Pipeline
- 
-from transformers import pipeline # le framework de huggingface
+from audio.audioanalyser_francais import AudioAnalyserAnglais
+from audio.audioanalyser_diarization import AudioAnalyserDiarization
 #from datasets import load_dataset, Audio # ça c'est pour entrainer mon modele
 
 
 app = FastAPI()
 
-#deepneurones = pipeline("text2text-generation", model="google/flan-t5-small")
-#deepneurones = pipeline("automatic-speech-recognition")# la liste des pipelines de huggingface est disponible ici :https://huggingface.co/docs/transformers/quicktour. pipeline() telecharge dans un cache local le modele deeplearning
-#deepneurones= pipeline("automatic-speech-recognition", model="facebook/wav2vec2-base-960h") # il faut choisir un modele
-deepneurones = Pipeline.from_pretrained("pyannote/speaker-diarization")
+
 @app.get("/healthcheck")
 def healthcheck():
     
@@ -27,11 +23,22 @@ def healthcheck():
 async def stt(file: str =  UploadFile(...)):
     #file_content = base64.b64decode(file)
     file_content = await file.read()
+    results = AudioAnalyserAnglais.stt(file_content)
     #dataset = load_dataset("PolyAI/minds14", name="en-US", split="train")
-    results = deepneurones(file_content)
+
     return {"output":results}
 #app.mount("/", StaticFiles(directory="static", html=True), name="static")
-    
+
+@app.post("/diarization")
+async def diarization(file: str =  UploadFile(...)):
+    #file_content = base64.b64decode(file)
+    file_content = await file.read()
+    results = AudioAnalyserDiarization.diarization(file_content)
+    #dataset = load_dataset("PolyAI/minds14", name="en-US", split="train")
+
+    return {"output":results}
+#app.mount("/", StaticFiles(directory="static", html=True), name="static")
+
 
 @app.get("/")
 def index() -> FileResponse:

audio/audioanalyser_anglais.py

@@ -0,0 +1,15 @@
+#deepneurones = pipeline("text2text-generation", model="google/flan-t5-small")
+#deepneurones = pipeline("automatic-speech-recognition")# la liste des pipelines de huggingface est disponible ici :https://huggingface.co/docs/transformers/quicktour. pipeline() telecharge dans un cache local le modele deeplearning
+from transformers import pipeline  # le framework de huggingface
+deepneurones= pipeline("automatic-speech-recognition", model="facebook/wav2vec2-base-960h") # il faut choisir un modele
+
+#from pyannote.audio import Pipeline
+#,use_auth_token="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")
+#deepneuronesdiarizatin = Pipeline.from_pretrained("pyannote/speaker-diarization",use_auth_token="test")
+
+
+class AudioAnalyserAnglais:
+
+    @classmethod
+    def stt(cls, file_content):
+        return deepneurones(file_content)

audio/audioanalyser_diarization.py

@@ -0,0 +1,14 @@
+#deepneurones = pipeline("text2text-generation", model="google/flan-t5-small")
+from pyannote.audio import Pipeline
+use_auth_token="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"
+deepneuronesdiarization = Pipeline.from_pretrained("pyannote/speaker-diarization", use_auth_token=use_auth_token)
+
+class AudioAnalyserDiarization:
+
+    @classmethod
+    def diarization(cls, file_content):
+        #        results = deepneuronesdiarizatin(file_content)
+        #        return results
+        return deepneuronesdiarization(file_content)
+
+

audio/audioanalyser_francais.py

@@ -0,0 +1,18 @@
+#from transformers import WhisperProcessor, WhisperForConditionalGeneration
+#from datasets import Audio, load_dataset
+#
+## load model and processor
+#processor = WhisperProcessor.from_pretrained("openai/whisper-base")
+#model = WhisperForConditionalGeneration.from_pretrained("openai/whisper-base")
+#forced_decoder_ids = processor.get_decoder_prompt_ids(language="french", task="transcribe")
+#
+## load streaming dataset and read first audio sample
+#input_speech = next(iter(ds))["audio"]
+#input_features = processor(input_speech["array"], sampling_rate=input_speech["sampling_rate"], return_tensors="pt").input_features
+#
+## generate token ids
+#predicted_ids = model.generate(input_features, forced_decoder_ids=forced_decoder_ids)
+## decode token ids to text
+#transcription = processor.batch_decode(predicted_ids)
+#
+#transcription = processor.batch_decode(predicted_ids, skip_special_tokens=True)

index.html

@@ -1,6 +1,12 @@
 <html>
+stt
 <form action="/stt"  method="post" enctype="multipart/form-data">
   <input type="file" name="file">
   <input type="submit" value="ok">
 </form>
+diarization
+<form action="/diarization"  method="post" enctype="multipart/form-data">
+  <input type="file" name="file">
+  <input type="submit" value="ok">
+</form>
 </html>

tests/audio/test_audio.py

@@ -0,0 +1,19 @@
+import unittest
+
+from audio.audioanalyser_diarization import AudioAnalyserDiarization
+from audio.audioanalyser_anglais import AudioAnalyserAnglais
+
+
+class MyTestCase(unittest.TestCase):
+    def test_stt_en(self):
+        with open("../testsdata/audio/en.wav", "rb") as filecontent:
+            #print(filecontent.read())
+            res = AudioAnalyserAnglais.stt(filecontent.read())
+            self.assertEqual(res, True)  # add assertion here
+
+    def test_diarization(self):
+        with open("../testsdata/audio/en.wav", "rb") as filecontent:
+            #print(filecontent.read())
+            res = AudioAnalyserDiarization.stt(filecontent.read())
+            self.assertEqual(res, True)  # add assertion here
+