Something like signals/slots from Qt http://doc.trolltech.com/2.3/signalsandslots.html, but much more simpler (you can't pass arguments.)
More documentation in the code. For the latest version, check out http://cvs.sourceforge.net/cgi-bin/viewcvs.cgi/bttotalcontrol/bttc-python/lib/flags.py.
code
#=Copyright Notice========================================================={{{1
#
# Copyright © 2001, 2002 Jonathan Gardner (jgardn@alumni.washington.edu)
#
# This file is part of BattleTech: Total Control (BTTC).
#
# BTTC is free software; you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation; either version 2 of the License, or any later version.
#
# BTTC is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with
# BTTC; if not, write to the Free Software Foundation, Inc., 59 Temple Place,
# Suite 330, Boston, MA 02111-1307 USA
#
#==============================================================================
# __doc__ {{{1
"""Flags
The flags module provides some simple and powerful signalling for your python
programming enjoyment.
A flag is an object that keeps a list of what is watching it. When the flag is
waved, it calls all of the things that are watching it. This means that only
callable objects can watch a flag.
The flag stores the objects that watch it with weakrefs. In order to store
bound instance methods, it uses a MethodWeakRef that keeps a weakref to the
instance and method. This code is based off of a post by Jason Orendorff
(jason@jorendorff.com) to comp.lang.python. You can view his post at
http://groups.google.com/groups?hl=en&selm=mailman.1008803665.26406.python-list%40python.org
The flagging code is optimized somewhat. It keeps its list of watchers clean by
clearing out the dead refs everytime a watcher is added or removed, or when the
flag is waved. The code for doing this runs in the loop that loops over the
watchers, so it is very efficient.
Also, waving a flag can cause other flags to be waved directly or indirectly.
If an object is watching two different flags, and the waving of one flag will
cause the other one to wave, then it would seem that the object would be called
twice. There is a way of avoiding this: keeping a list of all the objects that
need to be called, adding to it when another flag is waved, and removing from
it when the object is called. This way, the number of calls is greatly reduced.
How can you use a flag? The simplest way is to do something when it is waved.
For instance, if you are writing a GUI, and you want a window to be created
when a button is pushed, you can attach a flag to the button, and have the
function that creates the window watch the flag. This is very easy code to
maintain, because if you also want to change the contents of a text box when
that button is pressed, you can watch the flag with the functions that changes
the content of a text box.
A flagpole is a group of flags. It does a pretty good job of grouping the flags
together, and it has an efficient way of allowing a function to watch all the
flags simultaneously.
A flag_attr class is a special class that redefines __setattr__ to watch for
when an attribute it changed. When an attribute is changed, it will wave a flag
that has the same name as that attribute. This is very useful for GUI type
applications, where you want to keep the text in the GUI in sync with the text
in the box. Rather than putting code into your object to tell the GUI when to
update, you can have the GUI watch the object for when something changes. Add
to that the watch_all_flags ability of a flagpole, and you have a very
comfortable solution.
"""
# import {{{1
from weakref import ref
from new import instancemethod
from types import StringType, ListType, TupleType, MethodType
# classes {{{1
class MethodWeakRef: # {{{2
# __doc__ {{{3
"""A weakref for methods.
This extracts the class and function from the method. It holds on to a weak
ref of the instance and throws away the method.
When it is called, it reconstructs the method.
Thanks go to Jason Orendorff for this code.
"""
# methods {{{3
def __init__(self, o): # {{{4
self.s, self.f = ref(o.im_self), ref(o.im_func)
def __call__( self ): # {{{4
s = self.s()
f = self.f()
if s is None or f is None:
return None
else:
return instancemethod(f, s, s.__class__)
class flag: # {{{2
# __doc__ {{{3
"""A flag.
A flag is a piece of cloth that has bright colors on it so that people can
watch it during battle.
Well... this flag isn't made of cloth. It can be watched however, and
waved. Whenever it is waved, all the objects that are watching it get
called.
This uses weakrefs, but instance methods are okay.
"""
# attributes {{{3
_wave_stack = []
# methods {{{3
def __init__(self): # {{{4
self.watchers = []
def watch(self, w): # {{{4
"""Start watching the flag.
Whenever the flag is waved, w will be called. It must be a callable.
Note that this stores weakrefs, but instance methods are okay.
"""
def is_already_in(ref, me=w):
r = ref()
if r is None:
return 0
if r is me:
raise ValueError, "You are already watching this flag!"
return 1
self.watchers = filter(is_already_in, self.watchers)
# Make a new weakref, add it to the list of watchers
if type(w) == MethodType:
self.watchers.append(MethodWeakRef(w))
else:
if not callable(w):
raise ValueError, "Watcher is not callable."
self.watchers.append(ref(w))
def stop_watching(self, w): # {{{4
"""Remove the watcher from the list.
This can be a function, and instancemethod, or an instance. If it is an
instance, it will remove all instancemethods that are in here, as well
as any of itself.
"""
def not_me(ref, me=w):
if ref() is None:
return 0
if ref() is me:
return 0
if isinstance(ref, MethodWeakRef) and ref.s() is me:
return 0
return 1
self.watchers = filter(not_me, self.watchers)
def wave(self): # {{{4
"""Wave the flag.
There is a _watch_stack. It has a list of all the things that are going
to be called. This is setup so that something isn't called twice if it
is watching two different flags that are somehow connected.
"""
def add_to_stack(ref):
r = ref()
if r is None:
return 0
for i in flag._wave_stack:
if i == r:
return 1
flag._wave_stack.append(r)
return 1
self.watchers = filter(add_to_stack, self.watchers)
while flag._wave_stack:
flag._wave_stack.pop(0)()
class flagpole: # {{{2
# __doc__ {{{3
"""A bunch of flags with names.
You can raise a flag, lower a flag, watch a flag, find a flag, and wave a
flag.
Create a flag with "raise_flag". This create a new flag with a name. The
name can be any immutable.
Destroy a flag with "lower_flag". This removes the flag.
Find a flag with "find_flag". (Test for existence with "has_flag".)
Wave a flag with "wave_flag".
Watch a flag with "watch_flag".
"""
# methods {{{3
def __init__(self): # {{{4
if not hasattr(self, '_flagpole__flags'):
self.__flags = {}
self.__flags['_any'] = flag()
def has_flag(self, f): # {{{4
"""Is there a flag?"""
return self.get_flag(f) is not None
def get_flag(self, f): # {{{4
"""Get the flag object."""
try:
return self.__flags[f]
except KeyError:
return None
def raise_flag(self, f): # {{{4
"""Create a new flag."""
try:
self.__flags[f]
except KeyError:
self.__flags[f] = flag()
# Any flag waving will wave the _any flag.
self.__flags[f].watch(self.__flags['_any'].wave)
else:
raise ValueError, "There is already a flag named '%s'" % f
def lower_flag(self, f): # {{{4
"""Destroy the flag."""
if not self.has_flag(f):
raise ValueError, "No such flag: '%s'" % f
del self.__flags[f]
def wave_flag(self, f): # {{{4
"""Wave the flag."""
if not self.has_flag(f):
raise ValueError, "No such flag: '%s'" % f
self.get_flag(f).wave()
def watch_flag(self, f, w): # {{{4
"""Watch a flag."""
if not self.has_flag(f):
raise ValueError, "No such flag: '%s'" % f
self.get_flag(f).watch(w)
def watch_all_flags(self, w): # {{{4
"""Watch all the flags."""
self.watch_flag('_any', w)
def stop_watching_flag(self, f, w): # {{{4
"""Stop watching a flag."""
if not self.has_flag(f):
raise ValueError, "No such flag: '%s'" % f
self.get_flag(f).stop_watching(w)
def stop_watching_all_flags(self, w): # {{{4
for f in self.__flags.values():
f.stop_watching(w)
class flag_attr(flagpole): # {{{2
# __doc__ {{{3
"""Wave a flag when an attribute changes.
"""
# methods {{{3
def __init__(self): # {{{4
"""Set up the flagpole."""
flagpole.__init__(self)
def __setattr__(self, attr, value): # {{{4
"""Sets the attribute and waves a flag if necessary.
This is what happens when you set an attribute.
If there is no flag for it, nothing happens. It just gets set.
If there is a flag for it, then that means people are watching it. The
members are added to the stack after the variable is changed, and the
stack is executed.
If there is an _all flag, wave that as well.
"""
if attr == '_flags':
raise AttributeError, "What attribute are you talking about?"
elif self.__dict__.has_key(attr):
if value == self.__dict__[attr]:
# Don't want to have to write code that checks before assigning
# to a value
return
# Assign the new value
self.__dict__[attr] = value
if self.has_flag(attr):
self.wave_flag(attr)
if self.has_flag('_all'):
self.wave_flag('_all')
else:
# Humm, no attribute yet...
self.__dict__[attr] = value
def __delattr__(self, attr): # {{{4
if self.has_flag(attr):
self.lower_flag(attr)
def watch_attr(self, attr, w): # {{{4
"""Tells w when the attr changes.
Note: This guy is pretty smart. If you give it an attribute of
"foo.bar", it will watch self.foo.bar, not self."foo.bar".
"""
if attr.find('.') != -1:
(obj, attr) = attr.split('.', 1)
try:
obj = getattr(self, obj)
except AttributeError:
raise ValueError, "There is no attribute '%s'." % obj
obj.watch_attr(attr, w)
return
# If this raises an AttributeError, that is because there is no
# attribute.
getattr(self, attr)
if not self.has_flag(attr):
self.raise_flag(attr)
self.watch_flag(attr, w)
def stop_watching_attr(self, attr, w): # {{{4
"""Stops telling when an attribute changes."""
if attr.find('.') != -1:
(obj, attr) = attr.split('.', 1)
try:
obj = getattr(self, obj)
except AttributeError:
raise ValueError, "There is no attribute '%s'." % obj
obj.stop_watching_attr(attr, w)
return
# If this raises an AttributeError, that is because there is no
# attribute.
getattr(self, w)
self.stop_watching_flag(attr, w)
class rulebased(flag_attr): # {{{2
# __doc__ {{{3
"""A rulebased class.
This is a class that has several attributes that are rule-based.
To make an attribute apply to these rules, it has to staisfy the following:
1) It must exist before rulebased.__init__ is called.
2) It must be available through "getattr(self, <attr name>)"
3) It must be a key in the _attr of the parent class. If you want to
inherit the _attr from a base class, use copy.deepcopy. The value should be
a dict.
The attributes are limited by a function called the "check" function. Whenever
the attribute changes, the check function is called first. If the value
that the attribute is set to doesn't make you happy, raise a ValueError.
The attributes are calculated using the "calc" function. Whenever a flag
that it watches or an attribute that it depends on is changd, the calc
function is called, recalculating the value of the attribute. To tell
rulebased what attributes and flags it depends on, set _attr['<attr
name>']['depends'] = ('attributes', 'I', 'depend', 'on'). For flags, use
"watches" rather than depends.
Here is an example class.
from types import *
class bill(rulebased):
_attr = {}
_attr['total'] = {
'depends':('subtotal', 'tax')
}
def _check_total(self):
if type(self.total) is not FloatType:
raise ValueError, "'total' must be a float."
def _calc_total(self):
self.total = self.subtotal + self.tax
_attr['tax'] = {
'depends':'total',
}
def _check_tax(self):
if type(self.tax) is not FloatType:
raise ValueError, "'tax' must be a float."
def _calc_tax(self):
self.tax = self.subtotal * 0.1
_attr['subtotal'] = {
'watches':'items_changed',
}
def _check_subtotal(self):
if type(self.subtotal) is not FloatType:
raise ValueError, "'subtotal' must be a float."
def _calc_subtotal(self):
self.subtotal = reduce(lambda x,y:x+y, self.items, 0.0)
def __init__(self):
rulebased.__init__(self)
self.total = 0.0
self.subtotal = 0.0
self.tax = 0.0
self.items = []
self.raise_flag('items_changed')
self._apply_rules()
b = bill()
b.items.append(3.50)
b.items.append(4.75)
b.items.append(2.50)
b.items.append(12.00)
b.wave_flag('items_changed')
print "Items:"
for i in b.items:
print "$%.2f" % i
print "Subtotal: $%.2f" % b.subtotal
print "Tax: $%.2f" % b.tax
print "Total: $%.2f" % b.total
print "Have a nice day!"
Output should be:
Items:
$3.50
$4.75
$2.50
$12.00
Subtotal: $22.75
Tax: $2.27
Total: $25.02
Have a nice day!
"""
# attributes {{{3
_attr = {}
# methods {{{3
def __init__(self): # {{{4
flag_attr.__init__(self)
def __setattr__(self, attr, value): # {{{4
"""Call the check func if it exists."""
try:
getattr(self, "_check_%s"%attr)()
except AttributeError:
pass
flag_attr.__setattr__(self, attr, value)
def _apply_rules(self): # {{{4
# First pass: Create the attribute
for (attr, val) in self._attr.items():
if not hasattr(self, attr):
raise AttributeError, (
"You need to initialize '%s' before applying the rules."
) % attr
# Second pass: Watch flags and watch_attr about attributes
for (attr, val) in self._attr.items():
try:
meth = getattr(self, "_calc_%s"%attr)
meth()
except AttributeError:
pass
else:
try:
dep = val['depends']
except KeyError: pass
else:
if type(dep) is StringType:
self.watch_attr(dep, meth)
elif type(dep) is ListType or type(dep) is TupleType:
for a in dep: self.watch_attr(a, meth)
try:
watches = val['watches']
except KeyError: pass
else:
if type(watches) is StringType:
self.watch_flag(watches, meth)
elif type(watches) is ListType or \
type(watches) is TupleType:
for a in watches: self.watch_flag(a, meth)
# 1
}}}
comments on the code
There seems to be an error in the example (class bill) :
When I try this out at the prompt the init fails with the following traceback:
Traceback (most recent call last):
- File "E:\python.samples\flag.py", line 552, in ?
- b = bill()
File "E:\python.samples\flag.py", line 544, in init
- self.total = 0.0
File "E:\python.samples\flag.py", line 473, in setattr
- getattr(self, "_check_%s"%attr)(value)
TypeError: _check_total() takes exactly 1 argument (2 given)
And indeed the _check_xxx function is defined as referring to the already existing value, but it is called with the new value.
HTH Norbert.Klamann@klamann-software.de
Response from the author
I think I fixed it. Thanks for pointing that out. The new code is posted above.