test: update test cases for models modules, add preprocessAudioStandalone() function

.gitignore

@@ -280,3 +280,5 @@ fabric.properties
 
 # node_modules
 node_modules
+
+*.jit

aip_trainer/models/models.py

@@ -14,9 +14,7 @@ default_speaker_dict = {
 }
 
 
-def silero_tts(
-    language="en", version="latest", output_folder: Path | str = None, **kwargs
-):
+def silero_tts(language="en", version="latest", output_folder: Path | str = None, **kwargs):
     """Silero Text-To-Speech Models
     language (str): language of the model, now available are ['ru', 'en', 'de', 'es', 'fr']
     Returns a model and a set of utils
@@ -117,9 +115,10 @@ def init_jit_model(
     model_dir = (
         Path(output_folder)
         if output_folder is not None
-        else Path(os.path.dirname(__file__)) / "model"
+        else Path(torch.hub.get_dir())
     )
     os.makedirs(model_dir, exist_ok=True)
+    app_logger.info(f"downloading the models to model_dir: '{model_dir}' ...")
     model_path = model_dir / os.path.basename(model_url)
     app_logger.info(
         f"model_path exists? '{os.path.isfile(model_path)}' => '{model_path}' ..."
@@ -159,7 +158,7 @@ def get_models(language, output_folder, version, model_type):
     output_folder = (
         Path(output_folder)
         if output_folder is not None
-        else Path(os.path.dirname(__file__)) / ".." / ".."
+        else Path(os.path.dirname(__file__)).parent.parent
     )
     models_list_file = output_folder / f"latest_silero_model_{language}.yml"
     if not os.path.exists(models_list_file):

aip_trainer/pronunciationTrainer.py

@@ -33,6 +33,12 @@ def getTrainer(language: str):
     return trainer
 
 
+def preprocessAudioStandalone(audio: torch.tensor) -> torch.tensor:
+    audio = audio-torch.mean(audio)
+    audio = audio/torch.max(torch.abs(audio))
+    return audio
+
+
 class PronunciationTrainer:
     current_transcript: str
     current_ipa: str
@@ -185,6 +191,4 @@ class PronunciationTrainer:
         return np.argmin(abs(self.categories_thresholds-accuracy))
 
     def preprocessAudio(self, audio: torch.tensor) -> torch.tensor:
-        audio = audio-torch.mean(audio)
-        audio = audio/torch.max(torch.abs(audio))
-        return audio
+        return preprocessAudioStandalone(audio=audio)

aip_trainer/utils/split_cosmic_ray_report.py

@@ -1,7 +1,9 @@
 from pathlib import Path
 
 
-def get_cosmic_ray_report_filtered(input_filename, suffix="filtered", separator="============", filter_string="test outcome: TestOutcome.KILLED"):
+def get_cosmic_ray_report_filtered(input_filename, suffix="filtered", separator="============", filter_string_list: list = None):
+    if filter_string_list is None:
+        filter_string_list = ["test outcome: TestOutcome.KILLED"]
     filename, ext = Path(input_filename).stem, Path(input_filename).suffix
     working_dir = input_filename.parent
     # Read the input file
@@ -10,9 +12,11 @@ def get_cosmic_ray_report_filtered(input_filename, suffix="filtered", separator=
 
     # Split the content into sections
     sections = content.split(separator)
+    filtered_sections = [section for section in sections]
 
     # Filter out sections containing "test outcome: TestOutcome.KILLED"
-    filtered_sections = [section for section in sections if filter_string not in section]
+    for filter_string in filter_string_list:
+        filtered_sections = [section for section in filtered_sections if filter_string not in section]
 
     # Join the filtered sections back into a single string
     filtered_content = separator.join(filtered_sections)
@@ -25,5 +29,5 @@ def get_cosmic_ray_report_filtered(input_filename, suffix="filtered", separator=
 
 if __name__ == "__main__":
     from aip_trainer import PROJECT_ROOT_FOLDER
-    _input_filename =  "cosmic-ray-pronunciationtrainer1.txt"
+    _input_filename =  "cosmic-ray-models2.txt"
     get_cosmic_ray_report_filtered(PROJECT_ROOT_FOLDER / "tmp" / _input_filename)

cosmic_ray_config.toml

@@ -1,8 +1,8 @@
 [cosmic-ray]
-module-path = "aip_trainer/pronunciationTrainer.py"
+module-path = "aip_trainer/models/models.py"
 timeout = 30.0
 excluded-modules = []
-test-command = "python -m pytest tests/test_pronunciationtrainer.py"
+test-command = "python -m pytest tests/models/test_models.py"
 
 [cosmic-ray.distributor]
 name = "local"

tests/models/test_aimodels.py

@@ -0,0 +1,167 @@
+import json
+import unittest
+import torch
+
+from torchaudio.transforms import Resample
+
+from aip_trainer.lambdas.lambdaSpeechToScore import soundfile_load
+from aip_trainer.models import AIModels, models as mo
+from aip_trainer import sample_rate_start
+from aip_trainer.pronunciationTrainer import preprocessAudioStandalone
+from aip_trainer.utils import utilities
+from tests import EVENTS_FOLDER
+from tests.lambdas.test_lambdaSpeechToScore import set_seed
+
+
+device = torch.device('cpu')
+transform = Resample(orig_freq=sample_rate_start, new_freq=16000)
+
+
+def get_model(language):
+    model, decoder = mo.getASRModel(language)
+    model = model.to(device)
+    model.eval()
+    return AIModels.NeuralASR(model, decoder)
+
+signal_de, samplerate = soundfile_load(str(EVENTS_FOLDER / "test_de_easy.wav"))
+signal_en, samplerate = soundfile_load(str(EVENTS_FOLDER / "test_en_easy.wav"))
+signal_transformed_de = transform(torch.Tensor(signal_de)).unsqueeze(0)
+signal_transformed_de = preprocessAudioStandalone(signal_transformed_de)
+signal_transformed_en = transform(torch.Tensor(signal_en)).unsqueeze(0)
+signal_transformed_en = preprocessAudioStandalone(signal_transformed_en)
+
+
+class TestDeNeuralASR(unittest.TestCase):
+    def test_is_instance_of_NeuralASR_language_de(self):
+        asr_de = get_model("de")
+        self.assertIsInstance(asr_de, AIModels.NeuralASR)
+
+    def test_is_instance_of_NeuralASR_language_en(self):
+        asr_en = get_model("de")
+        self.assertIsInstance(asr_en, AIModels.NeuralASR)
+
+    def test_getTranscript_without_processing_audio_de(self):
+        with self.assertRaises(AssertionError):
+            try:
+                asr_de = get_model("de")
+                asr_de.getTranscript()
+            except AssertionError as ae:
+                assert "Can get audio transcripts without having processed the audio" in str(ae)
+                raise ae
+
+    def test_getTranscript_without_processing_audio_en(self):
+        with self.assertRaises(AssertionError):
+            try:
+                asr_en = get_model("en")
+                asr_en.getTranscript()
+            except AssertionError as ae:
+                assert "Can get audio transcripts without having processed the audio" in str(ae)
+                raise ae
+
+    def test_getWordLocations_without_processing_audio_de(self):
+        with self.assertRaises(AssertionError):
+            try:
+                asr_de = get_model("de")
+                asr_de.getWordLocations()
+            except AssertionError as ae:
+                assert "Can get word locations without having processed the audio" in str(ae)
+                raise ae
+
+    def test_getWordLocations_without_processing_audio_en(self):
+        with self.assertRaises(AssertionError):
+            try:
+                asr_en = get_model("en")
+                asr_en.getWordLocations()
+            except AssertionError as ae:
+                assert "Can get word locations without having processed the audio" in str(ae)
+                raise ae
+
+    def test_process_audio_de(self):
+        set_seed()
+        asr_de = get_model("de")
+        self.assertIsNone(asr_de.audio_transcript)
+        self.assertIsNone(asr_de.word_locations_in_samples)
+
+        asr_de.processAudio(signal_transformed_de)
+
+        self.assertEqual(asr_de.audio_transcript, 'hallo wie geht es dir')
+        self.assertEqual(
+            asr_de.word_locations_in_samples,
+            [
+                {"word": "hallo", "start_ts": 0.0, "end_ts": 6773.68},
+                {"word": "wie", "start_ts": 6773.68, "end_ts": 10468.42},
+                {"word": "geht", "start_ts": 10468.42, "end_ts": 13547.37},
+                {"word": "es", "start_ts": 13547.37, "end_ts": 16626.32},
+                {"word": "dir", "start_ts": 16626.32, "end_ts": 20321.05},
+            ],
+        )
+
+    def test_process_audio_en(self):
+        set_seed()
+        asr_en = get_model("en")
+        self.assertIsNone(asr_en.audio_transcript)
+        self.assertIsNone(asr_en.word_locations_in_samples)
+
+        asr_en.processAudio(signal_transformed_en)
+
+        self.assertEqual(asr_en.audio_transcript, 'i there how are you')
+        self.assertEqual(
+            asr_en.word_locations_in_samples,
+            [
+                {"word": "i", "start_ts": 0.0, "end_ts": 1800.0},
+                {"word": "there", "start_ts": 1800.0, "end_ts": 5400.0},
+                {"word": "how", "start_ts": 5400.0, "end_ts": 8400.0},
+                {"word": "are", "start_ts": 8400.0, "end_ts": 12000.0},
+                {"word": "you", "start_ts": 12000.0, "end_ts": 15000.0},
+            ],
+        )
+
+    def test_getTranscript_after_processing_audio_de(self):
+        set_seed()
+        asr_de = get_model("de")
+        asr_de.processAudio(signal_transformed_de)
+        transcript = asr_de.getTranscript()
+        self.assertEqual(transcript, 'hallo wie geht es dir')
+
+    def test_getTranscript_after_processing_audio_en(self):
+        set_seed()
+        asr_en = get_model("en")
+        asr_en.processAudio(signal_transformed_en)
+        transcript = asr_en.getTranscript()
+        self.assertEqual(transcript, 'i there how are you')
+
+    def test_getWordLocations_after_processing_audio_de(self):
+        set_seed()
+        asr_de = get_model("de")
+        asr_de.processAudio(signal_transformed_de)
+        word_locations = asr_de.getWordLocations()
+        self.assertEqual(
+            word_locations,
+            [
+                {"word": "hallo", "start_ts": 0.0, "end_ts": 6773.68},
+                {"word": "wie", "start_ts": 6773.68, "end_ts": 10468.42},
+                {"word": "geht", "start_ts": 10468.42, "end_ts": 13547.37},
+                {"word": "es", "start_ts": 13547.37, "end_ts": 16626.32},
+                {"word": "dir", "start_ts": 16626.32, "end_ts": 20321.05},
+            ],
+        )
+
+    def test_getWordLocations_after_processing_audio_en(self):
+        set_seed()
+        asr_en = get_model("en")
+        asr_en.processAudio(signal_transformed_en)
+        word_locations = asr_en.getWordLocations()
+        self.assertEqual(
+            word_locations,
+            [
+                {"word": "i", "start_ts": 0.0, "end_ts": 1800.0},
+                {"word": "there", "start_ts": 1800.0, "end_ts": 5400.0},
+                {"word": "how", "start_ts": 5400.0, "end_ts": 8400.0},
+                {"word": "are", "start_ts": 8400.0, "end_ts": 12000.0},
+                {"word": "you", "start_ts": 12000.0, "end_ts": 15000.0},
+            ],
+        )
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/models/test_models.py

@@ -0,0 +1,83 @@
+import tempfile
+import unittest
+import torch
+from pathlib import Path
+from silero.utils import Decoder
+from silero.silero import silero_tts
+import torch.package
+from aip_trainer import PROJECT_ROOT_FOLDER
+from aip_trainer.models import models as mo
+
+class TestModels(unittest.TestCase):
+
+    def setUp(self):
+        self.language_de = "de"
+        self.language_en = "en"
+        self.tmp_dir = torch.hub.get_dir()
+        self.device = torch.device("cpu")
+
+    def test_getASRModel_de(self):
+        model, decoder = mo.getASRModel(self.language_de)
+        self.assertIsInstance(model, torch.nn.Module)
+        self.assertIsInstance(decoder, Decoder)
+
+    def test_getASRModel_en(self):
+        model, decoder = mo.getASRModel(self.language_en)
+        self.assertIsInstance(model, torch.nn.Module)
+        self.assertIsInstance(decoder, Decoder)
+
+    def test_silero_stt_en(self):
+        model, decoder, utils = mo.silero_stt(language=self.language_en, output_folder=self.tmp_dir)
+        self.assertIsInstance(model, torch.jit.ScriptModule)
+        self.assertIsInstance(decoder, Decoder)
+        self.assertIsInstance(utils, tuple)
+
+    def test_silero_tts_en2(self):
+        model, example, speaker, sample_rate = mo.silero_tts(language=self.language_en, output_folder=self.tmp_dir)
+        assert model is not None
+        self.assertIsInstance(model, object)
+        self.assertIsInstance(example, str)
+        self.assertIsInstance(speaker, str)
+        self.assertIsInstance(sample_rate, int)
+        assert speaker == 'en_0'
+        assert sample_rate == 48000
+        assert example == 'Can you can a canned can into an un-canned can like a canner can can a canned can into an un-canned can?'
+
+    def test_init_jit_model_en(self):
+        name = "en_v5.jit"
+        model_url_en = f'https://models.silero.ai/models/en/{name}'
+        model_en1, decoder_en1 = mo.init_jit_model(model_url_en, device=self.device, output_folder=self.tmp_dir)
+        self.assertIsInstance(model_en1, torch.nn.Module)
+        self.assertIsInstance(decoder_en1, Decoder)
+
+        model_en2, decoder_en2 = mo.init_jit_model(model_url_en, device=self.device)
+        self.assertIsInstance(model_en2, torch.nn.Module)
+        self.assertIsInstance(decoder_en2, Decoder)
+        # model_en_filepath.unlink(missing_ok=False)
+        
+        model_en3, decoder_en3 = mo.init_jit_model(model_url_en)
+        self.assertIsInstance(model_en3, torch.nn.Module)
+        self.assertIsInstance(decoder_en3, Decoder)
+        # model_en_filepath.unlink(missing_ok=False)
+
+    def test_get_models_de(self):
+        models_de = mo.get_models(self.language_de, self.tmp_dir, "latest", "stt_models")
+        self.assertIn(self.language_de, models_de.stt_models)
+
+    def test_get_models_en(self):
+        models_en = mo.get_models(self.language_en, self.tmp_dir, "latest", "stt_models")
+        self.assertIn(self.language_en, models_en.stt_models)
+
+    def test_get_latest_model_de(self):
+        model_de, decoder_de = mo.get_latest_model(self.language_de, self.tmp_dir, "latest", "stt_models", "jit")
+        self.assertIsInstance(model_de, torch.nn.Module)
+        self.assertIsInstance(decoder_de, Decoder)
+
+    def test_get_latest_model_en(self):
+        model_en, decoder_en = mo.get_latest_model(self.language_en, self.tmp_dir, "latest", "stt_models", "jit")
+        self.assertIsInstance(model_en, torch.nn.Module)
+        self.assertIsInstance(decoder_en, Decoder)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_pronunciationtrainer.py

@@ -179,10 +179,20 @@ class TestScore(unittest.TestCase):
         output_hash = utilities.hash_calculate(signal_de, is_file=False)
         assert output_hash == b'D9pMFzYL1BSPPg89ZCQE61xzb7QICXolYtC9EJRpvS0='
         signal_transformed = transform(torch.Tensor(signal_de)).unsqueeze(0)
-        processed_audio = trainer_SST_lambda_de.preprocessAudio(signal_transformed)
-        self.assertIsInstance(processed_audio, torch.Tensor)
-        self.assertEqual(processed_audio.shape, (1, 23400))
-        output_hash = utilities.hash_calculate(processed_audio.numpy(), is_file=False)
+        preprocessed_audio = trainer_SST_lambda_de.preprocessAudio(signal_transformed)
+        self.assertIsInstance(preprocessed_audio, torch.Tensor)
+        self.assertEqual(preprocessed_audio.shape, (1, 23400))
+        output_hash = utilities.hash_calculate(preprocessed_audio.numpy(), is_file=False)
+        assert output_hash == b'Ri/1rmgYmRSWaAw/Y3PoLEu1woiczhSUdUCbaMf++EM='
+
+    def test_preprocessAudioStandalone_de(self):
+        output_hash = utilities.hash_calculate(signal_de, is_file=False)
+        assert output_hash == b'D9pMFzYL1BSPPg89ZCQE61xzb7QICXolYtC9EJRpvS0='
+        signal_transformed = transform(torch.Tensor(signal_de)).unsqueeze(0)
+        preprocessed_audio = pronunciationTrainer.preprocessAudioStandalone(signal_transformed)
+        self.assertIsInstance(preprocessed_audio, torch.Tensor)
+        self.assertEqual(preprocessed_audio.shape, (1, 23400))
+        output_hash = utilities.hash_calculate(preprocessed_audio.numpy(), is_file=False)
         assert output_hash == b'Ri/1rmgYmRSWaAw/Y3PoLEu1woiczhSUdUCbaMf++EM='
 
     def test_processAudioForGivenText_getTranscriptAndWordsLocations_en(self):
@@ -257,10 +267,20 @@ class TestScore(unittest.TestCase):
         output_hash = utilities.hash_calculate(signal_en, is_file=False)
         assert output_hash == b'zBAV/y7mecyPHLGiitHRP9vK7oU9hnYvyuatU0PQfts='
         signal_transformed = transform(torch.Tensor(signal_en)).unsqueeze(0)
-        processed_audio = trainer_SST_lambda_en.preprocessAudio(signal_transformed)
-        self.assertIsInstance(processed_audio, torch.Tensor)
-        self.assertEqual(processed_audio.shape, (1, 16800))
-        output_hash = utilities.hash_calculate(processed_audio.numpy(), is_file=False)
+        preprocessed_audio = trainer_SST_lambda_en.preprocessAudio(signal_transformed)
+        self.assertIsInstance(preprocessed_audio, torch.Tensor)
+        self.assertEqual(preprocessed_audio.shape, (1, 16800))
+        output_hash = utilities.hash_calculate(preprocessed_audio.numpy(), is_file=False)
+        assert output_hash == b'KsyH1MXIc+5e5B6CcijhitsGPUDRJjrJU2qg8bQi600='
+
+    def test_preprocessAudioStandalone_en(self):
+        output_hash = utilities.hash_calculate(signal_en, is_file=False)
+        assert output_hash == b'zBAV/y7mecyPHLGiitHRP9vK7oU9hnYvyuatU0PQfts='
+        signal_transformed = transform(torch.Tensor(signal_en)).unsqueeze(0)
+        preprocessed_audio = pronunciationTrainer.preprocessAudioStandalone(signal_transformed)
+        self.assertIsInstance(preprocessed_audio, torch.Tensor)
+        self.assertEqual(preprocessed_audio.shape, (1, 16800))
+        output_hash = utilities.hash_calculate(preprocessed_audio.numpy(), is_file=False)
         assert output_hash == b'KsyH1MXIc+5e5B6CcijhitsGPUDRJjrJU2qg8bQi600='
 
 

tests/utils/test_split_cosmic_ray_report.py

@@ -6,11 +6,11 @@ from tests import EVENTS_FOLDER
 
 
 class TestSplitCosmicRayReport(unittest.TestCase):
-    def test_get_cosmic_ray_report_filtered(self):
+    def test_get_cosmic_ray_report_filtered_only_string_killed(self):
         input_filename = EVENTS_FOLDER / "cosmic-ray-lambdagetsample.txt"
         output_filename = EVENTS_FOLDER / f"{input_filename.stem}_filtered{input_filename.suffix}"
         self.assertFalse(output_filename.exists())
-        get_cosmic_ray_report_filtered(input_filename, separator="============", filter_string="test outcome: TestOutcome.KILLED")
+        get_cosmic_ray_report_filtered(input_filename, separator="============", filter_string_list=["test outcome: TestOutcome.KILLED", ])
 
         # Check if the filtered file is created
         self.assertTrue(output_filename.exists() and output_filename.is_file())
@@ -22,5 +22,21 @@ class TestSplitCosmicRayReport(unittest.TestCase):
         output_filename.unlink(missing_ok=False)
 
 
+    def test_get_cosmic_ray_report_filtered_list_strings(self):
+        input_filename = EVENTS_FOLDER / "cosmic-ray-pronunciationtrainer.txt"
+        output_filename = EVENTS_FOLDER / f"{input_filename.stem}_filtered{input_filename.suffix}"
+        self.assertFalse(output_filename.exists())
+        get_cosmic_ray_report_filtered(input_filename, separator="============", filter_string_list=["test outcome: TestOutcome.KILLED", "-        duration = time.time()"])
+
+        # Check if the filtered file is created
+        self.assertTrue(output_filename.exists() and output_filename.is_file())
+
+        # Verify the filtered content
+        hash_output = utilities.hash_calculate(output_filename, True)
+        self.assertEqual(hash_output, b'Fk0KDWCbc8mPoZllQ7HfgMjuWQVvUdNl+eR56eJeSxg=')
+
+        output_filename.unlink(missing_ok=False)
+
+
 if __name__ == "__main__":
     unittest.main()