NATURESPEAK-ML-UTTER/utterance/osc.py

from osc4py3.as_eventloop import *
from osc4py3 import oscbuildparse
from threading import Lock

mutex = Lock()
temp_mutex = Lock()


def start_osc():
	osc_startup()

def terminate_osc():
	osc_terminate()

def temperature(temp, name):
	print(f'{temp} - {name}')

def update():
	global mutex
	with mutex:
		osc_process()

class OscBroadcaster:

	def __init__(self, name: str, host: str, port: str, command_channel: str):
		self.name = name
		self.host = host
		self.port = port
		self.cmd = command_channel
		osc_udp_client(self.host, self.port, self.name)		

	def utterance(self, utterance: str, channel: str):
		msg = oscbuildparse.OSCMessage(channel, None, [utterance])
		osc_send(msg, self.name)
		update()

	def command(self, command: str):
		msg = oscbuildparse.OSCMessage(self.cmd, None, [command])
		osc_send(msg, self.name)
		update()

	def temperature(self, temp: float, channel: str):
		global temp_mutex
		with temp_mutex:
			msg = oscbuildparse.OSCMessage(channel, None, [temp])			
			osc_send(msg, self.name)			
			osc_process()
			

class OscReceiver:

	def __init__(self, name: str, host: str, port: str, callback_fn=None):
		self.name = name
		self.host = host
		self.port = port
		# osc_udp_server(self.host, self.port, self.name)		
		osc_udp_server("0.0.0.0", self.port, self.name)
		if callback_fn:
			osc_method('/', callback_fn)
		else:
			osc_method('/', temperature)