Thought I had staged it
This commit is contained in:
Carlos-Mesquita
@@ -9,7 +9,7 @@ from app.repositories.abc import IFileStorage, IDocumentStore
|
||||
from app.services.abc import ISpeakingService, ILLMService, IVideoGeneratorService, ISpeechToTextService
|
||||
from app.configs.constants import (
|
||||
FieldsAndExercises, GPTModels, TemperatureSettings,
|
||||
ELAIAvatars, FilePaths
|
||||
FilePaths
|
||||
)
|
||||
from app.helpers import TextHelper
|
||||
|
||||
@@ -17,14 +17,12 @@ from app.helpers import TextHelper
|
||||
class SpeakingService(ISpeakingService):
|
||||
|
||||
def __init__(
|
||||
self, llm: ILLMService, vid_gen: IVideoGeneratorService,
|
||||
file_storage: IFileStorage, document_store: IDocumentStore,
|
||||
self, llm: ILLMService,
|
||||
file_storage: IFileStorage,
|
||||
stt: ISpeechToTextService
|
||||
):
|
||||
self._llm = llm
|
||||
self._vid_gen = vid_gen
|
||||
self._file_storage = file_storage
|
||||
self._document_store = document_store
|
||||
self._stt = stt
|
||||
self._logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -102,7 +100,7 @@ class SpeakingService(ISpeakingService):
|
||||
}
|
||||
|
||||
async def get_speaking_part(
|
||||
self, part: int, topic: str, difficulty: str, second_topic: Optional[str] = None
|
||||
self, part: int, topic: str, second_topic: str, difficulty: str
|
||||
) -> Dict:
|
||||
task_values = self._tasks[f'task_{part}']['get']
|
||||
|
||||
@@ -416,190 +414,6 @@ class SpeakingService(ISpeakingService):
|
||||
)
|
||||
return response["fixed_text"]
|
||||
|
||||
async def create_videos_and_save_to_db(self, exercises, template, req_id):
|
||||
template = await self._create_video_per_part(exercises, template, 1)
|
||||
template = await self._create_video_per_part(exercises, template, 2)
|
||||
template = await self._create_video_per_part(exercises, template, 3)
|
||||
|
||||
await self._document_store.save_to_db_with_id("speaking", template, req_id)
|
||||
self._logger.info(f'Saved speaking to DB with id {req_id} : {str(template)}')
|
||||
|
||||
async def _create_video_per_part(self, exercises: List[Dict], template: Dict, part: int):
|
||||
avatar = (random.choice(list(ELAIAvatars))).name
|
||||
template_index = part - 1
|
||||
|
||||
# Using list comprehension to find the element with the desired value in the 'type' field
|
||||
found_exercises = [element for element in exercises if element.get('type') == part]
|
||||
|
||||
# Check if any elements were found
|
||||
if found_exercises:
|
||||
exercise = found_exercises[0]
|
||||
self._logger.info(f'Creating video for speaking part {part}')
|
||||
if part in {1, 3}:
|
||||
questions = []
|
||||
for question in exercise["questions"]:
|
||||
result = await self._create_video(
|
||||
question,
|
||||
avatar,
|
||||
f'Failed to create video for part {part} question: {str(exercise["question"])}'
|
||||
)
|
||||
if result is not None:
|
||||
video = {
|
||||
"text": question,
|
||||
"video_path": result["video_path"],
|
||||
"video_url": result["video_url"]
|
||||
}
|
||||
questions.append(video)
|
||||
|
||||
template["exercises"][template_index]["prompts"] = questions
|
||||
if part == 1:
|
||||
template["exercises"][template_index]["first_title"] = exercise["first_topic"]
|
||||
template["exercises"][template_index]["second_title"] = exercise["second_topic"]
|
||||
else:
|
||||
template["exercises"][template_index]["title"] = exercise["topic"]
|
||||
else:
|
||||
result = await self._create_video(
|
||||
exercise["question"],
|
||||
avatar,
|
||||
f'Failed to create video for part {part} question: {str(exercise["question"])}'
|
||||
)
|
||||
if result is not None:
|
||||
template["exercises"][template_index]["prompts"] = exercise["prompts"]
|
||||
template["exercises"][template_index]["text"] = exercise["question"]
|
||||
template["exercises"][template_index]["title"] = exercise["topic"]
|
||||
template["exercises"][template_index]["video_url"] = result["video_url"]
|
||||
template["exercises"][template_index]["video_path"] = result["video_path"]
|
||||
|
||||
if not found_exercises:
|
||||
template["exercises"].pop(template_index)
|
||||
|
||||
return template
|
||||
|
||||
async def generate_video(
|
||||
self, part: int, avatar: str, topic: str, questions: list[str],
|
||||
*,
|
||||
second_topic: Optional[str] = None,
|
||||
prompts: Optional[list[str]] = None,
|
||||
suffix: Optional[str] = None,
|
||||
):
|
||||
params = locals()
|
||||
params.pop('self')
|
||||
|
||||
request_id = str(uuid.uuid4())
|
||||
self._logger.info(
|
||||
f'POST - generate_video_{part} - Received request to generate video {part}. '
|
||||
f'Use this id to track the logs: {request_id} - Request data: " + {params}'
|
||||
)
|
||||
|
||||
part_questions = self._get_part_questions(part, questions, avatar)
|
||||
videos = []
|
||||
|
||||
self._logger.info(f'POST - generate_video_{part} - {request_id} - Creating videos for speaking part {part}.')
|
||||
for question in part_questions:
|
||||
self._logger.info(f'POST - generate_video_{part} - {request_id} - Creating video for question: {question}')
|
||||
result = await self._create_video(
|
||||
question,
|
||||
avatar,
|
||||
'POST - generate_video_{p} - {r} - Failed to create video for part {p} question: {q}'.format(
|
||||
p=part, r=request_id, q=question
|
||||
)
|
||||
)
|
||||
if result is not None:
|
||||
self._logger.info(f'POST - generate_video_{part} - {request_id} - Video created')
|
||||
self._logger.info(
|
||||
f'POST - generate_video_{part} - {request_id} - Uploaded video to firebase: {result["video_url"]}'
|
||||
)
|
||||
video = {
|
||||
"text": question,
|
||||
"video_path": result["video_path"],
|
||||
"video_url": result["video_url"]
|
||||
}
|
||||
videos.append(video)
|
||||
|
||||
if part == 2 and len(videos) == 0:
|
||||
raise Exception(f'Failed to create video for part 2 question: {questions[0]}')
|
||||
|
||||
return self._get_part_response(part, topic, videos, second_topic, prompts, suffix)
|
||||
|
||||
@staticmethod
|
||||
def _get_part_questions(part: int, questions: list[str], avatar: str):
|
||||
part_questions: list[str] = []
|
||||
|
||||
if part == 1:
|
||||
id_to_name = {
|
||||
"5912afa7c77c47d3883af3d874047aaf": "MATTHEW",
|
||||
"9e58d96a383e4568a7f1e49df549e0e4": "VERA",
|
||||
"d2cdd9c0379a4d06ae2afb6e5039bd0c": "EDWARD",
|
||||
"045cb5dcd00042b3a1e4f3bc1c12176b": "TANYA",
|
||||
"1ae1e5396cc444bfad332155fdb7a934": "KAYLA",
|
||||
"0ee6aa7cc1084063a630ae514fccaa31": "JEROME",
|
||||
"5772cff935844516ad7eeff21f839e43": "TYLER",
|
||||
|
||||
}
|
||||
part_questions.extend(
|
||||
[
|
||||
"Hello my name is " + id_to_name.get(avatar) + ", what is yours?",
|
||||
"Do you work or do you study?",
|
||||
*questions
|
||||
]
|
||||
)
|
||||
elif part == 2:
|
||||
# Removed as the examiner should not say what is on the card.
|
||||
# question = question + " In your answer you should consider: " + " ".join(prompts) + suffix
|
||||
part_questions.append(f'{questions[0]}\nYou have 1 minute to take notes.')
|
||||
elif part == 3:
|
||||
part_questions = questions
|
||||
|
||||
return part_questions
|
||||
|
||||
@staticmethod
|
||||
def _get_part_response(
|
||||
part: int,
|
||||
topic: str,
|
||||
videos: list[dict],
|
||||
second_topic: Optional[str],
|
||||
prompts: Optional[list[str]],
|
||||
suffix: Optional[str]
|
||||
):
|
||||
response = {}
|
||||
if part == 1:
|
||||
response = {
|
||||
"prompts": videos,
|
||||
"first_title": topic,
|
||||
"second_title": second_topic,
|
||||
"type": "interactiveSpeaking"
|
||||
}
|
||||
if part == 2:
|
||||
response = {
|
||||
"prompts": prompts,
|
||||
"title": topic,
|
||||
"suffix": suffix,
|
||||
"type": "speaking",
|
||||
# includes text, video_url and video_path
|
||||
**videos[0]
|
||||
}
|
||||
if part == 3:
|
||||
response = {
|
||||
"prompts": videos,
|
||||
"title": topic,
|
||||
"type": "interactiveSpeaking",
|
||||
}
|
||||
|
||||
response["id"] = str(uuid.uuid4())
|
||||
return response
|
||||
|
||||
async def _create_video(self, question: str, avatar: str, error_message: str):
|
||||
result = await self._vid_gen.create_video(question, avatar)
|
||||
if result is not None:
|
||||
sound_file_path = FilePaths.VIDEO_FILES_PATH + result
|
||||
firebase_file_path = FilePaths.FIREBASE_SPEAKING_VIDEO_FILES_PATH + result
|
||||
url = await self._file_storage.upload_file_firebase_get_url(firebase_file_path, sound_file_path)
|
||||
return {
|
||||
"video_path": firebase_file_path,
|
||||
"video_url": url
|
||||
}
|
||||
self._logger.error(error_message)
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _grade_template():
|
||||
|
||||
Reference in New Issue
Block a user