feat: add request validations for lambdaGestSample and lambdaSpeechToScore backend modules

aip_trainer/lambdas/lambdaGetSample.py

@@ -7,6 +7,7 @@ import pandas as pd
 
 from aip_trainer import PROJECT_ROOT_FOLDER, app_logger
 from aip_trainer.models import RuleBasedModels
+from aip_trainer.utils.typing_hints import BodyGetSampleRequest
 
 
 class TextDataset:
@@ -53,19 +54,11 @@ lambda_ipa_converter['en'] = RuleBasedModels.EngPhonemConverter()
 
 
 def lambda_handler(event, context):
-    body = json.loads(event['body'])
-
-    try:
-        category = int(body['category'])
-    except KeyError:
-        category = 0
-    language = body['language']
-    try:
-        current_transcript = str(body["transcript"])
-    except KeyError:
-        current_transcript = get_random_selection(language, category, is_gradio_output=False)
-    current_transcript = current_transcript if isinstance(current_transcript, str) else current_transcript[0]
-    current_ipa = lambda_ipa_converter[language].convertToPhonem(current_transcript)
+    event_body = event["body"]
+    body = BodyGetSampleRequest.model_validate_json(event_body)
+    current_transcript = get_random_selection(body.language, body.category, is_gradio_output=False, transcript=body.transcript)
+    current_transcript = current_transcript[0] if isinstance(current_transcript, list) else current_transcript
+    current_ipa = lambda_ipa_converter[body.language].convertToPhonem(current_transcript)
 
     app_logger.info(f"real_transcript='{current_transcript}', ipa_transcript='{current_ipa}'.")
     result = {
@@ -77,7 +70,9 @@ def lambda_handler(event, context):
     return json.dumps(result)
 
 
-def get_random_selection(language: str, category: int, is_gradio_output=True):
+def get_random_selection(language: str, category: int, is_gradio_output=True, transcript=None):
+    if transcript is not None and isinstance(transcript, str) and len(transcript) > 0:
+        return transcript
     lambda_df_lang = lambda_database[language]
     current_transcript = lambda_df_lang.get_random_sample_from_df(language, category)
     app_logger.info(f"category={category}, language={language}, current_transcript={current_transcript}.")

aip_trainer/lambdas/lambdaSpeechToScore.py

@@ -12,6 +12,7 @@ from torchaudio.transforms import Resample
 
 from aip_trainer import WordMatching as wm, app_logger
 from aip_trainer import pronunciationTrainer, sample_rate_start
+from aip_trainer.utils.typing_hints import BodySpeechToScoreRequest
 
 
 trainer_SST_lambda = {
@@ -22,14 +23,14 @@ transform = Resample(orig_freq=sample_rate_start, new_freq=16000)
 
 
 def lambda_handler(event, context):
-    body = event['body']
-    data = json.loads(body)
+    event_body = event['body']
+    data = BodySpeechToScoreRequest.model_validate_json(event_body)
 
-    real_text = data['title']
-    base64_audio = data["base64Audio"]
+    real_text = data.title
+    base64_audio = data.base64Audio
     app_logger.debug(f"base64Audio:{base64_audio} ...")
     file_bytes_or_audiotmpfile = base64.b64decode(base64_audio[22:].encode('utf-8'))
-    language = data['language']
+    language = data.language
 
     if len(real_text) == 0:
         return {

aip_trainer/utils/typing_hints.py

@@ -0,0 +1,19 @@
+from typing import Annotated, Optional
+from pydantic import BaseModel
+
+import annotated_types
+
+
+type Category = Annotated[int, annotated_types.Ge(0), annotated_types.Le(4)]
+
+
+class BodyGetSampleRequest(BaseModel):
+    category: Optional[Category] = 0
+    language: str
+    transcript: Optional[str] = ""
+
+
+class BodySpeechToScoreRequest(BaseModel):
+    base64Audio: str
+    language: str
+    title: str