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.INFO)

UTTERANCE_LEN = 16				#<--------------- these should be in config
NUM_METRIC_GEN = 75
NUM_SAMPLE_VOICES = 3
RANDOM_SEED_TIMER_MIN = 2
broadcast = None
metric = None
exit = False
terminal = False
debug = 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, debug

	if debug:
		temp = format(v.temp, '.2f')
		text = f"{v.name.upper()}: {temp}\n"
	else:
		text = f""

	# Send all text to server to calculate bounds in advance
	broadcast.utterance(utterance, v.calculate)
	

	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:
		try:
			utterance.osc.update()
		except Exception as e:
			logging.error(e) 
			pass
		
		time.sleep(0.2)

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

def main() -> int:

	global broadcast, metric, terminal, debug

	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='store_true', 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'])
		voice.set_calculate(v['osc_channel']['root'], v['osc_channel']['calculate'])
		voice.set_temperature(v['osc_channel']['root'], v['osc_channel']['temperature'])
		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 receiver_cb(temp, name):
		global debug
		if type(temp) == str and temp == "DEBUG":
			debug = name
			logging.info(f"DEBUG MODE: {debug}")
		else: # temperature
			for v in voices:
				if v.name == name:
					print(f'{name} - {temp}')
					v.temp = temp
					broadcast.temperature(temp, v.temperature)

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


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

	#--------------------#
	# 		RANDOM
	#--------------------#

	def random_seed():
		global t_random_seed
		logging.info("INIT::SEEDING RANDOM")
		random.seed(time.time())
		if not exit:
			t_random_seed = threading.Timer(random.randint(60, 60 * RANDOM_SEED_TIMER_MIN), random_seed)
			t_random_seed.start()

	t_random_seed = threading.Timer(random.randint(60, 60 * RANDOM_SEED_TIMER_MIN), random_seed)
	t_random_seed.start()

	#--------------------#
	# 		  A
	#--------------------#
	logging.info(f"INIT::generating first utterance")
	
	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]
		# makse sure we don't say the same thing over and over again
		for r in results:
			v = r[2]
			u = r[1]
			if v.select(u):
				choice = r
				break
			else:
				logging.info(f"LOOP::reduncancy {v.name}")

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

	# if t_random_seed:
	logging.info(f"TERMINATE::random seed")
	t_random_seed.cancel()
	t_random_seed.join()

	logging.info(f"FIN")


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