Implementing text language Translation

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "python.analysis.extraPaths": ["${sunbird-ai-api-experimental}/"]
+}

app/api.py

@@ -1,4 +1,8 @@
 from fastapi import FastAPI
+from app.schemas.tasks import TranslationRequest, TranslationResponse
+from app.inference_services.translate import (translate_text,
+                                              long_text_translation,
+                                              predicted_language)
 
 
 app = FastAPI()
@@ -7,3 +11,22 @@
 @app.get("/")
 def read_root():
     return {"Hello": "World"}
+
+
+@app.post("/translate", response_model=TranslationResponse)
+def translate(translation_request: TranslationRequest):
+    # This is the pont where it checks if the source language is not Null
+    source_language = None
+    if translation_request.source_language == "":
+        source_language = predicted_language(translation_request.text)
+
+    if len(translation_request.text) < 200:
+        response = translate_text(translation_request.text,
+                                  translation_request.source_language,
+                                  translation_request.target_language)
+    else:
+        response = long_text_translation(translation_request.text,
+                                         translation_request.source_language,
+                                         translation_request.target_language)
+
+    return TranslationResponse(text=response, source_language=source_language)

app/inference_services/base.py

@@ -0,0 +1,48 @@
+import requests
+import os
+import time
+from dotenv import load_dotenv
+from tenacity import retry, wait_exponential
+
+load_dotenv()
+
+
+@retry(
+    wait=wait_exponential(multiplier=3, min=98, max=120)  # Exponential backoff
+)
+def inference_request_en_mul(payload):
+    url = 'https://api-inference.huggingface.co/m'\
+              'odels/Sunbird/sunbird-en-mul'
+    headers_en_mul = {"Authorization": os.getenv("HEADER_HUGGING_FACE_TOKEN")}
+    response = requests.post(url, headers=headers_en_mul, json=payload)
+    # TODOCreate a function that just calls it
+    # This is where i applied the exponential backoff
+    if response.status_code == 503:
+        estimated_time = response.json()['estimated_time']
+        time.sleep(estimated_time)
+        # logging.info(f"Model Loading ...{estimated_time}")
+        print(estimated_time)
+        response = requests.post(url, headers=headers_en_mul, json=payload)
+        return response.text
+    else:
+        return response.text
+
+
+@retry(
+    wait=wait_exponential(multiplier=3, min=98, max=120)  # Exponential backoff
+)
+def inference_request_mul_en(payload):
+    url = 'https://api-inference.huggingface.co/m'\
+              'odels/Sunbird/mbart-mul-en'
+    headers_mul_en = {"Authorization": os.getenv("HEADER_HUGGING_FACE_TOKEN")}
+    response = requests.post(url, headers=headers_mul_en, json=payload)
+    # This is where i applied the exponential backoff
+    if response.status_code == 503:
+        estimated_time = response.json()['estimated_time']
+        time.sleep(estimated_time)
+        # logging.info(f"Model Loading ...{estimated_time}")
+        print(estimated_time)
+        response = requests.post(url, headers=headers_mul_en, json=payload)
+        return response.text
+    else:
+        return response.text

app/inference_services/model.py

@@ -0,0 +1,14 @@
+import joblib
+from pathlib import Path
+
+BASE_DIR = Path(__file__).resolve(strict=True).parent
+
+with open(f'{BASE_DIR}/pipeline_MultinomialNB.pkl', "rb") as f:
+    model = joblib.load(f)
+
+
+def predicted_language(sentence):
+    samples = [sentence]
+    prediction = str(model.predict(samples))
+    prediction = prediction[2:-2]
+    return prediction

app/inference_services/translate.py

@@ -0,0 +1,78 @@
+from app.inference_services.base import (inference_request_en_mul,
+                                         inference_request_mul_en)
+from app.inference_services.model import predicted_language
+
+
+def create_payload_en_mul(text, target_language):
+    payload = {
+        "inputs": ">>" + str(target_language) + "<<" + str(text)
+    }
+    return payload
+
+
+def create_payload_mul_en(text):
+    payload = {
+        "inputs": str(text)
+    }
+    return payload
+
+
+def translate_text(text, source_language=None,  target_language=None):
+    response_translate = []
+    while source_language is None:
+        source_language = predicted_language(text)
+
+    if source_language != 'eng' and target_language != 'eng':
+        payload = create_payload_mul_en(text)
+        response_eng = inference_request_mul_en(payload)
+        response_eng = response_eng[20:-3]
+        payload = create_payload_en_mul(response_eng,
+                                        target_language)
+        response_translate = inference_request_en_mul(payload)
+
+    elif source_language == 'eng':
+        payload = create_payload_en_mul(text,
+                                        target_language)
+        response_translate = inference_request_en_mul(payload)
+
+    elif target_language == 'eng':
+        payload = create_payload_mul_en(text)
+        response_translate = inference_request_mul_en(payload)
+    response = response_translate[20:-3]
+
+    return response
+
+
+def create_chunks(text: str, chunk_size: int):
+    chunks = []
+    last_char_index = len(text)
+    chunk_start = 0
+    chunk_stop = 0
+
+    while chunk_stop != last_char_index:
+
+        chunk_stop += chunk_size
+
+        if chunk_stop > last_char_index:
+            chunk_stop = last_char_index
+
+        if chunk_stop != last_char_index:
+            while text[chunk_stop] != " ":
+                chunk_stop -= 1
+
+        chunks.append(text[chunk_start:chunk_stop])
+
+        chunk_start = chunk_stop+1
+
+    return chunks
+
+
+def long_text_translation(src_text: str, src_lang: str, trans_lang: str):
+    src_text_chunks = create_chunks(src_text, chunk_size=200)
+    trans_text_chunks = []
+    for chunk in src_text_chunks:
+        trans_text_chunks.append(
+                         translate_text(text=chunk,
+                                        target_language=trans_lang,
+                                        source_language=src_lang))
+    return " ".join(trans_text_chunks)

app/schemas/tasks.py

@@ -0,0 +1,22 @@
+from pydantic import BaseModel, Field
+# from enum import Enum
+
+# TODO: Include Enum
+# class Language(str, Enum):
+#     Acholi = "ach"
+#     Ateso = "teo"
+#     English = "eng"
+#     Luganda = "lug"
+#     Lugbara = "Lgg"
+#     Runyankole = "nyn"
+
+
+class TranslationResponse(BaseModel):
+    text: str
+    source_language: str = None
+
+
+class TranslationRequest(BaseModel):
+    source_language: str = None
+    target_language: str
+    text: str = Field(max_length=5000, min_length=3)

app/tests/test_api.py

@@ -1,10 +1,39 @@
 from fastapi.testclient import TestClient
 from app.api import app
-
+from app.inference_services.translate import translate_text, predicted_language
 client = TestClient(app)
 
 
 def test_root_api():
     response = client.get('/')
     assert response.status_code == 200
     assert response.json() == {"Hello": "World"}
+
+
+def test_mul_mul(mocker):
+    fake_response_eng = '[{"generated_text":"Where are we heading?"}]'
+    mocker.patch('requests.post').return_value.text = fake_response_eng
+
+    fake_response_mul = '[{"generated_text":"Nituza nkahi?"}]'
+    mocker.patch('requests.post').return_value.text = fake_response_mul
+
+    assert translate_text('Tuli wa', 'lug', 'nyn') == 'Nituza nkahi?'
+
+
+def test_mul_eng(mocker):
+    fake_response_eng = '[{"generated_text":"Where are we heading?"}]'
+    mocker.patch('requests.post').return_value.text = fake_response_eng
+    assert translate_text('Tuli wa', 'lug',
+                          'eng') == "Where are we heading?"
+
+
+def test_eng_mul(mocker):
+    fake_response_mul = '[{"generated_text":"Nituza nkahi?"}]'
+    mocker.patch('requests.post').return_value.text = fake_response_mul
+
+    assert translate_text('Where are we heading?',
+                          'eng', 'nyn') == "Nituza nkahi?"
+
+
+def test_language_detect():
+    assert predicted_language('Ninyisha omuka') == "nyn"

requirements.txt

@@ -1,9 +1,15 @@
 pre-commit==3.3.3
+pydantic<2.0.0
 fastapi
 uvicorn
 pytest
 httpx
 flake8
-
-# Task 2 - add github actions and badges
-coveralls
+tenacity==8.2.3
+python-dotenv==1.0.0
+joblib==1.3.2
+# Task 2 - add gitHub actions and badges
+coveralls
+scikit-learn==1.3.0
+requests
+pytest-mock~=3.10.0