ipaacalib/python/src/ipaaca/exception.py

+# -*- coding: utf-8 -*-
+
+# This file is part of IPAACA, the
+#  "Incremental Processing Architecture
+#   for Artificial Conversational Agents".
+#
+# Copyright (c) 2009-2022 Social Cognitive Systems Group
+#                         CITEC, Bielefeld University
+#
+# http://opensource.cit-ec.de/projects/ipaaca/
+# http://purl.org/net/ipaaca
+#
+# This file may be licensed under the terms of of the
+# GNU Lesser General Public License Version 3 (the ``LGPL''),
+# or (at your option) any later version.
+#
+# Software distributed under the License is distributed
+# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+# express or implied. See the LGPL for the specific language
+# governing rights and limitations.
+#
+# You should have received a copy of the LGPL along with this
+# program. If not, go to http://www.gnu.org/licenses/lgpl.html
+# or write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+#
+# The development of this software was supported by the
+# Excellence Cluster EXC 277 Cognitive Interaction Technology.
+# The Excellence Cluster EXC 277 is a grant of the Deutsche
+# Forschungsgemeinschaft (DFG) in the context of the German
+# Excellence Initiative.
+
+from __future__ import division, print_function
+
+
+__all__ = [
+	'IpaacaError',
+	'IUCommittedError',
+	'IUNotFoundError',
+	'IUPayloadLockedError',
+	'IUPayloadLockTimeoutError',
+	'IUPublishedError',
+	'IUReadOnlyError',
+	'IUResendRequestFailedError',
+	'IUUpdateFailedError',
+]
+
+
+class IpaacaError(Exception): pass
+
+class BackendInitializationError(IpaacaError):
+    """Error indicating that type marshalling cannot proceed
+    because no matching converter is known."""
+    def __init__(self, name=''):
+        super(BackendInitializationError, self).__init__(    \
+                'Failed to initialize selected backend '+str(name))
+
+class BackendSerializationError(IpaacaError):
+    """Error indicating that type marshalling cannot proceed
+    because no matching converter is known."""
+    def __init__(self, typ):
+        super(BackendSerializationError, self).__init__(    \
+                'Could not serialize type ' + str(typ.__name__)  \
+                + ' - no converter registered.')
+
+class BackendDeserializationError(IpaacaError):
+    """Error indicating that type unmarshalling cannot proceed
+    because no matching converter is known."""
+    def __init__(self, wire_schema):
+        super(BackendDeserializationError, self).__init__(           \
+                'Could not deserialize wire format "' + str(wire_schema) \
+                + '" - no converter registered.')
+
+class ConverterRegistrationError(IpaacaError):
+    '''Error indicating that a type or wire schema already had a registered converter.'''
+    def __init__(self, type_name_or_schema):
+        super(ConverterRegistrationError, self).__init__(
+                'Failed to register a converter: we already have one for ' \
+                + str(type_name_or_schema))
+
+class IUCommittedError(IpaacaError):
+	"""Error indicating that an IU is immutable because it has been committed to."""
+	def __init__(self, iu):
+		super(IUCommittedError, self).__init__('Writing to IU ' + str(iu.uid) + ' failed -- it has been committed to.')
+
+
+class IUNotFoundError(IpaacaError):
+	"""Error indicating that an IU UID was unexpectedly not found in an internal store."""
+	def __init__(self, iu_uid):
+		super(IUNotFoundError, self).__init__('Lookup of IU ' + str(iu_uid) + ' failed.')
+
+
+class IUPayloadLockTimeoutError(IpaacaError):
+	"""Error indicating that exclusive access to the Payload could not be obtained in time."""
+	def __init__(self, iu):
+		super(IUPayloadLockTimeoutError, self).__init__('Timeout while accessing payload of IU ' + str(iu.uid) + '.')
+
+
+class IUPayloadLockedError(IpaacaError):
+	"""Error indicating that exclusive access to the Payload could not be obtained because someone actually locked it."""
+	def __init__(self, iu):
+		super(IUPayloadLockedError, self).__init__('IU '+str(iu.uid)+' was locked during access attempt.')
+
+
+class IUPublishedError(IpaacaError):
+	"""Error publishing of an IU failed since it is already in the buffer."""
+	def __init__(self, iu):
+		super(IUPublishedError, self).__init__('IU ' + str(iu.uid) + ' is already present in the output buffer.')
+
+
+class IUReadOnlyError(IpaacaError):
+	"""Error indicating that an IU is immutable because it is 'read only'."""
+	def __init__(self, iu):
+		super(IUReadOnlyError, self).__init__(
+			'Writing to IU ' + str(iu.uid) + ' failed -- it is read-only.')
+
+
+class IUResendRequestFailedError(IpaacaError):
+	"""Error indicating that a remote IU resend failed."""
+	def __init__(self, iu_uid):
+		super(IUResendRequestFailedError, self).__init__(
+			'Remote resend failed for IU ' + str(iu_uid))
+
+class IUResendRequestRemoteServerUnknownError(IpaacaError):
+	"""Error indicating that a remote IU resend failed."""
+	def __init__(self, iu_uid):
+		super(IUResendRequestRemoteServerUnknownError, self).__init__(
+			'Remote resend request: remote server unknown for IU ' + str(iu_uid))
+
+class IURetractedError(IpaacaError):
+	"""Error indicating that an IU has been retracted."""
+	def __init__(self, iu):
+		super(IURetractedError, self).__init__('Writing to IU ' + str(iu.uid) + ' failed -- it has been retracted.')
+
+
+class IUUpdateFailedError(IpaacaError):
+	"""Error indicating that a remote IU update failed."""
+	def __init__(self, iu):
+		super(IUUpdateFailedError, self).__init__('Remote update failed for IU ' + str(iu.uid) + '.')

ipaacalib/python/src/ipaaca/iu.py

+# -*- coding: utf-8 -*-
+
+# This file is part of IPAACA, the
+#  "Incremental Processing Architecture
+#   for Artificial Conversational Agents".
+#
+# Copyright (c) 2009-2022 Social Cognitive Systems Group
+#                         CITEC, Bielefeld University
+#
+# http://opensource.cit-ec.de/projects/ipaaca/
+# http://purl.org/net/ipaaca
+#
+# This file may be licensed under the terms of of the
+# GNU Lesser General Public License Version 3 (the ``LGPL''),
+# or (at your option) any later version.
+#
+# Software distributed under the License is distributed
+# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+# express or implied. See the LGPL for the specific language
+# governing rights and limitations.
+#
+# You should have received a copy of the LGPL along with this
+# program. If not, go to http://www.gnu.org/licenses/lgpl.html
+# or write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+#
+# The development of this software was supported by the
+# Excellence Cluster EXC 277 Cognitive Interaction Technology.
+# The Excellence Cluster EXC 277 is a grant of the Deutsche
+# Forschungsgemeinschaft (DFG) in the context of the German
+# Excellence Initiative.
+
+from __future__ import division, print_function
+
+import collections
+import copy
+import threading
+import uuid
+
+import six
+
+import ipaaca.ipaaca_pb2
+import ipaaca.converter
+import ipaaca.exception
+import ipaaca.misc
+import ipaaca.payload
+
+
+__all__ = [
+	'IUAccessMode',
+	'IUEventType',
+	'IUPayloadType',
+	'IU',
+	'Message',
+]
+
+LOGGER = ipaaca.misc.get_library_logger()
+
+IUAccessMode = ipaaca.misc.enum(
+	PUSH = 'PUSH',
+	REMOTE = 'REMOTE',
+	MESSAGE = 'MESSAGE'
+)
+
+
+IUEventType = ipaaca.misc.enum(
+	ADDED = 'ADDED',
+	COMMITTED = 'COMMITTED',
+	DELETED = 'DELETED',
+	RETRACTED = 'RETRACTED',
+	UPDATED = 'UPDATED',
+	LINKSUPDATED = 'LINKSUPDATED',
+	MESSAGE = 'MESSAGE'
+)
+
+
+IUPayloadType = ipaaca.misc.enum(
+	JSON = 'JSON',
+	STR = 'STR'
+)
+
+
+class IUInterface(object):
+
+	"""Base class of all specialised IU classes."""
+
+	def __init__(self, uid, access_mode=IUAccessMode.PUSH, read_only=False, payload_type=None):
+		"""Creates an IU.
+
+		Keyword arguments:
+		uid -- unique ID of this IU
+		access_mode -- access mode of this IU
+		read_only -- flag indicating whether this IU is read_only or not
+		"""
+		self._uid = uid
+		self._revision = None
+		self._category = None
+		self._owner_name = None
+		self._committed = False
+		self._retracted = False
+		self._access_mode = access_mode
+		self._read_only = read_only
+		self._payload_type = payload_type if payload_type is not None else ipaaca.defaults.IPAACA_DEFAULT_IU_PAYLOAD_TYPE
+		self._buffer = None
+		# payload is not present here
+		self._links = collections.defaultdict(set)
+
+	def __str__(self):
+		s = str(self.__class__)+"{ "
+		s += "category="+("<None>" if self._category is None else self._category)+" "
+		s += "uid="+self._uid+" "
+		s += "(buffer="+(self.buffer.unique_name if self.buffer is not None else "<None>")+") "
+		s += "owner_name=" + ("<None>" if self.owner_name is None else self.owner_name) + " "
+		s += "payload={ "
+		for k,v in self.payload.items():
+			s += k+":'"+str(v)+"', "
+		s += "} "
+		s += "links={ "
+		for t, ids in self.get_all_links().items():
+			s += t+":'"+str(ids)+"', "
+		s += "} "
+		s += "}"
+		return s
+
+
+	def _add_and_remove_links(self, add, remove):
+		'''Just add and remove the new links in our links set, do not send an update here'''
+		'''Note: Also used for remotely enforced links updates.'''
+		for type in remove.keys(): self._links[type] -= set(remove[type])
+		for type in add.keys(): self._links[type] |= set(add[type])
+	def _replace_links(self, links):
+		'''Just wipe and replace our links set, do not send an update here'''
+		'''Note: Also used for remotely enforced links updates.'''
+		self._links = collections.defaultdict(set)
+		for type in links.keys(): self._links[type] |= set(links[type])
+
+	def add_links(self, type, targets, writer_name=None):
+		'''Attempt to add links if the conditions are met
+		and send an update message. Then call the local setter.'''
+		if isinstance(targets, six.string_types) and hasattr(targets, '__iter__'): targets=[targets]
+		self._modify_links(is_delta=True, new_links={type:targets}, links_to_remove={}, writer_name=writer_name)
+		self._add_and_remove_links( add={type:targets}, remove={} )
+	def remove_links(self, type, targets, writer_name=None):
+		'''Attempt to remove links if the conditions are met
+		and send an update message. Then call the local setter.'''
+		if isinstance(targets, six.string_types) and hasattr(targets, '__iter__'): targets=[targets]
+		self._modify_links(is_delta=True, new_links={}, links_to_remove={type:targets}, writer_name=writer_name)
+		self._add_and_remove_links( add={}, remove={type:targets} )
+	def modify_links(self, add, remove, writer_name=None):
+		'''Attempt to modify links if the conditions are met
+		and send an update message. Then call the local setter.'''
+		self._modify_links(is_delta=True, new_links=add, links_to_remove=remove, writer_name=writer_name)
+		self._add_and_remove_links( add=add, remove=remove )
+	def set_links(self, links, writer_name=None):
+		'''Attempt to set (replace) links if the conditions are met
+		and send an update message. Then call the local setter.'''
+		self._modify_links(is_delta=False, new_links=links, links_to_remove={}, writer_name=writer_name)
+		self._replace_links( links=links )
+	def get_links(self, type):
+		return set(self._links[type])
+	def get_all_links(self):
+		return copy.deepcopy(self._links)
+
+	def _get_revision(self):
+		return self._revision
+	revision = property(fget=_get_revision, doc='Revision number of the IU.')
+
+	def _get_category(self):
+		return self._category
+	category = property(fget=_get_category, doc='Category of the IU.')
+
+	def _get_payload_type(self):
+		return self._payload_type
+	def _set_payload_type(self, type):
+		if self._buffer is None:
+			self._payload_type = type
+		else:
+			raise ipaaca.exception.IpaacaError('The IU is already in a buffer, cannot change payload type anymore.')
+	payload_type = property(
+		fget=_get_payload_type,
+		fset=_set_payload_type,
+		 doc='Type of the IU payload')
+
+	def _get_committed(self):
+		return self._committed
+	committed = property(
+			fget=_get_committed,
+			doc='Flag indicating whether this IU has been committed to.')
+
+	def _get_retracted(self):
+		return self._retracted
+	retracted = property(
+			fget=_get_retracted,
+			doc='Flag indicating whether this IU has been retracted.')
+
+	def _get_uid(self):
+		return self._uid
+	uid = property(fget=_get_uid, doc='Unique ID of the IU.')
+
+	def _get_access_mode(self):
+		return self._access_mode
+	access_mode = property(fget=_get_access_mode, doc='Access mode of the IU.')
+
+	def _get_read_only(self):
+		return self._read_only
+	read_only = property(
+			fget=_get_read_only,
+			doc='Flag indicating whether this IU is read only.')
+
+	def _get_buffer(self):
+		return self._buffer
+	def _set_buffer(self, buffer):
+		if self._buffer is not None:
+			raise ipaaca.exception.IpaacaError('The IU is already in a buffer, cannot move it.')
+		self._buffer = buffer
+	buffer = property(
+			fget=_get_buffer,
+			fset=_set_buffer,
+			doc='Buffer this IU is held in.')
+
+	def _get_owner_name(self):
+		return self._owner_name
+	def _set_owner_name(self, owner_name):
+		if self._owner_name is not None:
+			raise ipaaca.exception.IpaacaError('The IU already has an owner name, cannot change it.')
+		self._owner_name = owner_name
+	owner_name = property(
+			fget=_get_owner_name,
+			fset=_set_owner_name,
+			doc="The IU's owner's name.")
+
+
+class IU(IUInterface):
+
+	"""A local IU."""
+
+	def __init__(self, category='undef', access_mode=IUAccessMode.PUSH, read_only=False, payload_type=None):
+		super(IU, self).__init__(uid=None, access_mode=access_mode, read_only=read_only, payload_type=payload_type)
+		self._revision = 1
+		self.uid = str(uuid.uuid4())
+		self._category = str(category)
+		self.revision_lock = threading.RLock()
+		self._payload = ipaaca.payload.Payload(iu=self)
+
+	def _modify_links(self, is_delta=False, new_links={}, links_to_remove={}, writer_name=None):
+		if self._retracted:
+			raise ipaaca.exception.IURetractedError(self)
+		if self._committed:
+			raise ipaaca.exception.IUCommittedError(self)
+		with self.revision_lock:
+			# modify links locally
+			self._increase_revision_number()
+			if self.is_published:
+				# send update to remote holders
+				self.buffer._send_iu_link_update(
+						self,
+						revision=self.revision,
+						is_delta=is_delta,
+						new_links=new_links,
+						links_to_remove=links_to_remove,
+						writer_name=self.owner_name if writer_name is None else writer_name)
+
+	def _modify_payload(self, is_delta=True, new_items={}, keys_to_remove=[], writer_name=None):
+		"""Modify the payload: add or remove items from this payload locally and send update."""
+		if self._retracted:
+			raise ipaaca.exception.IURetractedError(self)
+		if self._committed:
+			raise ipaaca.exception.IUCommittedError(self)
+		with self.revision_lock:
+			# set item locally
+			# FIXME: Is it actually set locally?
+			self._increase_revision_number()
+			if self.is_published:
+				#print('  _modify_payload: running send_iu_pl_upd with writer name '+str(writer_name))
+				# send update to remote holders
+				self.buffer._send_iu_payload_update(
+						self,
+						revision=self.revision,
+						is_delta=is_delta,
+						new_items=new_items,
+						keys_to_remove=keys_to_remove,
+						writer_name=self.owner_name if writer_name is None else writer_name)
+
+	def _increase_revision_number(self):
+		self._revision += 1
+
+	def _internal_commit(self, writer_name=None):
+		if self._committed:
+			return
+		if self._retracted:
+			raise ipaaca.exception.IURetractedError(self)
+		with self.revision_lock:
+			self._increase_revision_number()
+			self._committed = True
+			if self.buffer is not None:
+				self.buffer._send_iu_commission(self, writer_name=writer_name)
+
+	def commit(self):
+		"""Commit to this IU."""
+		return self._internal_commit()
+
+	def retract(self):
+		if self._buffer:
+			self._buffer.remove(self)
+			self._buffer = None
+		else:
+			self._retracted = True
+
+	def _get_payload(self):
+		return self._payload
+	def _set_payload(self, new_pl, writer_name=None):
+		if self._retracted:
+			raise ipaaca.exception.IURetractedError(self)
+		if self._committed:
+			raise ipaaca.exception.IUCommittedError(self)
+		with self.revision_lock:
+			self._increase_revision_number()
+			self._payload = ipaaca.payload.Payload(
+					iu=self,
+					writer_name=None if self.buffer is None else (self.buffer.unique_name if writer_name is None else writer_name),
+					new_payload=new_pl)
+	payload = property(
+			fget=_get_payload,
+			fset=_set_payload,
+			doc='Payload dictionary of this IU.')
+
+	def _get_is_published(self):
+		return self.buffer is not None
+	is_published = property(
+			fget=_get_is_published,
+			doc='Flag indicating whether this IU has been published or not.')
+
+	def _set_buffer(self, buffer):
+		if self._buffer is not None:
+			raise ipaaca.exception.IpaacaError('The IU is already in a buffer, cannot move it.')
+		self._buffer = buffer
+		self.owner_name = buffer.unique_name
+		self._payload.owner_name = buffer.unique_name
+	buffer = property(
+			fget=IUInterface._get_buffer,
+			fset=_set_buffer,
+			doc='Buffer this IU is held in.')
+
+	def _set_uid(self, uid):
+		if self._uid is not None:
+			raise AttributeError('The uid of IU ' + self.uid + ' has already been set, cannot change it.')
+		self._uid = uid
+	uid = property(
+			fget=IUInterface._get_uid,
+			fset=_set_uid,
+			doc='Unique ID of the IU.')
+
+
+class Message(IU):
+	"""Local IU of Message sub-type. Can be handled like a normal IU, but on the remote side it is only existent during the handler calls."""
+	def __init__(self, category='undef', access_mode=IUAccessMode.MESSAGE, read_only=True, payload_type=None):
+		super(Message, self).__init__(category=str(category), access_mode=access_mode, read_only=read_only, payload_type=payload_type)
+
+	def _modify_links(self, is_delta=False, new_links={}, links_to_remove={}, writer_name=None):
+		if self.is_published:
+			LOGGER.info('Info: modifying a Message after sending has no global effects')
+
+	def _modify_payload(self, is_delta=True, new_items={}, keys_to_remove=[], writer_name=None):
+		if self.is_published:
+			LOGGER.info('Info: modifying a Message after sending has no global effects')
+
+	def _increase_revision_number(self):
+		self._revision += 1
+
+	def _internal_commit(self, writer_name=None):
+		if self.is_published:
+			LOGGER.info('Info: committing to a Message after sending has no global effects')
+
+	def commit(self):
+		return self._internal_commit()
+
+	def _get_payload(self):
+		return self._payload
+	def _set_payload(self, new_pl, writer_name=None):
+		if self.is_published:
+			LOGGER.info('Info: modifying a Message after sending has no global effects')
+		else:
+			if self._retracted:
+				raise ipaaca.exception.IURetractedError(self)
+			if self._committed:
+				raise ipaaca.exception.IUCommittedError(self)
+			with self.revision_lock:
+				self._increase_revision_number()
+				self._payload = ipaaca.payload.Payload(
+						iu=self,
+						writer_name=None if self.buffer is None else (self.buffer.unique_name if writer_name is None else writer_name),
+						new_payload=new_pl)
+	payload = property(
+			fget=_get_payload,
+			fset=_set_payload,
+			doc='Payload dictionary of this IU.')
+
+	def _get_is_published(self):
+		return self.buffer is not None
+	is_published = property(
+			fget=_get_is_published,
+			doc='Flag indicating whether this IU has been published or not.')
+
+	def _set_buffer(self, buffer):
+		if self._buffer is not None:
+			raise ipaaca.exception.IpaacaError('The IU is already in a buffer, cannot move it.')
+		self._buffer = buffer
+		self.owner_name = buffer.unique_name
+		self._payload.owner_name = buffer.unique_name
+	buffer = property(
+			fget=IUInterface._get_buffer,
+			fset=_set_buffer,
+			doc='Buffer this IU is held in.')
+
+	def _set_uid(self, uid):
+		if self._uid is not None:
+			raise AttributeError('The uid of IU ' + self.uid + ' has already been set, cannot change it.')
+		self._uid = uid
+	uid = property(
+			fget=IUInterface._get_uid,
+			fset=_set_uid,
+			doc='Unique ID of the IU.')
+
+
+class RemoteMessage(IUInterface):
+
+	"""A remote IU with access mode 'MESSAGE'."""
+
+	def __init__(self, uid, revision, read_only, owner_name, category, payload_type, committed, payload, links):
+		super(RemoteMessage, self).__init__(uid=uid, access_mode=IUAccessMode.PUSH, read_only=read_only, payload_type=payload_type)
+		self._revision = revision
+		self._category = category
+		self.owner_name = owner_name
+		self._committed = committed
+		# NOTE Since the payload is an already-existant Payload which we didn't modify ourselves,
+		#  don't try to invoke any modification checks or network updates ourselves either.
+		#  We are just receiving it here and applying the new data.
+		self._payload = ipaaca.payload.Payload(iu=self, new_payload=payload, omit_init_update_message=True)
+		self._links = links
+
+	def _modify_links(self, is_delta=False, new_links={}, links_to_remove={}, writer_name=None):
+		LOGGER.info('Info: modifying a RemoteMessage only has local effects')
+
+	def _modify_payload(self, is_delta=True, new_items={}, keys_to_remove=[], writer_name=None):
+		LOGGER.info('Info: modifying a RemoteMessage only has local effects')
+
+	def commit(self):
+		LOGGER.info('Info: committing to a RemoteMessage only has local effects')
+
+	def _get_payload(self):
+		return self._payload
+	def _set_payload(self, new_pl):
+		LOGGER.info('Info: modifying a RemoteMessage only has local effects')
+		self._payload = ipaaca.payload.Payload(iu=self, new_payload=new_pl, omit_init_update_message=True)
+	payload = property(
+			fget=_get_payload,
+			fset=_set_payload,
+			doc='Payload dictionary of the IU.')
+
+	def _apply_link_update(self, update):
+		"""Apply a IULinkUpdate to the IU."""
+		LOGGER.warning('Warning: should never be called: RemoteMessage._apply_link_update')
+		self._revision = update.revision
+		if update.is_delta:
+			self._add_and_remove_links(add=update.new_links, remove=update.links_to_remove)
+		else:
+			self._replace_links(links=update.new_links)
+
+	def _apply_update(self, update):
+		"""Apply a IUPayloadUpdate to the IU."""
+		LOGGER.warning('Warning: should never be called: RemoteMessage._apply_update')
+		self._revision = update.revision
+		if update.is_delta:
+			for k in update.keys_to_remove: self.payload._remotely_enforced_delitem(k)
+			for k, v in update.new_items.items(): self.payload._remotely_enforced_setitem(k, v)
+		else:
+			# NOTE Please read the comment in the constructor
+			self._payload = ipaaca.payload.Payload(iu=self, new_payload=update.new_items, omit_init_update_message=True)
+
+	def _apply_commission(self):
+		"""Apply commission to the IU"""
+		LOGGER.warning('Warning: should never be called: RemoteMessage._apply_commission')
+		self._committed = True
+
+	def _apply_retraction(self):
+		"""Apply retraction to the IU"""
+		LOGGER.warning('Warning: should never be called: RemoteMessage._apply_retraction')
+		self._retracted = True
+
+
+class RemotePushIU(IUInterface):
+
+	"""A remote IU with access mode 'PUSH'."""
+
+	def __init__(self, uid, revision, read_only, owner_name, category, payload_type, committed, payload, links):
+		super(RemotePushIU, self).__init__(uid=uid, access_mode=IUAccessMode.PUSH, read_only=read_only, payload_type=payload_type)
+		self._revision = revision
+		self._category = category
+		self.owner_name = owner_name
+		self._committed = committed
+		# NOTE Since the payload is an already-existant Payload which we didn't modify ourselves,
+		#  don't try to invoke any modification checks or network updates ourselves either.
+		#  We are just receiving it here and applying the new data.
+		self._payload = ipaaca.payload.Payload(iu=self, new_payload=payload, omit_init_update_message=True)
+		self._links = links
+
+	def _modify_links(self, is_delta=False, new_links={}, links_to_remove={}, writer_name=None):
+		"""Modify the links: add or remove item from this payload remotely and send update."""
+		if self._retracted:
+			raise ipaaca.exception.IURetractedError(self)
+		if self._committed:
+			raise ipaaca.exception.IUCommittedError(self)
+		if self._read_only:
+			raise ipaaca.exception.IUReadOnlyError(self)
+		requested_update = ipaaca.converter.IULinkUpdate(
+				uid=self.uid,
+				revision=self.revision,
+				is_delta=is_delta,
+				writer_name=self.buffer.unique_name,
+				new_links=new_links,
+				links_to_remove=links_to_remove)
+		remote_server = self.buffer._get_remote_server(self)
+		new_revision = remote_server.updateLinks(requested_update)
+		if new_revision == 0:
+			raise ipaaca.exception.IUUpdateFailedError(self)
+		else:
+			self._revision = new_revision
+
+	def _modify_payload(self, is_delta=True, new_items={}, keys_to_remove=[], writer_name=None):
+		"""Modify the payload: add or remove item from this payload remotely and send update."""
+		if self._retracted:
+			raise ipaaca.exception.IURetractedError(self)
+		if self._committed:
+			raise ipaaca.exception.IUCommittedError(self)
+		if self._read_only:
+			raise ipaaca.exception.IUReadOnlyError(self)
+		requested_update = ipaaca.converter.IUPayloadUpdate(
+				uid=self.uid,
+				revision=self.revision,
+				payload_type=self.payload_type,
+				is_delta=is_delta,
+				writer_name=self.buffer.unique_name,
+				new_items=new_items,
+				keys_to_remove=keys_to_remove)
+		remote_server = self.buffer._get_remote_server(self)
+		new_revision = remote_server.updatePayload(requested_update)
+		if new_revision == 0:
+			raise ipaaca.exception.IUUpdateFailedError(self)
+		else:
+			self._revision = new_revision
+
+	def commit(self):
+		"""Commit to this IU."""
+		if self._committed:
+			return
+		if self._retracted:
+			raise ipaaca.exception.IURetractedError(self)
+		if self._read_only:
+			raise ipaaca.exception.IUReadOnlyError(self)
+		commission_request = ipaaca.ipaaca_pb2.IUCommission()
+		commission_request.uid = self.uid
+		commission_request.revision = self.revision
+		commission_request.writer_name = self.buffer.unique_name
+		remote_server = self.buffer._get_remote_server(self)
+		new_revision = remote_server.commit(commission_request)
+		if new_revision == 0:
+			raise ipaaca.exception.IUUpdateFailedError(self)
+		else:
+			self._revision = new_revision
+			self._committed = True
+
+	def _get_payload(self):
+		return self._payload
+	def _set_payload(self, new_pl):
+		if self._retracted:
+			raise ipaaca.exception.IURetractedError(self)
+		if self._committed:
+			raise ipaaca.exception.IUCommittedError(self)
+		if self._read_only:
+			raise ipaaca.exception.IUReadOnlyError(self)
+		requested_update = ipaaca.converter.IUPayloadUpdate(
+				uid=self.uid,
+				revision=self.revision,
+				payload_type=self.payload_type,
+				is_delta=False,
+				writer_name=self.buffer.unique_name,
+				new_items=new_pl,
+				keys_to_remove=[])
+		remote_server = self.buffer._get_remote_server(self)
+		new_revision = remote_server.updatePayload(requested_update)
+		if new_revision == 0:
+			raise ipaaca.exception.IUUpdateFailedError(self)
+		else:
+			self._revision = new_revision
+			# NOTE Please read the comment in the constructor
+			self._payload = ipaaca.payload.Payload(iu=self, new_payload=new_pl, omit_init_update_message=True)
+	payload = property(
+			fget=_get_payload,
+			fset=_set_payload,
+			doc='Payload dictionary of the IU.')
+
+	def _apply_link_update(self, update):
+		"""Apply a IULinkUpdate to the IU."""
+		self._revision = update.revision
+		if update.is_delta:
+			self._add_and_remove_links(add=update.new_links, remove=update.links_to_remove)
+		else:
+			self._replace_links(links=update.new_links)
+
+	def _apply_update(self, update):
+		"""Apply a IUPayloadUpdate to the IU."""
+		self._revision = update.revision
+		if update.is_delta:
+			for k in update.keys_to_remove: self.payload._remotely_enforced_delitem(k)
+			for k, v in update.new_items.items(): self.payload._remotely_enforced_setitem(k, v)
+		else:
+			# NOTE Please read the comment in the constructor
+			self._payload = ipaaca.payload.Payload(iu=self, new_payload=update.new_items, omit_init_update_message=True)
+
+	def _apply_commission(self):
+		"""Apply commission to the IU"""
+		self._committed = True
+
+	def _apply_retraction(self):
+		"""Apply retraction to the IU"""
+		self._retracted = True

ipaacalib/python/src/ipaaca/misc.py

+# -*- coding: utf-8 -*-
+
+# This file is part of IPAACA, the
+#  "Incremental Processing Architecture
+#   for Artificial Conversational Agents".
+#
+# Copyright (c) 2009-2022 Social Cognitive Systems Group
+#                         CITEC, Bielefeld University
+#
+# http://opensource.cit-ec.de/projects/ipaaca/
+# http://purl.org/net/ipaaca
+#
+# This file may be licensed under the terms of of the
+# GNU Lesser General Public License Version 3 (the ``LGPL''),
+# or (at your option) any later version.
+#
+# Software distributed under the License is distributed
+# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+# express or implied. See the LGPL for the specific language
+# governing rights and limitations.
+#
+# You should have received a copy of the LGPL along with this
+# program. If not, go to http://www.gnu.org/licenses/lgpl.html
+# or write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+#
+# The development of this software was supported by the
+# Excellence Cluster EXC 277 Cognitive Interaction Technology.
+# The Excellence Cluster EXC 277 is a grant of the Deutsche
+# Forschungsgemeinschaft (DFG) in the context of the German
+# Excellence Initiative.
+
+from __future__ import division, print_function
+
+
+import argparse
+import logging
+import sys
+
+import ipaaca.defaults
+
+__all__ = [
+	'enum',
+	'IpaacaArgumentParser',
+]
+
+
+def enum(*sequential, **named):
+	"""Create an enum type.
+
+	Based on suggestion of Alec Thomas on stackoverflow.com:
+	http://stackoverflow.com/questions/36932/
+		whats-the-best-way-to-implement-an-enum-in-python/1695250#1695250
+	"""
+	enums = dict(zip(sequential, range(len(sequential))), **named)
+	enums['_choices'] = enums.keys()
+	enums['_values'] = enums.values()  # RY e.g. see if raw int is valid
+	return type('Enum', (object,), enums)
+
+
+class IpaacaLoggingHandler(logging.Handler):
+	'''A logging handler that prints to stdout.'''
+
+	def __init__(self, prefix='IPAACA', level=logging.NOTSET):
+		logging.Handler.__init__(self, level)
+		self._prefix = prefix
+
+	def emit(self, record):
+		meta = '[%s: %s] ' % (self._prefix, str(record.levelname))
+		msg = str(record.msg.format(record.args))
+		print(meta + msg)
+
+class RSBLoggingHandler(logging.Handler):
+	'''A logging handler that prints to stdout, RSB version.'''
+
+	def __init__(self, prefix='IPAACA', level=logging.NOTSET):
+		logging.Handler.__init__(self, level)
+		self._prefix = prefix
+
+	def emit(self, record):
+		meta = '[%s: %s] ' % (self._prefix, str(record.levelname))
+		try:
+			msg = str(record.msg % record.args)
+		except:
+			msg = str(record.msg) + ' WITH ARGS: ' + str(record.args)
+		print(meta + msg)
+
+
+class GenericNoLoggingHandler(logging.Handler):
+	'''A logging handler that produces no output'''
+	def emit(self, record): pass
+
+
+def get_library_logger():
+	'''Get ipaaca's library-wide logger object.'''
+	return logging.getLogger(ipaaca.defaults.IPAACA_LOGGER_NAME)
+
+
+_IPAACA_LOGGING_HANDLER = IpaacaLoggingHandler('IPAACA')
+_GENERIC_NO_LOG_HANDLER = GenericNoLoggingHandler()
+
+# By default, suppress library logging
+# - for IPAACA
+get_library_logger().addHandler(_GENERIC_NO_LOG_HANDLER)
+# - for RSB
+logging.getLogger('rsb').addHandler(_GENERIC_NO_LOG_HANDLER)
+
+
+def enable_logging(level=None):
+	'''Enable ipaaca's 'library-wide logging.'''
+	ipaaca_logger = get_library_logger()
+	ipaaca_logger.addHandler(_IPAACA_LOGGING_HANDLER)
+	ipaaca_logger.removeHandler(_GENERIC_NO_LOG_HANDLER)
+	ipaaca_logger.setLevel(level=level if level is not None else
+		ipaaca.defaults.IPAACA_DEFAULT_LOGGING_LEVEL)
+
+
+class IpaacaArgumentParser(argparse.ArgumentParser):
+
+	class IpaacaDefaultChannelAction(argparse.Action):
+
+		def __call__(self, parser, namespace, values, option_string=None):
+			ipaaca.defaults.IPAACA_DEFAULT_CHANNEL = values
+
+	class IpaacaPayloadTypeAction(argparse.Action):
+
+		def __call__(self, parser, namespace, values, option_string=None):
+			ipaaca.defaults.IPAACA_DEFAULT_IU_PAYLOAD_TYPE = values
+
+	class IpaacaLoggingLevelAction(argparse.Action):
+
+		def __call__(self, parser, namespace, values, option_string=None):
+			enable_logging(values)
+
+	class IpaacaRSBLoggingLevelAction(argparse.Action):
+
+		def __call__(self, parser, namespace, values, option_string=None):
+			rsb_logger = logging.getLogger('rsb')
+			rsb_logger.addHandler(RSBLoggingHandler('RSB'))
+			rsb_logger.removeHandler(_GENERIC_NO_LOG_HANDLER)
+			rsb_logger.setLevel(level=values)
+
+	class IpaacaRSBHost(argparse.Action):
+
+		def __call__(self, parser, namespace, values, option_string=None):
+			ipaaca.defaults.IPAACA_DEFAULT_RSB_HOST = values
+
+	class IpaacaRSBPort(argparse.Action):
+
+		def __call__(self, parser, namespace, values, option_string=None):
+			ipaaca.defaults.IPAACA_DEFAULT_RSB_PORT = values
+
+	class IpaacaRSBTransport(argparse.Action):
+
+		def __call__(self, parser, namespace, values, option_string=None):
+			ipaaca.defaults.IPAACA_DEFAULT_RSB_TRANSPORT = values
+
+	class IpaacaRSBSocketServer(argparse.Action):
+
+		def __call__(self, parser, namespace, values, option_string=None):
+			ipaaca.defaults.IPAACA_DEFAULT_RSB_SOCKET_SERVER = values
+
+	def __init__(self, prog=None, usage=None, description=None, epilog=None,
+			parents=[], formatter_class=argparse.HelpFormatter,
+			prefix_chars='-', fromfile_prefix_chars=None, 
+			argument_default=None, conflict_handler='error', add_help=True):
+		super(IpaacaArgumentParser, self).__init__(prog=prog, usage=usage,
+			description=description, epilog=epilog, parents=parents,
+			formatter_class=formatter_class, prefix_chars=prefix_chars,
+			fromfile_prefix_chars=fromfile_prefix_chars,
+			argument_default=argument_default, 
+			conflict_handler=conflict_handler, add_help=add_help)
+
+	def _add_ipaaca_lib_arguments(self):
+		# CMD-arguments for ipaaca
+		ipaacalib_group = self.add_argument_group('IPAACA library arguments')
+		ipaacalib_group.add_argument(
+			'--ipaaca-payload-type', 
+			action=self.IpaacaPayloadTypeAction,
+			choices=['JSON', 'STR'], # one of ipaaca.iu.IUPayloadTypes
+			dest='_ipaaca_payload_type_',
+			default='JSON',
+			help="specify payload type (default: 'JSON')")
+		ipaacalib_group.add_argument(
+			'--ipaaca-default-channel', 
+			action=self.IpaacaDefaultChannelAction,
+			default='default',
+			metavar='NAME',
+			dest='_ipaaca_default_channel_',
+			help="specify default IPAACA channel name (default: 'default')")
+		ipaacalib_group.add_argument(
+			'--ipaaca-enable-logging', 
+			action=self.IpaacaLoggingLevelAction,
+			choices=['CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG'],
+			dest='_ipaaca_logging_level_',
+			help="enable IPAACA logging with threshold")
+		# CMD-arguments for rsb
+		rsblib_group = self.add_argument_group('RSB library arguments')
+		rsblib_group.add_argument(
+			'--rsb-enable-logging', 
+			action=self.IpaacaRSBLoggingLevelAction,
+			choices=['CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG'],
+			dest='_ipaaca_rsb_enable_logging_',
+			help="enable RSB logging with threshold")
+		rsblib_group.add_argument(
+			'--rsb-host', 
+			action=self.IpaacaRSBHost,
+			default=None,
+			dest='_ipaaca_rsb_set_host_',
+			metavar='HOST',
+			help="set RSB host")
+		rsblib_group.add_argument(
+			'--rsb-port', 
+			action=self.IpaacaRSBPort,
+			default=None,
+			dest='_ipaaca_rsb_set_port_',
+			metavar='PORT',
+			help="set RSB port")
+		rsblib_group.add_argument(
+			'--rsb-transport', 
+			action=self.IpaacaRSBTransport,
+			default=None,
+			dest='_ipaaca_rsb_set_transport_',
+			choices=['spread', 'socket'],
+			metavar='TRANSPORT',
+			help="set RSB transport")
+		rsblib_group.add_argument(
+			'--rsb-socket-server', 
+			action=self.IpaacaRSBSocketServer,
+			default=None,
+			dest='_ipaaca_rsb_set_socket_server_',
+			choices=['0', '1', 'auto'],
+			metavar='server',
+			help="act as server (only when --rsb-transport=socket)")
+
+	def parse_args(self, args=None, namespace=None):
+		self._add_ipaaca_lib_arguments() # Add ipaaca-args just before parsing
+		result = super(IpaacaArgumentParser, self).parse_args(args, namespace)
+		# Delete ipaaca specific arguments (beginning with '_ipaaca' and 
+		# ending with an underscore) from the resulting Namespace object.
+		for item in dir(result):
+			if item.startswith('_ipaaca') and item.endswith('_'):
+				delattr(result, item)
+		return result

ipaacalib/python/src/ipaaca/payload.py

+# -*- coding: utf-8 -*-
+
+# This file is part of IPAACA, the
+#  "Incremental Processing Architecture
+#   for Artificial Conversational Agents".
+#
+# Copyright (c) 2009-2022 Social Cognitive Systems Group
+#                         CITEC, Bielefeld University
+#
+# http://opensource.cit-ec.de/projects/ipaaca/
+# http://purl.org/net/ipaaca
+#
+# This file may be licensed under the terms of of the
+# GNU Lesser General Public License Version 3 (the ``LGPL''),
+# or (at your option) any later version.
+#
+# Software distributed under the License is distributed
+# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+# express or implied. See the LGPL for the specific language
+# governing rights and limitations.
+#
+# You should have received a copy of the LGPL along with this
+# program. If not, go to http://www.gnu.org/licenses/lgpl.html
+# or write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+#
+# The development of this software was supported by the
+# Excellence Cluster EXC 277 Cognitive Interaction Technology.
+# The Excellence Cluster EXC 277 is a grant of the Deutsche
+# Forschungsgemeinschaft (DFG) in the context of the German
+# Excellence Initiative.
+
+
+
+import threading
+import time
+
+import ipaaca.exception
+
+
+__all__ = [
+	'Payload',
+	'PayloadItemDictProxy',
+	'PayloadItemListProxy',
+]
+
+
+_DEFAULT_PAYLOAD_UPDATE_TIMEOUT = 0.1
+
+
+class Payload(dict):
+
+	def __init__(self, iu, writer_name=None, new_payload=None, omit_init_update_message=False, update_timeout=_DEFAULT_PAYLOAD_UPDATE_TIMEOUT):
+		self.iu = iu
+		_pl = {}
+		for k, v in ({} if new_payload is None else new_payload).items():
+			_pl[str(k) if type(k) == str else k] = str(v) if type(v) == str else v
+		# NOTE omit_init_update_message is necessary to prevent checking for
+		#   exceptions and sending updates in the case where we just receive
+		#   a whole new payload from the remote side and overwrite it locally.
+		for k, v in _pl.items():
+			dict.__setitem__(self, k, v)
+		if (not omit_init_update_message) and (self.iu.buffer is not None):
+			self.iu._modify_payload(
+				is_delta=False,
+				new_items=_pl,
+				keys_to_remove=[],
+				writer_name=writer_name)
+		self._update_on_every_change = True
+		self._update_timeout = update_timeout
+		self._collected_modifications = {}
+		self._collected_removals = []
+		self._batch_update_writer_name = None  # name of remote buffer or None
+		self._batch_update_lock = threading.RLock()
+		self._batch_update_cond = threading.Condition(threading.RLock())
+
+	def __getitem__(self, k):
+		value = dict.__getitem__(self, k)
+		if isinstance(value, dict):
+			return PayloadItemDictProxy(value, self, k)
+		elif isinstance(value, list):
+			return PayloadItemListProxy(value, self, k)
+		else:
+			return value
+
+	def __setitem__(self, k, v, writer_name=None):
+		with self._batch_update_lock:
+			k = str(k) if type(k) == str else k
+			v = str(v) if type(v) == str else v
+			if self._update_on_every_change:
+				self.iu._modify_payload(
+					is_delta=True,
+					new_items={k:v},
+					keys_to_remove=[],
+					writer_name=writer_name)
+			else: # Collect additions/modifications
+				self._batch_update_writer_name = writer_name
+				self._collected_modifications[k] = v
+				# revoke deletion of item since a new version has been added
+				self._collected_removals = [i for i in self._collected_removals if i!=k]
+			return dict.__setitem__(self, k, v)
+
+	def __delitem__(self, k, writer_name=None):
+		with self._batch_update_lock:
+			k = str(k) if type(k) == str else k
+			if self._update_on_every_change:
+				self.iu._modify_payload(
+					is_delta=True,
+					new_items={},
+					keys_to_remove=[k],
+					writer_name=writer_name)
+			else: # Collect additions/modifications
+				self._batch_update_writer_name = writer_name
+				self._collected_removals.append(k)
+				# revoke older update of item since more recent changes take precedence
+				if k in self._collected_modifications: del self._collected_modifications[k]
+			return dict.__delitem__(self, k)
+
+	# Context-manager based batch updates, not thread-safe (on remote updates)
+	def __enter__(self):
+		self._wait_batch_update_lock(self._update_timeout)
+		self._update_on_every_change = False
+
+	# Context-manager based batch updates, not thread-safe (on remote updates)
+	def __exit__(self, type, value, traceback):
+		self.iu._modify_payload(
+			is_delta=True,
+			new_items=self._collected_modifications,
+			keys_to_remove=self._collected_removals,
+			writer_name=self._batch_update_writer_name)
+		self._collected_modifications = {}
+		self._collected_removals = []
+		self._update_on_every_change = True
+		self._batch_update_writer_name = None
+		self._batch_update_lock.release()
+
+	def merge(self, payload, writer_name=None):
+		with self._batch_update_lock:
+			for k, v in payload.items():
+				k = str(k) if type(k) == str else k
+				v = str(v) if type(v) == str else v
+			self.iu._modify_payload(
+				is_delta=True,
+				new_items=payload,
+				keys_to_remove=[],
+				writer_name=writer_name)
+			return dict.update(self, payload) # batch update
+
+	def _remotely_enforced_setitem(self, k, v):
+		"""Sets an item when requested remotely."""
+		dict.__setitem__(self, k, v)
+
+	def _remotely_enforced_delitem(self, k):
+		"""Deletes an item when requested remotely."""
+		if k in self: dict.__delitem__(self, k)
+
+	def _wait_batch_update_lock(self, timeout):
+		# wait lock with time-out http://stackoverflow.com/a/8393033
+		with self._batch_update_cond:
+			current_time = start_time = time.time()
+			while current_time < start_time + timeout:
+				if self._batch_update_lock.acquire(False):
+					return True
+				else:
+					self._batch_update_cond.wait(timeout - current_time + start_time)
+					current_time = time.time()
+		raise ipaaca.exception.IUPayloadLockTimeoutError(self.iu)
+
+
+class PayloadItemProxy(object):
+
+	def __init__(self, content, payload, identifier_in_payload):
+		self.payload = payload
+		self.content = content
+		self.identifier_in_payload = identifier_in_payload
+
+	def _notify_payload(self):
+		try:
+			self.payload[self.identifier_in_payload] = self.content
+		except ipaaca.exception.IUUpdateFailedError as e:
+			# IU update failed. Use the ResendRequest mechanism
+			# to replace the altered RemotePushIU with the unchanged
+			# payload from its OutputBuffer.''
+			iu = self.payload.iu
+			iu.buffer._request_remote_resend(iu)
+			raise e # re-raise IUUpdateFailedError from aboves
+
+	def _create_proxy(self, obj, identifier_in_payload):
+		if isinstance(obj, dict):
+			return PayloadItemDictProxy(obj, self.payload, identifier_in_payload)
+		elif isinstance(obj, list):
+			return PayloadItemListProxy(obj, self.payload, identifier_in_payload)
+		else:
+			return obj
+
+	def __setitem__(self, k, v):
+		self.content.__setitem__(k,v)
+		self._notify_payload()
+
+	def __getitem__(self, k):
+		item = self.content.__getitem__(k)
+		return self._create_proxy(item, k)
+
+	def __delitem__(self, k):
+		self.content.__delitem__(k)
+		self._notify_payload()
+
+
+class PayloadItemDictProxy(PayloadItemProxy, dict):
+
+	def __init__(self, content, payload, identifier_in_payload):
+		dict.__init__(self, content)
+		PayloadItemProxy.__init__(self, content, payload, identifier_in_payload)
+
+	def clear(self):
+		self.content.clear()
+		self._notify_payload()
+
+	def get(self, key, default=None):
+		value = self.content.get(key, default)
+		return self._create_proxy(value, key)
+
+# py3port: were these used at all? 
+#	def items(self):
+#		return [(key, value) for key, value in self.items()]
+
+# py3port: was iteritems
+	def items(self):
+		for key, value in self.content.items():
+			yield key, self._create_proxy(value, key)
+
+# py3port: were these used at all? 
+#	def values(self):
+#		return [value for value in self.values()]
+
+# py3port: was itervalues 
+	def values(self):
+		for key, value in self.content.items():
+			yield self._create_proxy(value, key)
+
+	def pop(self, key, *args):
+		x = self.content.pop(key, *args)
+		self._notify_payload()
+		return x
+
+	def popitem(self):
+		x = self.content.popitem()
+		self._notify_payload()
+		return x
+
+	def setdefault(self, key, default=None):
+		notification_necessary = not key in self.content
+		x = self.content.setdefault(key, default)
+		if notification_necessary:
+			self._notify_payload()
+		return x
+
+	def update(self, *args, **kwargs):
+		self.content.update(*args, **kwargs)
+		self._notify_payload()
+
+
+class PayloadItemListProxy(PayloadItemProxy, list):
+
+	def __init__(self, content, payload, identifier_in_payload):
+		list.__init__(self, content)
+		PayloadItemProxy.__init__(self, content, payload, identifier_in_payload)
+
+	def __iter__(self):
+		for index, item in enumerate(self.content):
+			yield self._create_proxy(item, index)
+
+	def append(self, x):
+		self.content.append(x)
+		self._notify_payload()
+
+	def extend(self, l):
+		self.content.extend(l)
+		self._notify_payload()
+
+	def insert(self, i, x):
+		self.content.insert(i, x)
+		self._notify_payload()
+
+	def remove(self, x):
+		self.content.remove(x)
+		self._notify_payload()
+
+	def pop(self, *args, **kwargs):
+		x = self.content.pop(*args, **kwargs)
+		self._notify_payload()
+		return x
+
+	def sort(self, cmp=None, key=None, reverse=False):
+		self.content.sort(cmp, key, reverse)
+		self._notify_payload()
+
+	def reverse(self):
+		self.content.reverse()
+		self._notify_payload()

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

+# -*- coding: utf-8 -*-
+
+# This file is part of IPAACA, the
+#  "Incremental Processing Architecture
+#   for Artificial Conversational Agents".
+#
+# Copyright (c) 2009-2022 Social Cognitive Systems Group
+#                         CITEC, Bielefeld University
+#
+# http://opensource.cit-ec.de/projects/ipaaca/
+# http://purl.org/net/ipaaca
+#
+# This file may be licensed under the terms of of the
+# GNU Lesser General Public License Version 3 (the ``LGPL''),
+# or (at your option) any later version.
+#
+# Software distributed under the License is distributed
+# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+# express or implied. See the LGPL for the specific language
+# governing rights and limitations.
+#
+# You should have received a copy of the LGPL along with this
+# program. If not, go to http://www.gnu.org/licenses/lgpl.html
+# or write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+#
+# The development of this software was supported by the
+# Excellence Cluster EXC 277 Cognitive Interaction Technology.
+# The Excellence Cluster EXC 277 is a grant of the Deutsche
+# Forschungsgemeinschaft (DFG) in the context of the German
+# Excellence Initiative.
+
+from __future__ import division, print_function
+
+from .notifier import ComponentNotifier

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

+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+# This file is part of IPAACA, the
+#  "Incremental Processing Architecture
+#   for Artificial Conversational Agents".
+#
+# Copyright (c) 2009-2022 Social Cognitive Systems Group
+#                         CITEC, Bielefeld University
+#
+# http://opensource.cit-ec.de/projects/ipaaca/
+# http://purl.org/net/ipaaca
+#
+# This file may be licensed under the terms of of the
+# GNU Lesser General Public License Version 3 (the ``LGPL''),
+# or (at your option) any later version.
+#
+# Software distributed under the License is distributed
+# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+# express or implied. See the LGPL for the specific language
+# governing rights and limitations.
+#
+# You should have received a copy of the LGPL along with this
+# program. If not, go to http://www.gnu.org/licenses/lgpl.html
+# or write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+#
+# The development of this software was supported by the
+# Excellence Cluster EXC 277 Cognitive Interaction Technology.
+# The Excellence Cluster EXC 277 is a grant of the Deutsche
+# Forschungsgemeinschaft (DFG) in the context of the German
+# Excellence Initiative.
+
+from __future__ import division, print_function
+
+import datetime
+import subprocess
+import sys
+import threading
+import time
+import traceback
+import uuid
+
+import ipaaca
+import ipaaca.misc
+import six
+
+
+__all__ = [
+	'logger_send_ipaaca_logs',
+	'logger_set_log_filename',
+	'logger_set_module_name',
+	'logger_set_log_level',
+	'LOG_DEBUG',
+	'LOG_INFO',
+	'LOG_WARN',
+	'LOG_WARNING',
+	'LOG_ERROR',
+]
+
+
+LogLevel = ipaaca.misc.enum(
+	DEBUG = 0,
+	INFO = 1,
+	WARN = 2,
+	ERROR = 3,
+	SILENT = 4,
+)
+LOG_LEVEL_FROM_STRING_DICT = {
+	'DEBUG': LogLevel.DEBUG,
+	'INFO': LogLevel.INFO,
+	'WARN': LogLevel.WARN,
+	'WARNING': LogLevel.WARN,
+	'ERROR': LogLevel.ERROR,
+	'NONE': LogLevel.SILENT,
+	'SILENT': LogLevel.SILENT,
+}
+CURRENT_LOG_LEVEL = LogLevel.DEBUG
+
+LOGGER_LOCK = threading.RLock()
+
+MODULE_NAME = sys.argv[0]
+
+SEND_IPAACA_LOGS = True
+
+OUTPUT_BUFFER = None
+
+STANDARD_LOG_FILE_EXTENSION = '.log'
+
+LOG_MODES = ['append', 'timestamp', 'overwrite']
+
+
+def logger_set_log_filename(filename, existing=None):
+	global OUTPUT_BUFFER
+	if existing is not None and not existing in LOG_MODES:
+		raise Exception('Invalid log mode {mode} given. '
+			'Valid options are {options}.'.format(
+				mode=existing,
+				options=', '.join(LOG_MODES)))
+	with LOGGER_LOCK:
+		if OUTPUT_BUFFER is None:
+			OUTPUT_BUFFER = ipaaca.OutputBuffer('LogSender')
+		msg = ipaaca.Message('logcontrol')
+		msg.payload = {
+			'cmd': 'open_log_file',
+			'filename': filename}
+		if existing is not None:
+			msg.payload['existing'] = existing
+		OUTPUT_BUFFER.add(msg)
+
+
+def logger_set_module_name(name):
+	global MODULE_NAME
+	with LOGGER_LOCK:
+		MODULE_NAME = name
+
+
+def logger_send_ipaaca_logs(flag=True):
+	global SEND_IPAACA_LOGS
+	with LOGGER_LOCK:
+		SEND_IPAACA_LOGS = flag
+
+def logger_set_log_level(level=LogLevel.DEBUG):
+	global CURRENT_LOG_LEVEL
+	with LOGGER_LOCK:
+		if level in LogLevel._values:
+			CURRENT_LOG_LEVEL = level
+		elif isinstance(level, six.string_types) and level.upper() in LOG_LEVEL_FROM_STRING_DICT:
+			CURRENT_LOG_LEVEL = LOG_LEVEL_FROM_STRING_DICT[level.upper()]
+		else:
+			pass # leave previous setting untouched
+
+def LOG_IPAACA(lvl, text, now=0.0, fn='???', thread='???'):
+	global OUTPUT_BUFFER
+	uid = str(uuid.uuid4())[0:8]
+	with LOGGER_LOCK:
+		if OUTPUT_BUFFER is None:
+			OUTPUT_BUFFER = ipaaca.OutputBuffer('LogSender')
+		msg = ipaaca.Message('log')
+		msg.payload = {
+				'module': MODULE_NAME,
+				'function': fn,
+				'level': lvl,
+				'time':' %.3f'%now,
+				'thread': thread,
+				'uuid': uid,
+				'text': text,}
+		try:
+			OUTPUT_BUFFER.add(msg)
+		except Exception as e:
+			LOG_ERROR('Caught an exception while logging via ipaaca. '
+				+ ' str(e); ' 
+				+ traceback.format_exc())
+
+
+def LOG_CONSOLE(lvlstr, msg, fn_markup='', msg_markup='', now=0.0, fn='???', thread='???'):
+	if isinstance(msg, six.string_types):
+		lines = msg.split('\n')
+	else:
+		lines = [msg]
+	for line in lines:
+		text = lvlstr+' '+thread+' '+fn_markup+fn+'\033[m'+' '+msg_markup+str(line)+'\033[m'
+		print(text)
+		fn = ' '*len(fn)
+
+
+def LOG_ERROR(msg, now=None):
+	if CURRENT_LOG_LEVEL > LogLevel.ERROR: return
+	now = time.time() if now is None else now
+	f = sys._getframe(1)
+	classprefix = (f.f_locals['self'].__class__.__name__+'.') if 'self' in f.f_locals else ''
+	fn = classprefix + f.f_code.co_name
+	thread = threading.current_thread().getName()
+	with LOGGER_LOCK:
+		if SEND_IPAACA_LOGS: LOG_IPAACA('ERROR', msg, now=now, fn=fn, thread=thread)
+		LOG_CONSOLE('\033[38;5;9;1;4m[ERROR]\033[m', msg, fn_markup='\033[38;5;203m', msg_markup='\033[38;5;9;1;4m', now=now, fn=fn, thread=thread)
+
+
+def LOG_WARN(msg, now=None):
+	if CURRENT_LOG_LEVEL > LogLevel.WARN: return
+	now = time.time() if now is None else now
+	f = sys._getframe(1)
+	classprefix = (f.f_locals['self'].__class__.__name__+'.') if 'self' in f.f_locals else ''
+	fn = classprefix + f.f_code.co_name
+	thread = threading.current_thread().getName()
+	with LOGGER_LOCK:
+		if SEND_IPAACA_LOGS: LOG_IPAACA('WARN', msg, now=now, fn=fn, thread=thread)
+		LOG_CONSOLE('\033[38;5;208;1m[WARN]\033[m ', msg, fn_markup='\033[38;5;214m', msg_markup='\033[38;5;208;1m', now=now, fn=fn, thread=thread)
+
+
+LOG_WARNING = LOG_WARN
+
+
+def LOG_INFO(msg, now=None):
+	if CURRENT_LOG_LEVEL > LogLevel.INFO: return
+	now = time.time() if now is None else now
+	f = sys._getframe(1)
+	classprefix = (f.f_locals['self'].__class__.__name__+'.') if 'self' in f.f_locals else ''
+	fn = classprefix + f.f_code.co_name
+	thread = threading.current_thread().getName()
+	with LOGGER_LOCK:
+		if SEND_IPAACA_LOGS: LOG_IPAACA('INFO', msg, now=now, fn=fn, thread=thread)
+		LOG_CONSOLE('[INFO] ', msg, now=now, fn=fn, thread=thread)
+
+
+def LOG_DEBUG(msg, now=None):
+	if CURRENT_LOG_LEVEL > LogLevel.DEBUG: return
+	now = time.time() if now is None else now
+	f = sys._getframe(1)
+	classprefix = (f.f_locals['self'].__class__.__name__+'.') if 'self' in f.f_locals else ''
+	fn = classprefix + f.f_code.co_name
+	thread = threading.current_thread().getName()
+	with LOGGER_LOCK:
+		if SEND_IPAACA_LOGS: LOG_IPAACA('DEBUG', msg, now=now, fn=fn, thread=thread)
+		LOG_CONSOLE('\033[2m[DEBUG]\033[m', msg, fn_markup='\033[38;5;144m', msg_markup='\033[38;5;248m', now=now, fn=fn, thread=thread)
+
+
+class LoggerComponent(object):
+
+	def __init__(self, filename, log_mode='append'):
+		self.ib = ipaaca.InputBuffer('Logger', ['log', 'logcontrol'])
+		self.ib.register_handler(self._logger_handle_iu_event)
+		self.logfile = None
+		self.log_mode = log_mode
+		self.open_logfile(filename)
+		if self.logfile is None:
+			print('Logging to console only ...')
+		print('Press Ctrl-C at any time to terminate the logger.')
+
+	def open_logfile(self, filename):
+		with LOGGER_LOCK:
+			if filename is None or filename.strip() == '':
+				print('No log file name specified, not opening one.')
+				self.close_logfile()
+			else:
+				new_logfile = None
+				try:
+					# create dir first
+					ret = subprocess.call(
+							'mkdir -p `dirname ' + filename + '`',
+							shell=True)
+					# proceed with dir+file
+					if self.log_mode == 'timestamp':
+						t = datetime.datetime.now().strftime('%Y-%m-%d-%H%M%S')
+						if filename.endswith(STANDARD_LOG_FILE_EXTENSION):
+							# insert timestamp between filename and extension
+							# (only for standard extension)
+							filename = filename.replace(
+									STANDARD_LOG_FILE_EXTENSION,
+									'.' + t + STANDARD_LOG_FILE_EXTENSION)
+						else: # prepend timestamp
+							filename = t + '_' + filename
+					append_if_exist = not (self.log_mode == 'overwrite' or
+							self.log_mode == 'timestamp')
+					new_logfile = open(filename, 'a' if append_if_exist else 'w')
+					if self.logfile is not None:
+						text = u'Will now continue logging in log file ' + str(filename)
+						uid = str(uuid.uuid4())[0:8]
+						tim = time.time()
+						record = {
+							'uuid': uid, 
+							'time': tim,
+							'level': u'INFO',
+							'text': text,
+							'module': u'logger',
+							'function': u'LoggerComponent.open_logfile', 
+							'thread': '-', 
+							'logreceivedtime': tim}
+						self.logfile.write(str(record)+'\n')
+						self.logfile.close()
+					self.logfile = new_logfile
+					print('Logging to console and {filename} ...'.format(filename=filename))
+				except Exception as e:
+					print('Failed to open logfile {filename} for writing! Keeping previous configuration'.format(filename=filename))
+
+	def close_logfile(self):
+		if self.logfile is not None:
+			text = u'Closing of log file requested.'
+			uid = str(uuid.uuid4())[0:8]
+			tim = str(time.time())
+			record = {
+				'uuid': uid,
+				'time': tim,
+				'level': u'INFO',
+				'text': text,
+				'module': u'logger',
+				'function': u'LoggerComponent.close_logfile', 
+				'thread': u'-',
+				'logreceivedtime': tim}
+			self.logfile.write(str(record)+'\n')
+			self.logfile.close()
+			print('Closed current log file.')
+			self.logfile = None
+
+	def _logger_handle_iu_event(self, iu, event_type, own):
+		received_time = "%.3f" % time.time()
+		with LOGGER_LOCK:
+			try:
+				if iu.category == 'log':
+					pl = iu.payload
+					message = pl['text'] if 'text' in pl else '(No message.)'
+					uid = '????????' if 'uuid' not in pl else pl['uuid']
+					tim = '???' if 'time' not in pl else pl['time']
+					module = '???' if 'module' not in pl else pl['module']
+					function = '???' if 'function' not in pl else pl['function']
+					thread = '???' if 'thread' not in pl else pl['thread']
+					level = 'INFO' if 'level' not in pl else pl['level']
+					# dump to console
+					if level=='WARN':
+						level='WARNING'
+					if level not in ['DEBUG','INFO','WARNING','ERROR']:
+						level = 'INFO'
+					try:
+						print('%s %-8s {%s} {%s} {%s} %s'%(tim, ('['+level+']'), thread, module, function, message))
+					except:
+						print('Failed to format a string for printing!')
+					if self.logfile is not None:
+						try:
+							record = {
+								'uuid': uid, 
+								'time': tim, 
+								'level': level, 
+								'text': message,
+								'module': module,
+								'function': function, 
+								'thread': thread,
+								'logreceivedtime': received_time}
+							self.logfile.write(str(record) + '\n')
+						except:
+							print('Failed to write to logfile!')
+				elif iu.category == 'logcontrol':
+					cmd = iu.payload['cmd'] if 'cmd' in iu.payload else 'undef'
+					if cmd == 'open_log_file':
+						filename = iu.payload['filename'] if 'filename' in iu.payload else ''
+						if 'existing' in iu.payload:
+							log_mode_ = iu.payload['existing'].lower()
+							if log_mode_ not in LOG_MODES:
+								LOG_WARN(u'Value of "existing" should be "append", "timestamp", or "overwrite", continuing with mode {mode}'.format(mode=self.log_mode))
+							else:
+								self.log_mode = log_mode_
+						self.open_logfile(filename)
+					elif cmd == 'close_log_file':
+						self.close_logfile()
+					else:
+						LOG_WARN(u'Received unknown logcontrol command: '+str(cmd))
+			except Exception as e:
+				print('Exception while logging!')  # TODO write to file as well?
+				print(u'  '+str(traceback.format_exc()))

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

+# -*- coding: utf-8 -*-
+
+# This file is part of IPAACA, the
+#  "Incremental Processing Architecture
+#   for Artificial Conversational Agents".
+#
+# Copyright (c) 2009-2022 Social Cognitive Systems Group
+#                         CITEC, Bielefeld University
+#
+# http://opensource.cit-ec.de/projects/ipaaca/
+# http://purl.org/net/ipaaca
+#
+# This file may be licensed under the terms of of the
+# GNU Lesser General Public License Version 3 (the ``LGPL''),
+# or (at your option) any later version.
+#
+# Software distributed under the License is distributed
+# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+# express or implied. See the LGPL for the specific language
+# governing rights and limitations.
+#
+# You should have received a copy of the LGPL along with this
+# program. If not, go to http://www.gnu.org/licenses/lgpl.html
+# or write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+#
+# The development of this software was supported by the
+# Excellence Cluster EXC 277 Cognitive Interaction Technology.
+# The Excellence Cluster EXC 277 is a grant of the Deutsche
+# Forschungsgemeinschaft (DFG) in the context of the German
+# Excellence Initiative.
+
+
+from __future__ import division, print_function
+
+import os
+import threading
+
+import ipaaca.buffer
+import ipaaca.iu
+import ipaaca.misc
+import ipaaca.util.timesync
+
+
+__all__ = [
+	'NotificationState',
+	'ComponentError',
+	'ComponentNotifier'
+]
+
+
+NotificationState = ipaaca.misc.enum(
+	NEW = 'new',
+	OLD = 'old',
+	DOWN = 'down'
+)
+
+
+class ComponentError(Exception):
+	def __init__(self, msg):
+		super(ComponentError, self).__init__(msg)
+
+
+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
+		self.component_function = component_function
+		self.send_categories = frozenset(send_categories)
+		self.receive_categories = frozenset(receive_categories)
+		self.in_buffer = in_buffer if in_buffer is not None else ipaaca.buffer.InputBuffer(component_name + 'Notifier')
+		self.out_buffer = out_buffer if out_buffer is not None else ipaaca.buffer.OutputBuffer(component_name + 'Notifier')
+		self.terminated = False
+		self.initialized = False
+		self.notification_handlers = []
+		self.initialize_lock = threading.Lock()
+		self.notification_handler_lock = threading.Lock()
+		self.submit_lock = threading.Lock()
+		# clock sync code, sync slave/master pair will be installed when launched
+		self.timesync_slave = None
+		self.timesync_master = None
+		self.timesync_master_handlers = []
+		self.timesync_slave_handlers = []
+
+	def _submit_notify(self, is_new):
+		with self.submit_lock:
+			notify_iu = ipaaca.iu.Message(ComponentNotifier.NOTIFY_CATEGORY)
+			notify_iu.payload = {
+					ComponentNotifier.NAME: self.component_name,
+					ComponentNotifier.FUNCTION: self.component_function,
+					ComponentNotifier.SEND_CATEGORIES: ",".join(self.send_categories),
+					ComponentNotifier.RECEIVE_CATEGORIES:  ",".join(self.receive_categories),
+					ComponentNotifier.STATE: NotificationState.NEW if is_new else NotificationState.OLD,
+					}
+			self.out_buffer.add(notify_iu)
+
+	def terminate(self):
+		with self.submit_lock:
+			if self.terminated: return
+			self.terminated = True
+			notify_iu = ipaaca.iu.Message(ComponentNotifier.NOTIFY_CATEGORY)
+			notify_iu.payload = {
+					ComponentNotifier.NAME: self.component_name,
+					ComponentNotifier.FUNCTION: self.component_function,
+					ComponentNotifier.SEND_CATEGORIES: ",".join(self.send_categories),
+					ComponentNotifier.RECEIVE_CATEGORIES:  ",".join(self.receive_categories),
+					ComponentNotifier.STATE: NotificationState.DOWN,
+					}
+			self.out_buffer.add(notify_iu)
+
+	def _handle_iu_event(self, iu, event_type, local):
+		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:
+			self.notification_handlers.append(handler)
+
+	def launch_timesync_slave_handlers(self, master, slave, latency, offset):
+		for h in self.timesync_slave_handlers:
+			h(master, slave, latency, offset)
+
+	def launch_timesync_master_handlers(self, master, slave, latency, offset):
+		for h in self.timesync_master_handlers:
+			h(master, slave, latency, offset)
+
+	def add_timesync_slave_handler(self, handler):
+		self.timesync_slave_handlers.append(handler)
+
+	def add_timesync_master_handler(self, handler):
+		self.timesync_master_handlers.append(handler)
+
+	def send_master_timesync(self):
+		#if len(self.timesync_master_handlers)==0:
+		#	print('Warning: Sending a master timesync without a registered result callback.')
+		self.timesync_master.send_master_timesync()
+
+	def initialize(self):
+		with self.initialize_lock:
+			if self.terminated:
+				raise ComponentError('Attempted to reinitialize component '+component_name+' after termination')
+			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, 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

+# -*- coding: utf-8 -*-
+
+# This file is part of IPAACA, the
+#  "Incremental Processing Architecture
+#   for Artificial Conversational Agents".
+#
+# Copyright (c) 2009-2022 Social Cognitive Systems Group
+#                         CITEC, Bielefeld University
+#
+# http://opensource.cit-ec.de/projects/ipaaca/
+# http://purl.org/net/ipaaca
+#
+# This file may be licensed under the terms of of the
+# GNU Lesser General Public License Version 3 (the ``LGPL''),
+# or (at your option) any later version.
+#
+# Software distributed under the License is distributed
+# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+# express or implied. See the LGPL for the specific language
+# governing rights and limitations.
+#
+# You should have received a copy of the LGPL along with this
+# program. If not, go to http://www.gnu.org/licenses/lgpl.html
+# or write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+#
+# The development of this software was supported by the
+# Excellence Cluster EXC 277 Cognitive Interaction Technology.
+# The Excellence Cluster EXC 277 is a grant of the Deutsche
+# Forschungsgemeinschaft (DFG) in the context of the German
+# Excellence Initiative.
+
+from __future__ import division, print_function
+
+import threading
+import time
+
+import ipaaca.buffer
+import ipaaca.iu
+
+
+class TimesyncMaster(object):
+	def __init__(self, component_name=None, timing_handler=None, debug_offset=0):
+		self.ob = ipaaca.buffer.OutputBuffer(('' if component_name is None else component_name)+'TimesyncMaster')
+		self.ib = ipaaca.buffer.InputBuffer(('' if component_name is None else component_name)+'TimesyncMaster', ['timesyncReply'])
+		# 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.ib.register_handler(self.handle_timesync_master)
+		# master_t1 is identical for all slaves
+		self.master_t1 = None
+		self.slave_t1s = {}
+		self.master_t2s = {}
+		self.slave_t2s = {}
+		self.latencies = {}
+		self.time_offsets = {}
+		#
+		self.debug_offset = debug_offset
+		#
+		self.timing_handler = timing_handler
+	
+	def set_timing_handler(self, timing_handler):
+		self.timing_handler = timing_handler
+	
+	def send_master_timesync(self):
+		iu = ipaaca.iu.Message('timesyncRequest')
+		self.master_t1 = self.get_time()
+		iu.payload = {
+				'stage':'0',
+				'master_t1':str(self.master_t1),
+				'master':self.component_name,
+				}
+		self.ob.add(iu)
+	
+	def handle_timesync_master(self, iu, event_type, own):
+		master = iu.payload['master']
+		if not own and master == self.component_name:
+			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:
+						# other stages are handled by time slave handler
+						pass
+	
+	def get_time(self):
+		return time.time() + self.debug_offset
+
+
+class TimesyncSlave(object):
+	def __init__(self, component_name=None, timing_handler=None, debug_offset=0):
+		self.ob = ipaaca.buffer.OutputBuffer(('' if component_name is None else component_name)+'TimesyncSlave')
+		self.ib = ipaaca.buffer.InputBuffer(('' if component_name is None else component_name)+'TimesyncSlave', ['timesyncRequest'])
+		# component name to report (None => use buffer name)
+		self.component_name = component_name if component_name is not None else self.ib.unique_name
+		self.ob.register_handler(self.handle_timesync_slave)
+		self.ib.register_handler(self.handle_timesync_slave)
+		#self.master_t1 = None
+		#self.master_t2 = None
+		#self.master = None
+		self.latency = None
+		#
+		self.debug_offset = debug_offset
+		#
+		self.timing_handler = timing_handler
+	
+	def set_timing_handler(self, timing_handler):
+		self.timing_handler = timing_handler
+	
+	def handle_timesync_slave(self, iu, event_type, own):
+		master = iu.payload['master']
+		stage = iu.payload['stage']
+		if self.component_name != master:
+			if not own and stage=='0':
+				# reply only to IUs from others
+				#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()
+					iu.payload.merge({
+							'slave_t2':str(t2),
+							'stage':'3',
+							})
+				elif stage=='4':
+					latency = float(iu.payload['latency'])
+					offset = float(iu.payload['offset'])
+					if self.timing_handler is None:
+						print('Timesync master '+master+' determined our timing: ')
+						print('  Avg round-trip latency: %.3f s'%latency)
+						print('  Offset of our clock: %.3f s'%offset)
+					else:
+						self.timing_handler(master, self.component_name, latency, offset)
+	def get_time(self):
+		return time.time() + self.debug_offset

python/src/listener.py → ipaacalib/python/src/listener.py

 #!/usr/bin/env python
 
+# This file is part of IPAACA, the
+#  "Incremental Processing Architecture
+#   for Artificial Conversational Agents".	
+#
+# Copyright (c) 2009-2022 Sociable Agents Group
+#                         CITEC, Bielefeld University	
+#
+# http://opensource.cit-ec.de/projects/ipaaca/
+# http://purl.org/net/ipaaca
+#
+# This file may be licensed under the terms of of the
+# GNU Lesser General Public License Version 3 (the ``LGPL''),
+# or (at your option) any later version.
+#
+# Software distributed under the License is distributed
+# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+# express or implied. See the LGPL for the specific language
+# governing rights and limitations.
+#
+# You should have received a copy of the LGPL along with this
+# program. If not, go to http://www.gnu.org/licenses/lgpl.html
+# or write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.	
+#
+# The development of this software was supported by the
+# Excellence Cluster EXC 277 Cognitive Interaction Technology.
+# The Excellence Cluster EXC 277 is a grant of the Deutsche
+# Forschungsgemeinschaft (DFG) in the context of the German
+# Excellence Initiative.
+
+
 import time
-import logging
 import ipaaca
 
 iu_to_write = None
@@ -38,8 +68,8 @@ while True:
 			else:
 				iu.payload = {'a': 'reset'}
 			
-		except ipaaca.IUUpdateFailedError, e:
-			ipaaca.logger.warning("Payload update failed (IU changed in the mean time)")
+		except ipaaca.IUUpdateFailedError as e:
+			print("Payload update failed (IU changed in the mean time)")
 	time.sleep(0.1)
 
 exit(0)

python/src/run.py → ipaacalib/python/src/run.py

@@ -2,5 +2,5 @@
 
 import ipaaca
 
-print "{this is the IpaacaPython run.py doing nothing at all}"
+print("{this is the IpaacaPython run.py doing nothing at all}")
 

ipaacalib/python/test/ivy.xml

+<ivy-module version="2.0">
+   <info organisation="ipaaca" module="IpaacaPythonTest"/>
+   <dependencies>
+	<dependency org="hamcrest" name="hamcrest-python" rev="latest.release"/>
+        <dependency org="mockito" name="mockito-python"  rev="latest.release" />
+        <dependency org="nose" name="nose"  rev="latest.release" />
+   </dependencies>
+</ivy-module>
+

python/test/src/ipaacaspam.py → ipaacalib/python/test/src/ipaacaspam.py

-#!/usr/bin/env python
+# This file is part of IPAACA, the
+#  "Incremental Processing Architecture
+#   for Artificial Conversational Agents".	
+#
+# Copyright (c) 2009-2022 Sociable Agents Group
+#                         CITEC, Bielefeld University	
+#
+# http://opensource.cit-ec.de/projects/ipaaca/
+# http://purl.org/net/ipaaca
+#
+# This file may be licensed under the terms of of the
+# GNU Lesser General Public License Version 3 (the ``LGPL''),
+# or (at your option) any later version.
+#
+# Software distributed under the License is distributed
+# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+# express or implied. See the LGPL for the specific language
+# governing rights and limitations.
+#
+# You should have received a copy of the LGPL along with this
+# program. If not, go to http://www.gnu.org/licenses/lgpl.html
+# or write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.	
+#
+# The development of this software was supported by the
+# Excellence Cluster EXC 277 Cognitive Interaction Technology.
+# The Excellence Cluster EXC 277 is a grant of the Deutsche
+# Forschungsgemeinschaft (DFG) in the context of the German
+# Excellence Initiative.
 
 import sys
 import time
@@ -97,6 +125,24 @@ class Sender(object):
 				self.counter = 0
 				#print ".",
 				#sys.stdout.flush()
+
+class TestAcker(object):
+	def __init__(self):
+		self.ib = ipaaca.InputBuffer('TestIn', ['testcategory'])
+		self.ib.register_handler(self.handle_iu_event)
+	
+	def handle_iu_event(self, iu, event_type, local):
+		print "Received a testcategory event ", event_type
+		if event_type=='ADDED':
+			try:
+				iu.payload['ack'] = 'ack'
+			except ipaaca.IUUpdateFailedError, e:
+				print "== tried to send an initial update, but someone else was quicker."
+	
+	def run(self):
+		while True:
+			time.sleep(1)
+
 	
 if __name__ == '__main__':
 	if len(sys.argv)<2:
@@ -125,6 +171,9 @@ if __name__ == '__main__':
 			sys.exit(1)
 		s = Sender(send_frequency=freq)
 		s.run()
+	elif sys.argv[1] == 'testacker':
+		r = TestAcker()
+		r.run()
 	else:
 		print "specify either 'sender', 'receiver', 'ping' or 'pong' as an argument"
 		sys.exit(1)

python/test/src/testipaaca.py → ipaacalib/python/test/src/testipaaca.py

-#!/usr/bin/env python
+# This file is part of IPAACA, the
+#  "Incremental Processing Architecture
+#   for Artificial Conversational Agents".	
+#
+# Copyright (c) 2009-2022 Sociable Agents Group
+#                         CITEC, Bielefeld University	
+#
+# http://opensource.cit-ec.de/projects/ipaaca/
+# http://purl.org/net/ipaaca
+#
+# This file may be licensed under the terms of of the
+# GNU Lesser General Public License Version 3 (the ``LGPL''),
+# or (at your option) any later version.
+#
+# Software distributed under the License is distributed
+# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+# express or implied. See the LGPL for the specific language
+# governing rights and limitations.
+#
+# You should have received a copy of the LGPL along with this
+# program. If not, go to http://www.gnu.org/licenses/lgpl.html
+# or write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.	
+#
+# The development of this software was supported by the
+# Excellence Cluster EXC 277 Cognitive Interaction Technology.
+# The Excellence Cluster EXC 277 is a grant of the Deutsche
+# Forschungsgemeinschaft (DFG) in the context of the German
+# Excellence Initiative.
 
 import sys
 import time

ipaacalib/python/test/src/testnotifier.py

+# This file is part of IPAACA, the
+#  "Incremental Processing Architecture
+#   for Artificial Conversational Agents".  
+#
+# Copyright (c) 2009-2022 Sociable Agents Group
+#                         CITEC, Bielefeld University   
+#
+# http://opensource.cit-ec.de/projects/ipaaca/
+# http://purl.org/net/ipaaca
+#
+# This file may be licensed under the terms of of the
+# GNU Lesser General Public License Version 3 (the ``LGPL''),
+# or (at your option) any later version.
+#
+# Software distributed under the License is distributed
+# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+# express or implied. See the LGPL for the specific language
+# governing rights and limitations.
+#
+# You should have received a copy of the LGPL along with this
+# program. If not, go to http://www.gnu.org/licenses/lgpl.html
+# or write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.  
+#
+# The development of this software was supported by the
+# Excellence Cluster EXC 277 Cognitive Interaction Technology.
+# The Excellence Cluster EXC 277 is a grant of the Deutsche
+# Forschungsgemeinschaft (DFG) in the context of the German
+# Excellence Initiative.
+
+import os
+import time
+import unittest
+
+from mockito import mock
+from mockito import verify
+from mockito import any
+from mockito import when
+from mockito import times
+from hamcrest.core.base_matcher import BaseMatcher
+from ipaaca import IUEventType
+from ipaaca import Message
+from ipaaca import InputBuffer
+from ipaaca import OutputBuffer
+from ipaaca.util import ComponentNotifier
+
+class IUCategoryMatcher(BaseMatcher):
+    def __init__(self, expected_cat):
+        self.expected_cat = expected_cat
+    
+    def _matches_(self, iu):
+        return iu.category==self.expected_cat
+    
+    def describe_to(self, description):
+        description.append_text("IU with category :"+self.expected_cat)
+        
+class ComponentNotifierTest(unittest.TestCase):
+
+    RECV_CAT = set(["testrec1", "testrc2"])
+    SEND_CAT = set(["testsnd1", "testsnd2", "testsnd3"])
+    def setUp(self):
+        self.mockOutBuffer = mock()
+        self.mockInBuffer = mock()
+        self.notifier = ComponentNotifier("testcomp","testfunc", ComponentNotifierTest.SEND_CAT, ComponentNotifierTest.RECV_CAT, self.mockOutBuffer, self.mockInBuffer)
+        self.notifier.initialize()
+    
+    def tearDown(self):
+        pass
+    
+    def _sendNotify(self, state, receiveCats):        
+        mockIUNotify = Message(ComponentNotifier.NOTIFY_CATEGORY)
+        mockIUNotify.payload[ComponentNotifier.STATE] = state;
+        mockIUNotify.payload[ComponentNotifier.NAME] = "namex";
+        mockIUNotify.payload[ComponentNotifier.SEND_CATEGORIES] = "";
+        mockIUNotify.payload[ComponentNotifier.RECEIVE_CATEGORIES] = ",".join(receiveCats);
+        self.notifier._handle_iu_event(mockIUNotify, IUEventType.ADDED, False)        
+    
+    def testNotifyAtInit(self):        
+        verify(self.mockOutBuffer).add(any())   
+        #TODO: python mockito cannot yet use hamcrest matchers, so cannot easily test if the message is correct :(
+        #assertEquals(ComponentNotifier.NOTIFY_CATEGORY, iu.getCategory());
+        #assertEquals("new", iu.getPayload().get(ComponentNotifier.STATE));
+        #assertThat(ImmutableSet.copyOf(iu.getPayload().get(ComponentNotifier.RECEIVE_CATEGORIES).split(",")),
+        #        IsIterableContainingInAnyOrder.containsInAnyOrder(RECV_CAT.toArray(new String[0])));
+        #assertThat(ImmutableSet.copyOf(iu.getPayload().get(ComponentNotifier.SEND_CATEGORIES).split(",")),
+        #        IsIterableContainingInAnyOrder.containsInAnyOrder(SEND_CAT.toArray(new String[0])));
+      
+    def testNotifyAtNotifyNew(self):
+        self._sendNotify("new", set(["testsnd1"]));
+        verify(self.mockOutBuffer, times(2)).add(any())
+        #TODO: python mockito cannot yet use hamcrest matchers, so cannot easily test if the message is correct :(
+        #ArgumentCaptor<LocalIU> argument = ArgumentCaptor.forClass(LocalIU.class);
+        #verify(mockOutBuffer, times(2)).add(argument.capture());
+        #LocalIU iu = argument.getAllValues().get(1);
+        #assertEquals("componentNotify", iu.getCategory());
+        #assertEquals("old", iu.getPayload().get("state"));
+    
+    def testNoNotifyAtNotifyOld(self):
+        self._sendNotify("old", set(["testsnd1"]));
+        verify(self.mockOutBuffer, times(1)).add(any())
+
+class MyListener(object):
+    def __init__(self):
+        self.numCalled = 0
+    def handle(self, iu, mytype, local):
+        self.numCalled += 1 
+    
+class ComponentNotifierIntegrationTest(unittest.TestCase):
+    def setUp(self):
+        pass
+    
+    def tearDown(self):
+        pass
+    
+    def _setupCompNotifier(self, idx, sendList, recvList):
+        inBuffer = InputBuffer(idx+"in", set([ComponentNotifier.NOTIFY_CATEGORY]))
+        outBuffer = OutputBuffer(idx+"out")
+        return ComponentNotifier(idx, "testfunction", sendList, recvList, outBuffer, inBuffer)
+
+#    bug: this somehow remains active after running
+#    def testSelf(self):
+#        notifier = self._setupCompNotifier("not", {"a1","b1"}, {"a3","b1"});
+#        listener = MyListener()
+#        notifier.add_notification_handler(listener.handle);
+#        
+#        notifier.initialize();    
+#        time.sleep(0.5);
+#        
+#        self.assertEquals(0, listener.numCalled);
+    
+    def testTwo(self):
+        notifier1 = self._setupCompNotifier("not1", set(["a1", "b1"]), set(["a3", "b2"]));
+        notifier2 = self._setupCompNotifier("not2", set(["a2", "b2"]), set(["a3", "b1"]));
+        listener1 = MyListener()
+        listener2 = MyListener()
+        notifier1.add_notification_handler(listener1.handle)
+        notifier2.add_notification_handler(listener2.handle)
+        
+        notifier1.initialize()
+        time.sleep(0.5)
+        notifier2.initialize()
+        time.sleep(0.5)        
+        self.assertEqual(1, listener1.numCalled)
+        self.assertEqual(1, listener2.numCalled)        
+        
+if __name__ == "__main__":
+    #import sys;sys.argv = ['', 'Test.testName']
+    unittest.main()
+    os._exit(0)
+

python/test/src/textsender.py → ipaacalib/python/test/src/textsender.py

-#!/usr/bin/env python
+# This file is part of IPAACA, the
+#  "Incremental Processing Architecture
+#   for Artificial Conversational Agents".	
+#
+# Copyright (c) 2009-2022 Sociable Agents Group
+#                         CITEC, Bielefeld University	
+#
+# http://opensource.cit-ec.de/projects/ipaaca/
+# http://purl.org/net/ipaaca
+#
+# This file may be licensed under the terms of of the
+# GNU Lesser General Public License Version 3 (the ``LGPL''),
+# or (at your option) any later version.
+#
+# Software distributed under the License is distributed
+# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+# express or implied. See the LGPL for the specific language
+# governing rights and limitations.
+#
+# You should have received a copy of the LGPL along with this
+# program. If not, go to http://www.gnu.org/licenses/lgpl.html
+# or write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.	
+#
+# The development of this software was supported by the
+# Excellence Cluster EXC 277 Cognitive Interaction Technology.
+# The Excellence Cluster EXC 277 is a grant of the Deutsche
+# Forschungsgemeinschaft (DFG) in the context of the German
+# Excellence Initiative.
 
 import sys
 import time

python/test/src/wordsender.py → ipaacalib/python/test/src/wordsender.py

-#!/usr/bin/env python
+# This file is part of IPAACA, the
+#  "Incremental Processing Architecture
+#   for Artificial Conversational Agents".	
+#
+# Copyright (c) 2009-2022 Sociable Agents Group
+#                         CITEC, Bielefeld University	
+#
+# http://opensource.cit-ec.de/projects/ipaaca/
+# http://purl.org/net/ipaaca
+#
+# This file may be licensed under the terms of of the
+# GNU Lesser General Public License Version 3 (the ``LGPL''),
+# or (at your option) any later version.
+#
+# Software distributed under the License is distributed
+# on an ``AS IS'' basis, WITHOUT WARRANTY OF ANY KIND, either
+# express or implied. See the LGPL for the specific language
+# governing rights and limitations.
+#
+# You should have received a copy of the LGPL along with this
+# program. If not, go to http://www.gnu.org/licenses/lgpl.html
+# or write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.	
+#
+# The development of this software was supported by the
+# Excellence Cluster EXC 277 Cognitive Interaction Technology.
+# The Excellence Cluster EXC 277 is a grant of the Deutsche
+# Forschungsgemeinschaft (DFG) in the context of the German
+# Excellence Initiative.
 
 import sys
 import time

ipaacasoa/build.xml

+<project name="IpaacaSoa" default="build" basedir=".">
+  <import file="../../hmibuild/build-recurse.xml" />
+</project>

ipaacasoa/cpp/CMakeLists.txt

+cmake_minimum_required (VERSION 2.6)
+
+# project name
+project (ipaaca_soa_cpp)
+
+install(
+	DIRECTORY include
+	DESTINATION .
+	FILES_MATCHING PATTERN "*.h" PATTERN "*.hh" PATTERN "*.hpp" PATTERN "*.inl"
+	)
+