Merge 5d8d2cf into 8be934a

param/parameterized.py

@@ -787,6 +787,7 @@ def __init__(self_, cls, self=None):
         self_.cls = cls
         self_.self = self
         self_._BATCH_WATCH = False  # If true, Change and watcher objects are queued.
+        self_._TRIGGER = False
         self_._changes = []         # Queue of batched changed
         self_._watchers = []         # Queue of batched watchers
 
@@ -881,7 +882,9 @@ def _call_watcher(self_, watcher, change):
         """
         Invoke the given the watcher appropriately given a Change object.
         """
-        if watcher.onlychanged and (not self_._changed(change)):
+        if self_.cls.param._TRIGGER:
+            pass
+        elif watcher.onlychanged and (not self_._changed(change)):
             return
 
         if self_.cls.param._BATCH_WATCH:
@@ -894,6 +897,25 @@ def _call_watcher(self_, watcher, change):
             watcher.fn(**{change.name: change.new})
 
 
+    def trigger(self_, *param_names):
+        """
+        Trigger watchers for the given set of parameter names. Watchers
+        will be triggered whether or not the parameter values have
+        actually changed.
+        """
+        changes = self_.cls.param._changes
+        watchers = self_.cls.param._watchers
+        self_.cls.param._changes  = []
+        self_.cls.param._watchers = []
+        param_values = dict(self_.get_param_values())
+        params = {name: param_values[name] for name in param_names}
+        self_.cls.param._TRIGGER = True
+        self_.set_param(**params)
+        self_.cls.param._TRIGGER = False
+        self_.cls.param._changes = changes
+        self_.cls.param._changes = watchers
+
+
     def _batch_call_watchers(self_):
         """
         Batch call a set of watchers based on the parameter value

tests/API1/testwatch.py

@@ -321,6 +321,62 @@ def test_simple_batched_watch_values_callback_reuse(self):
                 raise Exception('Invalid number of arguments')
 
 
+
+class TestTrigger(API1TestCase):
+
+    def setUp(self):
+        super(TestTrigger, self).setUp()
+        self.accumulator = 0
+
+    def test_simple_trigger_one_param(self):
+        accumulator = Accumulator()
+        obj = SimpleWatchExample()
+        watcher = obj.param.watch(accumulator, ['a'])
+        obj.param.trigger('a')
+        self.assertEqual(accumulator.call_count(), 1)
+
+        args = accumulator.args_for_call(0)
+        self.assertEqual(args[0].name, 'a')
+        self.assertEqual(args[0].old, 0)
+        self.assertEqual(args[0].new, 0)
+
+    def test_simple_trigger_one_param_change(self):
+        accumulator = Accumulator()
+        obj = SimpleWatchExample()
+        watcher = obj.param.watch(accumulator, ['a'])
+        obj.a = 42
+        self.assertEqual(accumulator.call_count(), 1)
+
+        obj.param.trigger('a')
+        self.assertEqual(accumulator.call_count(), 2)
+
+        args = accumulator.args_for_call(0)
+        self.assertEqual(args[0].name, 'a')
+        self.assertEqual(args[0].old, 0)
+        self.assertEqual(args[0].new, 42)
+
+        args = accumulator.args_for_call(1)
+        self.assertEqual(args[0].name, 'a')
+        self.assertEqual(args[0].old, 42)
+        self.assertEqual(args[0].new, 42)
+
+    def test_simple_trigger_two_params(self):
+        accumulator = Accumulator()
+        obj = SimpleWatchExample()
+        watcher = obj.param.watch(accumulator, ['a','b'])
+        obj.param.trigger('a','b')
+        self.assertEqual(accumulator.call_count(), 1)
+
+        args = accumulator.args_for_call(0)
+        self.assertEqual(args[0].name, 'a')
+        self.assertEqual(args[0].old, 0)
+        self.assertEqual(args[0].new, 0)
+
+        self.assertEqual(args[1].name, 'b')
+        self.assertEqual(args[1].old, 0)
+        self.assertEqual(args[1].new, 0)
+
+
 if __name__ == "__main__":
     import nose
     nose.runmodule()