NATURESPEAK-ML-UTTER/speak_broadcast.py

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:
	t = utterance.utils.clean(text)
	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, exit

	text = v.name.upper() + ":\n"

	broadcast.utterance(text, v.channel)
	time.sleep(2)

	frags = v.fragments(utterance)

	for f in frags:
		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, 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']
		voice = utterance.voice.Voice(name=v["name"].upper(), model=model['model_dir'], tokenizer=model['tokeniser_file'], temp=float(model["temperature"]), lenght=UTTERANCE_LEN)
		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()	

	#--------------------#
	# 		METRIC
	#--------------------#
	logging.info(f"INIT::loading doc2vec metrics")
	metric = examine.metric.Metric(model_input='data/models/doc2vec.model')

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

		logging.info(f"LOOP::broadcasting {v.name}")
		broadcast_utterance(v, uv)
		
		logging.info(f"LOOP::finding candidates")
		start = time.time()
		candidates = random.sample(voices, NUM_SAMPLE_VOICES)
		results = []
		for c in candidates:
			if exit:
				break
			if c == v:
				continue
			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}")


	logging.info(f"TERMINATE::terminating OSC")
	broadcast.terminate_osc()

	logging.info(f"FIN")


if __name__ == '__main__':
	signal.signal(signal.SIGINT, signal_terminate)
	sys.exit(main())