gramps/gramps2/src/GrampsDBCallback.py

#
# Gramps - a GTK+/GNOME based genealogy program
#
# Copyright (C) 2000-2005  Donald N. Allingham
#
# This program 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
# (at your option) any later version.
#
# This program 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 this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
#

# $Id$

import types
import sys

#-------------------------------------------------------------------------
#
# Callback signal support for non-gtk parts of Gramps
#
#-------------------------------------------------------------------------

class GrampsDBCallback(object):
    """
    Callback and signal support for non-gtk parts of gramps.

    Classes that want to emit signals need to inherit from this
    class and call its __init__ method. They then need to declare
    the signals that they can emit and the types of their
    arguments.

    e.g.

        class TestSignals(GrampsDBCallback):

            __signals__ = {
                      'test-signal' : (int,)
                     }

            def __init__(self):
                GrampsDBCallback.__init__(self)

            def emit_signal(self):
                self.emit('test-signal',(1,))

    The signals will be inherited by any subclasses.

    Attaching a callback to the signals is similar to the gtk
    connect methods. e.g.

        class C(object):

            def cb_func(self, i):
                print "got class signal = ", 1

        def fn(i):
            print "got signal value = ", i

        t = TestSignals()

        # connect to a function.
        t.connect('test-signal', fn)
        
        t.emit_signal()

        r = R()
    """

    # If this True no signals will be emitted from any instance of
    # any class derived from this class. This should be toggled using
    # the class methods, dissable_all_signals() and enable_all_signals().
    __BLOCK_ALL_SIGNALS = False
    
    def __init__(self):
        self.__block_instance_signals = False # controls the blocking of
                                              # signals from this instance
        self.__callback_map = {} # dictionary containing all the connected
                                 # callback functions. The keys are the
                                 # signal names and the values are the
                                 # bound methods that will be called when
                                 # the signal is emitted
        self.__signal_map = {}   # dictionary contains all the signals that
                                 # this instance can emit. The keys are the
                                 # signal names and the values are tuples
                                 # containing the list of types of the arguments
                                 # that the callback methods must accept.
        
        # To speed up the signal type checking the signals declared by
        # each of the classes in the inheritance tree of this instance
        # are consolidated into a single dictionary.
        # The signals can't change so we only need to do this once.
        
        def trav(cls):
            """A traversal function to walk through all the classes in
            the inheritance tree. The return is a list of all the
            __signals__ dictionaries."""
            if cls.__dict__.has_key('__signals__'):
                signal_list = [cls.__signals__]
            else:
                signal_list = []

            for base_cls in cls.__bases__:
                base_list = trav(base_cls)
                if len(base_list) > 0:
                    signal_list = signal_list + base_list

            return signal_list

        # Build a signal dict from the list of signal dicts
        for s in trav(self.__class__):
            for (k,v) in s.items():
                if self.__signal_map.has_key(k):
                    # signal name clash
                    sys.err.write("Warning: signal name clash: ", str(k))
                self.__signal_map[k] = v

        # self.__signal_map now contains the connonical list
        # of signals that this instance can emit.
        

    def connect(self, signal_name, callback):
        # Check that signal exists.
        if signal_name not in self.__signal_map.keys():
            sys.stderr.write("Warning: attempt to connect to unknown signal: ", str(signal_name))
            return
        
        # Add callable to callback_map
        if signal_name not in self.__callback_map.keys():
            self.__callback_map[signal_name] = []
        self.__callback_map[signal_name].append(callback)

    def emit(self, signal_name, args=tuple()):
        # Check that signals are not blocked
        if GrampsDBCallback.__BLOCK_ALL_SIGNALS or \
               self.__block_instance_signals:
            return
        
        # Check signal exists
        if signal_name not in self.__signal_map.keys():
            sys.stderr.write("Warning: attempt to emit to unknown signal: ", str(signal_name))
            return

        # type check arguments
        arg_types = self.__signal_map[signal_name]
        if arg_types == None and len(args) > 0:
            sys.stderr.write("Warning: signal emitted with "\
                             "wrong number of args: ", str(signal_name))
            return

        if len(args) > 0:
            if len(args) != len(arg_types):
                sys.stderr.write("Warning: signal emitted with "\
                                 "wrong number of args: ", str(signal_name))
                return

            if arg_types != None:
                for i in range(0,len(arg_types)):
                    if not isinstance(args[i],arg_types[i]):
                        sys.stderr.write("Warning: signal emitted with "\
                                         "wrong arg types: %s" % (str(signal_name),))
                        sys.stderr.write("arg passed was: %s type should be: %s"
                                         % (args[i],repr(arg_types[i])))
                        return 
                
        for cb in self.__callback_map[signal_name]:
            try:
                if type(cb) == tuple: # call class method
                    cb[0](cb[1],*args)
                elif type(cb) == types.FunctionType or \
                         type(cb) == types.MethodType: # call func
                    cb(*args)
                else:
                    sys.stderr.write("Warning: badly formed entry in callback map")
            except:
                sys.stderr.write("Warning: exception occured in callback function.")

    #
    # instance signals control methods
    #
    def disable_signals(self):
        self.__block_instance_signals = True
        
    def enable_signals(self):
        self.__block_instance_signals = False
        
    #
    # Class methods
    #

    def __disable_all_signals(cls):
        GrampsDBCallback.__BLOCK_ALL_SIGNALS = True

    disable_all_signals = classmethod(__disable_all_signals)
    
    def __enable_all_signals(cls):
        GrampsDBCallback.__BLOCK_ALL_SIGNALS = False

    enable_all_signals = classmethod(__enable_all_signals)
    
#-------------------------------------------------------------------------
#
# Testing code below this point
#
#-------------------------------------------------------------------------

if __name__ == "__main__":

    import unittest

    class TestGrampsDBCallback(unittest.TestCase):

        def test_simple(self):

            class TestSignals(GrampsDBCallback):

                __signals__ = {
                          'test-signal' : (int,)
                         }

            rl = []
            def fn(i,r=rl):
                rl.append(i)

            t = TestSignals()
            t.connect('test-signal',fn)
            t.emit('test-signal',(1,))
    
            
            assert len(rl) == 1, "No signal emitted"
            assert rl[0] == 1, "Wrong argument recieved"

        def test_subclassing(self):

            class TestSignals(GrampsDBCallback):
                __signals__ = {
                          'test-signal' : (int,)
                         }
                
            class TestSignalsSubclass(TestSignals):
                __signals__ = {
                          'test-sub-signal' : (int,),
                         }

            rl = []
            def fn(i,r=rl):
                rl.append(i)

            t = TestSignalsSubclass()
            t.connect('test-signal',fn)
            t.emit('test-signal',(1,))
                
            assert len(rl) == 1, "No signal emitted"
            assert rl[0] == 1, "Wrong argument recieved"

            t.connect('test-sub-signal',fn)
            t.emit('test-sub-signal',(1,))
                
            assert len(rl) == 2, "No subclass signal emitted"
            assert rl[1] == 1, "Wrong argument recieved in subclass"

        def test_signal_block(self):

            class TestSignals(GrampsDBCallback):

                __signals__ = {
                          'test-signal' : (int,)
                         }

            rl = []
            def fn(i,r=rl):
                rl.append(i)

            t = TestSignals()
            t.connect('test-signal',fn)
            t.emit('test-signal',(1,))    
            
            assert len(rl) == 1, "No signal emitted"
            assert rl[0] == 1, "Wrong argument recieved"

            GrampsDBCallback.disable_all_signals()
            t.emit('test-signal',(1,))    
            assert len(rl) == 1, "Signal emitted while class blocked"

            GrampsDBCallback.enable_all_signals()
            t.emit('test-signal',(1,))    
            assert len(rl) == 2, "Signals not class unblocked"

            t.disable_signals()
            t.emit('test-signal',(1,))    
            assert len(rl) == 2, "Signal emitted while instance blocked"

            t.enable_signals()
            t.emit('test-signal',(1,))    
            assert len(rl) == 3, "Signals not instance unblocked"

        def test_type_checking(self):

            class TestSignals(GrampsDBCallback):
                __signals__ = {
                          'test-int' : (int,),
                          'test-list': (list,),
                          'test-object': (object,),
                          'test-str': (str,),
                          'test-float': (float,),
                          'test-dict': (dict,),
                          'test-lots': (int,str,list,object,float)
                         }

            rl = []
            def fn(i,r=rl):
                rl.append(i)

            t = TestSignals()
            t.connect('test-int',fn), t.emit('test-int',(1,))
            assert type(rl[0]) == int, "not int"

            t.connect('test-list',fn), t.emit('test-list',([1,2],))
            assert type(rl[1]) == list, "not list"

            t.connect('test-object',fn), t.emit('test-object',(t,))
            assert isinstance(rl[2],object), "not object"

            t.connect('test-float',fn), t.emit('test-float',(2.3,))
            assert type(rl[3]) == float, "not float"

            t.connect('test-dict',fn), t.emit('test-dict',({1:2},))
            assert type(rl[4]) == dict, "not dict"

            rl = []
            def fn2(i,s,l,o,f,r=rl):
                rl.append(i)

            t.connect('test-lots',fn2), t.emit('test-lots',(1,'a',[1,2],t,1.2))
            assert type(rl[0]) == int, "not lots"

            # This should fail because the type of arg1 is wrong
            res=[]
            class C:
                def write(self,s,r=res):
                    res.append(s)
            sys.stderr = C()
            t.connect('test-lots',fn2), t.emit('test-lots',('a','a',[1,2],t,1.2))
            assert res[0][0:8] == "Warning:", "Type error not detected"


    unittest.main()