Skip to content

2
2023-029
Commit de663530 authored by Sumudu-Himasha-Ranaweera's avatar Sumudu-Himasha-Ranaweera
Browse files
Options

fix: update 

Desc : update project
parent 269137bc
Hide whitespace changes
Inline Side-by-side
Showing with 146 additions and 90 deletions
Project/Backend/Server_Python/controllers/video_to_sign_language_controller.py 0 → 100644
View file @ de663530
from datetime import datetime
import moviepy.editor as mp
import requests
import speech_recognition as sr
from core import setup_logger
from fastapi import APIRouter, File, HTTPException, UploadFile
from fastapi.responses import JSONResponse
from pymongo.mongo_client import MongoClient
# Replace with your MongoDB Atlas credentials
username = "admin"
password = "JppbU6MZeHfOj7sp"
uri = f"mongodb+srv://{username}:{password}@researchmanagement-appl.vzhn4.mongodb.net/?retryWrites=true&w=majority"
client = MongoClient(uri)
db = client["test"]
items_collection = db["translated_items"]
items_collection_log = db["translated_items_log"]
TEMP_VIDEO_PATH = "C:/Users/himashara/Documents/SLIIT/1 SEMESTER/Research Project - IT4010/Research Project/2023-029/Project/Backend/Server_Python/resources/temp_video.mp4"
AUDIO_SAVE_PATH = "C:/Users/himashara/Documents/SLIIT/1 SEMESTER/Research Project - IT4010/Research Project/2023-029/Project/Backend/Server_Python/resources/audio.wav"
router = APIRouter()
logger = setup_logger()
@router.post("/rest_pyton/uploaded_video")
async def uploaded_video(file: UploadFile = File(...)):
try:
# Process the uploaded video
video_content = await file.read()
temp_video_path = TEMP_VIDEO_PATH
with open(temp_video_path, "wb") as temp_video_file:
temp_video_file.write(video_content)
audio_save_path = AUDIO_SAVE_PATH
video = mp.VideoFileClip(temp_video_path)
audio = video.audio
audio.write_audiofile(audio_save_path)
r = sr.Recognizer()
with sr.AudioFile(audio_save_path) as source:
audio = r.record(source)
recognized_text = r.recognize_google(audio, language="si-LK")
translated_text_si = translate_text(recognized_text, "si")
translated_text_en = translate_text(recognized_text, "en")
translated_integer_si = " ".join(
str(unicode_to_int_mapping.get(word, 0))
for word in translated_text_si.split()
)
print("Translated Integer (Si):", translated_integer_si)
# Send translated integer to MongoDB
send_to_mongodb(translated_integer_si)
return JSONResponse(
content={
"translated_text_si": translated_text_si,
"translated_text_en": translated_text_en,
}
)
# return JSONResponse(content={"translated_text_si": "test"})
except Exception as e:
return JSONResponse(content={"error": str(e)}, status_code=500)
unicode_to_int_mapping = {"මම": 1, "හෙට": 2, "යනවා": 3, "මං": 4}
def translate_text(text, target_language):
url = "https://translate.googleapis.com/translate_a/single"
params = {
"client": "gtx",
"sl": "auto",
"tl": target_language,
"dt": "t",
"q": text,
}
response = requests.get(url, params=params)
translation_data = response.json()
translated_text = translation_data[0][0][0]
return translated_text
# v1
# def send_to_mongodb(translated_integer_si):
# translated_item_data = {
# "translated_integer_si": translated_integer_si,
# "timestamp": datetime.utcnow(),
# }
# result = items_collection.insert_one(translated_item_data)
# if not result.inserted_id:
# raise HTTPException(status_code=500, detail="Failed to create translated item")
# v2
# def send_to_mongodb(translated_integer_si):
# translated_item_data = {
# "translated_integer_si": translated_integer_si,
# "timestamp": datetime.utcnow(),
# }
# # Save the previous record to the log before updating
# existing_record = items_collection.find_one()
# if existing_record:
# items_collection_log = db["translated_items_log"]
# items_collection_log.insert_one(existing_record)
# # Update the existing record or create a new one
# result = items_collection.replace_one({}, translated_item_data, upsert=True)
# if result.matched_count == 0 and result.modified_count == 0:
# raise HTTPException(
# status_code=500, detail="Failed to update or create translated item"
# )
# v3
def send_to_mongodb(translated_integer_si):
translated_item_data = {
"translated_integer_si": translated_integer_si,
"timestamp": datetime.utcnow(),
}
# Save the previous record to the log before updating
existing_record = items_collection.find_one()
if existing_record:
items_collection_log = db["translated_items_log"]
# Exclude the _id field to allow MongoDB to generate a new one
existing_record.pop("_id", None)
items_collection_log.insert_one(existing_record)
# Update the existing record or create a new one
result = items_collection.replace_one({}, translated_item_data, upsert=True)
if result.matched_count == 0 and result.modified_count == 0:
raise HTTPException(status_code=500, detail="Failed to update or create translated item")
Project/Backend/Server_Python/main.py
View file @ de663530
import io from controllers import (translate_controler, users_controller,
import os video_to_sign_language_controller)
import subprocess
import tkinter as tk
import urllib.parse
from datetime import datetime
from tkinter import filedialog, messagebox, scrolledtext
from typing import List
import moviepy.editor as mp
import requests
import speech_recognition as sr
from bson import ObjectId
from controllers import translate_controler, users_controller
from core import setup_logger from core import setup_logger
from fastapi import FastAPI, File, HTTPException, UploadFile from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, RedirectResponse
from pydantic import BaseModel
from pymongo.mongo_client import MongoClient from pymongo.mongo_client import MongoClient
app = FastAPI() app = FastAPI()
# Replace with your MongoDB Atlas credentials # Replace with your MongoDB Atlas credentials
username = "admin" username = "admin"
...@@ -49,7 +35,7 @@ logger = setup_logger() ...@@ -49,7 +35,7 @@ logger = setup_logger()
app.include_router(users_controller.router) app.include_router(users_controller.router)
app.include_router(translate_controler.router) app.include_router(translate_controler.router)
app.include_router(video_to_sign_language_controller.router)
# Add cores middleware # Add cores middleware
origins = [ origins = [
...@@ -68,74 +54,4 @@ app.add_middleware( ...@@ -68,74 +54,4 @@ app.add_middleware(
@app.get("/") @app.get("/")
async def read_root(): async def read_root():
return {"message": "FastAPI with MongoDB integration"} return {"message": "FastAPI with MongoDB integration"}
\ No newline at end of file
@app.post("/rest_pyton/uploaded_video")
async def uploaded_video(file: UploadFile = File(...)):
try:
# Process the uploaded video
video_content = await file.read()
temp_video_path = "C:/Users/himashara/Documents/SLIIT/1 SEMESTER/Research Project - IT4010/Research Project/2023-029/Project/Backend/Server_Python/resources/temp_video.mp4"
with open(temp_video_path, "wb") as temp_video_file:
temp_video_file.write(video_content)
audio_save_path = "C:/Users/himashara/Documents/SLIIT/1 SEMESTER/Research Project - IT4010/Research Project/2023-029/Project/Backend/Server_Python/resources/audio.wav"
video = mp.VideoFileClip(temp_video_path)
audio = video.audio
audio.write_audiofile(audio_save_path)
r = sr.Recognizer()
with sr.AudioFile(audio_save_path) as source:
audio = r.record(source)
recognized_text = r.recognize_google(audio, language="si-LK")
translated_text_si = translate_text(recognized_text, "si")
translated_text_en = translate_text(recognized_text, "en")
translated_integer_si = " ".join(
str(unicode_to_int_mapping.get(word, 0))
for word in translated_text_si.split()
)
print("Translated Integer (Si):", translated_integer_si)
# Send translated integer to MongoDB
send_to_mongodb(translated_integer_si)
return JSONResponse(
content={
"translated_text_si": translated_text_si,
"translated_text_en": translated_text_en,
}
)
# return JSONResponse(content={"translated_text_si": "test"})
except Exception as e:
return JSONResponse(content={"error": str(e)}, status_code=500)
class TranslatedItemCreate(BaseModel):
translated_integer_si: str
unicode_to_int_mapping = {"මම": 1, "හෙට": 2, "යනවා": 3, "මං": 4}
def translate_text(text, target_language):
url = "https://translate.googleapis.com/translate_a/single"
params = {
"client": "gtx",
"sl": "auto",
"tl": target_language,
"dt": "t",
"q": text,
}
response = requests.get(url, params=params)
translation_data = response.json()
translated_text = translation_data[0][0][0]
return translated_text
def send_to_mongodb(translated_integer_si):
translated_item_data = {
"translated_integer_si": translated_integer_si,
"timestamp": datetime.utcnow(),
}
result = items_collection.insert_one(translated_item_data)
if not result.inserted_id:
raise HTTPException(status_code=500, detail="Failed to create translated item")
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment