Fixed timesync and notifier; precursor for clean component orchestration

ipaacalib/python/src/ipaaca/util/notifier.py

@@ -4,7 +4,7 @@
 #  "Incremental Processing Architecture
 #   for Artificial Conversational Agents".
 #
-# Copyright (c) 2009-2014 Social Cognitive Systems Group
+# Copyright (c) 2009-2022 Social Cognitive Systems Group
 #                         CITEC, Bielefeld University
 #
 # http://opensource.cit-ec.de/projects/ipaaca/
@@ -33,9 +33,9 @@
 
 from __future__ import division, print_function
 
+import os
 import threading
 
-
 import ipaaca.buffer
 import ipaaca.iu
 import ipaaca.misc
@@ -64,11 +64,16 @@ class ComponentError(Exception):
 class ComponentNotifier(object):
 
 	NOTIFY_CATEGORY = "componentNotify"
+	CONTROL_CATEGORY = "componentControl"
 	SEND_CATEGORIES = "send_categories"
 	RECEIVE_CATEGORIES = "recv_categories"
+	CMD = "cmd"
 	STATE = "state"
 	NAME = "name"
+	WHO = "who" # list of names (or empty)
 	FUNCTION = "function"
+	PID = "pid"
+	CMD_REPORT = "report"
 
 	def __init__(self, component_name, component_function, send_categories, receive_categories, out_buffer=None, in_buffer=None):
 		self.component_name = component_name
@@ -116,14 +121,23 @@ class ComponentNotifier(object):
 			self.out_buffer.add(notify_iu)
 
 	def _handle_iu_event(self, iu, event_type, local):
-		if iu.payload[ComponentNotifier.NAME] == self.component_name:
-			return
-		with self.notification_handler_lock:
-			for h in self.notification_handlers:
-				h(iu, event_type, local)
-		if iu.payload[ComponentNotifier.STATE] == "new":
-			#print("submitting")
-			self._submit_notify(False)
+		if iu.category == ComponentNotifier.NOTIFY_CATEGORY:
+			if iu.payload[ComponentNotifier.NAME] == self.component_name:
+				return
+			with self.notification_handler_lock:
+				for h in self.notification_handlers:
+					h(iu, event_type, local)
+			if iu.payload[ComponentNotifier.STATE] == "new":
+				#print("submitting")
+				self._submit_notify(False)
+		elif iu.category == ComponentNotifier.CONTROL_CATEGORY:
+			cmd = iu.payload[ComponentNotifier.CMD]
+			if cmd=='report':
+				# Request to report (by component controller)
+				who = iu.payload[ComponentNotifier.WHO]
+				# If we are named specifically or it's a broadcast
+				if len(who)==0 or self.component_name in who:
+					self._submit_notify(False)
 
 	def add_notification_handler(self, handler):
 		with self.notification_handler_lock:
@@ -155,12 +169,14 @@ class ComponentNotifier(object):
 			if (not self.initialized):
 				self.timesync_slave = ipaaca.util.timesync.TimesyncSlave(component_name=self.component_name, timing_handler=self.launch_timesync_slave_handlers)
 				self.timesync_master = ipaaca.util.timesync.TimesyncMaster(component_name=self.component_name, timing_handler=self.launch_timesync_master_handlers)
-				self.in_buffer.register_handler(self._handle_iu_event, ipaaca.iu.IUEventType.MESSAGE, ComponentNotifier.NOTIFY_CATEGORY)
+				self.in_buffer.register_handler(self._handle_iu_event, ipaaca.iu.IUEventType.MESSAGE, [ComponentNotifier.NOTIFY_CATEGORY, ComponentNotifier.CONTROL_CATEGORY])
 				self._submit_notify(True)
 				self.initialized = True
 
 	def __enter__(self):
 		self.initialize()
+		return self
 
 	def __exit__(self, t, v, tb):
 		self.terminate()
+		return self

ipaacalib/python/src/ipaaca/util/timesync.py

@@ -4,7 +4,7 @@
 #  "Incremental Processing Architecture
 #   for Artificial Conversational Agents".
 #
-# Copyright (c) 2009-2014 Social Cognitive Systems Group
+# Copyright (c) 2009-2022 Social Cognitive Systems Group
 #                         CITEC, Bielefeld University
 #
 # http://opensource.cit-ec.de/projects/ipaaca/
@@ -32,6 +32,7 @@
 
 from __future__ import division, print_function
 
+import threading
 import time
 
 import ipaaca.buffer
@@ -45,7 +46,7 @@ class TimesyncMaster(object):
 		# component name to report (None => use buffer name)
 		self.component_name = component_name if component_name is not None else self.ob.unique_name
 		#
-		self.ob.register_handler(self.handle_timesync_master)
+		#self.ob.register_handler(self.handle_timesync_master)
 		self.ib.register_handler(self.handle_timesync_master)
 		# master_t1 is identical for all slaves
 		self.master_t1 = None
@@ -63,7 +64,7 @@ class TimesyncMaster(object):
 		self.timing_handler = timing_handler
 	
 	def send_master_timesync(self):
-		iu = ipaaca.iu.IU('timesyncRequest')
+		iu = ipaaca.iu.Message('timesyncRequest')
 		self.master_t1 = self.get_time()
 		iu.payload = {
 				'stage':'0',
@@ -71,47 +72,48 @@ class TimesyncMaster(object):
 				'master':self.component_name,
 				}
 		self.ob.add(iu)
-		print('Sent a stage 0 timesync as master '+self.component_name)
 	
 	def handle_timesync_master(self, iu, event_type, own):
 		master = iu.payload['master']
 		if not own and master == self.component_name:
-			if self.component_name == master:
-				# reply to our own initial IU
-				slave = iu.payload['slave']
-				stage = iu.payload['stage']
-				if stage=='1':
-					print('Received stage 1 from slave '+slave)
-					# initial reply by slave
-					t1 = iu.payload['slave_t1']
-					self.slave_t1s[slave] = float(t1)
-					t2 = self.master_t2s[slave] = self.get_time()
-					iu.payload.merge({'master_t2': str(t2), 'stage':'2'})
-					latency1 = t2 - self.master_t1
-					self.latencies[slave] = latency1
-					#print('Latency of round-trip 1: %.3f' % latency1)
-				elif stage=='3':
-					print('Received stage 3 from slave '+slave)
-					# second reply by slave
-					t2 = iu.payload['slave_t2']
-					self.slave_t2s[slave] = float(t2)
-					t_final = self.get_time()
-					latency1 = self.latencies[slave]
-					latency2 = t_final - self.master_t2s[slave]
-					latency = self.latencies[slave] = (latency1+latency2)/2.0
-					offset1 = (self.slave_t1s[slave]-self.master_t1)-latency/2.0
-					offset2 = (self.slave_t2s[slave]-self.master_t2s[slave])-latency/2.0
-					offset = (offset1+offset2)/2.0
-					iu.payload.merge({'stage':'4', 'latency': str(latency), 'offset':str(offset)})
-					if self.timing_handler is None:
-						print('Determined timing of timesync slave '+slave)
-						print('  Avg round-trip latency: %.3f s'%latency)
-						print('  Offset of their clock: %.3f s'%offset)
+			if event_type == ipaaca.IUEventType.ADDED or event_type == ipaaca.IUEventType.UPDATED:
+				if self.component_name == master:
+					# reply to our own initial IU
+					slave = iu.payload['slave']
+					stage = iu.payload['stage']
+					if stage=='1':
+						# initial reply by slave
+						t1 = iu.payload['slave_t1']
+						self.slave_t1s[slave] = float(t1)
+						t2 = self.master_t2s[slave] = self.get_time()
+						iu.payload.merge({'master_t2': str(t2), 'stage':'2'})
+						latency1 = t2 - self.master_t1
+						#print('Before stage 1 for '+slave+': '+str(self.latencies))
+						self.latencies[slave] = latency1
+						#print('After stage 1 for '+slave+': '+str(self.latencies))
+						#print('Latency of round-trip 1: %.3f' % latency1)
+					elif stage=='3':
+						#print('At stage 3 for '+slave+': '+str(self.latencies))
+						# second reply by slave
+						t2 = iu.payload['slave_t2']
+						self.slave_t2s[slave] = float(t2)
+						t_final = self.get_time()
+						latency1 = self.latencies[slave]
+						latency2 = t_final - self.master_t2s[slave]
+						latency = self.latencies[slave] = (latency1+latency2)/2.0
+						offset1 = (self.slave_t1s[slave]-self.master_t1)-latency/2.0
+						offset2 = (self.slave_t2s[slave]-self.master_t2s[slave])-latency/2.0
+						offset = (offset1+offset2)/2.0
+						iu.payload.merge({'stage':'4', 'latency': str(latency), 'offset':str(offset)})
+						if self.timing_handler is None:
+							print('Determined timing of timesync slave '+slave)
+							print('  Avg round-trip latency: %.3f s'%latency)
+							print('  Offset of their clock: %.3f s'%offset)
+						else:
+							self.timing_handler(self.component_name, slave, latency, offset)
 					else:
-						self.timing_handler(self.component_name, slave, latency, offset)
-				else:
-					# other stages are handled by time slave handler
-					pass
+						# other stages are handled by time slave handler
+						pass
 	
 	def get_time(self):
 		return time.time() + self.debug_offset
@@ -129,7 +131,6 @@ class TimesyncSlave(object):
 		#self.master_t2 = None
 		#self.master = None
 		self.latency = None
-		self.my_iu = None
 		#
 		self.debug_offset = debug_offset
 		#
@@ -142,30 +143,23 @@ class TimesyncSlave(object):
 		master = iu.payload['master']
 		stage = iu.payload['stage']
 		if self.component_name != master:
-			if not own:
+			if not own and stage=='0':
 				# reply only to IUs from others
-				if stage=='0':
-					#print('Received stage 0 from master '+master)
-					# initial reply to master
-					self.my_iu = ipaaca.iu.IU('timesyncReply')
-					# TODO: add grounded_in link too?
-					t1 = self.get_time()
-					self.my_iu.payload = iu.payload
-					self.my_iu.payload['slave'] = self.component_name
-					self.my_iu.payload['slave_t1'] = str(t1)
-					self.my_iu.payload['stage'] = '1'
-
-					#self.my_iu.payload.merge({
-					#		'slave':self.component_name,
-					#		'slave_t1':str(t1),
-					#		'stage':'1',
-					#		})
-					self.ob.add(self.my_iu)
-			else:
+				#print('Received stage 0 from master '+master)
+				# initial reply to master
+				myiu = ipaaca.iu.IU('timesyncReply')
+				# TODO: add grounded_in link too?
+				t1 = self.get_time()
+				myiu.payload = iu.payload
+				myiu.payload['slave'] = self.component_name
+				myiu.payload['slave_t1'] = str(t1)
+				myiu.payload['stage'] = '1'
+				self.ob.add(myiu)
+			elif iu.payload['slave'] == self.component_name:
 				if stage=='2':
 					#print('Received stage 2 from master '+master)
 					t2 = self.get_time()
-					self.my_iu.payload.merge({
+					iu.payload.merge({
 							'slave_t2':str(t2),
 							'stage':'3',
 							})