mirror of
https://github.com/gsi-upm/senpy
synced 2024-11-24 17:12:29 +00:00
371 lines
11 KiB
Python
371 lines
11 KiB
Python
#!/bin/env python
|
|
|
|
import os
|
|
import sys
|
|
import pickle
|
|
import shutil
|
|
import tempfile
|
|
|
|
from unittest import TestCase, skipIf
|
|
from senpy.models import Results, Entry, EmotionSet, Emotion, Plugins
|
|
from senpy import plugins
|
|
from senpy.plugins.conversion.emotion.centroids import CentroidConversion
|
|
|
|
import pandas as pd
|
|
|
|
|
|
class ShelfDummyPlugin(plugins.SentimentPlugin, plugins.ShelfMixin):
|
|
'''Dummy plugin for tests.'''
|
|
name = 'Shelf'
|
|
version = 0
|
|
author = 'the senpy community'
|
|
|
|
def activate(self, *args, **kwargs):
|
|
if 'counter' not in self.sh:
|
|
self.sh['counter'] = 0
|
|
self.save()
|
|
|
|
def deactivate(self, *args, **kwargs):
|
|
self.save()
|
|
|
|
def analyse(self, *args, **kwargs):
|
|
self.sh['counter'] = self.sh['counter'] + 1
|
|
e = Entry()
|
|
e.nif__isString = self.sh['counter']
|
|
r = Results()
|
|
r.entries.append(e)
|
|
return r
|
|
|
|
|
|
class PluginsTest(TestCase):
|
|
def tearDown(self):
|
|
if os.path.exists(self.shelf_dir):
|
|
shutil.rmtree(self.shelf_dir)
|
|
if os.path.isfile(self.shelf_file):
|
|
os.remove(self.shelf_file)
|
|
|
|
def setUp(self):
|
|
self.shelf_dir = tempfile.mkdtemp()
|
|
self.shelf_file = os.path.join(self.shelf_dir, "shelf")
|
|
|
|
def test_serialize(self):
|
|
'''A plugin should be serializable and de-serializable'''
|
|
dummy = ShelfDummyPlugin()
|
|
dummy.serialize()
|
|
|
|
def test_jsonld(self):
|
|
'''A plugin should be serializable and de-serializable'''
|
|
dummy = ShelfDummyPlugin()
|
|
dummy.jsonld()
|
|
|
|
def test_shelf_file(self):
|
|
a = ShelfDummyPlugin(
|
|
info={'name': 'default_shelve_file',
|
|
'description': 'Dummy plugin for tests',
|
|
'version': 'test'})
|
|
a.activate()
|
|
assert os.path.isfile(a.shelf_file)
|
|
os.remove(a.shelf_file)
|
|
|
|
def test_plugin_filter(self):
|
|
ps = Plugins()
|
|
for i in (plugins.SentimentPlugin,
|
|
plugins.EmotionPlugin,
|
|
plugins.AnalysisPlugin):
|
|
p = i(name='Plugin_{}'.format(i.__name__),
|
|
description='TEST',
|
|
version=0,
|
|
author='NOBODY')
|
|
ps.plugins.append(p)
|
|
assert len(ps.plugins) == 3
|
|
cases = [('AnalysisPlugin', 3),
|
|
('SentimentPlugin', 1),
|
|
('EmotionPlugin', 1)]
|
|
|
|
for name, num in cases:
|
|
res = list(plugins.pfilter(ps.plugins, plugin_type=name))
|
|
assert len(res) == num
|
|
|
|
def test_shelf(self):
|
|
''' A shelf is created and the value is stored '''
|
|
newfile = self.shelf_file + "new"
|
|
a = ShelfDummyPlugin(info={
|
|
'name': 'shelve',
|
|
'description': 'Shelf plugin for tests',
|
|
'version': 'test',
|
|
'shelf_file': newfile
|
|
})
|
|
assert a.sh == {}
|
|
a.activate()
|
|
assert a.sh == {'counter': 0}
|
|
assert a.shelf_file == newfile
|
|
|
|
a.sh['a'] = 'fromA'
|
|
assert a.sh['a'] == 'fromA'
|
|
|
|
a.save()
|
|
|
|
sh = pickle.load(open(newfile, 'rb'))
|
|
|
|
assert sh['a'] == 'fromA'
|
|
|
|
def test_dummy_shelf(self):
|
|
with open(self.shelf_file, 'wb') as f:
|
|
pickle.dump({'counter': 99}, f)
|
|
a = ShelfDummyPlugin(info={
|
|
'name': 'DummyShelf',
|
|
'description': 'Dummy plugin for tests',
|
|
'shelf_file': self.shelf_file,
|
|
'version': 'test'
|
|
})
|
|
a.activate()
|
|
|
|
assert a.shelf_file == self.shelf_file
|
|
res1 = a.analyse(input=1)
|
|
assert res1.entries[0].nif__isString == 100
|
|
a.deactivate()
|
|
del a
|
|
|
|
with open(self.shelf_file, 'rb') as f:
|
|
sh = pickle.load(f)
|
|
assert sh['counter'] == 100
|
|
|
|
def test_corrupt_shelf(self):
|
|
''' Reusing the values of a previous shelf '''
|
|
|
|
emptyfile = os.path.join(self.shelf_dir, "emptyfile")
|
|
invalidfile = os.path.join(self.shelf_dir, "invalid_file")
|
|
with open(emptyfile, 'w+b'), open(invalidfile, 'w+b') as inf:
|
|
inf.write(b'ohno')
|
|
|
|
files = {emptyfile: ['empty file', (EOFError, IndexError)],
|
|
invalidfile: ['invalid file', (pickle.UnpicklingError, IndexError)]}
|
|
|
|
for fn in files:
|
|
with open(fn, 'rb') as f:
|
|
msg, error = files[fn]
|
|
a = ShelfDummyPlugin(info={
|
|
'name': 'test_corrupt_shelf_{}'.format(msg),
|
|
'description': 'Dummy plugin for tests',
|
|
'version': 'test',
|
|
'shelf_file': f.name
|
|
})
|
|
assert os.path.isfile(a.shelf_file)
|
|
print('Shelf file: %s' % a.shelf_file)
|
|
with self.assertRaises(error):
|
|
a.sh['a'] = 'fromA'
|
|
a.save()
|
|
del a._sh
|
|
assert os.path.isfile(a.shelf_file)
|
|
a.force_shelf = True
|
|
a.sh['a'] = 'fromA'
|
|
a.save()
|
|
b = pickle.load(f)
|
|
assert b['a'] == 'fromA'
|
|
|
|
def test_reuse_shelf(self):
|
|
''' Reusing the values of a previous shelf '''
|
|
a = ShelfDummyPlugin(info={
|
|
'name': 'shelve',
|
|
'description': 'Dummy plugin for tests',
|
|
'version': 'test',
|
|
'shelf_file': self.shelf_file
|
|
})
|
|
a.activate()
|
|
print('Shelf file: %s' % a.shelf_file)
|
|
a.sh['a'] = 'fromA'
|
|
a.save()
|
|
|
|
b = ShelfDummyPlugin(info={
|
|
'name': 'shelve',
|
|
'description': 'Dummy plugin for tests',
|
|
'version': 'test',
|
|
'shelf_file': self.shelf_file
|
|
})
|
|
b.activate()
|
|
assert b.sh['a'] == 'fromA'
|
|
b.sh['a'] = 'fromB'
|
|
assert b.sh['a'] == 'fromB'
|
|
|
|
def test_extra_params(self):
|
|
''' Should be able to set extra parameters'''
|
|
a = ShelfDummyPlugin(info={
|
|
'name': 'shelve',
|
|
'description': 'Dummy shelf plugin for tests',
|
|
'version': 'test',
|
|
'shelf_file': self.shelf_file,
|
|
'extra_params': {
|
|
'example': {
|
|
'aliases': ['example', 'ex'],
|
|
'required': True,
|
|
'default': 'nonsense'
|
|
}
|
|
}
|
|
})
|
|
assert 'example' in a.extra_params
|
|
|
|
def test_box(self):
|
|
|
|
class MyBox(plugins.Box):
|
|
''' Vague description'''
|
|
|
|
author = 'me'
|
|
version = 0
|
|
|
|
def input(self, entry, **kwargs):
|
|
return entry.text
|
|
|
|
def predict_one(self, input):
|
|
return 'SIGN' in input
|
|
|
|
def output(self, output, entry, **kwargs):
|
|
if output:
|
|
entry.myAnnotation = 'DETECTED'
|
|
return entry
|
|
|
|
test_cases = [
|
|
{
|
|
'input': "nothing here",
|
|
'expected': {'myAnnotation': 'DETECTED'},
|
|
'should_fail': True
|
|
}, {
|
|
'input': "SIGN",
|
|
'expected': {'myAnnotation': 'DETECTED'}
|
|
}]
|
|
|
|
MyBox().test()
|
|
|
|
def test_sentimentbox(self):
|
|
|
|
class SentimentBox(plugins.MappingMixin, plugins.SentimentBox):
|
|
''' Vague description'''
|
|
|
|
author = 'me'
|
|
version = 0
|
|
|
|
mappings = {'happy': 'marl:Positive', 'sad': 'marl:Negative'}
|
|
|
|
def predict_one(self, input, **kwargs):
|
|
return 'happy' if ':)' in input else 'sad'
|
|
|
|
test_cases = [
|
|
{
|
|
'input': 'a happy face :)',
|
|
'polarity': 'marl:Positive'
|
|
}, {
|
|
'input': "Nothing",
|
|
'polarity': 'marl:Negative'
|
|
}]
|
|
|
|
SentimentBox().test()
|
|
|
|
def test_conversion_centroids(self):
|
|
info = {
|
|
"name": "CentroidTest",
|
|
"description": "Centroid test",
|
|
"version": 0,
|
|
"centroids": {
|
|
"c1": {"V1": 0.5,
|
|
"V2": 0.5},
|
|
"c2": {"V1": -0.5,
|
|
"V2": 0.5},
|
|
"c3": {"V1": -0.5,
|
|
"V2": -0.5},
|
|
"c4": {"V1": 0.5,
|
|
"V2": -0.5}},
|
|
"aliases": {
|
|
"V1": "X-dimension",
|
|
"V2": "Y-dimension"
|
|
},
|
|
"centroids_direction": ["emoml:big6", "emoml:fsre-dimensions"]
|
|
}
|
|
c = CentroidConversion(info)
|
|
print(c.serialize())
|
|
|
|
es1 = EmotionSet()
|
|
e1 = Emotion()
|
|
e1.onyx__hasEmotionCategory = "c1"
|
|
es1.onyx__hasEmotion.append(e1)
|
|
res = c._forward_conversion(es1)
|
|
assert res["X-dimension"] == 0.5
|
|
assert res["Y-dimension"] == 0.5
|
|
print(res)
|
|
|
|
e2 = Emotion()
|
|
e2.onyx__hasEmotionCategory = "c2"
|
|
es1.onyx__hasEmotion.append(e2)
|
|
res = c._forward_conversion(es1)
|
|
assert res["X-dimension"] == 0
|
|
assert res["Y-dimension"] == 1
|
|
print(res)
|
|
|
|
e = Emotion()
|
|
e["X-dimension"] = -0.2
|
|
e["Y-dimension"] = -0.3
|
|
res = c._backwards_conversion(e)
|
|
assert res["onyx:hasEmotionCategory"] == "c3"
|
|
print(res)
|
|
|
|
e = Emotion()
|
|
e["X-dimension"] = -0.2
|
|
e["Y-dimension"] = 0.3
|
|
res = c._backwards_conversion(e)
|
|
assert res["onyx:hasEmotionCategory"] == "c2"
|
|
|
|
@skipIf(sys.version_info < (3, 0),
|
|
reason="requires Python3")
|
|
def test_evaluation(self):
|
|
testdata = []
|
|
for i in range(50):
|
|
testdata.append(["good", 1])
|
|
for i in range(50):
|
|
testdata.append(["bad", 0])
|
|
dataset = pd.DataFrame(testdata, columns=['text', 'polarity'])
|
|
|
|
class DummyPlugin(plugins.TextBox):
|
|
description = 'Plugin to test evaluation'
|
|
version = 0
|
|
|
|
def predict_one(self, input):
|
|
return 0
|
|
|
|
class SmartPlugin(plugins.TextBox):
|
|
description = 'Plugin to test evaluation'
|
|
version = 0
|
|
|
|
def predict_one(self, input):
|
|
if input == 'good':
|
|
return 1
|
|
return 0
|
|
|
|
dpipe = DummyPlugin()
|
|
results = plugins.evaluate(datasets={'testdata': dataset}, plugins=[dpipe], flatten=True)
|
|
dumb_metrics = results[0].metrics[0]
|
|
assert abs(dumb_metrics['accuracy'] - 0.5) < 0.01
|
|
|
|
spipe = SmartPlugin()
|
|
results = plugins.evaluate(datasets={'testdata': dataset}, plugins=[spipe], flatten=True)
|
|
smart_metrics = results[0].metrics[0]
|
|
assert abs(smart_metrics['accuracy'] - 1) < 0.01
|
|
|
|
|
|
def make_mini_test(fpath):
|
|
def mini_test(self):
|
|
for plugin in plugins.from_path(fpath, install=True):
|
|
plugin.test()
|
|
return mini_test
|
|
|
|
|
|
def _add_tests():
|
|
root = os.path.join(os.path.dirname(__file__), '..')
|
|
print(root)
|
|
for fpath in plugins.find_plugins([root, ]):
|
|
pass
|
|
t_method = make_mini_test(fpath)
|
|
t_method.__name__ = 'test_plugin_{}'.format(fpath)
|
|
setattr(PluginsTest, t_method.__name__, t_method)
|
|
del t_method
|
|
|
|
|
|
_add_tests()
|