Accept plugin pipelines

Closes #15

senpy/blueprints.py

@@ -113,8 +113,11 @@ def basic_api(f):
 @api_blueprint.route('/', methods=['POST', 'GET'])
 @basic_api
 def api():
-    response = current_app.senpy.analyse(**request.params)
-    return response
+    try:
+        response = current_app.senpy.analyse(**request.params)
+        return response
+    except Error as ex:
+        return ex
 
 
 @api_blueprint.route('/plugins/', methods=['POST', 'GET'])

senpy/extensions.py

@@ -7,7 +7,7 @@ standard_library.install_aliases()
 
 from . import plugins
 from .plugins import SenpyPlugin
-from .models import Error, Entry, Results
+from .models import Error, Entry, Results, from_dict
 from .blueprints import api_blueprint, demo_blueprint, ns_blueprint
 from .api import API_PARAMS, NIF_PARAMS, parse_params
 
@@ -78,70 +78,101 @@ class Senpy(object):
         else:
             logger.debug("Not a folder: %s", folder)
 
-    def _find_plugin(self, params):
-        api_params = parse_params(params, spec=API_PARAMS)
-        algo = None
-        if "algorithm" in api_params and api_params["algorithm"]:
-            algo = api_params["algorithm"]
-        elif self.plugins:
-            algo = self.default_plugin and self.default_plugin.name
-        if not algo:
+    def _find_plugins(self, params):
+        if not self.analysis_plugins:
             raise Error(
                 status=404,
                 message=("No plugins found."
-                         " Please install one.").format(algo))
-        if algo not in self.plugins:
-            logger.debug(("The algorithm '{}' is not valid\n"
-                          "Valid algorithms: {}").format(algo,
-                                                         self.plugins.keys()))
+                         " Please install one."))
+        api_params = parse_params(params, spec=API_PARAMS)
+        algos = None
+        if "algorithm" in api_params and api_params["algorithm"]:
+            algos = api_params["algorithm"].split(',')
+        elif self.default_plugin:
+            algos = [self.default_plugin.name, ]
+        else:
             raise Error(
                 status=404,
-                message="The algorithm '{}' is not valid".format(algo))
-
-        if not self.plugins[algo].is_activated:
-            logger.debug("Plugin not activated: {}".format(algo))
-            raise Error(
-                status=400,
-                message=("The algorithm '{}'"
-                         " is not activated yet").format(algo))
-        return self.plugins[algo]
+                message="No default plugin found, and None provided")
+
+        plugins = list()
+        for algo in algos:
+            if algo not in self.plugins:
+                logger.debug(("The algorithm '{}' is not valid\n"
+                              "Valid algorithms: {}").format(algo,
+                                                             self.plugins.keys()))
+                raise Error(
+                    status=404,
+                    message="The algorithm '{}' is not valid".format(algo))
+
+            if not self.plugins[algo].is_activated:
+                logger.debug("Plugin not activated: {}".format(algo))
+                raise Error(
+                    status=400,
+                    message=("The algorithm '{}'"
+                             " is not activated yet").format(algo))
+            plugins.append(self.plugins[algo])
+        return plugins
 
-    def _get_params(self, params, plugin):
+    def _get_params(self, params, plugin=None):
         nif_params = parse_params(params, spec=NIF_PARAMS)
-        extra_params = plugin.get('extra_params', {})
-        specific_params = parse_params(params, spec=extra_params)
-        nif_params.update(specific_params)
+        if plugin:
+            extra_params = plugin.get('extra_params', {})
+            specific_params = parse_params(params, spec=extra_params)
+            nif_params.update(specific_params)
         return nif_params
 
     def _get_entries(self, params):
-        entry = None
         if params['informat'] == 'text':
+            results = Results()
             entry = Entry(text=params['input'])
+            results.entries.append(entry)
+        elif params['informat'] == 'json-ld':
+            results = from_dict(params['input'])
         else:
-            raise NotImplemented('Only text input format implemented')
-        yield entry
+            raise NotImplemented('Informat {} is not implemented'.format(params['informat']))
+        return results
+
+    def _process_entries(self, entries, plugins, nif_params):
+        if not plugins:
+            for i in entries:
+                yield i
+            return
+        plugin = plugins[0]
+        specific_params = self._get_params(nif_params, plugin)
+        results = plugin.analyse_entries(entries, specific_params)
+        for i in self._process_entries(results, plugins[1:], nif_params):
+            yield i
+
+    def _process_response(self, resp, plugins, nif_params):
+        entries = resp.entries
+        resp.entries = []
+        for plug in plugins:
+            resp.analysis.append(plug.id)
+        for i in self._process_entries(entries, plugins, nif_params):
+            resp.entries.append(i)
+        return resp
 
     def analyse(self, **api_params):
+        """
+        Main method that analyses a request, either from CLI or HTTP.
+        It uses a dictionary of parameters, provided by the user.
+        """
         logger.debug("analysing with params: {}".format(api_params))
-        plugin = self._find_plugin(api_params)
-        nif_params = self._get_params(api_params, plugin)
-        resp = Results()
+        plugins = self._find_plugins(api_params)
+        nif_params = self._get_params(api_params)
+        resp = self._get_entries(nif_params)
         if 'with_parameters' in api_params:
             resp.parameters = nif_params
         try:
-            entries = []
-            for i in self._get_entries(nif_params):
-                entries += list(plugin.analyse_entry(i, nif_params))
-            resp.entries = entries
-            self.convert_emotions(resp, plugin, nif_params)
-            resp.analysis.append(plugin.id)
+            resp = self._process_response(resp, plugins, nif_params)
+            self.convert_emotions(resp, plugins, nif_params)
             logger.debug("Returning analysis result: {}".format(resp))
-        except Error as ex:
-            logger.exception('Error returning analysis result')
-            resp = ex
-        except Exception as ex:
+        except (Error, Exception) as ex:
+            if not isinstance(ex, Error):
+                ex = Error(message=str(ex), status=500)
             logger.exception('Error returning analysis result')
-            resp = Error(message=str(ex), status=500)
+            raise ex
         return resp
 
     def _conversion_candidates(self, fromModel, toModel):
@@ -155,7 +186,7 @@ class Senpy(object):
                     # logging.debug('Found candidate: {}'.format(candidate))
                     yield candidate
 
-    def convert_emotions(self, resp, plugin, params):
+    def convert_emotions(self, resp, plugins, params):
         """
         Conversion of all emotions in a response.
         In addition to converting from one model to another, it has
@@ -163,29 +194,35 @@ class Senpy(object):
         Needless to say, this is far from an elegant solution, but it works.
         @todo refactor and clean up
         """
-        fromModel = plugin.get('onyx:usesEmotionModel', None)
         toModel = params.get('emotionModel', None)
-        output = params.get('conversion', None)
-        logger.debug('Asked for model: {}'.format(toModel))
-        logger.debug('Analysis plugin uses model: {}'.format(fromModel))
-
         if not toModel:
             return
-        try:
-            candidate = next(self._conversion_candidates(fromModel, toModel))
-        except StopIteration:
-            e = Error(('No conversion plugin found for: '
-                       '{} -> {}'.format(fromModel, toModel)))
-            e.original_response = resp
-            e.parameters = params
-            raise e
+
+        logger.debug('Asked for model: {}'.format(toModel))
+        output = params.get('conversion', None)
+        candidates = {}
+        for plugin in plugins:
+            try:
+                fromModel = plugin.get('onyx:usesEmotionModel', None)
+                candidates[plugin.id] = next(self._conversion_candidates(fromModel, toModel))
+                logger.debug('Analysis plugin {} uses model: {}'.format(plugin.id, fromModel))
+            except StopIteration:
+                e = Error(('No conversion plugin found for: '
+                           '{} -> {}'.format(fromModel, toModel)))
+                e.original_response = resp
+                e.parameters = params
+                raise e
         newentries = []
+        resp.analysis = set(resp.analysis)
         for i in resp.entries:
             if output == "full":
                 newemotions = copy.deepcopy(i.emotions)
             else:
                 newemotions = []
             for j in i.emotions:
+                plugname = j['prov:wasGeneratedBy']
+                candidate = candidates[plugname]
+                resp.analysis.add(candidate.id)
                 for k in candidate.convert(j, fromModel, toModel, params):
                     k.prov__wasGeneratedBy = candidate.id
                     if output == 'nested':
@@ -194,7 +231,6 @@ class Senpy(object):
             i.emotions = newemotions
             newentries.append(i)
         resp.entries = newentries
-        resp.analysis.append(candidate.id)
 
     @property
     def default_plugin(self):

senpy/plugins/__init__.py

@@ -57,6 +57,12 @@ class AnalysisPlugin(SenpyPlugin):
         for i in results.entries:
             yield i
 
+    def analyse_entries(self, entries, parameters):
+        for entry in entries:
+            logger.debug('Analysing entry with plugin {}: {}'.format(self, entry))
+            for result in self.analyse_entry(entry, parameters):
+                yield result
+
 
 class ConversionPlugin(SenpyPlugin):
     pass

tests/plugins/dummy_plugin/dummy.py

@@ -4,4 +4,5 @@ from senpy.plugins import SentimentPlugin
 class DummyPlugin(SentimentPlugin):
     def analyse_entry(self, entry, params):
         entry.text = entry.text[::-1]
+        entry.reversed = entry.get('reversed', 0) + 1
         yield entry

tests/test_extensions.py

@@ -10,7 +10,7 @@ except ImportError:
 
 from functools import partial
 from senpy.extensions import Senpy
-from senpy.models import Error, Results, Entry, EmotionSet, Emotion
+from senpy.models import Error, Results, Entry, EmotionSet, Emotion, Plugin
 from flask import Flask
 from unittest import TestCase
 
@@ -98,17 +98,26 @@ class ExtensionsTest(TestCase):
 
     def test_analyse_error(self):
         mm = mock.MagicMock()
-        mm.analyse_entry.side_effect = Error('error on analysis', status=900)
+        mm.id = 'magic_mock'
+        mm.analyse_entries.side_effect = Error('error on analysis', status=500)
         self.senpy.plugins['MOCK'] = mm
-        resp = self.senpy.analyse(input='nothing', algorithm='MOCK')
-        assert resp['message'] == 'error on analysis'
-        assert resp['status'] == 900
+        try:
+            self.senpy.analyse(input='nothing', algorithm='MOCK')
+            assert False
+        except Error as ex:
+            assert ex['message'] == 'error on analysis'
+            assert ex['status'] == 500
+
         mm.analyse.side_effect = Exception('generic exception on analysis')
-        mm.analyse_entry.side_effect = Exception(
+        mm.analyse_entries.side_effect = Exception(
             'generic exception on analysis')
-        resp = self.senpy.analyse(input='nothing', algorithm='MOCK')
-        assert resp['message'] == 'generic exception on analysis'
-        assert resp['status'] == 500
+
+        try:
+            self.senpy.analyse(input='nothing', algorithm='MOCK')
+            assert False
+        except Error as ex:
+            assert ex['message'] == 'generic exception on analysis'
+            assert ex['status'] == 500
 
     def test_filtering(self):
         """ Filtering plugins """
@@ -125,11 +134,12 @@ class ExtensionsTest(TestCase):
 
     def test_convert_emotions(self):
         self.senpy.activate_all()
-        plugin = {
+        plugin = Plugin({
             'id': 'imaginary',
             'onyx:usesEmotionModel': 'emoml:fsre-dimensions'
-        }
+        })
         eSet1 = EmotionSet()
+        eSet1.prov__wasGeneratedBy = plugin['id']
         eSet1['onyx:hasEmotion'].append(Emotion({
             'emoml:arousal': 1,
             'emoml:potency': 0,
@@ -145,19 +155,19 @@ class ExtensionsTest(TestCase):
                   'conversion': 'full'}
         r1 = deepcopy(response)
         self.senpy.convert_emotions(r1,
-                                    plugin,
+                                    [plugin, ],
                                     params)
         assert len(r1.entries[0].emotions) == 2
         params['conversion'] = 'nested'
         r2 = deepcopy(response)
         self.senpy.convert_emotions(r2,
-                                    plugin,
+                                    [plugin, ],
                                     params)
         assert len(r2.entries[0].emotions) == 1
         assert r2.entries[0].emotions[0]['prov:wasDerivedFrom'] == eSet1
         params['conversion'] = 'filtered'
         r3 = deepcopy(response)
         self.senpy.convert_emotions(r3,
-                                    plugin,
+                                    [plugin, ],
                                     params)
         assert len(r3.entries[0].emotions) == 1