main.py

from fastapi import FastAPI, File, UploadFile, Path
import uvicorn
from fastapi.responses import FileResponse

import os
from typing import List


from wav_to_label import wav_to_label
from kantts.bin.train_sambert import train as train_sambert
from text_to_wav_trans import text_to_wav as text_to_wav_trans
from text_to_wav_onnx import text_to_wav_onnx


wav_dir = "./Data/ptts_spk0_wav"
txt_dir = "./Data"
output_dir = "./res/ptts_syn"


# 创建FastAPI实例
app = FastAPI()

@app.get("/")
def get_root():
    """
    注册一个根路径
    """
    return {"message": "Welcome to try: Personal Text To Speech !"}


@app.post("/uploadwavs")
async def uploadwavs(files: List[UploadFile] = File(...)):
    # 将文件保存到指定目录；文件路径=目录+文件名
    if not os.path.exists(wav_dir):
        os.makedirs(wav_dir, exist_ok=True)
    for file in files:
        with open(os.path.join(wav_dir, file.filename), "wb") as f:  
            f.write(await file.read())
    return {"msg": "File upload success in directory 'Data/ptts_spk0_wav/'"}

@app.post("/uploadtxt")
async def uploadtxt(file: UploadFile = File(...)):
    # 将文件保存到指定目录；文件路径=目录+文件名
    if not os.path.exists(txt_dir):
        os.makedirs(txt_dir, exist_ok=True)
    with open(os.path.join(txt_dir, file.filename), "wb") as f:  
        f.write(await file.read())
    return {"msg": "File upload success in directory 'Data/'"}


@app.get("/listfiles/{dirpath:path}")
async def listfiles(dirpath: str):
    # return {"file_path": file_path}
    res = os.listdir(dirpath)
    return {"files": res}


@app.get("/cleardir")
async def cleardir():
    for filename in os.listdir(wav_dir):  
        file_path = os.path.join(wav_dir, filename)  
        try:  
            if os.path.isfile(file_path):  
                os.remove(file_path)  
        except Exception as e:  
            print(f"Error: {e}")  
    return {"msg": "Director cleared and empty"}

@app.get("/deletetxt")
async def deletetxt():
    # 指定要删除的文件路径  
    file_path = "./Data/test.txt"  
    # 检查文件是否存在  
    if os.path.isfile(file_path):  
        # 删除文件  
        os.remove(file_path)
        return {"msg": "File remoeved"}
    else:
        return {"msg": "File not exist"}

@app.get("/downloadfile/{filename}")
async def downloadfile(filename: str):
    file_path = os.path.join("res/ptts_syn/res_wavs", filename)
    if os.path.exists(file_path):
        return FileResponse(file_path)
    else:
        return {
            "msg": "File not exis"
        }


@app.get("/wav2label")
async def wav2label():
    report = wav_to_label(wav_dir)
    return report

@app.get("/featsextract")
async def featsextract():
    # 执行Shell脚本
    f = os.popen('./feats_extract.sh')
    return f.read()

@app.get("/trainsambert")
async def trainsambert():
    train_sambert(
        "speech_personal_sambert-hifigan_nsf_tts_zh-cn_pretrain_16k/basemodel_16k/sambert/config.yaml",
        "training_stage/ptts_feats",
        "training_stage/ptts_sambert_ckpt",
        "speech_personal_sambert-hifigan_nsf_tts_zh-cn_pretrain_16k/basemodel_16k/sambert/ckpt/checkpoint_2400000.pth"
    )
    return {"msg": "Traing finished"}


@app.get("/modeltransform")
async def modeltransform():
    text_to_wav_trans(
        "./Data/test.txt",
        "res/ptts_syn_one",
        "speech_personal_sambert-hifigan_nsf_tts_zh-cn_pretrain_16k/resource.zip",
        "training_stage/ptts_sambert_ckpt/ckpt/checkpoint_2402200.pth",
        "speech_personal_sambert-hifigan_nsf_tts_zh-cn_pretrain_16k//basemodel_16k/hifigan/ckpt/checkpoint_2400000.pth",
        "F7",
        "training_stage/ptts_feats/se/se.npy"
    )

    return {"msg": "Model transform finished"}


# from enum import Enum

# class TargetRate(Enum):
#     rate0 = 1.0
#     rate1 = 0.5
#     rate2 = 0.75   
#     rate3 = 1.25
#     rate4 = 1.5
#     rate5= 1.75
#     rate6 = 2.0

@app.get("/text2wav/{targetrate}")
async def text2wav(targetrate: float=1.0):
    text_to_wav_onnx(
        "./Data/test.txt",
        output_dir,
        "speech_personal_sambert-hifigan_nsf_tts_zh-cn_pretrain_16k/resource.zip",
        "sambert_onnx/text_encoder.onnx",
        "sambert_onnx/variance_adaptor_dict.pt",
        "sambert_onnx/mel_decoder_dict.pt",
        "sambert_onnx/mel_postnet.onnx",
        "training_stage/ptts_sambert_ckpt/config.yaml",
        "hifigan_onnx/hifigan.onnx",
        "speech_personal_sambert-hifigan_nsf_tts_zh-cn_pretrain_16k//basemodel_16k/hifigan/config.yaml", 
        targetrate,
        "F7",
        "training_stage/ptts_feats/se/se.npy"
    )

    return {"msg": "Text to wav finished"}



@app.get("/oneclickstart/{targetrate}")
async def oneclickstart(targetrate: float=1.0):
    report = wav_to_label(wav_dir)

    f = os.popen('./feats_extract.sh')

    train_sambert(
        "speech_personal_sambert-hifigan_nsf_tts_zh-cn_pretrain_16k/basemodel_16k/sambert/config.yaml",
        "training_stage/ptts_feats",
        "training_stage/ptts_sambert_ckpt",
        "speech_personal_sambert-hifigan_nsf_tts_zh-cn_pretrain_16k/basemodel_16k/sambert/ckpt/checkpoint_2400000.pth"
    )


    text_to_wav_trans(
        "./Data/test.txt",
        "res/ptts_syn_one",
        "speech_personal_sambert-hifigan_nsf_tts_zh-cn_pretrain_16k/resource.zip",
        "training_stage/ptts_sambert_ckpt/ckpt/checkpoint_2402200.pth",
        "speech_personal_sambert-hifigan_nsf_tts_zh-cn_pretrain_16k//basemodel_16k/hifigan/ckpt/checkpoint_2400000.pth",
        "F7",
        "training_stage/ptts_feats/se/se.npy"
    )

    text_to_wav_onnx(
        "./Data/test.txt",
        output_dir,
        "speech_personal_sambert-hifigan_nsf_tts_zh-cn_pretrain_16k/resource.zip",
        "sambert_onnx/text_encoder.onnx",
        "sambert_onnx/variance_adaptor_dict.pt",
        "sambert_onnx/mel_decoder_dict.pt",
        "sambert_onnx/mel_postnet.onnx",
        "training_stage/ptts_sambert_ckpt/config.yaml",
        "hifigan_onnx/hifigan.onnx",
        "speech_personal_sambert-hifigan_nsf_tts_zh-cn_pretrain_16k//basemodel_16k/hifigan/config.yaml", 
        targetrate,
        "F7",
        "training_stage/ptts_feats/se/se.npy"
    )

    return {"msg": "Text to wav finished"}



if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)