New API for the algorithms

beat/backend/python/algorithm.py

@@ -41,6 +41,8 @@ from . import loader
 from . import utils
 
 
+#----------------------------------------------------------
+
 
 class Storage(utils.CodeStorage):
     """Resolves paths for algorithms
@@ -67,6 +69,8 @@ class Storage(utils.CodeStorage):
         super(Storage, self).__init__(path, language)
 
 
+#----------------------------------------------------------
+
 
 class Runner(object):
     '''A special loader class for algorithms, with specialized methods
@@ -79,13 +83,13 @@ class Runner(object):
       obj_name (str): The name of the object within the module you're interested
         on
 
+      algorithm (object): The algorithm instance that is used for parameter
+        checking.
+
       exc (class): The class to use as base exception when translating the
         exception from the user code. Read the documention of :py:func:`run`
         for more details.
 
-      algorithm (object): The algorithm instance that is used for parameter
-        checking.
-
       *args: Constructor parameters for the algorithm (normally none)
 
       **kwargs: Constructor parameters for the algorithm (normally none)
@@ -125,17 +129,15 @@ class Runner(object):
         if not user_keys.issubset(valid_keys):
             err_keys = user_keys - valid_keys
             message = "parameters `%s' are not declared for algorithm `%s' - " \
-                    "valid parameters are `%s'" % (
+                      "valid parameters are `%s'" % (
                             ','.join(err_keys),
                             self.name,
                             ','.join(valid_keys),
-                            )
+                      )
             exc = self.exc or KeyError
             raise exc(message)
 
-        # checks all values set by the user are in range (if a range is set)
-
-        retval = dict() #dictionary with checked user parameters and defaults
+        retval = dict()
 
         for key, definition in algo_parameters.items():
 
@@ -160,17 +162,29 @@ class Runner(object):
     def setup(self, parameters, *args, **kwargs):
         '''Sets up the algorithm, only effective on the first call'''
 
-        if self.ready: return self.ready
+        if self.ready:
+            return self.ready
 
-        completed_parameters = self._check_parameters(parameters) #may raise
+        completed_parameters = self._check_parameters(parameters)
         kwargs['parameters'] = completed_parameters
 
-        if hasattr(self.obj, 'setup'):
-            self.ready = loader.run(self.obj, 'setup', self.exc, *args, **kwargs)
-            return self.ready
-        else:
+        self.ready = loader.run(self.obj, 'setup', self.exc, *args, **kwargs)
+        return self.ready
+
+
+    def prepare(self, parameters, *args, **kwargs):
+        '''Let the algorithm process the data on the non-principal channels'''
+
+        if not self.ready:
+            message = "algorithm `%s' is not yet setup" % (self.name,)
+            exc = self.exc or RuntimeError
+            raise self.exc(message)
+
+        if not hasattr(self.obj, 'prepare'):
             return True
 
+        return loader.run(self.obj, 'prepare', self.exc, *args, **kwargs)
+
 
     def process(self, *args, **kwargs):
         '''Runs through data'''
@@ -188,6 +202,8 @@ class Runner(object):
         return getattr(self.obj, key)
 
 
+#----------------------------------------------------------
+
 
 class Algorithm(object):
     """Algorithms represent runnable components within the platform.
@@ -262,6 +278,11 @@ class Algorithm(object):
 
     """
 
+    LEGACY     = 'legacy'
+    SEQUENTIAL = 'sequential'
+    AUTONOMOUS = 'autonomous'
+
+
     def __init__(self, prefix, name, dataformat_cache=None, library_cache=None):
 
         self._name = None
@@ -451,6 +472,21 @@ class Algorithm(object):
         return self.data['language']
 
 
+    @property
+    def api_version(self):
+        """Returns the API version"""
+        return self.data.get('api_version', 1)
+
+
+    @property
+    def type(self):
+        """Returns the type of algorithm"""
+        if self.api_version == 1:
+            return Algorithm.LEGACY
+
+        return self.data.get('type', Algorithm.SEQUENTIAL)
+
+
     @language.setter
     def language(self, value):
         """Sets the current executable code programming language"""

beat/backend/python/data_loaders.py

+#!/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.backend.python 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 time
+import logging
+logger = logging.getLogger(__name__)
+
+from functools import reduce
+
+import six
+import zmq
+
+from .data import mixDataIndices
+
+
+#----------------------------------------------------------
+
+
+class DataView(object):
+    """Provides access to a subset of data from a group of inputs synchronized
+    together
+
+    Data views are created from a data loader
+    (see :py:class:`beat.backend.python.data_loaders.DataLoader`), which are
+    provided to the algorithms of types 'sequential' and 'autonomous'
+    (see :py:class:`beat.backend.python.data_loaders.DataLoaderList`).
+
+    Example:
+
+      .. code-block:: python
+
+         view = data_loader.view('input1', 0)
+
+         for i in range(view.count())
+             (data, start_index, end_index) = view[i]
+
+
+    Parameters:
+
+      data_loader (:py:class:`beat.backend.python.data_loaders.DataLoader`):
+                Name of the data channel of the group of inputs
+
+      data_indices (list of tuples): Data indices to consider
+
+
+    Attributes:
+
+      data_index_start (int): Lower data index across all inputs
+        (see the section *Inputs synchronization* of the User's Guide)
+
+      data_index_end (int): Bigger data index across all inputs
+        (see the section *Inputs synchronization* of the User's Guide)
+
+    """
+
+    def __init__(self, data_loader, data_indices):
+        self.infos             = {}
+        self.data_indices      = data_indices
+        self.nb_data_units     = len(data_indices)
+        self.data_index_start  = data_indices[0][0]
+        self.data_index_end    = data_indices[-1][1]
+
+        for input_name, infos in data_loader.infos.items():
+            input_data_indices = []
+            current_start = self.data_index_start
+
+            for i in range(self.data_index_start, self.data_index_end + 1):
+                for indices in infos['data_indices']:
+                    if indices[1] == i:
+                        input_data_indices.append( (current_start, i) )
+                        current_start = i + 1
+                        break
+
+            if (len(input_data_indices) == 0) or (input_data_indices[-1][1] != self.data_index_end):
+                input_data_indices.append( (current_start, self.data_index_end) )
+
+            self.infos[input_name] = dict(
+                cached_file = infos['cached_file'],
+                data_indices = input_data_indices,
+                data = None,
+                start_index = -1,
+                end_index = -1,
+            )
+
+
+    def count(self, input_name=None):
+        if input_name is not None:
+            try:
+                return len(self.infos[input_name]['data_indices'])
+            except:
+                return None
+        else:
+            return self.nb_data_units
+
+
+    def __getitem__(self, index):
+        if index < 0:
+            return (None, None, None)
+
+        try:
+            indices = self.data_indices[index]
+        except:
+            return (None, None, None)
+
+        result = {}
+
+        for input_name, infos in self.infos.items():
+            if (indices[0] < infos['start_index']) or (infos['end_index'] < indices[0]):
+                (infos['data'], infos['start_index'], infos['end_index']) = \
+                        infos['cached_file'].getAtDataIndex(indices[0])
+
+            result[input_name] = infos['data']
+
+        return (result, indices[0], indices[1])
+
+
+#----------------------------------------------------------
+
+
+class DataLoader(object):
+    """Provides access to data from a group of inputs synchronized together
+
+    Data loaders are provided to the algorithms of types 'sequential' and
+    'autonomous' (see :py:class:`beat.backend.python.data_loaders.DataLoaderList`).
+
+    Example:
+
+      .. code-block:: python
+
+         # Iterate through all the data
+         for i in range(data_loader.count())
+             (data, start_index, end_index) = data_loader[i]
+             print data['input1'].data
+
+         # Restrict to a subset of the data
+         view = data_loader.view('input1', 0)
+         for i in range(view.count())
+             (data, start_index, end_index) = view[i]
+
+
+    Parameters:
+
+      channel (str): Name of the data channel of the group of inputs
+
+
+    Attributes:
+
+      data_index_start (int): Lower data index across all inputs
+        (see the section *Inputs synchronization* of the User's Guide)
+
+      data_index_end (int): Bigger data index across all inputs
+        (see the section *Inputs synchronization* of the User's Guide)
+
+      channel (str): Name of the data channel of the group
+
+    """
+
+    def __init__(self, channel):
+        self.channel            = str(channel)
+        self.infos              = {}
+        self.mixed_data_indices = None
+        self.nb_data_units      = 0
+        self.data_index_start   = -1  # Lower index across all inputs
+        self.data_index_end     = -1  # Bigger index across all inputs
+
+
+    def add(self, input_name, cached_file):
+        self.infos[input_name] = dict(
+            cached_file = cached_file,
+            data_indices = cached_file.data_indices(),
+            data = None,
+            start_index = -1,
+            end_index = -1,
+        )
+
+        self.mixed_data_indices = mixDataIndices([ x['data_indices'] for x in self.infos.values() ])
+        self.nb_data_units = len(self.mixed_data_indices)
+        self.data_index_start = self.mixed_data_indices[0][0]
+        self.data_index_end = self.mixed_data_indices[-1][1]
+
+
+    def input_names(self):
+        return self.infos.keys()
+
+
+    def count(self, input_name=None):
+        if input_name is not None:
+            try:
+                return len(self.infos[input_name]['data_indices'])
+            except:
+                return 0
+        else:
+            return self.nb_data_units
+
+
+    def view(self, input_name, index):
+        if index < 0:
+            return None
+
+        try:
+            indices = self.infos[input_name]['data_indices'][index]
+        except:
+            return None
+
+        limited_data_indices = [ x for x in self.mixed_data_indices
+                                 if (indices[0] <= x[0]) and (x[1] <= indices[1]) ]
+
+        return DataView(self, limited_data_indices)
+
+
+    def __getitem__(self, index):
+        if index < 0:
+            return (None, None, None)
+
+        try:
+            indices = self.mixed_data_indices[index]
+        except:
+            return (None, None, None)
+
+        result = {}
+
+        for input_name, infos in self.infos.items():
+            if (indices[0] < infos['start_index']) or (infos['end_index'] < indices[0]):
+                (infos['data'], infos['start_index'], infos['end_index']) = \
+                        infos['cached_file'].getAtDataIndex(indices[0])
+
+            result[input_name] = infos['data']
+
+        return (result, indices[0], indices[1])
+
+
+#----------------------------------------------------------
+
+
+class DataLoaderList(object):
+    """Represents a list of data loaders
+
+    Inputs are organized by groups. The inputs inside a group are all
+    synchronized together (see the section *Inputs synchronization* of the User's
+    Guide). A data loader provides access to data from a group of inputs.
+
+    A list implementing this interface is provided to the algorithms of types
+    'sequential' and 'autonomous'.
+
+    One group of inputs is always considered as the **main** one, and is used to
+    drive the algorithm. The usage of the other groups is left to the algorithm.
+
+    See :py:class:`beat.backend.python.data_loaders.DataLoader`
+
+
+    Example:
+
+      .. code-block:: python
+
+         data_loaders = DataLoaderList()
+         ...
+
+         # Retrieve a data loader by name
+         data_loader = data_loaders['labels']
+
+         # Retrieve a data loader by index
+         for index in range(0, len(data_loaders)):
+             data_loader = data_loaders[index]
+
+         # Iteration over all data loaders
+         for data_loader in data_loaders:
+             ...
+
+         # Retrieve the data loader an input belongs to, by input name
+         data_loader = data_loaders.loaderOf('label')
+
+
+    Attributes:
+
+      main_loader (beat.backend.python.data_loaders.DataLoader): Main data loader
+
+    """
+
+    def __init__(self):
+        self._loaders = []
+        self.main_loader = None
+
+
+    def add(self, data_loader):
+        """Add a data loader to the list
+
+        :param beat.backend.python.data_loaders.DataLoader data_loader: The data
+            loader to add
+        """
+        if self.main_loader is None:
+            self.main_loader = data_loader
+
+        self._loaders.append(data_loader)
+
+
+    def __getitem__(self, name_or_index):
+        try:
+            if isinstance(name_or_index, six.string_types):
+                return [x for x in self._loaders if x.channel == name_or_index][0]
+
+            elif isinstance(index, int):
+                return self._loaders[index]
+        except:
+            pass
+
+        return None
+
+
+    def __iter__(self):
+        for i in range(len(self._loaders)):
+            yield self._loaders[i]
+
+
+    def __len__(self):
+        return len(self._loaders)
+
+
+    def loaderOf(self, input_name):
+        try:
+            return [ k for k in self._loaders if input_name in k.input_names() ][0]
+        except:
+            return None

beat/backend/python/inputs.py

@@ -495,147 +495,6 @@ class InputGroup:
 #----------------------------------------------------------
 
 
-class DataView(object):
-
-    def __init__(self, data_loader_group, data_indices):
-        self.infos             = {}
-        self.data_indices      = data_indices
-        self.nb_data_units     = len(data_indices)
-        self.data_index        = data_indices[0][0]
-        self.data_index_end    = data_indices[-1][1]
-
-        for input_name, infos in data_loader_group.infos.items():
-            input_data_indices = []
-            current_start = self.data_index
-
-            for i in range(self.data_index, self.data_index_end + 1):
-                for indices in infos['data_indices']:
-                    if indices[1] == i:
-                        input_data_indices.append( (current_start, i) )
-                        current_start = i + 1
-                        break
-
-            if (len(input_data_indices) == 0) or (input_data_indices[-1][1] != self.data_index_end):
-                input_data_indices.append( (current_start, self.data_index_end) )
-
-            self.infos[input_name] = dict(
-                cached_file = infos['cached_file'],
-                data_indices = input_data_indices,
-                data = None,
-                start_index = -1,
-                end_index = -1,
-            )
-
-
-    def count(self, input_name=None):
-        if input_name is not None:
-            try:
-                return len(self.infos[input_name]['data_indices'])
-            except:
-                return None
-        else:
-            return self.nb_data_units
-
-
-    def __getitem__(self, index):
-        if index < 0:
-            return (None, None, None)
-
-        try:
-            indices = self.data_indices[index]
-        except:
-            return (None, None, None)
-
-        result = {}
-
-        for input_name, infos in self.infos.items():
-            if (indices[0] < infos['start_index']) or (infos['end_index'] < indices[0]):
-                (infos['data'], infos['start_index'], infos['end_index']) = \
-                        infos['cached_file'].getAtDataIndex(indices[0])
-
-            result[input_name] = infos['data']
-
-        return (result, indices[0], indices[1])
-
-
-#----------------------------------------------------------
-
-
-class DataLoaderGroup(object):
-
-    def __init__(self, channel):
-        self.channel            = str(channel)
-        self.infos              = {}
-        self.mixed_data_indices = None
-        self.nb_data_units      = 0
-        self.data_index         = -1  # Lower index across all inputs
-        self.data_index_end     = -1  # Bigger index across all inputs
-
-
-    def add(self, input_name, cached_file):
-        self.infos[input_name] = dict(
-            cached_file = cached_file,
-            data_indices = cached_file.data_indices(),
-            data = None,
-            start_index = -1,
-            end_index = -1,
-        )
-
-        self.mixed_data_indices = mixDataIndices([ x['data_indices'] for x in self.infos.values() ])
-        self.nb_data_units = len(self.mixed_data_indices)
-        self.data_index = self.mixed_data_indices[0][0]
-        self.data_index_end = self.mixed_data_indices[-1][1]
-
-
-    def count(self, input_name=None):
-        if input_name is not None:
-            try:
-                return len(self.infos[input_name]['data_indices'])
-            except:
-                return 0
-        else:
-            return self.nb_data_units
-
-
-    def view(self, input_name, index):
-        if index < 0:
-            return None
-
-        try:
-            indices = self.infos[input_name]['data_indices'][index]
-        except:
-            return None
-
-        limited_data_indices = [ x for x in self.mixed_data_indices
-                                 if (indices[0] <= x[0]) and (x[1] <= indices[1]) ]
-
-        return DataView(self, limited_data_indices)
-
-
-    def __getitem__(self, index):
-        if index < 0:
-            return (None, None, None)
-
-        try:
-            indices = self.mixed_data_indices[index]
-        except:
-            return (None, None, None)
-
-        result = {}
-
-        for input_name, infos in self.infos.items():