diff --git a/speak_broadcast.py b/speak_broadcast.py index a9c5ff7..8f26861 100644 --- a/speak_broadcast.py +++ b/speak_broadcast.py @@ -1,19 +1,44 @@ -import argparse, json, sys, time, random +import argparse, json, sys, time, random, logging, signal import utterance.voice import utterance.utils import utterance.osc import examine.metric UTTERANCE_LEN = 16 +NUM_METRIC_GEN = 50 +NUM_SAMPLE_VOICES = 3 broadcast = None +metric = None +exit = False -def format_str(text: str) -> str: +def format_str(text) -> str: t = utterance.utils.clean(text) - return utterance.utils.format(t) + return utterance.utils.format(t) + +def utter_one(v) -> str: + u = v.utter_one() + return format_str(u) + +def utter_one_vectorise(v): + global metric + uv = utter_one(v) + uv_vec = metric.vector(uv) + return uv, uv_vec + +def utter_n_vectorise_distance(v, n, vec): + global metric + results = [] + texts = v.utter_n(n=n) + for t in texts: + t = format_str(t) + t_vec = metric.vector(t) + d = examine.metric.cos_dist(vec, t_vec) + results.append([d, t, v]) + return results def broadcast_utterance(v, utterance): - global broadcast + global broadcast, exit text = v.name.upper() + ":\n" @@ -26,23 +51,41 @@ def broadcast_utterance(v, utterance): text += f broadcast.utterance(text, v.channel) time.sleep(2) + if exit: + return broadcast.command('clear') time.sleep(2) - + +def signal_terminate(signum, frame): + global exit + logging.warning("::SIGNAL TERMINATE::") + exit = True + def main() -> int: - global broadcast + global broadcast, metric p = argparse.ArgumentParser() p.add_argument("-c", "--config", type=str, default="voice.config.json", help="configuratin file") p.add_argument("-i", "--iterations", type=int, default=10, help="number of iterations") args = p.parse_args() + logging.basicConfig(level=logging.INFO) + + logging.info(f"INIT::loading config file - {args.config}") + with open(args.config) as f: conf = json.load(f) + logging.info(conf) + + #--------------------# + # VOICES + #--------------------# + logging.info(f"INIT::creating voices") + voices = [] for v in conf['voices']: model = v['model'] @@ -50,64 +93,63 @@ def main() -> int: voice.set_channel(v['osc_channel']['root'], v['osc_channel']['utterance']) voices.append(voice) + + #--------------------# + # NET + #--------------------# + logging.info(f"INIT::setting up OSC") broadcast = utterance.osc.OscBroadcaster(name="osc_broadcast", host=conf['host'], port=conf['port'], command_channel=conf['command_osc_channel']) - broadcast.start_osc() - - nbr_voices = len(voices) - - state = 'c' + broadcast.start_osc() + #--------------------# + # METRIC + #--------------------# + logging.info(f"INIT::loading doc2vec metrics") metric = examine.metric.Metric(model_input='data/models/doc2vec.model') - s = set(range(0, nbr_voices - 1)) - - rindex = random.sample(s, 1)[0] - - v = voices[rindex] - uv = v.utter_one() - uv = format_str(uv) - - v_vec = metric.vector(uv) + #--------------------# + # A + #--------------------# + logging.info(f"INIT::generating first utterance") + random.seed(time.time()) + v = random.choice(voices) + uv, uv_vec = utter_one_vectorise(v) + while not exit: - while state == 'c': + logging.info(f"LOOP::broadcasting {v.name}") + broadcast_utterance(v, uv) - candidates = random.sample(s, 3) - + logging.info(f"LOOP::finding candidates") + start = time.time() + candidates = random.sample(voices, NUM_SAMPLE_VOICES) results = [] for c in candidates: - if c == rindex: + if exit: + break + if c == v: continue - vc = voices[c] - vc_texts = vc.utter_n(n=50) - for t in vc_texts: - t = format_str(t) - t_vec = metric.vector(t) - d = examine.metric.cos_dist(v_vec, t_vec) - results.append([d, t, c]) - + results += utter_n_vectorise_distance(c, NUM_METRIC_GEN, uv_vec) results.sort(key=lambda t: t[0], reverse=True) + lapse = time.time() - start + logging.info(f"LOOP::done - {lapse} secs") + + # ok here we need to randomise maybe...?! + # ([d, t, v]) + choice = results[0] + v = choice[2] + uv = choice[1] + uv_vec = metric.vector(uv) + logging.info(f"LOOP::next {v.name}") - broadcast_utterance(v, uv) - - - # new round - - top = results[0] - rindex = top[2] - v = voices[rindex] - uv = top[1] - v_vec = metric.vector(top[1]) - - - # state = input("Continue? ") - - - + logging.info(f"TERMINATE::terminating OSC") broadcast.terminate_osc() + logging.info(f"FIN") + if __name__ == '__main__': - sys.exit(main()) \ No newline at end of file + signal.signal(signal.SIGINT, signal_terminate) + sys.exit(main()) \ No newline at end of file