Python Forum
Ysignal - WeakRef Signal/Slots
Thread Rating:
  • 0 Vote(s) - 0 Average
  • 1
  • 2
  • 3
  • 4
  • 5
Ysignal - WeakRef Signal/Slots
#1
signal.py
'''
@author: Yoriz
'''
import inspect
import weakref

class Ysignal(object):

    '''WeakRef Signal/Slots'''

    def __init__(self):
        '''Initialise attributes to store observers'''
        self._functions = weakref.WeakSet()
        self._methods = weakref.WeakKeyDictionary()

    def emit_slot(self, slot, *args, **kwargs):
        '''emit a signal to the passed in slot only'''
        slot(*args, **kwargs)

    def emit(self, *args, **kwargs):
        '''emit a signal to all slots'''
        self._emit_functions(*args, **kwargs)
        self._emit_methods(*args, **kwargs)

    def _emit_functions(self, *args, **kwargs):
        '''Emits a signal to any Function slots'''
        for func in tuple(self._functions):
            func(*args, **kwargs)

    def _emit_methods(self, *args, **kwargs):
        '''Emits a signal to any Method slots'''
        for obj, funcs in self._methods.items():
            for func in tuple(funcs):
                method = getattr(obj, func.__name__)
                method(*args, **kwargs)

    def bind(self, slot):
        '''Add a slot to the list of listeners'''
        if inspect.ismethod(slot):
            self._bind_method(slot)
        else:
            self._bind_function(slot)

    def _bind_function(self, slot):
        '''Add a Function slot'''
        self._functions.add(slot)

    def _bind_method(self, slot):
        '''Add a Method slot'''
        try:
            self._methods[slot.__self__].add(slot.__func__)
        except KeyError:
            self._methods[slot.__self__] = set()
            self._bind_method(slot)

    def unbind(self, slot):
        '''Remove slot from the list of listeners'''
        if inspect.ismethod(slot):
            self._unbind_method(slot)
        else:
            self._unbind_function(slot)

    def _unbind_function(self, slot):
        '''Remove a Function slot'''
        try:
            self._functions.remove(slot)
        except (ValueError, KeyError):
            pass

    def _unbind_method(self, slot):
        '''Remove a Method slot'''
        try:
            self._methods[slot.__self__].remove(slot.__func__)
        except (ValueError, KeyError):
            pass

    def unbind_all(self):
        '''Remove all slots'''
        self._functions.clear()
        self._methods.clear()
test_ysignal.py
'''
@author: Yoriz
'''
import unittest
import ysignal

class TestYsignal(unittest.TestCase):

    def setUp(self):
        self.signal = ysignal.Ysignal()
        self.attr1 = None
        self.attr2 = None
        self.attr3 = 0

    def tearDown(self):
        pass

    def set_attr1(self, value):
        self.attr1 = value

    def method_for_test(self):
        pass

    def test_connect_function(self):
        def test_function():
            pass
        func = test_function
        self.signal.bind(func)
        self.assertSetEqual(set((func,)), self.signal._functions)

    def test_connect_method(self):
        self.signal.bind(self.set_attr1)
        self.assertIn(self, self.signal._methods)
        funcs = self.signal._methods.get(self, set())
        self.assertIn(self.set_attr1.__func__, funcs)

    def test_emit_function(self):
        value = 'EmitFunction'
        def test_function(value):
            self.attr1 = value
        self.signal.bind(test_function)
        self.signal.emit(value=value)
        self.assertEqual(self.attr1, value)

    def test_emit_method(self):
        value = 'EmitMethod'
        self.signal.bind(self.set_attr1)
        self.signal.emit(value=value)
        self.assertEqual(self.attr1, value)

    def test_emit_function_and_method(self):
        value = 'EmitBothTypes'
        def test_function(value):
            self.attr2 = value
        self.signal.bind(test_function)
        self.signal.bind(self.set_attr1)
        self.signal.emit(value=value)
        self.assertEqual(self.attr1, value)
        self.assertEqual(self.attr2, value)

    def test_emit_slot_function(self):
        value = 'EmitSlotFunction'
        def test_function(value):
            self.attr1 = value
        self.signal.emit_slot(test_function, value=value)
        self.assertEqual(self.attr1, value)

    def test_emit_slot_method(self):
        value = 'EmitMethod'
        self.signal.emit_slot(self.set_attr1, value=value)
        self.assertEqual(self.attr1, value)

    def test_disconnect_slot_function(self):
        def test_function():
            pass
        def test_function2():
            pass
        func = test_function
        func2 = test_function2
        self.signal.bind(func)
        self.signal.bind(func2)
        self.assertSetEqual(set((func, func2)), self.signal._functions)
        self.signal.unbind(func2)
        self.assertSetEqual(set((func,)), self.signal._functions)

    def test_disconnect_slot_method(self):
        self.signal.bind(self.set_attr1)
        self.signal.bind(self.method_for_test)
        self.assertIn(self, self.signal._methods)
        funcs = self.signal._methods.get(self, [])
        self.assertSetEqual(set((self.set_attr1.__func__,
                                self.method_for_test.__func__)), funcs)
        self.signal.unbind(self.method_for_test)
        funcs = self.signal._methods.get(self, set())
        self.assertSetEqual(set((self.set_attr1.__func__,)), funcs)

    def test_disconnect_all(self):
        def test_function():
            pass
        def test_function2():
            pass
        func = test_function
        func2 = test_function2
        self.signal.bind(func)
        self.signal.bind(func2)
        self.signal.bind(self.set_attr1)
        self.signal.bind(self.method_for_test)
        self.signal.unbind_all()
        self.assertSetEqual(set(), self.signal._functions)
        funcs = self.signal._methods.get(self, set())
        self.assertSetEqual(set(), funcs)

if __name__ == '__main__':
    unittest.main()
#2
Could you elaborate on what this is about?
#3
It is my implementation of the Observer pattern, I don't generally use it directly but import it into another script which is my implementation of the Model–view–controller pattern.
Which is then used with my GUI programming.

Here is an example of it in action using a forum thread subscription example.
import ysignal

class ForumThread:

    def __init__(self):
        self.ysignal = ysignal.Ysignal()

    def post_to_thread(self, post):
        self.ysignal.emit(post)

    def subscribe_to_thread(self, slot):
        self.ysignal.bind(slot)

class Forum:

    def update_forum_on_new_post(self, post):
        print('The forum updated its new post list: {}'.format(post))

def forum_dweller(post):
    print('forum_dweller got the post: {}'.format(post))

def another_forum_dweller(post):
    print('another_forum_dweller got the post: {}'.format(post))

forum = Forum()
forum_thread = ForumThread()
forum_thread.subscribe_to_thread(forum.update_forum_on_new_post)
forum_thread.subscribe_to_thread(forum_dweller)
forum_thread.subscribe_to_thread(another_forum_dweller)
forum_thread.post_to_thread('someone posted something')
del forum_dweller #  this forum dweller got banned
forum_thread.post_to_thread('something else was posted')
Output:
forum_dweller got the post: someone posted something another_forum_dweller got the post: someone posted something The forum updated its new post list: someone posted something another_forum_dweller got the post: something else was posted The forum updated its new post list: something else was posted


Forum Jump:

User Panel Messages

Announcements
Announcement #1 8/1/2020
Announcement #2 8/2/2020
Announcement #3 8/6/2020