diff --git a/beat/core/execution.py b/beat/core/execution.py
index 1fc7d2d599098e7502463e8c40c68b0bc4fa1d89..466d30905d33bce44821590f68a02f1107c7402d 100755
--- a/beat/core/execution.py
+++ b/beat/core/execution.py
@@ -292,6 +292,7 @@ class Executor(object):
if group is None:
group = inputs.InputGroup(
details['channel'],
+ synchronization_listener=outputs.SynchronizationListener(),
restricted_access=(details['channel'] == self.data['channel'])
)
self.input_list.add(group)
diff --git a/beat/core/test/prefix/algorithms/user/labelled_integers_sum/1.json b/beat/core/test/prefix/algorithms/user/labelled_integers_sum/1.json
new file mode 100644
index 0000000000000000000000000000000000000000..47002d9941ee59d716de41215a5f7ef7d85f16d9
--- /dev/null
+++ b/beat/core/test/prefix/algorithms/user/labelled_integers_sum/1.json
@@ -0,0 +1,24 @@
+{
+ "language": "python",
+ "splittable": true,
+ "groups": [
+ {
+ "name": "main",
+ "inputs": {
+ "value": {
+ "type": "user/single_integer/1"
+ },
+ "label": {
+ "type": "user/single_string/1"
+ }
+ },
+ "outputs": {
+ "result": {
+ "type": "user/single_integer/1"
+ }
+ }
+ }
+ ],
+ "parameters": {
+ }
+}
diff --git a/beat/core/test/prefix/algorithms/user/labelled_integers_sum/1.py b/beat/core/test/prefix/algorithms/user/labelled_integers_sum/1.py
new file mode 100755
index 0000000000000000000000000000000000000000..db5207be666ff7ad5fc1c5e36eb2f70ee35c5677
--- /dev/null
+++ b/beat/core/test/prefix/algorithms/user/labelled_integers_sum/1.py
@@ -0,0 +1,44 @@
+#!/usr/bin/env python
+# vim: set fileencoding=utf-8 :
+
+###############################################################################
+# #
+# Copyright (c) 2017 Idiap Research Institute, http://www.idiap.ch/ #
+# Contact: beat.support@idiap.ch #
+# #
+# This file is part of the beat.core module of the BEAT platform. #
+# #
+# Commercial License Usage #
+# Licensees holding valid commercial BEAT licenses may use this file in #
+# accordance with the terms contained in a written agreement between you #
+# and Idiap. For further information contact tto@idiap.ch #
+# #
+# Alternatively, this file may be used under the terms of the GNU Affero #
+# Public License version 3 as published by the Free Software and appearing #
+# in the file LICENSE.AGPL included in the packaging of this file. #
+# The BEAT platform is distributed in the hope that it will be useful, but #
+# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY #
+# or FITNESS FOR A PARTICULAR PURPOSE. #
+# #
+# You should have received a copy of the GNU Affero Public License along #
+# with the BEAT platform. If not, see http://www.gnu.org/licenses/. #
+# #
+###############################################################################
+
+import numpy as np
+
+
+class Algorithm:
+
+ def __init__(self):
+ self.sum = np.int32(0)
+
+ def process(self, inputs, outputs):
+ self.sum += inputs['value'].data.value
+
+ if inputs['label'].isDataUnitDone():
+ outputs['result'].write({
+ 'value': self.sum
+ })
+
+ return True
diff --git a/beat/core/test/prefix/algorithms/user/synchronisation_analyzer/1.json b/beat/core/test/prefix/algorithms/user/synchronisation_analyzer/1.json
new file mode 100644
index 0000000000000000000000000000000000000000..7a7eb9b77ff5a7c242cca653d7de728ede8e5942
--- /dev/null
+++ b/beat/core/test/prefix/algorithms/user/synchronisation_analyzer/1.json
@@ -0,0 +1,23 @@
+{
+ "language": "python",
+ "groups": [
+ {
+ "name": "main",
+ "inputs": {
+ "input": {
+ "type": "user/single_integer/1"
+ }
+ }
+ }
+ ],
+ "results": {
+ "nb_data_units": {
+ "type": "int32",
+ "display": true
+ },
+ "indices": {
+ "type": "string",
+ "display": true
+ }
+ }
+}
diff --git a/beat/core/test/prefix/algorithms/user/synchronisation_analyzer/1.py b/beat/core/test/prefix/algorithms/user/synchronisation_analyzer/1.py
new file mode 100755
index 0000000000000000000000000000000000000000..d7e98b4c57b43eec592a165da3e3374e21614917
--- /dev/null
+++ b/beat/core/test/prefix/algorithms/user/synchronisation_analyzer/1.py
@@ -0,0 +1,51 @@
+#!/usr/bin/env python
+# vim: set fileencoding=utf-8 :
+
+###############################################################################
+# #
+# Copyright (c) 2016 Idiap Research Institute, http://www.idiap.ch/ #
+# Contact: beat.support@idiap.ch #
+# #
+# This file is part of the beat.core module of the BEAT platform. #
+# #
+# Commercial License Usage #
+# Licensees holding valid commercial BEAT licenses may use this file in #
+# accordance with the terms contained in a written agreement between you #
+# and Idiap. For further information contact tto@idiap.ch #
+# #
+# Alternatively, this file may be used under the terms of the GNU Affero #
+# Public License version 3 as published by the Free Software and appearing #
+# in the file LICENSE.AGPL included in the packaging of this file. #
+# The BEAT platform is distributed in the hope that it will be useful, but #
+# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY #
+# or FITNESS FOR A PARTICULAR PURPOSE. #
+# #
+# You should have received a copy of the GNU Affero Public License along #
+# with the BEAT platform. If not, see http://www.gnu.org/licenses/. #
+# #
+###############################################################################
+
+import numpy as np
+
+
+class Algorithm:
+
+ def __init__(self):
+ self.nb_data_units = 0
+ self.indices = ''
+
+ def process(self, inputs, output):
+ self.nb_data_units += 1
+
+ self.indices += '%d - %d\n' % (
+ inputs['input'].data_index,
+ inputs['input'].data_index_end
+ )
+
+ if not inputs['input'].hasMoreData():
+ output.write({
+ 'nb_data_units': np.int32(self.nb_data_units),
+ 'indices': self.indices
+ })
+
+ return True
diff --git a/beat/core/test/prefix/databases/integers_db/1.json b/beat/core/test/prefix/databases/integers_db/1.json
index 35a7b706914dd90673fc43129ef5a8807c5eb785..33ee808a7121307e2366dc31f06d2022ee46b282 100644
--- a/beat/core/test/prefix/databases/integers_db/1.json
+++ b/beat/core/test/prefix/databases/integers_db/1.json
@@ -60,6 +60,21 @@
}
}
]
+ },
+ {
+ "name": "labelled",
+ "template": "labelled",
+ "sets": [
+ {
+ "name": "labelled",
+ "template": "labelled",
+ "view": "Labelled",
+ "outputs": {
+ "value": "user/single_integer/1",
+ "label": "user/single_string/1"
+ }
+ }
+ ]
}
]
}
diff --git a/beat/core/test/prefix/databases/integers_db/1.py b/beat/core/test/prefix/databases/integers_db/1.py
old mode 100644
new mode 100755
index 7dd0d7a9dca0fd15e43e5bd69ae57027e652e59c..978324aa0ea6874971fb136b400a629b8c1a0bd3
--- a/beat/core/test/prefix/databases/integers_db/1.py
+++ b/beat/core/test/prefix/databases/integers_db/1.py
@@ -59,6 +59,7 @@ class Double:
return True
+
class Triple:
def setup(self, root_folder, outputs, parameters):
@@ -92,3 +93,50 @@ class Triple:
})
return True
+
+
+
+class Labelled:
+
+ def setup(self, root_folder, outputs, parameters):
+ self.outputs = outputs
+ self.remaining = [
+ ['A', [1, 2, 3, 4, 5]],
+ ['B', [10, 20, 30, 40, 50]],
+ ['C', [100, 200, 300, 400, 500]],
+ ]
+ self.current_label = None
+ return True
+
+
+ def done(self):
+ return (len(self.remaining) == 0)
+
+
+ def next(self):
+ # Ensure that we are not done
+ if len(self.remaining) == 0:
+ return False
+
+ # Retrieve the next label and value
+ label = self.remaining[0][0]
+ value = self.remaining[0][1][0]
+
+ # Only write each label once on the output, with the correct range of indexes
+ if self.current_label != label:
+ self.outputs['label'].write({
+ 'value': label,
+ }, self.outputs['label'].last_written_data_index + len(self.remaining[0][1]))
+ self.current_label = label
+
+ # Write the value
+ self.outputs['value'].write({
+ 'value': numpy.int32(value),
+ })
+
+ # Remove the value (and if needed the label) from the list of remaining data
+ self.remaining[0][1] = self.remaining[0][1][1:]
+ if len(self.remaining[0][1]) == 0:
+ self.remaining = self.remaining[1:]
+
+ return True
diff --git a/beat/core/test/prefix/experiments/user/user/integers_labelled/1/test.json b/beat/core/test/prefix/experiments/user/user/integers_labelled/1/test.json
new file mode 100644
index 0000000000000000000000000000000000000000..1570695d257548eae76e5a4aeb491daa5fb5f229
--- /dev/null
+++ b/beat/core/test/prefix/experiments/user/user/integers_labelled/1/test.json
@@ -0,0 +1,40 @@
+{
+ "datasets": {
+ "integers": {
+ "database": "integers_db/1",
+ "protocol": "labelled",
+ "set": "labelled"
+ }
+ },
+ "blocks": {
+ "processing": {
+ "algorithm": "user/labelled_integers_sum/1",
+ "parameters": {
+ },
+ "inputs": {
+ "value": "value",
+ "label": "label"
+ },
+ "outputs": {
+ "result": "result"
+ }
+ }
+ },
+ "analyzers": {
+ "analysis": {
+ "algorithm": "user/synchronisation_analyzer/1",
+ "parameters": {
+ },
+ "inputs": {
+ "input": "input"
+ }
+ }
+ },
+ "globals": {
+ "environment": {
+ "name": "environment",
+ "version": "1"
+ },
+ "queue": "queue"
+ }
+}
diff --git a/beat/core/test/prefix/toolchains/user/integers_labelled/1.json b/beat/core/test/prefix/toolchains/user/integers_labelled/1.json
new file mode 100644
index 0000000000000000000000000000000000000000..039bd467ff4f989e5bd483062da05c10367fd90d
--- /dev/null
+++ b/beat/core/test/prefix/toolchains/user/integers_labelled/1.json
@@ -0,0 +1,59 @@
+{
+ "datasets": [
+ {
+ "name": "integers",
+ "outputs": [
+ "value",
+ "label"
+ ]
+ }
+ ],
+ "blocks": [
+ {
+ "name": "processing",
+ "inputs": [
+ "value",
+ "label"
+ ],
+ "outputs": [
+ "result"
+ ],
+ "synchronized_channel": "integers"
+ }
+ ],
+ "analyzers": [
+ {
+ "name": "analysis",
+ "inputs": [
+ "input"
+ ],
+ "synchronized_channel": "integers"
+ }
+ ],
+ "connections": [
+ {
+ "from": "integers.value",
+ "to": "processing.value",
+ "channel": "integers"
+ },
+ {
+ "from": "integers.label",
+ "to": "processing.label",
+ "channel": "integers"
+ },
+ {
+ "from": "processing.result",
+ "to": "analysis.input",
+ "channel": "integers"
+ }
+ ],
+ "representation": {
+ "connections": {
+ },
+ "blocks": {
+ },
+ "channel_colors": {
+ "integers": "#5555ff"
+ }
+ }
+}
diff --git a/beat/core/test/test_execution.py b/beat/core/test/test_execution.py
old mode 100644
new mode 100755
index adc5be93412b6cd59273d29658267fce52ffa209..f74d3e4e4432c8e8c924740b20dde9a350d9fd78
--- a/beat/core/test/test_execution.py
+++ b/beat/core/test/test_execution.py
@@ -255,3 +255,11 @@ class TestExecution(unittest.TestCase):
def test_inputs_mix_4b(self):
assert self.execute('user/user/inputs_mix/4/test2', [{'sum': 25170, 'nb': 10}]) is None
+
+ def test_integers_labelled_1(self):
+ assert self.execute('user/user/integers_labelled/1/test', [
+ {
+ 'nb_data_units': 3,
+ 'indices': '0 - 4\n5 - 9\n10 - 14\n'
+ }
+ ]) is None