scripts/runSQ.py

import os
from scripts.ctcalign import aligner, wav16m
from scripts.tapi import tiro

# 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):
    #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.

    corpus_meta = '../human_data/SQL1adult_metadata.tsv'
    speech_dir = '../human_data/audio/squeries/'
    speech_aligns = '../human_data/aligns/squeries/'
    speech_f0 = '../human_data/f0/squeries/'
    align_model_path ="carlosdanielhernandezmena/wav2vec2-large-xlsr-53-icelandic-ep10-1000h"

    tts_dir = '../tts_data/'
    
    
    norm_sentencd = snorm(sentence)

    meta = get_recordings(norm_sentence, corpus_meta)
    if meta:
        align_human(meta,speech_aligns,speech_dir,align_model_path)
        f0_human(meta, speech_f0, speech_dir, 'TODO path to reaper')
    if voices:
        temp_a_sample = get_tts(sentence,voices,tts_dir)
        f0_tts(sentence, voices, tts_dir, 'TODO path to reaper')
        
    return temp_a_sample


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
    

# find all the recordings of a given sentence
# listed in the corpus metadata.
# sentence should be provided lowercase without punctuation
def get_recordings(sentence, corpusdb):
    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
    smeta = [l for l in meta if l[4] == sentence]
    
    if len(smeta) < 10:
        if len(smeta) < 1:
            print('This sentence does not exist in the corpus')
        else:
            print('Under 10 copies of the sentence: skipping.')
        return []
    else:
        print(f'{len(smeta)} recordings of sentence <{sentence}>')
        return smeta



# check if word alignments exist for a set of human speech recordings
# if not, warn, and make them with ctcalign.
def align_human(meta,align_dir,speech_dir,model_path):

    model_word_sep = '|'
    model_blank_tk = '[PAD]'
    
    no_align = []
    
    for rec in meta:
        apath = align_dir + rec[2].replace('.wav','.tsv')
        if not os.path.exists(apath):
            no_align.append(rec)
            
    if no_align:
        print(f'Need to run alignment for {len(no_align)} files')
        caligner = aligner(model_path,model_word_sep,model_blank_tk)
        for rec in no_align:
            wav_path = f'{speech_dir}{rec[1]}/{rec[2]}'
            word_aln = caligner(wav16m(wav_path),rec[4],is_normed=True)
            apath = align_dir + rec[2].replace('.wav','.tsv')
            word_aln = [[str(x) for x in l] for l in word_aln]
            with open(apath,'w') as handle:
                handle.write(''.join(['\t'.join(l)+'\n' for l in word_aln]))
    else:
        print('All alignments existed')
        
        

# check if f0s exist for all of those files.
# if not, warn, and make them with TODO reaper
def f0_human(meta, f0_dir, speech_dir, reaper_path):
    no_f0 = []
    
    for rec in meta:
        fpath = f0_dir + rec[2].replace('.wav','.f0')
        if not os.path.exists(fpath):
            no_f0.append(rec)
            
    if no_f0:
        print(f'Need to estimate pitch for {len(no_f0)} recordings')
        #TODO
        
    else:
        print('All speech pitch trackings existed')



# # # # # # # # # 
#################
# TODO
# IMPLEMENT GOOD 2 STEP PITCH ESTIMATION
# TODO
#################
# # # # # # # # # 




# check if the TTS wavs + align jsons exist for this sentence
# if not, warn and make them with TAPI ******
def get_tts(sentence,voices,ttsdir):

    # assume the first 64 chars of sentence are enough
    dpath = sentence.replace(' ','_')[:65]
    
    no_voice = []
    
    temp_sample_path = ''

    for v in voices:
        wpath = f'{ttsdir}{dpath}/{v}.wav'
        jpath = f'{ttsdir}{dpath}/{v}.json'
        if not (os.path.exists(wpath) and os.path.exists(jpath)):
            no_voice.append(v)
        if not temp_sample_path:
            temp_sample_path = wpath
            
    if no_voice:
        print(f'Need to generate TTS for {len(no_voice)} voices')
        if not os.path.exists(f'{ttsdir}{dpath}'):
            os.mkdir(f'{ttsdir}{dpath}')
        for v in voices:
            wf, af = tiro(sentence,v,save=f'{ttsdir}{dpath}/')
        
    else:
        print('TTS for all voices existed')
        
    return temp_sample_path
    


# check if the TTS f0s exist
# if not warn + make
# TODO collapse functions
def f0_tts(sentence, voices, ttsdir, reaper_path):

    # assume the first 64 chars of sentence are enough
    dpath = sentence.replace(' ','_')[:65]
    
    no_f0 = []
    
    for v in voices:
        fpath = f'{ttsdir}{dpath}/{v}.f0'
        if not os.path.exists(fpath):
            no_f0.append(v)
            
    if no_f0:
        print(f'Need to estimate pitch for {len(no_f0)} voices')
        #TODO
        
    else:
        print('All TTS pitch trackings existed')






run()






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

# CLUSTER the humans
# - read energy and pitch, to alignments
# - dtw based with selected chunking ? code should exist.

# ... experimental variants? 
# ** 1 dimension at a time vs 2 on top of each other
# ** 25 points resampling (euclidean, kmeans, i guess....) vs all points dtw kmediods
#  +/or maybe some intermediate parts of that??? like 25 points dtw medoids particularly **
# --different normings for pitch? different settings for energy (tbqh i hope not too much?)
# TODO '''replacement with a constant low value''' ********
# errrrrrrrm duration?
#   duration feature vector will have a different length than the others, BUT, 
 # besides the single clustering,, 
 # i SUPPOSE one could TRY assigning the phone's 'speech rate' value to every frame of the phone, so it doesn't change while the other 2 values do change.... like it would still VAGUELY represent that 2 people elongating the same vowel/syllable are doing similar things with duration while someone eliding that vowel is doing a different durational thing right there?
 # might want to z-score this dimension across ALL speakers tho not within a speaker
  # try doing it both ways at least. bc not sure to what extent i want absolute vs. relative rate info here.
   #(note - unless chengs dur metric is of a kind where only rel makes sense in the first place. idr.)




# GRAPH the humans.
# - probably modify this code a bit to centre on boundary.
# - idk.


# TEST each TTS
# - structure its features
# - find its avg dist for each human cluster
# - find the lowest dist cluster
# - report the dist for i guess this and all clusters
# - GRAPH the tts with its best cluster



# 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.