scripts/runSQ.py

import os, unicodedata, string, random
from scripts.ctcalign import aligner, wav16m
from scripts.tapi import tiro, grammatek
from scripts.reaper2pass import estimate_pitch, save_pitch
import scripts.clusterprosody as cl




# given a Sentence string,
# using a metadata file of SQ, // SQL1adult_metadata.tsv
# get every file from SQ of a L1 adult with that sentence
#  report how many, or if 0.


def run(sentence, voices, start_end_word_ix):
    #sentence = 'hvaða sjúkdómar geta fylgt óbeinum reykingum'
    #voices = ['Alfur','Dilja','Karl', 'Dora']
    # On tts.tiro.is speech marks are only available 
    # for the voices: Alfur, Dilja, Karl and Dora.
    # in practise, only for alfur and dilja.

    corpus_meta = '/home/user/app/human_data/SQL1adult10s_metadata.tsv'
    speech_dir = '/home/user/app/human_data/audio/squeries/'
    playable_dir = 'https://huggingface.co/spaces/clr/pce/resolve/main/human_data/audio/squeries/'
    speech_aligns = '/home/user/app/human_data/align/squeries/'
    speech_f0 = '/home/user/app/human_data/f0/squeries/'
    align_model_path ="carlosdanielhernandezmena/wav2vec2-large-xlsr-53-icelandic-ep10-1000h"

    tts_dir = '/home/user/app/tts_data/'

    
    norm_sentence = snorm(sentence)
    sentence = sentence.replace('\t', ' ')

    human_rec_ids = get_samromur_queries(norm_sentence, corpus_meta, speech_dir, speech_aligns, align_model_path, speech_f0)

    if voices:
        temp_tts_sample, tts_sent_dir = get_tts(sentence,voices,tts_dir,align_model_path)

        #voices = [voices[0]] # TODO. now limit one voice at a time.
        #score, f0_fig_c0, f0_fig_c1, f0_fig_c2, en_fig_c0, en_fig_c1, en_fig_c2, html, f0_fig_tts, en_fig_tts = cl.cluster(norm_sentence, sentence, human_rec_ids, speech_aligns, speech_f0, speech_dir, playable_dir, tts_sent_dir, voices, start_end_word_ix)
        f0_fig_c0, f0_fig_c1, f0_fig_c2, en_fig_c0, en_fig_c1, en_fig_c2, html, tts_results = cl.cluster(norm_sentence, sentence, human_rec_ids, speech_aligns, speech_f0, speech_dir, playable_dir, tts_sent_dir, voices, start_end_word_ix)
    # also stop forgetting duration.
        
    #return temp_tts_sample, score, f0_fig_c0, f0_fig_c1, f0_fig_c2, en_fig_c0, en_fig_c1, en_fig_c2, html, f0_fig_tts, en_fig_tts
    return f0_fig_c0, f0_fig_c1, f0_fig_c2, en_fig_c0, en_fig_c1, en_fig_c2, html, tts_results






def snorm(s):
    s = ''.join([c.lower() for c in s if not unicodedata.category(c).startswith("P") ])
    while '  ' in s:
        s = s.replace('  ', ' ')
    return s
   
    

def create_temp_sent_list():
    corpusdb = '/home/user/app/human_data/SQL1adult10s_metadata.tsv'
    with open(corpusdb,'r') as handle:
        meta = handle.read().splitlines()
    meta = [l.split('\t')[3] for l in meta[1:]]
    meta = sorted(list(set(meta)))
    return meta
    
    
    
def align_file(wav_path, output_path, norm_sentence, word_aligner = None, model_path = "carlosdanielhernandezmena/wav2vec2-large-xlsr-53-icelandic-ep10-1000h"):

    model_word_sep = '|'
    model_blank_tk = '[PAD]'
    
    if not word_aligner:
        print('initiating forced alignment, can take some time...')
        word_aligner = aligner(model_path,model_word_sep,model_blank_tk)
  
    word_aln = word_aligner(wav16m(wav_path),norm_sentence,is_normed=True)
    word_aln = [[str(x) for x in l] for l in word_aln]
    
    with open(output_path,'w') as handle:
        handle.write(''.join(['\t'.join(l)+'\n' for l in word_aln]))
        
    return word_aligner
    
    


# find all the recordings of a given sentence
# listed in the corpus metadata.
# find or create their alignments and f0 tracking.
# sentence should be provided lowercase without punctuation
# TODO something not fatal to interface if <10 --
#   metadata file for SQ is already filtered.
# TODO handle audio that is not originally .wav
#   not an issue for SQ
def get_samromur_queries(sentence, corpusdb, speech_dir, align_dir, align_model_path, f0_dir, reaper_path = "REAPER/build/reaper"):
    with open(corpusdb,'r') as handle:
        meta = handle.read().splitlines()
    meta = [l.split('\t') for l in meta[1:]]
    
    
    # column index 4 of db is normalised sentence text
    meta = [l for l in meta if l[4] == sentence]
    
    if len(meta) < 10:
        if len(meta) < 1:
            print('This sentence does not exist in the corpus')
        else:
            print('Under 10 copies of the sentence: skipping.')
        return []
    else:
        print(f'{len(meta)} recordings of sentence <{sentence}>')
        
    
    word_aligner = None
    
    if not os.path.exists(align_dir):
            os.makedirs(align_dir)
    if not os.path.exists(f0_dir):
            os.makedirs(f0_dir)


    for rec in meta:
        wpath = f'{speech_dir}{rec[2]}'
        apath = align_dir + rec[2].replace('.wav','.tsv')
        if not os.path.exists(apath):
            word_aligner = align_file(wpath,apath, rec[4], word_aligner = word_aligner, model_path = align_model_path)
            
        fpath = f0_dir + rec[2].replace('.wav','.f0')
        if not os.path.exists(fpath):
            fpath = f0_dir + rec[2].replace('.wav','.f0')
            f0_data = estimate_pitch(wpath, reaper_path)
            save_pitch(f0_data,fpath)
      
    
    human_rec_ids = sorted([l[2].split('.wav')[0] for l in meta])
    return human_rec_ids
    


# check if the TTS wavs, alignments, f0 exist for this sentence
# if not, make them
def get_tts(sentence,voices,ttsdir,align_model_path,reaper_path = "REAPER/build/reaper"):

    dpath = setup_tts_sent(sentence,ttsdir)
    

    sample_paths = []

    word_aligner = None

    for v in voices:
        wpath = f'{dpath}/{v}.wav'
        apath = f'{dpath}/{v}.tsv'
        fpath = f'{dpath}/{v}.f0'
        
        if not os.path.exists(wpath):
            #wf = tiro(sentence,v,save=f'{dpath}/')
            wf = grammatek(sentence,v,save=f'{dpath}/')
            
        if not os.path.exists(apath):
            word_aligner = align_file(wpath, apath, snorm(sentence), word_aligner = word_aligner, model_path = align_model_path)


        if not os.path.exists(fpath):
            f0_data = estimate_pitch(wpath, reaper_path)
            save_pitch(f0_data,fpath)
        
        sample_paths.append(wpath)
        
        
        # TODO TEMP
        # return for single last voice
        temp_sample_path = wpath
        
    return temp_sample_path, dpath
    
    


# find if dir for this sentence exists yet
#   or make one, and record it.
# punctuation can affect synthesis
#   so index by original sentence, not normed text
def setup_tts_sent(sentence,ttsdir,meta_path = 'tts_meta.tsv'):
    
    if not os.path.exists(f'{ttsdir}'):
        os.makedirs(f'{ttsdir}')
    
    sentence = sentence.replace('\n',' ')
            
    with open(f'{ttsdir}{meta_path}','a+') as handle:
        handle.seek(0)
        tts_meta = handle.read().splitlines()
        tts_meta = [l.split('\t') for l in tts_meta]
            
        tts_meta = {sent:s_id for s_id,sent in tts_meta}
        
        if sentence not in tts_meta.keys():
            sent_id = sentence.replace(' ','_')[:33]
            rand_id = ''.join(random.choices(string.ascii_uppercase + string.digits, k=6))
            while f'{sent_id}_{rand_id}' in tts_meta.values():
                rand_id = ''.join(random.choices(string.ascii_uppercase + string.digits, k=6))
            sent_id = f'{sent_id}_{rand_id}'
            
            handle.write(f'{sent_id}\t{sentence}\n')
                
        else:
            sent_id = tts_meta[sentence]
        
    sent_dir = f'{ttsdir}{sent_id}'
    if not os.path.exists(f'{sent_dir}'):
        os.makedirs(f'{sent_dir}')
    return sent_dir




# speedup
def precompute(corpusdb, speech_dir, align_dir, align_model_path, f0_dir, reaper_path, fromi=None,toi=None):
    with open(corpusdb,'r') as handle:
        meta = handle.read().splitlines()
    meta = [l.split('\t') for l in meta[1:]]
    
    word_aligner = None
    
    if not os.path.exists(align_dir):
            os.makedirs(align_dir)
    if not os.path.exists(f0_dir):
            os.makedirs(f0_dir)


    if (fromi and toi):
        meta = meta[fromi:toi]

    for rec in meta:
        wpath = f'{speech_dir}{rec[2]}'
        apath = align_dir + rec[2].replace('.wav','.tsv')
        if not os.path.exists(apath):
            word_aligner = align_file(wpath,apath, rec[4], word_aligner = word_aligner, model_path = align_model_path)
            
        fpath = f0_dir + rec[2].replace('.wav','.f0')
        if not os.path.exists(fpath):
            fpath = f0_dir + rec[2].replace('.wav','.f0')
            f0_data = estimate_pitch(wpath, reaper_path)
            save_pitch(f0_data,fpath)
            
    return max(toi,len(meta))
    

def localtest():

    # TODO
    # En hvað veldur þá þessari miklu fjölgun snjógæsa?
    sentence= "Hann spyr: Hvað get ég vitað?"
    #sentence = 'En er hægt að taka orðalagið bókstaflega?'#'Ef svo er, hvað heita þau þá?'#'Var það ekki nóg?'
    voices = ['Alfur_v2'] #,'Dilja']
    # make for now the interface allows max one voice
    
    start_end_word_ix = '1-3'#'5-7'

    locl = '/home/caitlinr/work/peval/pce/'
    corpus_meta = locl+'human_data/SQL1adult10s_metadata.tsv'
    speech_dir = locl+'human_data/audio/squeries/'
    playable_dir = 'https://huggingface.co/spaces/clr/pce/resolve/main/human_data/audio/squeries/'
    speech_aligns = locl+'human_data/align/squeries/'
    speech_f0 = locl+'human_data/f0/squeries/'
    align_model_path ="/home/caitlinr/work/models/LVL/wav2vec2-large-xlsr-53-icelandic-ep10-1000h"

    tts_dir = locl+'tts_data/'
    
    reaper_exc = '/home/caitlinr/work/notterra/REAPER/build/reaper'
    
    norm_sentence = snorm(sentence)
    
    human_rec_ids = get_samromur_queries(norm_sentence, corpus_meta, speech_dir, speech_aligns, align_model_path, speech_f0, reaper_path = reaper_exc)
    
    if voices:

        one_audio_sample, tts_sent_dir = get_tts(sentence,voices,tts_dir,align_model_path,reaper_path = reaper_exc)
        
        voices = [voices[0]] # TODO. now limit one voice at a time.

        score, f0_fig_c0, f0_fig_c1, f0_fig_c2, en_fig_c0, en_fig_c1, en_fig_c2, html, f0_fig_tts, en_fig_tts = cl.cluster(norm_sentence, sentence, human_rec_ids, speech_aligns, speech_f0, speech_dir, playable_dir, tts_sent_dir, voices, start_end_word_ix)


#localtest()
# torch matplotlib librosa sklearn_extra pydub
# env pclustr

#lp = '/home/caitlinr/work/peval/pce/human_data/'
#fnum = precompute(f'{lp}SQL1adult10s_metadata.tsv',f'{lp}audio/squeries/',f'{lp}align/squeries/','/home/caitlinr/work/models/LVL/wav2vec2-large-xlsr-53-icelandic-ep10-1000h',f'{lp}f0/squeries/', '/home/caitlinr/work/notterra/REAPER/build/reaper')
#print(f'have alignments and f0 files for {fnum} recordings')


# https://colab.research.google.com/drive/1RApnJEocx3-mqdQC2h5SH8vucDkSlQYt?authuser=1#scrollTo=410ecd91fa29bc73

# EVALUATION
# - of the tts
# - of the method: consistency? coherency / interpretability of 'best' voice across different features; alt. ability to recover good & problematic features from a combined method if that is chosen as the best? 
# - how similar are the results across different sentences? are any voices consistently good or bad; if multiple are good, are they good in the same way or good in different ways; do humans agree.
#  >> bc hey THAT could at least be an argument for the method, u might have to take time for human judgement once but then you can keep re using it free for new voices. or to select among alternative generations given you might know a context and know what you're going for in that context. etc.