Ysignal - WeakRef Signal/Slots - Printable Version +- Python Forum (https://python-forum.io) +-- Forum: General (https://python-forum.io/forum-1.html) +--- Forum: Code sharing (https://python-forum.io/forum-5.html) +--- Thread: Ysignal - WeakRef Signal/Slots (/thread-430.html) |
Ysignal - WeakRef Signal/Slots - Yoriz - Oct-10-2016 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() RE: Ysignal - WeakRef Signal/Slots - micseydel - Oct-12-2016 Could you elaborate on what this is about? RE: Ysignal - WeakRef Signal/Slots - Yoriz - Oct-12-2016 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')
|