Upload folder using huggingface_hub

app/app.py

@@ -9,182 +9,18 @@ load_dotenv()
 from retry import retry
 from typing import List, Dict, Optional
 from tqdm import tqdm
+from .llm import LLM, get_languages, get_level_names
+from .speech import Speech
 
-countries = {"nl": {"language":"Dutch", "country":"the Netherlands"},
-            "de": {"language":"German", "country":"Germany"},
-            "se": {"language":"Swedish", "country":"Sweden"},
-            "en": {"language":"English", "country":"England"},
-            "fr": {"language":"French", "country":"France"},
-            "it": {"language":"Italian", "country":"Italy"},
-            "es": {"language":"Spanish", "country":"Spain"},
-            "pl": {"language":"Polish", "country":"Poland"},
-            "hu": {"language":"Hungarian", "country":"Hungary"},
-            }
-
-moods = ["hilarious", "funny", "serious", "poetic"]
-
-levels = {"easy": [4, 8], 
-        "medium": [7, 16],
-        "complicated": [12, 25],
-        "very complicated": [20, 40]}
-
-def get_languages():
-    return list(countries.keys())
-
-def get_level_names():
-    return list(levels.keys())
+# n = 10
+# input_country = "nl"
+# target_country = "de"
+# input_length = "easy"
 
 languages = get_languages()
 #languages_flags = [ emoji.emojize(f":flag-{countries[language]['country'].lower().replace(' ', '-')}:") for language in languages]
 level_names = get_level_names()
 
-# 
-#  -- llm.py --
-#  
-
-class LLM():
-
-    def __init__(self):
-
-        load_dotenv()
-        openai.api_key = os.getenv("OPENAI_API_KEY")
-        self.min_length = 10
-
-    def parse_response(self, response:str) -> List[Dict[str, str]]:
-
-        logger.info(f"{response = }")
-
-        logger.info(f"Parsing response from OpenAI API.")
-
-        text = response
-        tokens = text.split("Sentence ")
-
-        logger.debug(f"Number of tokens: {len(tokens)}")
-        logger.debug(f"{tokens = }")
-
-
-        qas = []
-        for token in tokens:
-            if len(token) == 0:
-                continue
-            lines = token.split("\n")
-            #print(f"{lines = }")
-            original = lines[0][3:]
-
-            true_trans = ""
-            for line in lines:
-                if line.startswith("True: "):
-                    true_trans = line[6:]
-
-            # get line starting with "False: "
-            false_trans = ""
-            for line in lines:
-                if line.startswith("False: "):
-                    false_trans = line[7:]
-
-            # get line starting with "Funny: "
-            funny_trans = ""
-            for line in lines:
-                if line.startswith("Funny: "):
-                    funny_trans = line[7:]
-
-            qa = {"original": original, "true": true_trans, "false": false_trans, "funny": funny_trans}
-
-            logger.debug(f"------------------------")
-            logger.debug(f"{qa = }")
-            logger.debug(f"------------------------")
-
-            if len(qa["original"]) > self.min_length and \
-                len(qa["true"]) > self.min_length and \
-                len(qa["false"]) > self.min_length and \
-                len(qa["funny"]) > self.min_length:
-                qas.append(qa)
-
-        logger.info(f"Returning {len(qas)} valid QA pairs.")
-
-        return qas
-
-    @retry(delay=0, backoff=2, max_delay=20, tries=10, logger=logger)
-    def generate(self, n:int, 
-                 input_country:str, 
-                 target_country:str, 
-                 level:str, 
-                 temperature:float=0.80, 
-                 streaming:bool=False) -> Dict[str, str]:
-        input_length = levels[level]
-
-        language = countries[input_country]['language']
-        country = countries[input_country]['country']
-        target_language = countries[target_country]['language']
-        mood = random.choice(moods)
-
-        logger.info(f"Generating {str(n)} QA pairs for {language} to {target_language} with level: {level}.")
-
-
-        assert temperature >= 0.0 and temperature <= 1.0, "temperature must be between 0 and 1"
-        response = openai.ChatCompletion.create(
-            model="gpt-3.5-turbo",
-            messages=[
-                {"role": "system", "content": 
-                f"You are a helpful assistant at creating a translation game. \
-You create sentences in {language} with a length that is exactly between {str(input_length[0])} to {str(input_length[1])} words. \
-You create translations in {target_language}. Do not include a translation in the original sentence. \
-Ensure that each translation is quite different from the other ones and reduce repetition. \
-Formulate your answer in exactly this format: Sentence N: [X],\True: [A],\nFalse: [B],\nFunny: [C]."},
-                {"role": "assistant", "content": 
-                f"Sometimes, create original sentences with words, locations, concepts and phrases are typical for {country}. Otherwise, create sentences that are unlreated to the country."},
-                {"role": "user", "content": 
-                f"Generate {str(n)} {level} and {mood} sentences in {language} and 3 corresponding translations in {target_language}. \
-You create 1 correct translation, 1 incorrect translations, and 1 which very wrong and funny. \
-"},
-            ],
-            temperature=temperature,
-            stream=streaming
-        )
-
-        if streaming:
-            import time
-            start_time = time.time()
-            collected_chunks = []
-            collected_messages = []
-            for chunk in tqdm(response, total=250):
-                chunk_time = time.time() - start_time  # calculate the time delay of the chunk
-                collected_chunks.append(chunk)  # save the event response
-                chunk_message = chunk['choices'][0]['delta']  # extract the message
-                collected_messages.append(chunk_message)  # save the message
-                #print(f"Message received {chunk_time:.2f} seconds after request: {chunk_message}")  # print the delay and text
-
-            full_reply_content = ''.join([m.get('content', '') for m in collected_messages])
-            return full_reply_content
-        else:
-            return response.choices[0].message.content
-
-    def get_QAs(self, n:int, input_countries:List[str], target_country:str, level:str, debug:bool=False):
-    
-        logger.info(f"Generating new sentences...")
-        # gr.Info(f"Generating new Q&A in {input_countries} with level: {level}.")
-
-
-        if debug:
-            return [ {"original": "The Netherlands is a country in Europe.",
-                        "true": "Nederland is een land in Europa.",
-                        "false": "Nederland is een land in Azië.",
-                        "funny": "Nederland is een aap in Europa."},
-                    {"original": "Aap, noot, mies.",
-                        "true": "Aap.",
-                        "false": "Noot.",
-                        "funny": "Mies."} ]
-        else:
-            return self.parse_response(
-                        self.generate(n=n, 
-                                      input_country=random.choice(input_countries),
-                                      target_country=target_country, 
-                                      level=level))
-
-# n = 10
-# input_country = "nl"
-# target_country = "de"
-# input_length = "easy"
 
 class Game:
 
@@ -195,6 +31,7 @@ class Game:
                  debug:bool=False):
 
         self.llm = LLM()
+        # self.speech = Speech()
         self.n = n
         self.input_countries = input_countries
         self.target_country = target_country
@@ -206,11 +43,19 @@ class Game:
         self.score_relative = {'correct': 0., 'incorrect': 0.}
         self.total = 0
 
-        self.qas = self.llm.get_QAs(n=self.n,
-                                input_countries=self.input_countries,
-                                target_country=self.target_country,
-                                level=self.level,
-                                debug=self.debug)
+
+        # self.qas = self.llm.get_QAs(n=self.n,
+        #                         input_countries=self.input_countries,
+        #                         target_country=self.target_country,
+        #                         level=self.level,
+        #                         debug=self.debug)
+
+        self.qas = [{
+            "original": "So you think you can translate?",
+            "true": "Ja",
+            "false": "Sure",
+            "funny": "Igen"
+        }]
 
 
     def get_qa(self):
@@ -232,28 +77,31 @@ class Game:
         true = qa["true"]
         random.shuffle(options)
         options_md = f"""### A: {options[0]}\n\n### B: {options[1]}\n\n### C: {options[2]}"""
+
         # get correct answer
         self.correct_answer = [letter for letter, option in zip(["A", "B", "C"], options) if option == true][0]
 
-        logger.info(f"get_qa()")
-        logger.info(f"Question: {question_md}")
-        logger.info(f"Options: {options}")
-        logger.info(f"Correct answer: {self.correct_answer}")
+        logger.debug(f"get_qa()")
+        logger.debug(f"Question: {question_md}")
+        logger.debug(f"Options: {options}")
+        logger.debug(f"Correct answer: {self.correct_answer}")
+
         return question_md, options_md
         
-    def update(self, answer):
+    def check_answer(self, answer):
 
         # check answer
         logger.info(f"Checking answer \"{answer}\" against \"{self.correct_answer}\"")
         result = answer == self.correct_answer
+
         # display output
         gr.Info(emoji.emojize(":thumbs_up:")+" Correct!") if result else gr.Info(emoji.emojize(":thumbs_down:"+f" Wrong! Correct answer was: {self.correct_answer}"))
         #output = emoji.emojize("# :thumbs_up:") if result else emoji.emojize("# :thumbs_down:")
 
-        self.score["correct"] += 1 if result else 0
-        self.score["incorrect"] += 0 if result else 1
-        self.total += 1
-        score_relative = {'correct': self.score['correct']/self.total, 'incorrect': self.score['incorrect']/self.total}
+        # self.score["correct"] += 1 if result else 0
+        # self.score["incorrect"] += 0 if result else 1
+        # self.total += 1
+        # score_relative = {'correct': self.score['correct']/self.total, 'incorrect': self.score['incorrect']/self.total}
 
 
         # get new question and answer
@@ -263,7 +111,7 @@ class Game:
         logger.info(f"Question: {question}")
         logger.info(f"Options: {options}")
 
-        return question, options, score_relative
+        return question, options
 
     def update_settings(self, new_input_countries, new_target_country, new_level):
 
@@ -290,7 +138,7 @@ class Game:
 
         if changed:
             logger.info(f"Setting changed, generating new QAs...")
-            gr.Info("Setting changed, generating new QAs...")
+            gr.Info("Setting changed, generating new sentences...")
             self.qas = self.llm.get_QAs(n=1, 
                                         input_countries=self.input_countries, 
                                         target_country=self.target_country, 
@@ -311,6 +159,10 @@ class Game:
                     # show question
                     question_md = gr.Markdown(question)
 
+                    # # add audio
+                    # self.speech.text_to_speech(question, self.input_countries[0])
+                    # speech = gr.Audio("output.wav", interactive=False)
+             
                     # show answers
                     answers_md = gr.Markdown(options)
 
@@ -323,13 +175,13 @@ class Game:
                     # # show output box
                     # output = gr.Markdown()
 
-                    # show scores
-                    if score_relative is None:
-                        score_relative = {'correct': 0., 'incorrect': 0.}
-                    scores = gr.Label(score_relative, label="Scores", show_label=False) #, visible=True if self.total > 0 else False)
+                    # # show scores
+                    # if score_relative is None:
+                    #     score_relative = {'correct': 0., 'incorrect': 0.}
+                    # scores = gr.Label(score_relative, label="Scores", show_label=False) #, visible=True if self.total > 0 else False)
 
                     # event listener
-                    button.click(fn=self.update, inputs=radio, outputs=[question_md, answers_md, scores], scroll_to_output=True)
+                    button.click(fn=self.check_answer, inputs=radio, outputs=[question_md, answers_md], scroll_to_output=True)
 
             with gr.Tab("Settings"):
                 # Settings
@@ -344,6 +196,6 @@ class Game:
 
 
 if __name__ == "__main__":
-    game = Game(n=1, input_countries=["nl", "en", "se"], target_country="de", level="medium", debug=False)
+    game = Game(n=1, input_countries=["en"], target_country="de", level="complicated", debug=False)
     question, options = game.get_qa()
     game.run(question, options, share=False)

app/llm.py

@@ -0,0 +1,175 @@
+import os
+import openai
+import random
+from loguru import logger
+from dotenv import load_dotenv
+load_dotenv()
+from retry import retry
+from typing import List, Dict, Optional
+from tqdm import tqdm
+
+
+# speech ISO codes from: https://dl.fbaipublicfiles.com/mms/tts/all-tts-languages.html
+countries = {"nl": {"language":"Dutch", "country":"the Netherlands", "flag": "🇳🇱", "iso": "nld"}, 
+            "de": {"language":"German", "country":"Germany", "flag": "🇩🇪", "iso": "deu"},
+            "se": {"language":"Swedish", "country":"Sweden", "flag": "🇸🇪", "iso": "swe"},
+            "en": {"language":"English", "country":"England", "flag": "🇬🇧", "iso": "eng"},
+            "fr": {"language":"French", "country":"France", "flag": "🇫🇷", "iso": "fra"},
+            "it": {"language":"Italian", "country":"Italy", "flag": "🇮🇹", "iso": None},
+            "es": {"language":"Spanish", "country":"Spain", "flag": "🇪🇸", "iso": "spa"},
+            "pl": {"language":"Polish", "country":"Poland", "flag": "🇵🇱", "iso": "pol"},
+            "hu": {"language":"Hungarian", "country":"Hungary", "flag": "🇭🇺", "iso": "hun"},
+            "fi": {"language":"Finnish", "country":"Finland", "flag": "🇫🇮", "iso": "fin"},
+            }
+
+moods = ["hilarious", "funny", "serious", "poetic"]
+
+levels = {"easy": [4, 8], 
+        "medium": [7, 16],
+        "complicated": [12, 25],
+        "very complicated": [20, 40]}
+
+def get_languages():
+    return list(countries.keys())
+
+def get_level_names():
+    return list(levels.keys())
+
+class LLM():
+
+    def __init__(self):
+
+        load_dotenv()
+        openai.api_key = os.getenv("OPENAI_API_KEY")
+        self.min_length = 10
+
+    def parse_response(self, response:str) -> List[Dict[str, str]]:
+
+        logger.info(f"{response = }")
+
+        logger.info(f"Parsing response from OpenAI API.")
+
+        text = response
+        tokens = text.split("Sentence ")
+
+        logger.debug(f"Number of tokens: {len(tokens)}")
+        logger.debug(f"{tokens = }")
+
+
+        qas = []
+        for token in tokens:
+            if len(token) == 0:
+                continue
+            lines = token.split("\n")
+            #print(f"{lines = }")
+            original = lines[0][3:]
+
+            true_trans = ""
+            for line in lines:
+                if line.startswith("True: "):
+                    true_trans = line[6:]
+
+            # get line starting with "False: "
+            false_trans = ""
+            for line in lines:
+                if line.startswith("False: "):
+                    false_trans = line[7:]
+
+            # get line starting with "Funny: "
+            funny_trans = ""
+            for line in lines:
+                if line.startswith("Funny: "):
+                    funny_trans = line[7:]
+
+            qa = {"original": original, "true": true_trans, "false": false_trans, "funny": funny_trans}
+
+            logger.debug(f"------------------------")
+            logger.debug(f"{qa = }")
+            logger.debug(f"------------------------")
+
+            if len(qa["original"]) > self.min_length and \
+                len(qa["true"]) > self.min_length and \
+                len(qa["false"]) > self.min_length and \
+                len(qa["funny"]) > self.min_length:
+                qas.append(qa)
+
+        logger.info(f"Returning {len(qas)} valid QA pairs.")
+
+        return qas
+
+    @retry(delay=0, backoff=2, max_delay=20, tries=10, logger=logger)
+    def generate(self, n:int, 
+                 input_country:str, 
+                 target_country:str, 
+                 level:str, 
+                 temperature:float=0.80, 
+                 streaming:bool=False) -> Dict[str, str]:
+        input_length = levels[level]
+
+        language = countries[input_country]['language']
+        country = countries[input_country]['country']
+        target_language = countries[target_country]['language']
+        mood = random.choice(moods)
+
+        logger.info(f"Generating {str(n)} QA pairs for {language} to {target_language} with level: {level}.")
+
+
+        assert temperature >= 0.0 and temperature <= 1.0, "temperature must be between 0 and 1"
+        response = openai.ChatCompletion.create(
+            model="gpt-3.5-turbo",
+            messages=[
+                {"role": "system", "content": 
+                f"You are a helpful assistant at creating a translation game. \
+You create sentences in {language} with a length that is exactly between {str(input_length[0])} to {str(input_length[1])} words. \
+You create translations in {target_language}. Do not include a translation in the original sentence. \
+Ensure that each translation is quite different from the other ones and reduce repetition. \
+Formulate your answer in exactly this format: Sentence N: [X],\True: [A],\nFalse: [B],\nFunny: [C]."},
+                {"role": "assistant", "content": 
+                f"Sometimes, create original sentences with words, locations, concepts and phrases are typical for {country}. Otherwise, create sentences that are unlreated to the country."},
+                {"role": "user", "content": 
+                f"Generate {str(n)} {level} and {mood} sentences in {language} and 3 corresponding translations in {target_language}. \
+You create 1 correct translation, 1 incorrect translations, and 1 which very wrong and funny. \
+"},
+            ],
+            temperature=temperature,
+            stream=streaming
+        )
+
+        if streaming:
+            import time
+            start_time = time.time()
+            collected_chunks = []
+            collected_messages = []
+            for chunk in tqdm(response, total=250):
+                chunk_time = time.time() - start_time  # calculate the time delay of the chunk
+                collected_chunks.append(chunk)  # save the event response
+                chunk_message = chunk['choices'][0]['delta']  # extract the message
+                collected_messages.append(chunk_message)  # save the message
+                #print(f"Message received {chunk_time:.2f} seconds after request: {chunk_message}")  # print the delay and text
+
+            full_reply_content = ''.join([m.get('content', '') for m in collected_messages])
+            return full_reply_content
+        else:
+            return response.choices[0].message.content
+
+    def get_QAs(self, n:int, input_countries:List[str], target_country:str, level:str, debug:bool=False):
+    
+        logger.info(f"Generating new sentences...")
+        # gr.Info(f"Generating new Q&A in {input_countries} with level: {level}.")
+
+
+        if debug:
+            return [ {"original": "The Netherlands is a country in Europe.",
+                        "true": "Nederland is een land in Europa.",
+                        "false": "Nederland is een land in Azië.",
+                        "funny": "Nederland is een aap in Europa."},
+                    {"original": "Aap, noot, mies.",
+                        "true": "Aap.",
+                        "false": "Noot.",
+                        "funny": "Mies."} ]
+        else:
+            return self.parse_response(
+                        self.generate(n=n, 
+                                      input_country=random.choice(input_countries),
+                                      target_country=target_country, 
+                                      level=level))

app/speech.py

@@ -0,0 +1,41 @@
+import torch
+from TTS.api import TTS
+from loguru import logger
+
+class Speech():
+    
+    def __init__(self) -> None:
+        
+        # List available 🐸TTS models
+        # print(TTS().list_models())
+
+        # Get device
+        device = "cuda" if torch.cuda.is_available() else "cpu"
+
+        # Init TTS (and download model if necessary)
+        self.tts = TTS("tts_models/multilingual/multi-dataset/xtts_v1").to(device)
+
+        logger.info(f"Initialized TTS: {self.tts}")
+
+    def text_to_speech(self, text:str, language:str):
+
+        logger.info(f"Text to speech: {text} to {language}")
+
+        # remove ## from text
+        text = text.replace("##", "")
+
+        # language =  en, de, es, fr, it, pt, pl.
+        valid_languages = ["en", "de", "es", "fr", "it", "pt", "pl"]
+        if language not in valid_languages:
+            raise ValueError(f"Language '{language}' is not one of the supported languages: {valid_languages}")
+            return None 
+        else:
+            # Run TTS
+            # ❗ Since this model is multi-lingual voice cloning model, we must set the target speaker_wav and language
+            # Text to speech list of amplitude values as output
+            # wav = self.tts.tts(text=text, language=language)
+            # # Text to speech to a file
+            self.tts.tts_to_file(text=text, language=language, file_path="output.wav")
+
+            logger.info(f"Converted text to speech")
+            return "output.wav"

app/speech_google.py

@@ -0,0 +1,48 @@
+"""Synthesizes speech from the input string of text or ssml.
+Make sure to be working in a virtual environment.
+
+Note: ssml must be well-formed according to:
+    https://www.w3.org/TR/speech-synthesis/
+"""
+from google.cloud import texttospeech
+from loguru import logger
+
+class Speech:
+
+    def __init__(self) -> None:
+
+        # Instantiates a client
+        self.client = texttospeech.TextToSpeechClient()
+
+        # Select the type of audio file you want returned
+        self.audio_config = texttospeech.AudioConfig(
+            audio_encoding=texttospeech.AudioEncoding.MP3
+        )
+
+        logger.info(f"Initialized Speech()")
+
+    def text_to_speech(self, text:str, language:str):
+
+        logger.info(f"Converting text to speech.")
+
+        # Set the text input to be synthesized
+        synthesis_input = texttospeech.SynthesisInput(text=text)
+
+        # Build the voice request, select the language code ("en-US") and the ssml
+        # voice gender ("neutral")
+        voice = texttospeech.VoiceSelectionParams(
+            language_code="en-US", 
+            ssml_gender=texttospeech.SsmlVoiceGender.NEUTRAL
+        )
+
+        # Perform the text-to-speech request on the text input with the selected
+        # voice parameters and audio file type
+        response = self.client.synthesize_speech(
+            input=synthesis_input, voice=voice, audio_config=self.audio_config
+        )
+
+        # The response's audio_content is binary.
+        with open("output.mp3", "wb") as out:
+            # Write the response to the output file.
+            out.write(response.audio_content)
+            logger.info('Audio content written to file "output.mp3"')

requirements.txt

@@ -1,6 +1,7 @@
-emoji
-gradio
-python-dotenv
-openai
-loguru
-retry
+openai # openai api for chatGPT: https://platform.openai.com/docs/introduction
+gradio # UI: https://gradio.app/
+python-dotenv # dotenv: https://github.com/theskumar/python-dotenv
+emoji # convert emoji to text: https://github.com/carpedm20/emoji/
+loguru # logging: https://loguru.readthedocs.io/en/stable/index.html
+retry # retry decorator: https://github.com/invl/retry
+TTS # text to speech: https://github.com/coqui-ai/TTS