Update app.py

app.py

@@ -1,333 +1,333 @@
-from flask import Flask, render_template, request, jsonify, send_from_directory , send_file
-from flask_cors import CORS
-import time
-import uuid
-import threading
-from collections import deque
-import subprocess
-import os
-
-app = Flask(__name__)
-CORS(app)
-
-# Set the paths for espeak-ng and the script directory
-script_path = os.path.dirname(os.path.abspath(__file__))
-espeak_path = "C:\\Program Files\\eSpeak NG\\espeak-ng.exe"  # Change this to the full path if needed
-static_folder = os.path.join(script_path, "static")
-
-# Ensure the static folder exists
-os.makedirs(static_folder, exist_ok=True)
-
-class Component:
-    def __init__(self, component_type, label, options=None, value=None):
-        self.component_type = component_type
-        self.label = label
-        self.options = options
-        self.value = value
-
-    def render(self):
-        if self.component_type == "textbox":
-            return f'<textarea name="{self.label}" class="form-control" placeholder="{self.label}">{self.value}</textarea>'
-        elif self.component_type == "radio":
-            options_html = ''.join([f'''
-            <div class="form-check form-check-inline">
-            <input class="form-check-input" type="radio" name="{self.label}" id="{self.label}-{index}" value="{self.value}">
-            <label class="form-check-label" for="{self.label}-{index}">{option}</label></div>''' for index,option in enumerate(self.options)])
-            return options_html
-        elif self.component_type == "audio":
-            return f'<audio controls id="{self.label}"><source src="" type="audio/wav">Your browser does not support the audio element.</audio>'
-        elif self.component_type == "output":
-            return f'<div id="{self.label}">Placeholder for {self.label}</div>'
-        elif self.component_type == "better_textbox":
-            return f'<div contenteditable="true" id="{self.label}" class="better-textbox" placeholder="{self.value}">{self.value}</div>'
-        else:
-            return ""
-
-class Row:
-    def __init__(self, components):
-        self.components = components
-
-    def render(self):
-        components_html = ''.join([component.render() for component in self.components])
-        return f'<div class="row mb-3">{components_html}</div>'
-
-class Column:
-    def __init__(self, components):
-        self.components = components
-
-    def render(self):
-        components_html = ''.join([component.render() for component in self.components])
-        return f'<div class="col">{components_html}</div>'
-
-
-class GradioInterface:
-    def __init__(self, fn, input_components, output_components, max_parallel_tasks=5):
-        self.fn = fn
-        self.input_components = input_components
-        self.output_components = output_components
-        self.max_parallel_tasks = max_parallel_tasks
-        self.task_queue = deque()
-        self.results = {}
-        self.current_tasks = {}
-        self.task_threads = {}
-        self.stop_flags = {}
-
-    def submit(self, input_data):
-        task_id = str(uuid.uuid4())
-        if len(self.current_tasks) < self.max_parallel_tasks:
-            self.task_queue.append(task_id)
-            self.current_tasks[task_id] = input_data
-            
-            stop_flag = threading.Event()
-            self.stop_flags[task_id] = stop_flag
-            
-            thread = threading.Thread(target=self.process_task, args=(task_id, input_data, stop_flag))
-            self.task_threads[task_id] = thread
-            thread.start()
-            return task_id
-        else:
-            return None
-
-    def process_task(self, task_id, input_data, stop_flag):
-        result = self.fn(input_data)
-        self.results[task_id] = result
-        self.task_queue.remove(task_id)
-        del self.current_tasks[task_id]
-        del self.task_threads[task_id]
-        del self.stop_flags[task_id]
-
-    def get_result(self, task_id):
-        if task_id in self.results:
-            return {"status": "completed", "result": self.results[task_id]}
-        return {"status": "pending"}
-
-    def render_components(self):
-        input_html = ''.join([component.render() for component in self.input_components])
-        output_html = ''.join([component.render() for component in self.output_components])
-        return input_html #f"{input_html}{output_html}"
-
-    def get_queue_position(self):
-        return len(self.task_queue)
-
-    def stop_task(self, task_id):
-        if task_id in self.stop_flags:
-            self.stop_flags[task_id].set()
-            return True
-        return False
-
-
-
-import os
-import shutil
-import subprocess
-
-def find_espeak_ng_data_folder():
-    try:
-        # https://github.com/espeak-ng/espeak-ng/blob/b006f6d4f997fbfe4016cf29767743e6b397d0fb/src/espeak-ng.c#L309
-        data_folder = subprocess.check_output([espeak_path, '--version']).decode().split('Data at: ')[1].strip()
-        return data_folder
-    except Exception as e:
-        print(f"Error: {str(e)}")
-        return None
-
-def copy_lang_dict(data_folder, destination,lang):
-    dict_file=lang+'_dict'
-    en_dict_path = os.path.join(data_folder,dict_file )
-    try:
-        # Copy the en_dict folder to the destination
-        shutil.copyfile(en_dict_path, os.path.join(destination, dict_file))
-        print(f"Successfully copied {dict_file} to {destination}")
-    except Exception as e:
-        print(f"Error copying {dict_file}: {str(e)}")
-
-def compile_espeak():
-    compile_command = [espeak_path, "--compile=fa"]
-    cwd_path = r"K:\code\dev\python\other\tts\espeak dictionary\dictsource"
-    subprocess.run(compile_command, cwd=cwd_path)
-    print("Compilation done!")  # Giving feedback in the console for now
-
-
-
-import gzip
-phonetc_data_file_path = os.path.join(static_folder, 'fa_extra.txt.gz')
-generations_file_path = os.path.join(static_folder, 'generations.txt.gz')
-
-import time
-import datetime
-
-def get_iran_time_no_packages():
-    iran_offset_seconds = 3.5 * 3600  # 3.5 hours * seconds/hour
-    utc_time = time.gmtime()
-    iran_time_tuple = time.localtime(time.mktime(utc_time) + iran_offset_seconds)
-    iran_datetime = datetime.datetime(*iran_time_tuple[:6])  # Create datetime object
-    formatted_time = iran_datetime.strftime("%Y-%m-%d %H:%M:%S")
-    return formatted_time
-
-with gzip.open(phonetc_data_file_path, 'at',encoding='utf-8') as f:  # 'at' for text mode
-    f.write(f'// started at : {get_iran_time_no_packages()}\n')
-
-
-def tts(input_data):
-    text = input_data.get('text')
-    voice = input_data.get('voice')
-    input_file_path = os.path.join(script_path, "input_text.txt")  # Path for the input text file
-    audio_file = os.path.join(static_folder, f'output_{uuid.uuid4()}.wav')  # Path for the output audio file
-    
-    with gzip.open(generations_file_path, 'at',encoding='utf-8') as f:
-        f.write(f'{voice}\t{text.strip()}\n')
-
-    # Write the input text to a file
-    with open(input_file_path, 'w', encoding='utf-8') as file:
-        file.write(text)
-
-    # Phonemize the text
-    phonemes_command = [espeak_path, "-v", voice, "-x", "-q", "-f", input_file_path]
-    phonemes_result = subprocess.run(phonemes_command, capture_output=True, text=True)
-    phonemized_text = phonemes_result.stdout.strip()
-
-    # Generate audio file
-    audio_cmd = f'"{espeak_path}" -v {voice} -w "{audio_file}" -f "{input_file_path}"'
-    subprocess.run(audio_cmd, shell=True)
-    
-    status = f'مدل : espeak-ng\nآوانگاشت: {phonemized_text}'
-
-    return {"audio": f"/static/{os.path.basename(audio_file)}",  # Return the static path
-            "status": status
-            }
-
-
-def phonemize(input_data):
-    text = input_data.get('word')
-    voice = 'fa'
-    task = input_data.get('task')
-    phonetic = input_data.get('phonetic',"").strip()
-    input_file_path = os.path.join(script_path, "input_text.txt")  # Path for the input text file
-    audio_file = os.path.join(static_folder, f'output_{uuid.uuid4()}.wav')  # Path for the output audio file
-
-    
-    if phonetic=='' or task=='phonemize' :
-        phonetic=text
-    else:
-        phonetic=f'[[{phonetic.strip()}]]'
-
-    # Write the input text to a file
-    with open(input_file_path, 'w', encoding='utf-8') as file:
-        file.write(phonetic)
-
-    # Phonemize the text
-    
-    phonemes_command = [espeak_path, "-v", voice, "-x", "-q", "-f", input_file_path]
-    phonemes_result = subprocess.run(phonemes_command, capture_output=True, text=True)
-    phonemized_text = phonemes_result.stdout.strip()
-
-    # Generate audio file
-    audio_cmd = f'"{espeak_path}" -v {voice} -w "{audio_file}" -f "{input_file_path}"'
-    subprocess.run(audio_cmd, shell=True)
-    
-    status = f'متن      : {text}\nآوانگاشت: {phonemized_text}'
-    if task=='send' and input_data.get('phonetic',False):
-        with gzip.open(phonetc_data_file_path, 'at',encoding='utf-8') as f:  # 'at' for text mode
-            f.write(f'{text.strip()}\t{input_data.get("phonetic",False).strip()}\n')
-        
-
-    return {"audio": f"/static/{os.path.basename(audio_file)}",
-            "text": text,
-            "phonemes":phonemized_text,
-            "status": status
-            }
-
-
-def listWords(input_data):
-    data = []    
-    with gzip.open(phonetc_data_file_path, 'rt',encoding='utf-8') as f:  # 'rt' for text mode
-        for line in f:
-            if not line.startswith('//'):  # Skip commented lines
-                parts = line.strip().split('\t')
-                if len(parts) == 2:
-                    data.append({'word': parts[0].strip(), 'phonetic': parts[1].strip()})
-
-    return data
-    
-def example_function(input_data):
-    return input_data
-
-input_components = [
-    Component("radio", "voice", options=["fa", "en", "ar"], value="fa")
-]
-
-output_components = [
-    Component("output", "phonemized_output"),
-    Component("audio", "audio_output")
-]
-
-ifaces = {
-'tts' : GradioInterface(
-        fn=tts,
-        input_components=input_components,
-        output_components=output_components,
-        max_parallel_tasks=5
-        )
-    ,
-'phonemize' : GradioInterface(
-        fn=phonemize,
-        input_components=input_components,
-        output_components=output_components,
-        max_parallel_tasks=5
-        )
-    ,
-'words' : GradioInterface(
-        fn=listWords,
-        input_components=input_components,
-        output_components=output_components,
-        max_parallel_tasks=5
-        )
-    ,
-
-}
-
-
-@app.route("/", methods=["GET"])
-def index():
-    return render_template("tab-2-6-template.html", components=ifaces['tts'].render_components())
-
-@app.route("/submit", methods=["POST"])
-def submit():
-    tab = request.form.get("tab")
-    task_id = ifaces[tab].submit(request.form)
-    
-    if task_id:
-        position = ifaces[tab].get_queue_position()
-        return jsonify({"task_id": task_id, "position": position})
-    else:
-        return jsonify({"error": "Task queue is full"}), 429
-
-@app.route("/result/<task_id>/<tab>", methods=["GET"])
-def result(task_id,tab):
-    result_data = ifaces[tab].get_result(task_id)
-    if result_data["status"] == "completed":
-        return jsonify(result_data)
-    return jsonify({"status": result_data["status"]})
-
-@app.route("/queue", methods=["GET"])
-def queue():
-    return jsonify({"queue": list(iface.task_queue)})
-
-@app.route("/stop/<task_id>", methods=["POST"])
-def stop(task_id):
-    stopped = iface.stop_task(task_id)
-    if stopped:
-        return jsonify({"status": "stopped"})
-    return jsonify({"status": "not found"}), 404
-
-@app.route('/static/<path:filename>')
-def send_static(filename):
-    return send_from_directory(static_folder, filename)
-@app.route('/download/fa_extra')
-def download():
-    return send_file(phonetc_data_file_path, as_attachment=True)
-@app.route('/download/fa_dict')
-def download1():
-    dict_file='fa_dict'
-    return send_file(os.path.join(find_espeak_ng_data_folder(),dict_file ), as_attachment=True)
-if __name__ == "__main__":
-    app.run(debug=True)
+from flask import Flask, render_template, request, jsonify, send_from_directory , send_file
+from flask_cors import CORS
+import time
+import uuid
+import threading
+from collections import deque
+import subprocess
+import os
+
+app = Flask(__name__)
+CORS(app)
+
+# Set the paths for espeak-ng and the script directory
+script_path = os.path.dirname(os.path.abspath(__file__))
+espeak_path = "C:\\Program Files\\eSpeak NG\\espeak-ng.exe"  # Change this to the full path if needed
+static_folder = os.path.join(script_path, "static")
+
+# Ensure the static folder exists
+os.makedirs(static_folder, exist_ok=True)
+
+class Component:
+    def __init__(self, component_type, label, options=None, value=None):
+        self.component_type = component_type
+        self.label = label
+        self.options = options
+        self.value = value
+
+    def render(self):
+        if self.component_type == "textbox":
+            return f'<textarea name="{self.label}" class="form-control" placeholder="{self.label}">{self.value}</textarea>'
+        elif self.component_type == "radio":
+            options_html = ''.join([f'''
+            <div class="form-check form-check-inline">
+            <input class="form-check-input" type="radio" name="{self.label}" id="{self.label}-{index}" value="{self.value}">
+            <label class="form-check-label" for="{self.label}-{index}">{option}</label></div>''' for index,option in enumerate(self.options)])
+            return options_html
+        elif self.component_type == "audio":
+            return f'<audio controls id="{self.label}"><source src="" type="audio/wav">Your browser does not support the audio element.</audio>'
+        elif self.component_type == "output":
+            return f'<div id="{self.label}">Placeholder for {self.label}</div>'
+        elif self.component_type == "better_textbox":
+            return f'<div contenteditable="true" id="{self.label}" class="better-textbox" placeholder="{self.value}">{self.value}</div>'
+        else:
+            return ""
+
+class Row:
+    def __init__(self, components):
+        self.components = components
+
+    def render(self):
+        components_html = ''.join([component.render() for component in self.components])
+        return f'<div class="row mb-3">{components_html}</div>'
+
+class Column:
+    def __init__(self, components):
+        self.components = components
+
+    def render(self):
+        components_html = ''.join([component.render() for component in self.components])
+        return f'<div class="col">{components_html}</div>'
+
+
+class GradioInterface:
+    def __init__(self, fn, input_components, output_components, max_parallel_tasks=5):
+        self.fn = fn
+        self.input_components = input_components
+        self.output_components = output_components
+        self.max_parallel_tasks = max_parallel_tasks
+        self.task_queue = deque()
+        self.results = {}
+        self.current_tasks = {}
+        self.task_threads = {}
+        self.stop_flags = {}
+
+    def submit(self, input_data):
+        task_id = str(uuid.uuid4())
+        if len(self.current_tasks) < self.max_parallel_tasks:
+            self.task_queue.append(task_id)
+            self.current_tasks[task_id] = input_data
+            
+            stop_flag = threading.Event()
+            self.stop_flags[task_id] = stop_flag
+            
+            thread = threading.Thread(target=self.process_task, args=(task_id, input_data, stop_flag))
+            self.task_threads[task_id] = thread
+            thread.start()
+            return task_id
+        else:
+            return None
+
+    def process_task(self, task_id, input_data, stop_flag):
+        result = self.fn(input_data)
+        self.results[task_id] = result
+        self.task_queue.remove(task_id)
+        del self.current_tasks[task_id]
+        del self.task_threads[task_id]
+        del self.stop_flags[task_id]
+
+    def get_result(self, task_id):
+        if task_id in self.results:
+            return {"status": "completed", "result": self.results[task_id]}
+        return {"status": "pending"}
+
+    def render_components(self):
+        input_html = ''.join([component.render() for component in self.input_components])
+        output_html = ''.join([component.render() for component in self.output_components])
+        return input_html #f"{input_html}{output_html}"
+
+    def get_queue_position(self):
+        return len(self.task_queue)
+
+    def stop_task(self, task_id):
+        if task_id in self.stop_flags:
+            self.stop_flags[task_id].set()
+            return True
+        return False
+
+
+
+import os
+import shutil
+import subprocess
+
+def find_espeak_ng_data_folder():
+    try:
+        # https://github.com/espeak-ng/espeak-ng/blob/b006f6d4f997fbfe4016cf29767743e6b397d0fb/src/espeak-ng.c#L309
+        data_folder = subprocess.check_output([espeak_path, '--version']).decode().split('Data at: ')[1].strip()
+        return data_folder
+    except Exception as e:
+        print(f"Error: {str(e)}")
+        return None
+
+def copy_lang_dict(data_folder, destination,lang):
+    dict_file=lang+'_dict'
+    en_dict_path = os.path.join(data_folder,dict_file )
+    try:
+        # Copy the en_dict folder to the destination
+        shutil.copyfile(en_dict_path, os.path.join(destination, dict_file))
+        print(f"Successfully copied {dict_file} to {destination}")
+    except Exception as e:
+        print(f"Error copying {dict_file}: {str(e)}")
+
+def compile_espeak():
+    compile_command = [espeak_path, "--compile=fa"]
+    cwd_path = r"K:\code\dev\python\other\tts\espeak dictionary\dictsource"
+    subprocess.run(compile_command, cwd=cwd_path)
+    print("Compilation done!")  # Giving feedback in the console for now
+
+
+
+import gzip
+phonetc_data_file_path = os.path.join(static_folder, 'fa_extra.txt.gz')
+generations_file_path = os.path.join(static_folder, 'generations.txt.gz')
+
+import time
+import datetime
+
+def get_iran_time_no_packages():
+    iran_offset_seconds = 3.5 * 3600  # 3.5 hours * seconds/hour
+    utc_time = time.gmtime()
+    iran_time_tuple = time.localtime(time.mktime(utc_time) + iran_offset_seconds)
+    iran_datetime = datetime.datetime(*iran_time_tuple[:6])  # Create datetime object
+    formatted_time = iran_datetime.strftime("%Y-%m-%d %H:%M:%S")
+    return formatted_time
+
+with gzip.open(phonetc_data_file_path, 'at',encoding='utf-8') as f:  # 'at' for text mode
+    f.write(f'// started at : {get_iran_time_no_packages()}\n')
+
+
+def tts(input_data):
+    text = input_data.get('text')
+    voice = input_data.get('voice')
+    input_file_path = os.path.join(script_path, "input_text.txt")  # Path for the input text file
+    audio_file = os.path.join(static_folder, f'output_{uuid.uuid4()}.wav')  # Path for the output audio file
+    
+    with gzip.open(generations_file_path, 'at',encoding='utf-8') as f:
+        f.write(f'{voice}\t{text.strip()}\n')
+
+    # Write the input text to a file
+    with open(input_file_path, 'w', encoding='utf-8') as file:
+        file.write(text)
+
+    # Phonemize the text
+    phonemes_command = [espeak_path, "-v", voice, "-x", "-q", "-f", input_file_path]
+    phonemes_result = subprocess.run(phonemes_command, capture_output=True, text=True)
+    phonemized_text = phonemes_result.stdout.strip()
+
+    # Generate audio file
+    audio_cmd = f'"{espeak_path}" -v {voice} -w "{audio_file}" -f "{input_file_path}"'
+    subprocess.run(audio_cmd, shell=True)
+    
+    status = f'مدل : espeak-ng\nآوانگاشت: {phonemized_text}'
+
+    return {"audio": f"/static/{os.path.basename(audio_file)}",  # Return the static path
+            "status": status
+            }
+
+
+def phonemize(input_data):
+    text = input_data.get('word')
+    voice = 'fa'
+    task = input_data.get('task')
+    phonetic = input_data.get('phonetic',"").strip()
+    input_file_path = os.path.join(script_path, "input_text.txt")  # Path for the input text file
+    audio_file = os.path.join(static_folder, f'output_{uuid.uuid4()}.wav')  # Path for the output audio file
+
+    
+    if phonetic=='' or task=='phonemize' :
+        phonetic=text
+    else:
+        phonetic=f'[[{phonetic.strip()}]]'
+
+    # Write the input text to a file
+    with open(input_file_path, 'w', encoding='utf-8') as file:
+        file.write(phonetic)
+
+    # Phonemize the text
+    
+    phonemes_command = [espeak_path, "-v", voice, "-x", "-q", "-f", input_file_path]
+    phonemes_result = subprocess.run(phonemes_command, capture_output=True, text=True)
+    phonemized_text = phonemes_result.stdout.strip()
+
+    # Generate audio file
+    audio_cmd = f'"{espeak_path}" -v {voice} -w "{audio_file}" -f "{input_file_path}"'
+    subprocess.run(audio_cmd, shell=True)
+    
+    status = f'متن      : {text}\nآوانگاشت: {phonemized_text}'
+    if task=='send' and input_data.get('phonetic',False):
+        with gzip.open(phonetc_data_file_path, 'at',encoding='utf-8') as f:  # 'at' for text mode
+            f.write(f'{text.strip()}\t{input_data.get("phonetic",False).strip()}\n')
+        
+
+    return {"audio": f"/static/{os.path.basename(audio_file)}",
+            "text": text,
+            "phonemes":phonemized_text,
+            "status": status
+            }
+
+
+def listWords(input_data):
+    data = []    
+    with gzip.open(phonetc_data_file_path, 'rt',encoding='utf-8') as f:  # 'rt' for text mode
+        for line in f:
+            if not line.startswith('//'):  # Skip commented lines
+                parts = line.strip().split('\t')
+                if len(parts) == 2:
+                    data.append({'word': parts[0].strip(), 'phonetic': parts[1].strip()})
+
+    return data
+    
+def example_function(input_data):
+    return input_data
+
+input_components = [
+    Component("radio", "voice", options=["fa", "en", "ar"], value="fa")
+]
+
+output_components = [
+    Component("output", "phonemized_output"),
+    Component("audio", "audio_output")
+]
+
+ifaces = {
+'tts' : GradioInterface(
+        fn=tts,
+        input_components=input_components,
+        output_components=output_components,
+        max_parallel_tasks=5
+        )
+    ,
+'phonemize' : GradioInterface(
+        fn=phonemize,
+        input_components=input_components,
+        output_components=output_components,
+        max_parallel_tasks=5
+        )
+    ,
+'words' : GradioInterface(
+        fn=listWords,
+        input_components=input_components,
+        output_components=output_components,
+        max_parallel_tasks=5
+        )
+    ,
+
+}
+
+
+@app.route("/", methods=["GET"])
+def index():
+    return render_template("tab-2-6-template.html", components=ifaces['tts'].render_components())
+
+@app.route("/submit", methods=["POST"])
+def submit():
+    tab = request.form.get("tab")
+    task_id = ifaces[tab].submit(request.form)
+    
+    if task_id:
+        position = ifaces[tab].get_queue_position()
+        return jsonify({"task_id": task_id, "position": position})
+    else:
+        return jsonify({"error": "Task queue is full"}), 429
+
+@app.route("/result/<task_id>/<tab>", methods=["GET"])
+def result(task_id,tab):
+    result_data = ifaces[tab].get_result(task_id)
+    if result_data["status"] == "completed":
+        return jsonify(result_data)
+    return jsonify({"status": result_data["status"]})
+
+@app.route("/queue", methods=["GET"])
+def queue():
+    return jsonify({"queue": list(iface.task_queue)})
+
+@app.route("/stop/<task_id>", methods=["POST"])
+def stop(task_id):
+    stopped = iface.stop_task(task_id)
+    if stopped:
+        return jsonify({"status": "stopped"})
+    return jsonify({"status": "not found"}), 404
+
+@app.route('/static/<path:filename>')
+def send_static(filename):
+    return send_from_directory(static_folder, filename)
+@app.route('/download/fa_extra')
+def download():
+    return send_file(phonetc_data_file_path, as_attachment=True)
+@app.route('/download/fa_dict')
+def download1():
+    dict_file='fa_dict'
+    return send_file(os.path.join(find_espeak_ng_data_folder(),dict_file ), as_attachment=True)
+if __name__ == "__main__":
+    app.run(host='0.0.0.0',  port= 7860))