Velocity Reviews - Computer Hardware Reviews

Velocity Reviews > Newsgroups > Programming > Python > Signals/Slots support in Python

Thread Tools

Signals/Slots support in Python

Brian Vanderburg II
Posts: n/a
I don't know if any such support is already built in, so I ended up
making my own simple signals/slots like mechanism. If anyone is
interested then here it is, along with a simple test. It can connect to
normal functions as well as instance methods. It also supports weak
connections where when an object is gone, the slot is gone as well, the
slot just holds a weak reference to the object.

Brian Vanderburg II

# Begin Signal
import weakref
import random

class Signal:
class Slot:
def __init__(self, fn):
self.__fn = fn

def __call__(self, accum, *args, **kwargs):
result = self.__fn(*args, **kwargs)
return accum(result)

class WeakSlot:
def __init__(self, conn, parent, fn, obj):
self.__conn = conn
# Avoid circular references so deleting a signal will
# allow deletion of the signal since the slot doesn't ref
# back to it but only weakefs back to it
self.__parent = weakref.ref(parent)

self.__fn = fn
self.__obj = weakref.ref(obj, self.Cleanup)

def __call__(self, accum, *args, **kwargs):
obj = self.__obj()
if obj is None:
return True

result = self.__fn(obj, *args, **kwargs)
return accum(result)

def Cleanup(self, ref):
parent = self.__parent()
if parent is not None:

class Accumulator:
def __call__(self, *args, **kwargs):
return True

def Finalize(self):
return None

def __init__(self):
self.__slots = [ ]

# This connects a signal to a slot, but stores a strong reference so
# The object will not be deleted as long as the signal is connected
def Connect(self, fn):
conn = self.NewConn()
self.__slots.append([conn, Signal.Slot(fn)])
return conn

# This connects a signal to a slot, but store a weak reference so
# when the object is gone the slot will not be called. Because of
# the implemenations, it is not possible to do WeakConnect(obj.Fn),
# since obj.Fn is a new object and would go to 0 refcount soon after
# the call to WeakConnect completes. Instead we must do a call as
# WeakConnect(ObjClass.Fn, obj)
# Only the object is weak-referenced. The function object is still
# a normal reference, this ensures that as long as the object exists
# the function will also exist. When the object dies, the slot will
# be removed
def WeakConnect(self, fn, obj):
conn = self.NewConn()
self.__slots.append([conn, Signal.WeakSlot(conn, self, fn, obj)])
return conn

# Disconnect a slot
def Disconnect(self, conn):
result = self.Find(conn)
if result >= 0:
del self.__slots[result]

# Disconnect all slots
def DisconnectAll(self):
self.__slots = [ ]

# Create an accumulator. Accumulator will be called as a callable
# for each return value of the executed slots. Execution of slots
# continues as long as the reutrn value of the accumulator call is
# True. The 'Finalize'function will be called to get the result
# A custom accumulator can be created by deriving from Signal and
# Creating a custom 'Accumulator' class, or by deriving from Singal
# and creating CreateAccumulator
def CreateAccumulator(self):
return self.Accumulator()

# Execute the slots
def __call__(self, *args, **kwargs):
accum = self.CreateAccumulator()
for conn in xrange(len(self.__slots)):
if not self.__slots[conn][1](accum, *args, **kwargs):
return accum.Finalize()

# Create a connection name
def NewConn(self):
value = 0
while self.Find(value) >= 0:
value = random.randint(1, 100000000)
return value

def Find(self, conn):
for i in xrange(len(self.__slots)):
if self.__slots[i][0] == conn:
return i

return -1

# End Signal

def fn1():
print "Hello World"

def fn2():
print "Goodbye Space"

class O:
def __init__(self, value):
self.value = value

def Action(self):
print "O %d" % self.value

a = Signal()


print "Part 1"


o1 = O(4)
o2 = O(12)

a.WeakConnect(O.Action, o1)

print "Part 2"

print "Part 3"
o1 = None

print "Part 4"
o2 = None


def f1():
print "Hello Neighbor"

def f2():
print "Back to Work"

c1 = a.Connect(f1)
c2 = a.Connect(f2)

print "Part 5"

print "Part 6"


def f1(name):
print "Hello %s" % name

def f2(name):
print "Goodbye %s" % name


print "Part 7"
#a() # Error


class MySignal(Signal):
class Accumulator:
def __init__(self):
self.value = 0
def __call__(self, value):
self.value += value
return bool(value != 0)
def Finalize(self):
return self.value

def f1(x):
return x * x

def f2(x):
return x + x

def f3(x):
return 0

a = MySignal()

print "Part 8"
print a(5)

Reply With Quote

Thread Tools

Posting Rules
You may not post new threads
You may not post replies
You may not post attachments
You may not edit your posts

BB code is On
Smilies are On
[IMG] code is On
HTML code is Off
Trackbacks are On
Pingbacks are On
Refbacks are Off

Similar Threads
Thread Thread Starter Forum Replies Last Post
IEEE 1284.4 generic printing support / HP LaserJet printer support Manuel Lopez Windows 64bit 13 10-18-2009 12:51 PM
Question for IT Support/Managers - Providing Personal PC Support Big Dog Microsoft Certification 4 01-16-2008 01:33 AM
Anti Virus Support + Japanese character support =?Utf-8?B?Q2hyaXMgQnVzaA==?= Windows 64bit 6 07-31-2005 09:57 AM
ANN: SCons.0.96 adds Fortran 90/95 support, better Qt support,platform-independent file system actions, improved debugging, lots more Steven Knight Python 0 08-18-2004 03:57 PM
Getting Third Party Component Suppliers to support NUnit and NUnitASP to support test driven development in web pages Nick Zdunic ASP .Net 0 11-05-2003 10:45 AM