NATURESPEAK-ML-UTTER/speak_broadcast.py

import argparse, json, sys, time, random, logging, signal, threading
import utterance.voice
import utterance.utils
import utterance.osc
import examine.metric

logging.basicConfig(level=logging.ERROR)

UTTERANCE_LEN = 16				#<--------------- these should be in config
NUM_METRIC_GEN = 50
NUM_SAMPLE_VOICES = 3
broadcast = None
metric = None
exit = False
terminal = 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 terminal_utterance(utterance):
	if terminal:
		print(utterance, end="")

def broadcast_utterance(v, utterance):	

	global broadcast, exit

	text = f"{v.name.upper()} : {v.temp}\n"

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

	frags = v.fragments(utterance)

	for f in frags:
		text += f
		broadcast.utterance(text, v.channel)
		terminal_utterance(f)
		time.sleep(2)
		if exit:
			return

	broadcast.command('clear')
	print("==========")

	time.sleep(2)

def find_candidates(v, uv_vec, voices, results):
	logging.info(f"LOOP::finding candidates")

	start = time.time()
	candidates = random.sample(voices, NUM_SAMPLE_VOICES)
	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")

def update():
	global exit
	while not exit:
		utterance.osc.update()
		time.sleep(0.2)


def signal_terminate(signum, frame):
	global exit
	logging.warning("::SIGNAL TERMINATE::")
	exit = True

def main() -> int:

	global broadcast, metric, terminal

	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")
	p.add_argument("-t", "--terminal", action=argparse.BooleanOptionalAction, help="print to terminal")
	args = p.parse_args()	

	logging.info(f"INIT::loading config file - {args.config}")

	with open(args.config) as f:
		conf = json.load(f)

	logging.info(conf)

	terminal = args.terminal

	#--------------------#
	# 		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")

	utterance.osc.start_osc()

	broadcast = utterance.osc.OscBroadcaster(name="osc_broadcast", host=conf['host_voicemachine'], port=conf['port_voicemachine'], command_channel=conf['command_osc_channel'])

	def temperature_cb(temp, name):
		for v in voices:
			if v.name == name:
				v.temp = temp

	receiver = utterance.osc.OscReceiver(name="osc_receiver", host=conf['host_machinespeak'], port=conf['port_machinespeak'], callback_fn=temperature_cb)


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

	t_update = threading.Thread(target=update)
	t_update.start()
	
	while not exit:

		results = []
		t = threading.Thread(target=find_candidates, args=[v, uv_vec, voices, results])
		t.start()

		logging.info(f"LOOP::broadcasting {v.name}")
		broadcast_utterance(v, uv)
		
		t.join()

		# 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}")

	t_update.join()

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

	logging.info(f"FIN")


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