From 7764df2e743e48875fe391b52a570e9439b95e46 Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Tue, 15 Oct 2019 18:23:28 +0100
Subject: [PATCH 01/19] Various improvements for Pipeline

---
 panel/pipeline.py | 143 +++++++++++++++++++++++++++++++++-------------
 1 file changed, 104 insertions(+), 39 deletions(-)

diff --git a/panel/pipeline.py b/panel/pipeline.py
index fd93bbfeb4..95ed98786e 100644
--- a/panel/pipeline.py
+++ b/panel/pipeline.py
@@ -1,16 +1,26 @@
 from __future__ import absolute_import, division, unicode_literals
 
 import os
+import sys
+import traceback as tb
 
 import param
 import numpy as np
 
-from .layout import Row, Column, HSpacer, VSpacer
-from .pane import HoloViews, Markdown, Pane
+from .layout import Row, Column, HSpacer, VSpacer, Spacer
+from .pane import HoloViews, HTML, Pane
+from .widgets import Button
 from .param import Param
 from .util import param_reprs
 
 
+class PipelineError(RuntimeError):
+    """
+    Custom error type which can be raised to display custom error
+    message in a Pipeline.
+    """
+
+
 class Pipeline(param.Parameterized):
     """
     Allows connecting a linear series of panels to define a workflow.
@@ -19,49 +29,66 @@ class Pipeline(param.Parameterized):
     outputs using the param.output decorator.
     """
 
+    auto_advance = param.Boolean(default=False, precedence=-1, doc="""
+        Whether to automatically advance if the ready parameter is True.""")
+
     debug = param.Boolean(default=False, precedence=-1, doc="""
         Whether to raise errors, useful for debugging while building an application.""")
 
     inherit_params = param.Boolean(default=True, precedence=-1, doc="""
         Whether parameters should be inherited between pipeline stages""")
 
+    ready_parameter = param.String(default=None, doc="""
+        Parameter name to watch to check whether a stage is ready.""")
+
     next = param.Action(default=lambda x: x.param.trigger('next'))
 
     previous = param.Action(default=lambda x: x.param.trigger('previous'))
 
-    def __init__(self, stages=None, **params):
+    def __init__(self, stages=[], **params):
         try:
             import holoviews as hv
         except:
             raise ImportError('Pipeline requires holoviews to be installed')
 
-        self._stages = [] if stages is None else list(stages)
+        self._stages = []
+        for stage in stages:
+            kwargs = {}
+            if len(stage) == 2:
+                name, stage = stage
+            elif len(stage) == 3:
+                name, stage, kwargs = stage
+            self.add_stage(name, stage, **kwargs)
         self._stage = 0
         super(Pipeline, self).__init__(**params)
-        self._error = Markdown('')
         self._states = []
         self._state = None
+        self._block = False
         self._progress_sel = hv.streams.Selection1D()
         self._progress_sel.add_subscriber(self._set_stage)
         prev_button =  Param(self.param.previous, width=100)
         next_button =  Param(self.param.next, width=100)
         prev_button.layout[0].disabled = True
-        self._progress_bar = Row(self._make_progress, prev_button, next_button)
+        self._progress_bar = Row(Spacer(width=100), self._make_progress(), prev_button, next_button)
         spinner = Pane(os.path.join(os.path.dirname(__file__), 'assets', 'spinner.gif'))
         self._spinner_layout = Row(HSpacer(), Column(VSpacer(), spinner, VSpacer()), HSpacer())
         stage_layout = Row()
-        if stages:
-            stage_layout.append(self._init_stage())
-        self._layout = Column(self._progress_bar, self._error, stage_layout)
+        if len(stages):
+            stage = self._init_stage()
+            stage_layout.append(stage)
+            self._update_button(stage)
+        self._layout = Column(self._progress_bar, stage_layout)
 
-    def add_stage(self, name, stage):
+    def add_stage(self, name, stage, **kwargs):
         self._validate(stage)
-        self._stages.append((name, stage))
+        self._stages.append((name, stage, kwargs))
         if len(self._stages) == 1:
-            self._layout[2].append(self._init_stage())
+            stage = self._init_stage()
+            self._layout[1].append(stage)
+            self._update_button(stage)
 
     def _validate(self, stage):
-        if any(stage is s for n, s in self._stages):
+        if any(stage is s for n, s, kw in self._stages):
             raise ValueError('Stage %s is already in pipeline' % stage)
         elif not ((isinstance(stage, type) and issubclass(stage, param.Parameterized))
                   or isinstance(stage, param.Parameterized)):
@@ -83,15 +110,31 @@ def __getitem__(self, index):
 
     @property
     def layout(self):
-        self._progress_bar[0] = self._make_progress
+        self._progress_bar[1] = self._make_progress
         return self._layout
 
+    def _unblock(self, event):
+        if self._state is not event.obj or self._block:
+            self._block = False
+            return
+
+        button = self._progress_bar[-1][0]
+        if button.disabled and event.new:
+            button.disabled = False
+        elif not button.disabled and not event.new:
+            button.disabled = True
+
+        stage_kwargs = self._stages[self._stage][-1]
+        if event.new and stage_kwargs.get('auto_advance', self.auto_advance):
+            self._next()
+
     def _init_stage(self):
-        name, stage = self._stages[self._stage]
+        _, stage, stage_kwargs = self._stages[self._stage]
         kwargs = {}
-        if self._state:
+        if self._state is not None:
             results = {}
             for name, (_, method, index) in self._state.param.outputs().items():
+                print(name, method)
                 if name not in stage.param:
                     continue
                 if method not in results:
@@ -100,12 +143,16 @@ def _init_stage(self):
                 if index is not None:
                     result = result[index]
                 kwargs[name] = result
-            if self.inherit_params:
+            if stage_kwargs.get('inherit_params', self.inherit_params):
                 params = [k for k, v in self._state.param.objects('existing').items()
                           if v.precedence is None or v.precedence >= 0]
                 kwargs.update({k: v for k, v in self._state.param.get_param_values()
                                if k in stage.param and k != 'name' and k in params})
 
+        ready_param = stage_kwargs.get('ready_parameter', self.ready_parameter)
+        if ready_param and ready_param in stage.param:
+            stage.param.watch(self._unblock, ready_param, onlychanged=False)
+
         if isinstance(stage, param.Parameterized):
             stage.set_param(**kwargs)
             self._state = stage
@@ -131,59 +178,77 @@ def _set_stage(self, index):
                 if self._error.object:
                     break
 
-    def _update_button(self):
+    def _update_button(self, stage):
         # Disable previous button
         if self._stage == 0:
-            self._progress_bar[1].layout[0].disabled = True
+            self._progress_bar[2].layout[0].disabled = True
         else:
-            self._progress_bar[1].layout[0].disabled = False
+            self._progress_bar[2].layout[0].disabled = False
 
         # Disable next button
         if self._stage == len(self._stages)-1:
-            self._progress_bar[2].layout[0].disabled = True
+            self._progress_bar[3].layout[0].disabled = True
         else:
-            self._progress_bar[2].layout[0].disabled = False
+            kwargs = self._stages[self._stage][2]
+            ready = kwargs.get('ready_parameter', self.ready_parameter)
+            disabled = (not getattr(stage, ready)) if ready in stage.param else False
+            self._progress_bar[3].layout[0].disabled = disabled
+
+    def _get_error_button(cls, e):
+        msg = str(e) if isinstance(e, PipelineError) else ""
+        type, value, trb = sys.exc_info()
+        tb_list = tb.format_tb(trb, None) + tb.format_exception_only(type, value)
+        traceback = (("%s\n\nTraceback (innermost last):\n" + "%-20s %s") %
+                     (msg, ''.join(tb_list[-5:-1]), tb_list[-1]))
+        button = Button(name='Error', button_type='danger', width=100,
+                        align='center')
+        button.jslink(button, code={'clicks': "alert(`{tb}`)".format(tb=traceback)})
+        return button
 
     @param.depends('next', watch=True)
     def _next(self):
         self._stage += 1
-        prev_state = self._layout[2][0]
-        self._layout[2][0] = self._spinner_layout
+        prev_state = self._state
+        self._layout[1][0] = self._spinner_layout
         try:
             new_stage = self._init_stage()
-            self._layout[2][0] = new_stage
-            self._update_button()
+            self._state = self._states[self._stage]
+            self._layout[1][0] = new_stage
+            self._update_button(new_stage)
         except Exception as e:
             self._stage -= 1
             self._state = prev_state
-            self._error.object = ('Next stage raised following error:\n\n\t%s: %s'
-                                  % (type(e).__name__, str(e)))
-            self._layout[2][0] = prev_state
+            self._layout[1][0] = prev_state
+            self._progress_bar[0] = self._get_error_button(e)
             if self.debug:
                 raise e
             return e
         else:
-            self._error.object = ''
+            self._progress_bar[0] = Spacer(width=100)
+        finally:
+            self._progress_bar[1] = self._make_progress()
+
 
     @param.depends('previous', watch=True)
     def _previous(self):
         self._stage -= 1
-        old_stage = self._layout[2][0]
+        prev_state = self._state
         try:
             self._state = self._states[self._stage]
-            self._layout[2][0] = self._state.panel()
-            self._update_button()
+            self._block = True
+            self._layout[1][0] = self._state.panel()
+            self._update_button(self._state)
         except Exception as e:
             self._stage += 1
-            self._state = old_stage
-            self._error.object = ('Previous stage raised following error:\n\n\t%s: %s'
-                                  % (type(e).__name__, str(e)))
+            self._state = prev_state
+            self._progress_bar[0] = self._get_error_button(e)
             if self.debug:
                 raise e
         else:
-            self._error.object = ''
+            self._progress_bar[0] = Spacer(width=100)
+        finally:
+            self._progress_bar[1] = self._make_progress()
 
-    @param.depends('previous', 'next')
     def _make_progress(self):
         import holoviews as hv
         import holoviews.plotting.bokeh # noqa
@@ -198,7 +263,7 @@ def _make_progress(self):
             show_legend=False, size=20, default_tools=[], tools=['tap'],
             nonselection_alpha=1, backend='bokeh'
         )
-        point_labels = points.add_dimension('text', 0, [n for n, _ in self._stages], vdim=True)
+        point_labels = points.add_dimension('text', 0, [s[0] for s in self._stages], vdim=True)
         labels = hv.Labels(point_labels).options(yoffset=-2.5, backend='bokeh')
         self._progress_sel.source = points
         hv_plot = (line * points * labels).options(

From fa923a0f31e525c3573568f42b73251bdd160e44 Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Wed, 16 Oct 2019 11:31:21 +0100
Subject: [PATCH 02/19] Minor fixes

---
 panel/pipeline.py | 28 ++++++++++++++++------------
 1 file changed, 16 insertions(+), 12 deletions(-)

diff --git a/panel/pipeline.py b/panel/pipeline.py
index 95ed98786e..5e710d0614 100644
--- a/panel/pipeline.py
+++ b/panel/pipeline.py
@@ -8,7 +8,7 @@
 import numpy as np
 
 from .layout import Row, Column, HSpacer, VSpacer, Spacer
-from .pane import HoloViews, HTML, Pane
+from .pane import HoloViews, Pane
 from .widgets import Button
 from .param import Param
 from .util import param_reprs
@@ -110,7 +110,7 @@ def __getitem__(self, index):
 
     @property
     def layout(self):
-        self._progress_bar[1] = self._make_progress
+        self._progress_bar[1] = self._make_progress()
         return self._layout
 
     def _unblock(self, event):
@@ -194,12 +194,15 @@ def _update_button(self, stage):
             disabled = (not getattr(stage, ready)) if ready in stage.param else False
             self._progress_bar[3].layout[0].disabled = disabled
 
-    def _get_error_button(cls, e):
+    def _get_error_button(self, e):
         msg = str(e) if isinstance(e, PipelineError) else ""
-        type, value, trb = sys.exc_info()
-        tb_list = tb.format_tb(trb, None) + tb.format_exception_only(type, value)
-        traceback = (("%s\n\nTraceback (innermost last):\n" + "%-20s %s") %
-                     (msg, ''.join(tb_list[-5:-1]), tb_list[-1]))
+        if self.debug:
+            type, value, trb = sys.exc_info()
+            tb_list = tb.format_tb(trb, None) + tb.format_exception_only(type, value)
+            traceback = (("%s\n\nTraceback (innermost last):\n" + "%-20s %s") %
+                         (msg, ''.join(tb_list[-5:-1]), tb_list[-1]))
+        else:
+            traceback = msg or "Undefined error, enable debug mode."
         button = Button(name='Error', button_type='danger', width=100,
                         align='center')
         button.jslink(button, code={'clicks': "alert(`{tb}`)".format(tb=traceback)})
@@ -254,21 +257,22 @@ def _make_progress(self):
         import holoviews.plotting.bokeh # noqa
         stages = len(self._stages)
         line = hv.Path([[(0, 0), (stages-1, 0)]]).options(
-            line_width=6, color='black', backend='bokeh'
+            line_width=6, color='black', default_tools=[], backend='bokeh'
         )
         vals = np.arange(stages)
         active = [1 if v == self._stage else 0 for v in vals]
         points = hv.Points((vals, np.zeros(stages), active), vdims=['active']).options(
             color_index='active', line_color='black', cmap={0: 'white', 1: '#5cb85c'},
-            show_legend=False, size=20, default_tools=[], tools=['tap'],
+            show_legend=False, size=20, default_tools=['tap'],
             nonselection_alpha=1, backend='bokeh'
         )
         point_labels = points.add_dimension('text', 0, [s[0] for s in self._stages], vdim=True)
-        labels = hv.Labels(point_labels).options(yoffset=-2.5, backend='bokeh')
+        labels = hv.Labels(point_labels).options(yoffset=-2.5, default_tools=[], backend='bokeh')
         self._progress_sel.source = points
         hv_plot = (line * points * labels).options(
-            xaxis=None, yaxis=None, width=800, show_frame=False, toolbar=None,
+            xaxis=None, yaxis=None, width=800, show_frame=False,
             height=80, xlim=(-0.5, stages-0.5), ylim=(-4, 1.5),
-            clone=False, backend='bokeh'
+            clone=False, default_tools=[], active_tools=['tap'],
+            toolbar=None, backend='bokeh'
         )
         return HoloViews(hv_plot, backend='bokeh')

From f2b6c6bd48d7cacad10a2497b002c043f19af300 Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Wed, 16 Oct 2019 12:02:16 +0100
Subject: [PATCH 03/19] Fixes for existing tests

---
 panel/pipeline.py            | 21 +++++++++---------
 panel/tests/test_pipeline.py | 42 ++++++++++++++++++++----------------
 2 files changed, 34 insertions(+), 29 deletions(-)

diff --git a/panel/pipeline.py b/panel/pipeline.py
index 5e710d0614..3d4757affa 100644
--- a/panel/pipeline.py
+++ b/panel/pipeline.py
@@ -51,15 +51,8 @@ def __init__(self, stages=[], **params):
         except:
             raise ImportError('Pipeline requires holoviews to be installed')
 
-        self._stages = []
-        for stage in stages:
-            kwargs = {}
-            if len(stage) == 2:
-                name, stage = stage
-            elif len(stage) == 3:
-                name, stage, kwargs = stage
-            self.add_stage(name, stage, **kwargs)
         self._stage = 0
+        self._stages = []
         super(Pipeline, self).__init__(**params)
         self._states = []
         self._state = None
@@ -73,11 +66,19 @@ def __init__(self, stages=[], **params):
         spinner = Pane(os.path.join(os.path.dirname(__file__), 'assets', 'spinner.gif'))
         self._spinner_layout = Row(HSpacer(), Column(VSpacer(), spinner, VSpacer()), HSpacer())
         stage_layout = Row()
+        self._layout = Column(self._progress_bar, stage_layout)
+        for stage in stages:
+            kwargs = {}
+            if len(stage) == 2:
+                name, stage = stage
+            elif len(stage) == 3:
+                name, stage, kwargs = stage
+            self.add_stage(name, stage, **kwargs)
         if len(stages):
             stage = self._init_stage()
             stage_layout.append(stage)
             self._update_button(stage)
-        self._layout = Column(self._progress_bar, stage_layout)
+
 
     def add_stage(self, name, stage, **kwargs):
         self._validate(stage)
@@ -96,7 +97,7 @@ def _validate(self, stage):
 
     def __repr__(self):
         repr_str = 'Pipeline:'
-        for i, (name, stage) in enumerate(self._stages):
+        for i, (name, stage, _) in enumerate(self._stages):
             if isinstance(stage, param.Parameterized):
                 cls_name = type(stage).__name__
             else:
diff --git a/panel/tests/test_pipeline.py b/panel/tests/test_pipeline.py
index 034ec3e6da..c6b16d8b60 100644
--- a/panel/tests/test_pipeline.py
+++ b/panel/tests/test_pipeline.py
@@ -3,7 +3,8 @@
 import pytest
 import param
 
-from panel.layout import Row, Column
+from panel.layout import Row, Column, Spacer
+from panel.pane import HoloViews
 from panel.param import Param, ParamMethod
 from panel.pipeline import Pipeline
 from panel.tests.util import hv_available
@@ -53,13 +54,14 @@ def test_pipeline_from_classes():
 
     assert isinstance(layout, Column)
     assert isinstance(layout[0], Row)
-    progress, prev_button, next_button = layout[0].objects
+    error, progress, prev_button, next_button = layout[0].objects
 
+    assert isinstance(error, Spacer)
     assert isinstance(prev_button, Param)
     assert isinstance(next_button, Param)
-    assert isinstance(progress, ParamMethod)
+    assert isinstance(progress, HoloViews)
 
-    hv_obj = progress.object().object
+    hv_obj = progress.object
     points = hv_obj.get(1)
     assert isinstance(points, hv.Points)
     assert len(points) == 2
@@ -67,19 +69,19 @@ def test_pipeline_from_classes():
     assert isinstance(labels, hv.Labels)
     assert list(labels['text']) == ['Stage 1', 'Stage 2']
 
-    stage = layout[2][0]
+    stage = layout[1][0]
     assert isinstance(stage, Row)
     assert isinstance(stage[1], ParamMethod)
     assert stage[1].object() == '5 * 5 = 25'
 
     pipeline.param.trigger('next')
-    stage = layout[2][0]
+    stage = layout[1][0]
     assert isinstance(stage, Row)
     assert isinstance(stage[1], ParamMethod)
     assert stage[1].object() == '25^0.1=1.380'
 
     pipeline.param.trigger('previous')
-    stage = layout[2][0]
+    stage = layout[1][0]
     assert isinstance(stage, Row)
     assert isinstance(stage[1], ParamMethod)
     assert stage[1].object() == '5 * 5 = 25'
@@ -95,13 +97,14 @@ def test_pipeline_from_instances():
 
     assert isinstance(layout, Column)
     assert isinstance(layout[0], Row)
-    progress, prev_button, next_button = layout[0].objects
+    error, progress, prev_button, next_button = layout[0].objects
 
+    assert isinstance(error, Spacer)
     assert isinstance(prev_button, Param)
     assert isinstance(next_button, Param)
-    assert isinstance(progress, ParamMethod)
+    assert isinstance(progress, HoloViews)
 
-    hv_obj = progress.object().object
+    hv_obj = progress.object
     points = hv_obj.get(1)
     assert isinstance(points, hv.Points)
     assert len(points) == 2
@@ -109,19 +112,19 @@ def test_pipeline_from_instances():
     assert isinstance(labels, hv.Labels)
     assert list(labels['text']) == ['Stage 1', 'Stage 2']
 
-    stage = layout[2][0]
+    stage = layout[1][0]
     assert isinstance(stage, Row)
     assert isinstance(stage[1], ParamMethod)
     assert stage[1].object() == '5 * 5 = 25'
 
     pipeline.param.trigger('next')
-    stage = layout[2][0]
+    stage = layout[1][0]
     assert isinstance(stage, Row)
     assert isinstance(stage[1], ParamMethod)
     assert stage[1].object() == '25^0.1=1.380'
 
     pipeline.param.trigger('previous')
-    stage = layout[2][0]
+    stage = layout[1][0]
     assert isinstance(stage, Row)
     assert isinstance(stage[1], ParamMethod)
     assert stage[1].object() == '5 * 5 = 25'
@@ -139,13 +142,14 @@ def test_pipeline_from_add_stages():
 
     assert isinstance(layout, Column)
     assert isinstance(layout[0], Row)
-    progress, prev_button, next_button = layout[0].objects
+    error, progress, prev_button, next_button = layout[0].objects
 
+    assert isinstance(error, Spacer)
     assert isinstance(prev_button, Param)
     assert isinstance(next_button, Param)
-    assert isinstance(progress, ParamMethod)
+    assert isinstance(progress, HoloViews)
 
-    hv_obj = progress.object().object
+    hv_obj = progress.object
     points = hv_obj.get(1)
     assert isinstance(points, hv.Points)
     assert len(points) == 2
@@ -153,19 +157,19 @@ def test_pipeline_from_add_stages():
     assert isinstance(labels, hv.Labels)
     assert list(labels['text']) == ['Stage 1', 'Stage 2']
 
-    stage = layout[2][0]
+    stage = layout[1][0]
     assert isinstance(stage, Row)
     assert isinstance(stage[1], ParamMethod)
     assert stage[1].object() == '5 * 5 = 25'
 
     pipeline.param.trigger('next')
-    stage = layout[2][0]
+    stage = layout[1][0]
     assert isinstance(stage, Row)
     assert isinstance(stage[1], ParamMethod)
     assert stage[1].object() == '25^0.1=1.380'
 
     pipeline.param.trigger('previous')
-    stage = layout[2][0]
+    stage = layout[1][0]
     assert isinstance(stage, Row)
     assert isinstance(stage[1], ParamMethod)
     assert stage[1].object() == '5 * 5 = 25'

From 978892d551d98cdc1490676f7da58bac9442ce24 Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Mon, 21 Oct 2019 15:16:08 -0500
Subject: [PATCH 04/19] API cleanup on Pipeline

---
 panel/pipeline.py | 126 ++++++++++++++++++++++++++++++++--------------
 1 file changed, 87 insertions(+), 39 deletions(-)

diff --git a/panel/pipeline.py b/panel/pipeline.py
index 3d4757affa..ee35a8de62 100644
--- a/panel/pipeline.py
+++ b/panel/pipeline.py
@@ -45,7 +45,7 @@ class Pipeline(param.Parameterized):
 
     previous = param.Action(default=lambda x: x.param.trigger('previous'))
 
-    def __init__(self, stages=[], **params):
+    def __init__(self, stages=[], graph={}, **params):
         try:
             import holoviews as hv
         except:
@@ -59,14 +59,24 @@ def __init__(self, stages=[], **params):
         self._block = False
         self._progress_sel = hv.streams.Selection1D()
         self._progress_sel.add_subscriber(self._set_stage)
-        prev_button =  Param(self.param.previous, width=100)
-        next_button =  Param(self.param.next, width=100)
-        prev_button.layout[0].disabled = True
-        self._progress_bar = Row(Spacer(width=100), self._make_progress(), prev_button, next_button)
+        self._prev_button = Param(self.param.previous).layout[0]
+        self._prev_button.width = 100
+        self._next_button = Param(self.param.next).layout[0]
+        self._next_button.width = 100
+        self._prev_button.disabled = True
+        self._progress_bar = Row(
+            Spacer(width=100),
+            self._make_progress(),
+            self._prev_button,
+            self._next_button
+        )
         spinner = Pane(os.path.join(os.path.dirname(__file__), 'assets', 'spinner.gif'))
-        self._spinner_layout = Row(HSpacer(), Column(VSpacer(), spinner, VSpacer()), HSpacer())
-        stage_layout = Row()
-        self._layout = Column(self._progress_bar, stage_layout)
+        self._spinner_layout = Row(
+            HSpacer(),
+            Column(VSpacer(), spinner, VSpacer()),
+            HSpacer()
+        )
+        self._layout = Column(self._progress_bar, Row())
         for stage in stages:
             kwargs = {}
             if len(stage) == 2:
@@ -74,19 +84,7 @@ def __init__(self, stages=[], **params):
             elif len(stage) == 3:
                 name, stage, kwargs = stage
             self.add_stage(name, stage, **kwargs)
-        if len(stages):
-            stage = self._init_stage()
-            stage_layout.append(stage)
-            self._update_button(stage)
-
-
-    def add_stage(self, name, stage, **kwargs):
-        self._validate(stage)
-        self._stages.append((name, stage, kwargs))
-        if len(self._stages) == 1:
-            stage = self._init_stage()
-            self._layout[1].append(stage)
-            self._update_button(stage)
+        self.define_graph(graph)
 
     def _validate(self, stage):
         if any(stage is s for n, s, kw in self._stages):
@@ -109,11 +107,6 @@ def __repr__(self):
     def __getitem__(self, index):
         return self._stages[index][1]
 
-    @property
-    def layout(self):
-        self._progress_bar[1] = self._make_progress()
-        return self._layout
-
     def _unblock(self, event):
         if self._state is not event.obj or self._block:
             self._block = False
@@ -159,6 +152,7 @@ def _init_stage(self):
             self._state = stage
         else:
             self._state = stage(**kwargs)
+
         if len(self._states) <= self._stage:
             self._states.append(self._state)
         else:
@@ -179,21 +173,21 @@ def _set_stage(self, index):
                 if self._error.object:
                     break
 
-    def _update_button(self, stage):
+    def _update_button(self):
         # Disable previous button
         if self._stage == 0:
-            self._progress_bar[2].layout[0].disabled = True
+            self._prev_button.disabled = True
         else:
-            self._progress_bar[2].layout[0].disabled = False
+            self._prev_button.disabled = False
 
         # Disable next button
         if self._stage == len(self._stages)-1:
-            self._progress_bar[3].layout[0].disabled = True
+            self._next_button.disabled = True
         else:
-            kwargs = self._stages[self._stage][2]
+            stage, kwargs = self._stages[self._stage][1:]
             ready = kwargs.get('ready_parameter', self.ready_parameter)
             disabled = (not getattr(stage, ready)) if ready in stage.param else False
-            self._progress_bar[3].layout[0].disabled = disabled
+            self._next_button.disabled = disabled
 
     def _get_error_button(self, e):
         msg = str(e) if isinstance(e, PipelineError) else ""
@@ -215,14 +209,12 @@ def _next(self):
         prev_state = self._state
         self._layout[1][0] = self._spinner_layout
         try:
-            new_stage = self._init_stage()
-            self._state = self._states[self._stage]
-            self._layout[1][0] = new_stage
-            self._update_button(new_stage)
+            self._layout[1][0] = self._init_stage()
+            self._update_button()
         except Exception as e:
             self._stage -= 1
             self._state = prev_state
-            self._layout[1][0] = prev_state
+            self._layout[1][0] = prev_state.panel()
             self._progress_bar[0] = self._get_error_button(e)
             if self.debug:
                 raise e
@@ -232,7 +224,6 @@ def _next(self):
         finally:
             self._progress_bar[1] = self._make_progress()
 
-
     @param.depends('previous', watch=True)
     def _previous(self):
         self._stage -= 1
@@ -241,7 +232,7 @@ def _previous(self):
             self._state = self._states[self._stage]
             self._block = True
             self._layout[1][0] = self._state.panel()
-            self._update_button(self._state)
+            self._update_button()
         except Exception as e:
             self._stage += 1
             self._state = prev_state
@@ -277,3 +268,60 @@ def _make_progress(self):
             toolbar=None, backend='bokeh'
         )
         return HoloViews(hv_plot, backend='bokeh')
+
+    #----------------------------------------------------------------
+    # Public API
+    #----------------------------------------------------------------
+
+    def add_stage(self, name, stage, **kwargs):
+        """
+        Adds a new, named stage to the Pipeline.
+
+        Arguments
+        ---------
+        name: str
+          A string name for the Pipeline stage
+        stage: param.Parameterized
+          A Parameterized object which represents the Pipeline stage.
+        **kwargs: dict
+          Additional arguments declaring the behavior of the stage.
+        """
+        self._validate(stage)
+        self._stages.append((name, stage, kwargs))
+        if len(self._stages) == 1:
+            stage = self._init_stage()
+            self._layout[1].append(stage)
+            self._update_button()
+
+    def define_graph(self, graph):
+        """
+        Declares a custom graph structure for the Pipeline overriding
+        the default linear flow. The graph should be defined as an
+        adjacency mapping.
+
+        Arguments
+        ---------
+        graph: dict
+          Dictionary declaring the relationship between different
+          pipeline stages. Should map from a single stage name to
+          one or more stage names.
+        """
+        stages = [stage[0] for stage in self._stages]
+        not_found = []
+        for source, targets in graph.items():
+            if source not in stages:
+                not_found.append(source)
+            not_found += [t for t in targets if t not in stages]
+        if not_found:
+            raise ValueError(
+                'Pipeline stage(s) %s not found, ensure all stages '
+                'referenced in the graph have been added.' %
+                (not_found[0] if len(not_found) == 1 else not_found)
+            )
+        self._graph = graph
+
+    @property
+    def layout(self):
+        self._progress_bar[1] = self._make_progress()
+        self._update_button()
+        return self._layout

From 10e7e8605576555fb666b10cd386de9310e49117 Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Tue, 22 Oct 2019 13:30:04 -0500
Subject: [PATCH 05/19] Add Graph support to Pipeline

---
 panel/pipeline.py | 263 +++++++++++++++++++++++++++++++++++-----------
 1 file changed, 201 insertions(+), 62 deletions(-)

diff --git a/panel/pipeline.py b/panel/pipeline.py
index ee35a8de62..556d75c0e6 100644
--- a/panel/pipeline.py
+++ b/panel/pipeline.py
@@ -4,12 +4,14 @@
 import sys
 import traceback as tb
 
+from collections import OrderedDict, defaultdict
+
 import param
 import numpy as np
 
 from .layout import Row, Column, HSpacer, VSpacer, Spacer
 from .pane import HoloViews, Pane
-from .widgets import Button
+from .widgets import Button, Select
 from .param import Param
 from .util import param_reprs
 
@@ -21,6 +23,70 @@ class PipelineError(RuntimeError):
     """
 
 
+def traverse(graph, v, visited):
+    """
+    Traverse the graph from a node and mark visited vertices.
+    """
+    visited[v] = True
+    # Recur for all the vertices adjacent to this vertex
+    for i in graph.get(v, []):
+        if visited[i] == False:
+            traverse(graph, i, visited)
+
+
+def get_root(graph):
+    """
+    Search for the root not by finding nodes without inputs.
+    """
+    # Find root node
+    roots = []
+    targets = [t for ts in graph.values() for t in ts]
+    for src in graph:
+        if src not in targets:
+            roots.append(src)
+
+    if len(roots) > 1:
+        raise ValueError("Graph has more than one node with no "
+                         "incoming edges. Ensure that the graph "
+                         "only has a single source node.")
+    elif len(roots) == 0:
+        raise ValueError("Graph has no source node. Ensure that the "
+                         "graph is not cyclic and has a single "
+                         "starting point.")
+    return roots[0]
+
+
+def is_traversable(root, graph, stages):
+    """
+    Check if the graph is fully traversable from the root node.
+    """
+    # Ensure graph is traverable from root
+    int_graph = {stages.index(s): tuple(stages.index(t) for t in tgts)
+                 for s, tgts in graph.items()}
+    visited = [False]*len(stages)
+    traverse(int_graph, stages.index(root), visited)
+    return all(visited)
+
+
+def get_depth(node, graph, depth=0):
+    depths = []
+    for sub in graph.get(node, []):
+        depths.append(get_depth(sub, graph, depth+1))
+    return max(depths) if depths else depth+1
+
+
+def get_breadths(node, graph, depth=0, breadths=None):
+    if breadths is None:
+        breadths = defaultdict(list)
+        breadths[depth].append(node)
+    for sub in graph.get(node, []):
+        if sub not in breadths[depth+1]:
+            breadths[depth+1].append(sub)
+        get_breadths(sub, graph, depth+1, breadths)
+    return breadths
+
+
+
 class Pipeline(param.Parameterized):
     """
     Allows connecting a linear series of panels to define a workflow.
@@ -51,24 +117,29 @@ def __init__(self, stages=[], graph={}, **params):
         except:
             raise ImportError('Pipeline requires holoviews to be installed')
 
-        self._stage = 0
-        self._stages = []
+        self._stage = None
+        self._stages = OrderedDict()
         super(Pipeline, self).__init__(**params)
-        self._states = []
+        self._states = {}
         self._state = None
+        self._linear = True
         self._block = False
+        self._graph = {}
         self._progress_sel = hv.streams.Selection1D()
         self._progress_sel.add_subscriber(self._set_stage)
         self._prev_button = Param(self.param.previous).layout[0]
-        self._prev_button.width = 100
+        self._prev_button.width = 125
+        self._prev_selector = Select(width=125)
         self._next_button = Param(self.param.next).layout[0]
-        self._next_button.width = 100
+        self._next_button.width = 125
+        self._next_selector = Select(width=125)
         self._prev_button.disabled = True
         self._progress_bar = Row(
             Spacer(width=100),
             self._make_progress(),
             self._prev_button,
-            self._next_button
+            self._next_button,
+            sizing_mode='stretch_width'
         )
         spinner = Pane(os.path.join(os.path.dirname(__file__), 'assets', 'spinner.gif'))
         self._spinner_layout = Row(
@@ -76,7 +147,7 @@ def __init__(self, stages=[], graph={}, **params):
             Column(VSpacer(), spinner, VSpacer()),
             HSpacer()
         )
-        self._layout = Column(self._progress_bar, Row())
+        self._layout = Column(self._progress_bar, Row(), sizing_mode='stretch_width')
         for stage in stages:
             kwargs = {}
             if len(stage) == 2:
@@ -87,7 +158,7 @@ def __init__(self, stages=[], graph={}, **params):
         self.define_graph(graph)
 
     def _validate(self, stage):
-        if any(stage is s for n, s, kw in self._stages):
+        if any(stage is s for n, (s, kw) in self._stages.items()):
             raise ValueError('Stage %s is already in pipeline' % stage)
         elif not ((isinstance(stage, type) and issubclass(stage, param.Parameterized))
                   or isinstance(stage, param.Parameterized)):
@@ -95,7 +166,7 @@ def _validate(self, stage):
 
     def __repr__(self):
         repr_str = 'Pipeline:'
-        for i, (name, stage, _) in enumerate(self._stages):
+        for i, (name, (stage, _)) in enumerate(self._stages.items()):
             if isinstance(stage, param.Parameterized):
                 cls_name = type(stage).__name__
             else:
@@ -123,12 +194,17 @@ def _unblock(self, event):
             self._next()
 
     def _init_stage(self):
-        _, stage, stage_kwargs = self._stages[self._stage]
-        kwargs = {}
-        if self._state is not None:
-            results = {}
-            for name, (_, method, index) in self._state.param.outputs().items():
-                print(name, method)
+        stage, stage_kwargs = self._stages[self._stage]
+
+        previous = []
+        for src, tgts in self._graph.items():
+            if self._stage in tgts:
+                previous.append(src)
+        prev_states = [self._states[prev] for prev in previous if prev in self._states]
+
+        kwargs, results = {}, {}
+        for state in prev_states:
+            for name, (_, method, index) in state.param.outputs().items():
                 if name not in stage.param:
                     continue
                 if method not in results:
@@ -138,9 +214,9 @@ def _init_stage(self):
                     result = result[index]
                 kwargs[name] = result
             if stage_kwargs.get('inherit_params', self.inherit_params):
-                params = [k for k, v in self._state.param.objects('existing').items()
-                          if v.precedence is None or v.precedence >= 0]
-                kwargs.update({k: v for k, v in self._state.param.get_param_values()
+                params = [k for k, v in state.param.objects('existing').items()
+                              if v.precedence is None or v.precedence >= 0]
+                kwargs.update({k: v for k, v in state.param.get_param_values()
                                if k in stage.param and k != 'name' and k in params})
 
         ready_param = stage_kwargs.get('ready_parameter', self.ready_parameter)
@@ -153,10 +229,7 @@ def _init_stage(self):
         else:
             self._state = stage(**kwargs)
 
-        if len(self._states) <= self._stage:
-            self._states.append(self._state)
-        else:
-            self._states[self._stage] = self._state
+        self._states[self._stage] = self._state
         return self._state.panel()
 
     def _set_stage(self, index):
@@ -173,18 +246,38 @@ def _set_stage(self, index):
                 if self._error.object:
                     break
 
+    @property
+    def _next_stage(self):
+        return self._next_selector.value
+
+    @property
+    def _prev_stage(self):
+        return self._prev_selector.value
+
     def _update_button(self):
+        options = list(self._graph.get(self._stage, []))
+        self._next_selector.options = options
+        self._next_selector.value = options[0] if options else None
+        self._next_selector.disabled = not bool(options)
+        previous = []
+        for src, tgts in self._graph.items():
+            if self._stage in tgts:
+                previous.append(src)
+        self._prev_selector.options = previous
+        self._prev_selector.value = previous[0] if previous else None
+        self._prev_selector.disabled = not bool(previous)
+
         # Disable previous button
-        if self._stage == 0:
+        if self._prev_stage is None:
             self._prev_button.disabled = True
         else:
             self._prev_button.disabled = False
 
         # Disable next button
-        if self._stage == len(self._stages)-1:
+        if self._next_stage is None:
             self._next_button.disabled = True
         else:
-            stage, kwargs = self._stages[self._stage][1:]
+            stage, kwargs = self._stages[self._stage]
             ready = kwargs.get('ready_parameter', self.ready_parameter)
             disabled = (not getattr(stage, ready)) if ready in stage.param else False
             self._next_button.disabled = disabled
@@ -205,14 +298,13 @@ def _get_error_button(self, e):
 
     @param.depends('next', watch=True)
     def _next(self):
-        self._stage += 1
+        self._stage = self._next_stage
         prev_state = self._state
         self._layout[1][0] = self._spinner_layout
         try:
             self._layout[1][0] = self._init_stage()
-            self._update_button()
         except Exception as e:
-            self._stage -= 1
+            self._stage = self._prev_stage
             self._state = prev_state
             self._layout[1][0] = prev_state.panel()
             self._progress_bar[0] = self._get_error_button(e)
@@ -221,53 +313,77 @@ def _next(self):
             return e
         else:
             self._progress_bar[0] = Spacer(width=100)
+            self._update_button()
         finally:
             self._progress_bar[1] = self._make_progress()
 
     @param.depends('previous', watch=True)
     def _previous(self):
-        self._stage -= 1
+        self._stage = self._prev_stage
         prev_state = self._state
         try:
-            self._state = self._states[self._stage]
+            if self._stage in self._states:
+                self._state = self._states[self._stage]
+                self._layout[1][0] = self._state.panel()
+            else:
+                self._layout[1][0] = self._init_stage()
             self._block = True
-            self._layout[1][0] = self._state.panel()
-            self._update_button()
         except Exception as e:
-            self._stage += 1
+            self._stage = self._next_stage
             self._state = prev_state
             self._progress_bar[0] = self._get_error_button(e)
             if self.debug:
                 raise e
         else:
             self._progress_bar[0] = Spacer(width=100)
+            self._update_button()
         finally:
             self._progress_bar[1] = self._make_progress()
 
     def _make_progress(self):
         import holoviews as hv
         import holoviews.plotting.bokeh # noqa
-        stages = len(self._stages)
-        line = hv.Path([[(0, 0), (stages-1, 0)]]).options(
-            line_width=6, color='black', default_tools=[], backend='bokeh'
-        )
-        vals = np.arange(stages)
-        active = [1 if v == self._stage else 0 for v in vals]
-        points = hv.Points((vals, np.zeros(stages), active), vdims=['active']).options(
-            color_index='active', line_color='black', cmap={0: 'white', 1: '#5cb85c'},
-            show_legend=False, size=20, default_tools=['tap'],
-            nonselection_alpha=1, backend='bokeh'
-        )
-        point_labels = points.add_dimension('text', 0, [s[0] for s in self._stages], vdim=True)
-        labels = hv.Labels(point_labels).options(yoffset=-2.5, default_tools=[], backend='bokeh')
-        self._progress_sel.source = points
-        hv_plot = (line * points * labels).options(
-            xaxis=None, yaxis=None, width=800, show_frame=False,
-            height=80, xlim=(-0.5, stages-0.5), ylim=(-4, 1.5),
-            clone=False, default_tools=[], active_tools=['tap'],
-            toolbar=None, backend='bokeh'
+
+        if self._graph:
+            root = get_root(self._graph)
+            depth = get_depth(root, self._graph)
+            breadths = get_breadths('A', self._graph)
+            max_breadth = max(len(v) for v in breadths.values())
+        else:
+            root = None
+            max_breadth, depth = 0, 0
+            breadths = {}
+
+        height = 80 + (max_breadth-1) * 20
+
+        edges = []
+        for src, tgts in self._graph.items():
+            for t in tgts:
+                edges.append((src, t))
+
+        nodes = []
+        for depth, subnodes in breadths.items():
+            breadth = len(subnodes)
+            step = 1./breadth
+            for i, n in enumerate(subnodes):
+                nodes.append((depth, step/2.+i*step, n, n==self._stage))
+
+        nodes = hv.Nodes(nodes, ['x', 'y', 'Stage'], 'Active')
+        graph = hv.Graph((edges, nodes)).opts(
+            node_color='Active', cmap={'False': 'white', 'True': '#5cb85c'},
+            tools=[], default_tools=['hover'], selection_policy=None,
+            edge_hover_line_color='black', node_hover_fill_color='gray',
+            backend='bokeh')
+        labels = hv.Labels(nodes, ['x', 'y'], 'Stage').opts(
+            yoffset=-.2, backend='bokeh')
+        plot = (graph * labels) if self._linear else graph
+        plot.opts(
+            xaxis=None, yaxis=None, min_width=600, responsive=True,
+            show_frame=False, height=height, xlim=(-0.25, depth+0.25), ylim=(0, 1),
+            default_tools=['hover'], toolbar=None,
+            backend='bokeh'
         )
-        return HoloViews(hv_plot, backend='bokeh')
+        return HoloViews(plot, backend='bokeh')
 
     #----------------------------------------------------------------
     # Public API
@@ -287,13 +403,9 @@ def add_stage(self, name, stage, **kwargs):
           Additional arguments declaring the behavior of the stage.
         """
         self._validate(stage)
-        self._stages.append((name, stage, kwargs))
-        if len(self._stages) == 1:
-            stage = self._init_stage()
-            self._layout[1].append(stage)
-            self._update_button()
+        self._stages[name] = (stage, kwargs)
 
-    def define_graph(self, graph):
+    def define_graph(self, graph, force=False):
         """
         Declares a custom graph structure for the Pipeline overriding
         the default linear flow. The graph should be defined as an
@@ -306,7 +418,11 @@ def define_graph(self, graph):
           pipeline stages. Should map from a single stage name to
           one or more stage names.
         """
-        stages = [stage[0] for stage in self._stages]
+        stages = list(self._stages)
+        if not stages:
+            self._graph = {}
+            return
+
         not_found = []
         for source, targets in graph.items():
             if source not in stages:
@@ -318,10 +434,33 @@ def define_graph(self, graph):
                 'referenced in the graph have been added.' %
                 (not_found[0] if len(not_found) == 1 else not_found)
             )
+
+        if graph:
+            if not (self._linear or force):
+                raise ValueError("Graph has already been defined, "
+                                 "cannot override existing graph.")
+            graph = {k: v if isinstance(v, tuple) else (v,)
+                     for k, v in graph.items()}
+            self._linear = False
+        else:
+            graph = {s: (t,) for s, t in zip(stages[:-1], stages[1:])}
+
+        root = get_root(graph)
+        if not is_traversable(root, graph, stages):
+            raise ValueError('Graph is not fully traversable from stage: %s.'
+                             % root)
+
+        self._stage = root
         self._graph = graph
+        if not self._linear:
+            self._progress_bar[2] = Column(self._prev_selector, self._prev_button)
+            self._progress_bar[3] = Column(self._next_selector, self._next_button)
 
     @property
     def layout(self):
-        self._progress_bar[1] = self._make_progress()
+        if self._linear or not self._graph:
+            self.define_graph(self._graph)
+        self._layout[1][:] = [self._init_stage()]
         self._update_button()
+        self._progress_bar[1] = self._make_progress()
         return self._layout

From 70c069d5663bb6425ef5fecc0d5b7e71f223bc59 Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Tue, 22 Oct 2019 15:49:48 -0500
Subject: [PATCH 06/19] Various improvements after Pipeline graph introduction

---
 panel/pipeline.py | 86 ++++++++++++++++++++++++++++-------------------
 1 file changed, 51 insertions(+), 35 deletions(-)

diff --git a/panel/pipeline.py b/panel/pipeline.py
index 556d75c0e6..b7a18a5b43 100644
--- a/panel/pipeline.py
+++ b/panel/pipeline.py
@@ -10,7 +10,7 @@
 import numpy as np
 
 from .layout import Row, Column, HSpacer, VSpacer, Spacer
-from .pane import HoloViews, Pane
+from .pane import HoloViews, Pane, Markdown
 from .widgets import Button, Select
 from .param import Param
 from .util import param_reprs
@@ -124,6 +124,7 @@ def __init__(self, stages=[], graph={}, **params):
         self._state = None
         self._linear = True
         self._block = False
+        self._error = None
         self._graph = {}
         self._progress_sel = hv.streams.Selection1D()
         self._progress_sel.add_subscriber(self._set_stage)
@@ -134,8 +135,12 @@ def __init__(self, stages=[], graph={}, **params):
         self._next_button.width = 125
         self._next_selector = Select(width=125)
         self._prev_button.disabled = True
+        self._next_selector.param.watch(self._update_progress, 'value')
         self._progress_bar = Row(
-            Spacer(width=100),
+            Column(
+                Markdown('# Header', margin=(0, 0, 0, 5)),
+                Spacer(width=100),
+            ),
             self._make_progress(),
             self._prev_button,
             self._next_button,
@@ -233,18 +238,7 @@ def _init_stage(self):
         return self._state.panel()
 
     def _set_stage(self, index):
-        idx = index[0]
-        steps = idx-self._stage
-        if steps < 0:
-            for i in range(abs(steps)):
-                self.param.trigger('previous')
-                if self._error.object:
-                    break
-        else:
-            for i in range(steps):
-                self.param.trigger('next')
-                if self._error.object:
-                    break
+        self._next_selector.value = list(self._stages)[index[0]]
 
     @property
     def _next_stage(self):
@@ -292,35 +286,37 @@ def _get_error_button(self, e):
         else:
             traceback = msg or "Undefined error, enable debug mode."
         button = Button(name='Error', button_type='danger', width=100,
-                        align='center')
+                        align='center', margin=(0, 0, 0, 5))
         button.jslink(button, code={'clicks': "alert(`{tb}`)".format(tb=traceback)})
         return button
 
     @param.depends('next', watch=True)
     def _next(self):
+        prev_state, prev_stage = self._state, self._stage
         self._stage = self._next_stage
-        prev_state = self._state
         self._layout[1][0] = self._spinner_layout
         try:
             self._layout[1][0] = self._init_stage()
         except Exception as e:
-            self._stage = self._prev_stage
+            self._error = self._stage
+            self._stage = prev_stage
             self._state = prev_state
             self._layout[1][0] = prev_state.panel()
-            self._progress_bar[0] = self._get_error_button(e)
+            self._progress_bar[0][1] = self._get_error_button(e)
             if self.debug:
                 raise e
             return e
         else:
-            self._progress_bar[0] = Spacer(width=100)
+            self._error = None
+            self._progress_bar[0][1] = Spacer(width=100)
             self._update_button()
         finally:
-            self._progress_bar[1] = self._make_progress()
+            self._update_progress()
 
     @param.depends('previous', watch=True)
     def _previous(self):
+        prev_state, prev_stage = self._state, self._stage
         self._stage = self._prev_stage
-        prev_state = self._state
         try:
             if self._stage in self._states:
                 self._state = self._states[self._stage]
@@ -329,16 +325,22 @@ def _previous(self):
                 self._layout[1][0] = self._init_stage()
             self._block = True
         except Exception as e:
-            self._stage = self._next_stage
+            self._error = self._stage
+            self._stage = prev_xstage
             self._state = prev_state
-            self._progress_bar[0] = self._get_error_button(e)
+            self._progress_bar[0][1] = self._get_error_button(e)
             if self.debug:
                 raise e
         else:
-            self._progress_bar[0] = Spacer(width=100)
+            self._error = None
+            self._progress_bar[0][1] = Spacer(width=100)
             self._update_button()
         finally:
-            self._progress_bar[1] = self._make_progress()
+            self._update_progress()
+
+    def _update_progress(self, *args):
+        self._progress_bar[0][0].object = '# Stage: ' + self._stage
+        self._progress_bar[1] = self._make_progress()
 
     def _make_progress(self):
         import holoviews as hv
@@ -347,7 +349,7 @@ def _make_progress(self):
         if self._graph:
             root = get_root(self._graph)
             depth = get_depth(root, self._graph)
-            breadths = get_breadths('A', self._graph)
+            breadths = get_breadths(root, self._graph)
             max_breadth = max(len(v) for v in breadths.values())
         else:
             root = None
@@ -365,17 +367,29 @@ def _make_progress(self):
         for depth, subnodes in breadths.items():
             breadth = len(subnodes)
             step = 1./breadth
-            for i, n in enumerate(subnodes):
-                nodes.append((depth, step/2.+i*step, n, n==self._stage))
-
-        nodes = hv.Nodes(nodes, ['x', 'y', 'Stage'], 'Active')
+            for i, n in enumerate(subnodes[::-1]):
+                if n == self._stage:
+                    state = 'active'
+                elif n == self._error:
+                    state = 'error'
+                elif n == self._next_stage:
+                    state = 'next'
+                else:
+                    state = 'inactive'
+                nodes.append((depth, step/2.+i*step, n, state))
+
+        cmap = {'inactive': 'white', 'active': '#5cb85c', 'error': 'red',
+                'next': 'yellow'}
+
+        nodes = hv.Nodes(nodes, ['x', 'y', 'Stage'], 'State').opts(
+            alpha=0, selection_alpha=0, nonselection_alpha=0, backend='bokeh')
+        self._progress_sel.source = nodes
         graph = hv.Graph((edges, nodes)).opts(
-            node_color='Active', cmap={'False': 'white', 'True': '#5cb85c'},
+            edge_hover_line_color='black', node_color='State', cmap=cmap,
             tools=[], default_tools=['hover'], selection_policy=None,
-            edge_hover_line_color='black', node_hover_fill_color='gray',
-            backend='bokeh')
+            node_hover_fill_color='gray', backend='bokeh')
         labels = hv.Labels(nodes, ['x', 'y'], 'Stage').opts(
-            yoffset=-.2, backend='bokeh')
+            yoffset=-.30, backend='bokeh')
         plot = (graph * labels) if self._linear else graph
         plot.opts(
             xaxis=None, yaxis=None, min_width=600, responsive=True,
@@ -455,12 +469,14 @@ def define_graph(self, graph, force=False):
         if not self._linear:
             self._progress_bar[2] = Column(self._prev_selector, self._prev_button)
             self._progress_bar[3] = Column(self._next_selector, self._next_button)
+        self._update_progress()
 
     @property
     def layout(self):
         if self._linear or not self._graph:
             self.define_graph(self._graph)
+        else:
+            self._update_progress()
         self._layout[1][:] = [self._init_stage()]
         self._update_button()
-        self._progress_bar[1] = self._make_progress()
         return self._layout

From 00d8969de54803d72d2f40db220d0eb32a016227 Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Wed, 23 Oct 2019 16:30:20 -0500
Subject: [PATCH 07/19] Made Pipeline composable

---
 panel/pipeline.py | 165 +++++++++++++++++++++++++++++-----------------
 1 file changed, 106 insertions(+), 59 deletions(-)

diff --git a/panel/pipeline.py b/panel/pipeline.py
index b7a18a5b43..4a26921d31 100644
--- a/panel/pipeline.py
+++ b/panel/pipeline.py
@@ -89,24 +89,51 @@ def get_breadths(node, graph, depth=0, breadths=None):
 
 class Pipeline(param.Parameterized):
     """
-    Allows connecting a linear series of panels to define a workflow.
-    Each stage in a pipeline should declare a panel method which
-    returns a panel object that can be displayed and annotate its
-    outputs using the param.output decorator.
+    A Pipeline represents a directed graph of stages, which each
+    returns a panel object to render. A pipeline therefore represents
+    a UI workflow of multiple linear or branching stages.
+
+    The Pipeline layout consists of a number of sub-components:
+
+    * header:
+      - title: The name of the current stage.
+      - error: A field to display the error state.
+      - network: A network diagram representing the pipeline.
+      - buttons: All navigation buttons and selectors.
+      - prev_button: The button to go to the previous stage.
+      - prev_selector: The selector widget to select between previous
+                       branching stages.
+      - next_button:   The button to go to the previous stage
+      - next_selector: The selector widget to select the next
+                       branching stages.
+    * stage: The contents of the current pipeline stage.
+
+    By default any outputs of one stage annotated with the
+    param.output decorator are fed into the next stage. Additionally,
+    if the inherit_params parameter is set any parameters which are
+    declared on both the previous and next stage are also inherited.
+
+    The stages are declared using the add_stage method and must each
+    be given a unique name. By default any stages will simply be
+    connected linearly, however an explicit graph can be declared using
+    the define_graph method.
     """
 
-    auto_advance = param.Boolean(default=False, precedence=-1, doc="""
+    auto_advance = param.Boolean(default=False, doc="""
         Whether to automatically advance if the ready parameter is True.""")
 
-    debug = param.Boolean(default=False, precedence=-1, doc="""
+    debug = param.Boolean(default=False, doc="""
         Whether to raise errors, useful for debugging while building an application.""")
 
-    inherit_params = param.Boolean(default=True, precedence=-1, doc="""
+    inherit_params = param.Boolean(default=True, doc="""
         Whether parameters should be inherited between pipeline stages""")
 
     ready_parameter = param.String(default=None, doc="""
         Parameter name to watch to check whether a stage is ready.""")
 
+    show_header = param.Boolean(default=True, doc="""
+        Whether to show the header with the title, network diagram, and buttons.""")
+
     next = param.Action(default=lambda x: x.param.trigger('next'))
 
     previous = param.Action(default=lambda x: x.param.trigger('previous'))
@@ -117,42 +144,51 @@ def __init__(self, stages=[], graph={}, **params):
         except:
             raise ImportError('Pipeline requires holoviews to be installed')
 
+        super(Pipeline, self).__init__(**params)
+
+        # Initialize internal state
         self._stage = None
         self._stages = OrderedDict()
-        super(Pipeline, self).__init__(**params)
         self._states = {}
         self._state = None
         self._linear = True
         self._block = False
         self._error = None
         self._graph = {}
+        self._route = []
+
+        # Declare UI components
         self._progress_sel = hv.streams.Selection1D()
         self._progress_sel.add_subscriber(self._set_stage)
-        self._prev_button = Param(self.param.previous).layout[0]
-        self._prev_button.width = 125
-        self._prev_selector = Select(width=125)
-        self._next_button = Param(self.param.next).layout[0]
-        self._next_button.width = 125
-        self._next_selector = Select(width=125)
-        self._prev_button.disabled = True
-        self._next_selector.param.watch(self._update_progress, 'value')
-        self._progress_bar = Row(
-            Column(
-                Markdown('# Header', margin=(0, 0, 0, 5)),
-                Spacer(width=100),
-            ),
-            self._make_progress(),
-            self._prev_button,
-            self._next_button,
+        self.prev_button = Param(self.param.previous).layout[0]
+        self.prev_button.width = 125
+        self.prev_selector = Select(width=125)
+        self.next_button = Param(self.param.next).layout[0]
+        self.next_button.width = 125
+        self.next_selector = Select(width=125)
+        self.prev_button.disabled = True
+        self.next_selector.param.watch(self._update_progress, 'value')
+        self.network = HoloViews(backend='bokeh')
+        self.title = Markdown('# Header', margin=(0, 0, 0, 5))
+        self.error = Row(width=100)
+        self.buttons = Row(self.prev_button, self.next_button)
+        self.header = Row(
+            Column(self.title, self.error),
+            self.network,
+            self.buttons,
             sizing_mode='stretch_width'
         )
+        self.network.object = self._make_progress()
         spinner = Pane(os.path.join(os.path.dirname(__file__), 'assets', 'spinner.gif'))
         self._spinner_layout = Row(
             HSpacer(),
             Column(VSpacer(), spinner, VSpacer()),
             HSpacer()
         )
-        self._layout = Column(self._progress_bar, Row(), sizing_mode='stretch_width')
+        self.stage = Row()
+        self._layout = Column(self.header, self.stage, sizing_mode='stretch_width')
+
+        # Initialize stages and the graph
         for stage in stages:
             kwargs = {}
             if len(stage) == 2:
@@ -188,7 +224,7 @@ def _unblock(self, event):
             self._block = False
             return
 
-        button = self._progress_bar[-1][0]
+        button = self.next_button
         if button.disabled and event.new:
             button.disabled = False
         elif not button.disabled and not event.new:
@@ -238,43 +274,43 @@ def _init_stage(self):
         return self._state.panel()
 
     def _set_stage(self, index):
-        self._next_selector.value = list(self._stages)[index[0]]
+        self.next_selector.value = list(self._stages)[index[0]]
 
     @property
     def _next_stage(self):
-        return self._next_selector.value
+        return self.next_selector.value
 
     @property
     def _prev_stage(self):
-        return self._prev_selector.value
+        return self.prev_selector.value
 
     def _update_button(self):
         options = list(self._graph.get(self._stage, []))
-        self._next_selector.options = options
-        self._next_selector.value = options[0] if options else None
-        self._next_selector.disabled = not bool(options)
+        self.next_selector.options = options
+        self.next_selector.value = options[0] if options else None
+        self.next_selector.disabled = not bool(options)
         previous = []
         for src, tgts in self._graph.items():
             if self._stage in tgts:
                 previous.append(src)
-        self._prev_selector.options = previous
-        self._prev_selector.value = previous[0] if previous else None
-        self._prev_selector.disabled = not bool(previous)
+        self.prev_selector.options = previous
+        self.prev_selector.value = self._route[-1] if previous else None
+        self.prev_selector.disabled = not bool(previous)
 
         # Disable previous button
         if self._prev_stage is None:
-            self._prev_button.disabled = True
+            self.prev_button.disabled = True
         else:
-            self._prev_button.disabled = False
+            self.prev_button.disabled = False
 
         # Disable next button
         if self._next_stage is None:
-            self._next_button.disabled = True
+            self.next_button.disabled = True
         else:
             stage, kwargs = self._stages[self._stage]
             ready = kwargs.get('ready_parameter', self.ready_parameter)
             disabled = (not getattr(stage, ready)) if ready in stage.param else False
-            self._next_button.disabled = disabled
+            self.next_button.disabled = disabled
 
     def _get_error_button(self, e):
         msg = str(e) if isinstance(e, PipelineError) else ""
@@ -294,22 +330,23 @@ def _get_error_button(self, e):
     def _next(self):
         prev_state, prev_stage = self._state, self._stage
         self._stage = self._next_stage
-        self._layout[1][0] = self._spinner_layout
+        self.stage[0] = self._spinner_layout
         try:
-            self._layout[1][0] = self._init_stage()
+            self.stage[0] = self._init_stage()
         except Exception as e:
             self._error = self._stage
             self._stage = prev_stage
             self._state = prev_state
-            self._layout[1][0] = prev_state.panel()
-            self._progress_bar[0][1] = self._get_error_button(e)
+            self.stage[0] = prev_state.panel()
+            self.error[:] = [self._get_error_button(e)]
             if self.debug:
                 raise e
             return e
         else:
+            self.error = [Spacer(width=100)]
             self._error = None
-            self._progress_bar[0][1] = Spacer(width=100)
             self._update_button()
+            self._route.append(self._stage)
         finally:
             self._update_progress()
 
@@ -320,27 +357,28 @@ def _previous(self):
         try:
             if self._stage in self._states:
                 self._state = self._states[self._stage]
-                self._layout[1][0] = self._state.panel()
+                self.stage[0] = self._state.panel()
             else:
-                self._layout[1][0] = self._init_stage()
+                self.stage[0] = self._init_stage()
             self._block = True
         except Exception as e:
+            self.error[:] = [self._get_error_button(e)]
             self._error = self._stage
-            self._stage = prev_xstage
+            self._stage = prev_stage
             self._state = prev_state
-            self._progress_bar[0][1] = self._get_error_button(e)
             if self.debug:
                 raise e
         else:
+            self.error[:] = [Spacer(width=100)]
             self._error = None
-            self._progress_bar[0][1] = Spacer(width=100)
             self._update_button()
+            self._route.pop()
         finally:
             self._update_progress()
 
     def _update_progress(self, *args):
-        self._progress_bar[0][0].object = '# Stage: ' + self._stage
-        self._progress_bar[1] = self._make_progress()
+        self.title.object = '## Stage: ' + self._stage
+        self.network.object = self._make_progress()
 
     def _make_progress(self):
         import holoviews as hv
@@ -397,7 +435,7 @@ def _make_progress(self):
             default_tools=['hover'], toolbar=None,
             backend='bokeh'
         )
-        return HoloViews(plot, backend='bokeh')
+        return plot
 
     #----------------------------------------------------------------
     # Public API
@@ -437,6 +475,8 @@ def define_graph(self, graph, force=False):
             self._graph = {}
             return
 
+        graph = {k: v if isinstance(v, tuple) else (v,) for k, v in graph.items()}
+
         not_found = []
         for source, targets in graph.items():
             if source not in stages:
@@ -453,8 +493,6 @@ def define_graph(self, graph, force=False):
             if not (self._linear or force):
                 raise ValueError("Graph has already been defined, "
                                  "cannot override existing graph.")
-            graph = {k: v if isinstance(v, tuple) else (v,)
-                     for k, v in graph.items()}
             self._linear = False
         else:
             graph = {s: (t,) for s, t in zip(stages[:-1], stages[1:])}
@@ -466,17 +504,26 @@ def define_graph(self, graph, force=False):
 
         self._stage = root
         self._graph = graph
+        self._route = [root]
         if not self._linear:
-            self._progress_bar[2] = Column(self._prev_selector, self._prev_button)
-            self._progress_bar[3] = Column(self._next_selector, self._next_button)
+            self.buttons[:] = [
+                Column(self.prev_selector, self.prev_button),
+                Column(self.next_selector, self.next_button)
+            ]
+        self.stage[:] = [self._init_stage()]
         self._update_progress()
 
-    @property
-    def layout(self):
+    def init(self):
+        """
+        Initialize the Pipeline before first display.
+        """
         if self._linear or not self._graph:
             self.define_graph(self._graph)
         else:
             self._update_progress()
-        self._layout[1][:] = [self._init_stage()]
         self._update_button()
+        
+    @property
+    def layout(self):
+        self.init()
         return self._layout

From 8907e322840489db2a07cf6e339ffc14948820db Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Wed, 23 Oct 2019 17:23:58 -0500
Subject: [PATCH 08/19] Added docs for new Pipeline features

---
 examples/user_guide/Pipelines.ipynb | 255 +++++++++++++++++++++++++++-
 1 file changed, 248 insertions(+), 7 deletions(-)

diff --git a/examples/user_guide/Pipelines.ipynb b/examples/user_guide/Pipelines.ipynb
index 5cfb34e03e..59158124f6 100644
--- a/examples/user_guide/Pipelines.ipynb
+++ b/examples/user_guide/Pipelines.ipynb
@@ -44,7 +44,9 @@
     "\n",
     "* ``param.output(c=param.Number, d=param.String)`` or ``param.output(('c', param.Number), ('d', param.String))``\n",
     "\n",
-    "In the example below the output is simply the result of multiplying the two inputs (``a`` and ``b``) which will produce output ``c``. Additionally we declare a ``view`` method which returns a ``LaTeX`` pane which will render the equation to ``LaTeX``. Finally a ``panel`` method declares returns a Panel object rendering both the parameters and the view; this is the second convention that a ``Pipeline`` expects.\n",
+    "In the example below the output is simply the result of multiplying the two inputs (``a`` and ``b``) which will produce output ``c``. Additionally we declare a ``view`` method which returns a ``LaTeX`` pane which will render the equation to ``LaTeX``. Finally a ``panel`` method should be implemented to return a Panel object providing a visual representation of the stage; this is the second convention that a ``Pipeline`` expects.\n",
+    "\n",
+    "In addition to the output parameters, as long as ``inherit_params`` is set any parameters which exist between two different stages will also be passed along.\n",
     "\n",
     "Let's start by displaying this stage on its own:"
    ]
@@ -56,24 +58,28 @@
    "outputs": [],
    "source": [
     "class Stage1(param.Parameterized):\n",
-    "    \n",
+    "\n",
     "    a = param.Number(default=5, bounds=(0, 10))\n",
     "\n",
     "    b = param.Number(default=5, bounds=(0, 10))\n",
     "    \n",
+    "    ready = param.Boolean(default=False, precedence=-1)\n",
+    "\n",
     "    @param.output(('c', param.Number), ('d', param.Number))\n",
     "    def output(self):\n",
     "        return self.a * self.b, self.a ** self.b\n",
     "    \n",
     "    @param.depends('a', 'b')\n",
     "    def view(self):\n",
+    "        if self.a > 5:\n",
+    "            self.ready = True\n",
     "        c, d = self.output()\n",
     "        return pn.pane.LaTeX('${a} * {b} = {c}$\\n${a}^{{{b}}} = {d}$'.format(\n",
     "            a=self.a, b=self.b, c=c, d=d), style={'font-size': '2em'})\n",
     "\n",
     "    def panel(self):\n",
     "        return pn.Row(self.param, self.view)\n",
-    "    \n",
+    "\n",
     "stage1 = Stage1()\n",
     "stage1.panel()"
    ]
@@ -84,9 +90,9 @@
    "source": [
     "To summarize we have followed several conventions when setting up this stage of our ``Pipeline``:\n",
     "\n",
-    "1. Declare a Parameterized class with some input parameters\n",
-    "2. Declare one or more output methods and name them appropriately\n",
-    "3. Declare a ``panel`` method that returns a view of the object that the ``Pipeline`` can render\n",
+    "1. Declare a Parameterized class with some input parameters.\n",
+    "2. Declare one or more methods decorated with the `param.output` decorator.\n",
+    "3. Declare a ``panel`` method that returns a view of the object that the ``Pipeline`` can render.\n",
     "\n",
     "Now that the object has been instantiated we can also query it for its outputs:"
    ]
@@ -116,6 +122,13 @@
     "pipeline.add_stage('Stage 1', stage1)"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "The ``add_stage`` method takes the name of the stage as its first argument, the stage class or instance as the second parameter and any additional keyword arguments to override default behavior."
+   ]
+  },
   {
    "cell_type": "markdown",
    "metadata": {},
@@ -178,6 +191,9 @@
    "metadata": {},
    "outputs": [],
    "source": [
+    "pipeline = pn.pipeline.Pipeline(debug=True)\n",
+    "pipeline.add_stage('Stage 1', Stage1(), ready_parameter='ready', auto_advance=True)\n",
+    "pipeline.add_stage('Stage 2', Stage2)\n",
     "pipeline.layout"
    ]
   },
@@ -211,6 +227,231 @@
    "source": [
     "As you will note the Pipeline stages may be either ``Parameterized`` instances or ``Parameterized`` classes, but when working with instances you must ensure that updating the parameters of the class is sufficient to update the current state of the class."
    ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Non-linear pipelines\n",
+    "\n",
+    "Pipelines are not limited to simple linear UI workflows, they support any arbitrary branching structures, or put another way any acyclic graph. A simple example might be a workflow with two alternative stages which rejoin at a later point. In the very simple example below we declare four stages, an `Input`, `Multiply` and `Add` and finally a `Result` stage."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "class Input(param.Parameterized):\n",
+    "    \n",
+    "    value1 = param.Integer(default=0)\n",
+    "\n",
+    "    value2 = param.Integer(default=0)\n",
+    "    \n",
+    "    def panel(self):\n",
+    "        return pn.Row(self.param.value1, self.param.value2)\n",
+    "\n",
+    "class Multiply(Input):\n",
+    "    \n",
+    "    def panel(self):\n",
+    "        return '%.3f * %.3f' % (self.value1, self.value2)\n",
+    "\n",
+    "    @param.output('result')\n",
+    "    def output(self):\n",
+    "        return self.value1 * self.value2\n",
+    "    \n",
+    "class Add(Input):\n",
+    "    \n",
+    "    def panel(self):\n",
+    "        return '%d + %d' % (self.value1, self.value2)\n",
+    "    \n",
+    "    @param.output('result')\n",
+    "    def output(self):\n",
+    "        return self.value1 + self.value2\n",
+    "    \n",
+    "class Result(Input):\n",
+    "    \n",
+    "    result = param.Number(default=0)\n",
+    "\n",
+    "    def panel(self):\n",
+    "        return self.result\n",
+    "    \n",
+    "dag = pn.pipeline.Pipeline()\n",
+    "\n",
+    "dag.add_stage('Input', Input)\n",
+    "dag.add_stage('Multiply', Multiply)\n",
+    "dag.add_stage('Add', Add)\n",
+    "dag.add_stage('Result', Result)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "After adding all the stages we have to express the relationship between these stages. To declare the graph we can use the ``define_graph`` method and provide a adjacency map, which declares which stage feeds into which other stages. In this case the `Input` feeds into both ``Multiply`` and ``Add`` and both those stages feed into the ``Result``:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "dag.define_graph({'Input': ('Multiply', 'Add'), 'Multiply': 'Result', 'Add': 'Result'})"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "This is of course a very simple example but it demonstrates the ability to express arbitrary workflows with branching and converging steps:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "dag.layout"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Custom layout\n",
+    "\n",
+    "The `Pipeline` has a default `layout` with a `header` bar containing the navigation elements and the main `stage`. To achieve more custom layouts, each of the components can be arranged manually:\n",
+    "\n",
+    "* `layout`: The overall layout of the header and stage.\n",
+    "* `header`: The navigation components and network diagram.\n",
+    "  - `title`: The name of the current stage.\n",
+    "  - `network`: A network diagram representing the pipeline.\n",
+    "  - `buttons`: All navigation buttons and selectors.\n",
+    "  - `prev_button`: The button to go to the previous stage.\n",
+    "  - `prev_selector`: The selector widget to select between previous branching stages.\n",
+    "  - `next_button`:   The button to go to the previous stage\n",
+    "  - `next_selector`: The selector widget to select the next branching stages.\n",
+    "* `stage`: The contents of the current pipeline stage.\n",
+    "\n",
+    "When displaying a pipeline in this way it is important to first initialize it:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "dag.init()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "Now we can compose the components in any way that is desired:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "pn.Column(\n",
+    "    pn.Row(dag.title, pn.layout.HSpacer(), dag.buttons),\n",
+    "    dag.network,\n",
+    "    dag.stage\n",
+    ")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Programmatic flow control\n",
+    "\n",
+    "By default controlling the flow between different stages is done using the \"Previous\" and \"Next\" buttons. However often we want to control the UI flow programmatically from within a stage. A `Pipeline` allows programmatic control by declaring a `ready_param` either per stage or globally on the Pipeline, which blocks and unblocks the buttons and allow advancing automatically when combined with the ``auto_advance`` parameter. In this way we can control the workflow programmatically from inside the stages.\n",
+    "\n",
+    "In the example below we create a version of the previous workflow which can be used without the buttons by declaring ``ready`` parameters on each of the stages, which we can toggle with a custom button or simply set to `True` by default to automatically skip the stage."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "class AutoInput(Input):\n",
+    "\n",
+    "    operator = param.Selector(default='+', objects=['*', '+'])\n",
+    "    \n",
+    "    ready = param.Boolean(default=False)\n",
+    "    \n",
+    "    def panel(self):\n",
+    "        button = pn.widgets.Button(name='Go', button_type='success')\n",
+    "        button.on_click(lambda event: setattr(self, 'ready', True))\n",
+    "        return pn.Column(\n",
+    "            pn.Row(self.param.value1, self.param.operator, self.param.value2),\n",
+    "            button\n",
+    "        )\n",
+    "\n",
+    "class AutoMultiply(Multiply):\n",
+    "    \n",
+    "    ready = param.Boolean(default=True)\n",
+    "\n",
+    "class AutoAdd(Add):\n",
+    "    \n",
+    "    ready = param.Boolean(default=True)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "Now that we have declared these stages let us set up the pipeline ensuring that we declare the `ready_parameter` and `auto_advance` settings:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "dag = pn.pipeline.Pipeline() # could set ready_parameter='ready' and auto_advance=True globally\n",
+    "\n",
+    "dag.add_stage('Input', AutoInput, ready_parameter='ready', auto_advance=True)\n",
+    "dag.add_stage('Multiply', AutoMultiply, ready_parameter='ready', auto_advance=True)\n",
+    "dag.add_stage('Add', AutoAdd, ready_parameter='ready', auto_advance=True)\n",
+    "dag.add_stage('Result', Result)\n",
+    "\n",
+    "dag.define_graph({'Input': ('Multiply', 'Add'), 'Multiply': 'Result', 'Add': 'Result'})\n",
+    "\n",
+    "dag.init()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "Finally we display the pipeline, without the buttons because we don't need them:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "pn.Column(\n",
+    "    dag.title,\n",
+    "    dag.network,\n",
+    "    dag.stage\n",
+    ")"
+   ]
   }
  ],
  "metadata": {
@@ -220,5 +461,5 @@
   }
  },
  "nbformat": 4,
- "nbformat_minor": 2
+ "nbformat_minor": 4
 }

From 19b04f7b98e8aa4895f730facb22f7d3a4e67619 Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Wed, 23 Oct 2019 17:24:34 -0500
Subject: [PATCH 09/19] Fixed minor issues

---
 panel/pipeline.py | 20 ++++++++++++++------
 1 file changed, 14 insertions(+), 6 deletions(-)

diff --git a/panel/pipeline.py b/panel/pipeline.py
index 4a26921d31..4192865b00 100644
--- a/panel/pipeline.py
+++ b/panel/pipeline.py
@@ -217,7 +217,7 @@ def __repr__(self):
         return repr_str
 
     def __getitem__(self, index):
-        return self._stages[index][1]
+        return self._stages[index][0]
 
     def _unblock(self, event):
         if self._state is not event.obj or self._block:
@@ -260,16 +260,16 @@ def _init_stage(self):
                 kwargs.update({k: v for k, v in state.param.get_param_values()
                                if k in stage.param and k != 'name' and k in params})
 
-        ready_param = stage_kwargs.get('ready_parameter', self.ready_parameter)
-        if ready_param and ready_param in stage.param:
-            stage.param.watch(self._unblock, ready_param, onlychanged=False)
-
         if isinstance(stage, param.Parameterized):
             stage.set_param(**kwargs)
             self._state = stage
         else:
             self._state = stage(**kwargs)
 
+        ready_param = stage_kwargs.get('ready_parameter', self.ready_parameter)
+        if ready_param and ready_param in stage.param:
+            self._state.param.watch(self._unblock, ready_param, onlychanged=False)
+
         self._states[self._stage] = self._state
         return self._state.panel()
 
@@ -347,6 +347,11 @@ def _next(self):
             self._error = None
             self._update_button()
             self._route.append(self._stage)
+            stage_kwargs = self._stages[self._stage][-1]
+            ready_param = stage_kwargs.get('ready_parameter', self.ready_parameter)
+            if (ready_param and getattr(self._state, ready_param, False) and
+                stage_kwargs.get('auto_advance', self.auto_advance)):
+                self._next()
         finally:
             self._update_progress()
 
@@ -455,6 +460,9 @@ def add_stage(self, name, stage, **kwargs):
           Additional arguments declaring the behavior of the stage.
         """
         self._validate(stage)
+        for k in kwargs:
+            if k not in self.param:
+                raise ValueError("Keyword argument %s is not a valid parameter. " % k)
         self._stages[name] = (stage, kwargs)
 
     def define_graph(self, graph, force=False):
@@ -522,7 +530,7 @@ def init(self):
         else:
             self._update_progress()
         self._update_button()
-        
+
     @property
     def layout(self):
         self.init()

From 3743eb095f57f0f2d37f4707914bcbc5c1ef6f7b Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Fri, 25 Oct 2019 13:55:39 -0500
Subject: [PATCH 10/19] Add control over branching stages from within stages

---
 panel/pipeline.py | 23 +++++++++++++++++++++--
 1 file changed, 21 insertions(+), 2 deletions(-)

diff --git a/panel/pipeline.py b/panel/pipeline.py
index 4192865b00..91a5eabdd5 100644
--- a/panel/pipeline.py
+++ b/panel/pipeline.py
@@ -128,6 +128,9 @@ class Pipeline(param.Parameterized):
     inherit_params = param.Boolean(default=True, doc="""
         Whether parameters should be inherited between pipeline stages""")
 
+    next_parameter = param.String(default=None, doc="""
+        Parameter name to watch to switch between different branching stages""")
+
     ready_parameter = param.String(default=None, doc="""
         Parameter name to watch to check whether a stage is ready.""")
 
@@ -234,6 +237,12 @@ def _unblock(self, event):
         if event.new and stage_kwargs.get('auto_advance', self.auto_advance):
             self._next()
 
+    def _select_next(self, event):
+        if self._state is not event.obj:
+            return
+        self.next_selector.value = event.new
+        self._update_progress()
+
     def _init_stage(self):
         stage, stage_kwargs = self._stages[self._stage]
 
@@ -270,6 +279,11 @@ def _init_stage(self):
         if ready_param and ready_param in stage.param:
             self._state.param.watch(self._unblock, ready_param, onlychanged=False)
 
+        next_param = stage_kwargs.get('next_parameter', self.next_parameter)
+        print(next_param)
+        if next_param and next_param in stage.param:
+            self._state.param.watch(self._select_next, next_param, onlychanged=False)
+
         self._states[self._stage] = self._state
         return self._state.panel()
 
@@ -285,9 +299,15 @@ def _prev_stage(self):
         return self.prev_selector.value
 
     def _update_button(self):
+        stage, kwargs = self._stages[self._stage]
         options = list(self._graph.get(self._stage, []))
+        next_param = kwargs.get('next_parameter', self.next_parameter)
+        option = getattr(self._state, next_param) if next_param and next_param in stage.param else None
+        print(next_param, option)
+        if option is None:
+            option = options[0]
         self.next_selector.options = options
-        self.next_selector.value = options[0] if options else None
+        self.next_selector.value = option if options else None
         self.next_selector.disabled = not bool(options)
         previous = []
         for src, tgts in self._graph.items():
@@ -307,7 +327,6 @@ def _update_button(self):
         if self._next_stage is None:
             self.next_button.disabled = True
         else:
-            stage, kwargs = self._stages[self._stage]
             ready = kwargs.get('ready_parameter', self.ready_parameter)
             disabled = (not getattr(stage, ready)) if ready in stage.param else False
             self.next_button.disabled = disabled

From 8019848a5c789eeba710f416077315a87d3d72b1 Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Fri, 25 Oct 2019 14:00:10 -0500
Subject: [PATCH 11/19] Update docs

---
 examples/user_guide/Pipelines.ipynb | 26 +++++++++++++-------------
 1 file changed, 13 insertions(+), 13 deletions(-)

diff --git a/examples/user_guide/Pipelines.ipynb b/examples/user_guide/Pipelines.ipynb
index 59158124f6..9ec7269dc7 100644
--- a/examples/user_guide/Pipelines.ipynb
+++ b/examples/user_guide/Pipelines.ipynb
@@ -71,8 +71,6 @@
     "    \n",
     "    @param.depends('a', 'b')\n",
     "    def view(self):\n",
-    "        if self.a > 5:\n",
-    "            self.ready = True\n",
     "        c, d = self.output()\n",
     "        return pn.pane.LaTeX('${a} * {b} = {c}$\\n${a}^{{{b}}} = {d}$'.format(\n",
     "            a=self.a, b=self.b, c=c, d=d), style={'font-size': '2em'})\n",
@@ -374,9 +372,11 @@
    "source": [
     "## Programmatic flow control\n",
     "\n",
-    "By default controlling the flow between different stages is done using the \"Previous\" and \"Next\" buttons. However often we want to control the UI flow programmatically from within a stage. A `Pipeline` allows programmatic control by declaring a `ready_param` either per stage or globally on the Pipeline, which blocks and unblocks the buttons and allow advancing automatically when combined with the ``auto_advance`` parameter. In this way we can control the workflow programmatically from inside the stages.\n",
+    "By default controlling the flow between different stages is done using the \"Previous\" and \"Next\" buttons. However often we want to control the UI flow programmatically from within a stage. A `Pipeline` allows programmatic control by declaring a `ready_parameter` either per stage or globally on the Pipeline, which blocks and unblocks the buttons and allow advancing automatically when combined with the ``auto_advance`` parameter. In this way we can control the workflow programmatically from inside the stages.\n",
     "\n",
-    "In the example below we create a version of the previous workflow which can be used without the buttons by declaring ``ready`` parameters on each of the stages, which we can toggle with a custom button or simply set to `True` by default to automatically skip the stage."
+    "In the example below we create a version of the previous workflow which can be used without the buttons by declaring ``ready`` parameters on each of the stages, which we can toggle with a custom button or simply set to `True` by default to automatically skip the stage.\n",
+    "\n",
+    "Lastly, we can also control which branching stage to switch to from within a stage. To do so we declare a parameter which will hold the name of the next stage to switch to, in this case selecting between 'Add' and 'Multiply', later we will point the pipeline to this parameter using the `next_parameter` argument."
    ]
   },
   {
@@ -387,17 +387,17 @@
    "source": [
     "class AutoInput(Input):\n",
     "\n",
-    "    operator = param.Selector(default='+', objects=['*', '+'])\n",
-    "    \n",
+    "    operator = param.Selector(default='Add', objects=['Multiply', 'Add'])\n",
+    "\n",
     "    ready = param.Boolean(default=False)\n",
     "    \n",
     "    def panel(self):\n",
     "        button = pn.widgets.Button(name='Go', button_type='success')\n",
     "        button.on_click(lambda event: setattr(self, 'ready', True))\n",
-    "        return pn.Column(\n",
-    "            pn.Row(self.param.value1, self.param.operator, self.param.value2),\n",
-    "            button\n",
-    "        )\n",
+    "        widgets = pn.Row(self.param.value1, self.param.operator, self.param.value2)\n",
+    "        for w in widgets:\n",
+    "            w.width = 85\n",
+    "        return pn.Column(widgets, button)\n",
     "\n",
     "class AutoMultiply(Multiply):\n",
     "    \n",
@@ -412,7 +412,7 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "Now that we have declared these stages let us set up the pipeline ensuring that we declare the `ready_parameter` and `auto_advance` settings:"
+    "Now that we have declared these stages let us set up the pipeline ensuring that we declare the `ready_parameter`, `next_parameter` and `auto_advance` settings appropriately:"
    ]
   },
   {
@@ -423,7 +423,7 @@
    "source": [
     "dag = pn.pipeline.Pipeline() # could set ready_parameter='ready' and auto_advance=True globally\n",
     "\n",
-    "dag.add_stage('Input', AutoInput, ready_parameter='ready', auto_advance=True)\n",
+    "dag.add_stage('Input', AutoInput, ready_parameter='ready', auto_advance=True, next_parameter='operator')\n",
     "dag.add_stage('Multiply', AutoMultiply, ready_parameter='ready', auto_advance=True)\n",
     "dag.add_stage('Add', AutoAdd, ready_parameter='ready', auto_advance=True)\n",
     "dag.add_stage('Result', Result)\n",
@@ -437,7 +437,7 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "Finally we display the pipeline, without the buttons because we don't need them:"
+    "Finally we display the pipeline, without the buttons because all the flow control is now handled from within the stages:"
    ]
   },
   {

From d544e7b7a2c01b9f8d0bfad3bd9f1d29e1a1aea3 Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Fri, 25 Oct 2019 14:30:00 -0500
Subject: [PATCH 12/19] Minor fixes

---
 panel/pipeline.py | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/panel/pipeline.py b/panel/pipeline.py
index 91a5eabdd5..8cd2dd61f2 100644
--- a/panel/pipeline.py
+++ b/panel/pipeline.py
@@ -303,11 +303,10 @@ def _update_button(self):
         options = list(self._graph.get(self._stage, []))
         next_param = kwargs.get('next_parameter', self.next_parameter)
         option = getattr(self._state, next_param) if next_param and next_param in stage.param else None
-        print(next_param, option)
         if option is None:
-            option = options[0]
+            option = options[0] if options else None
         self.next_selector.options = options
-        self.next_selector.value = option if options else None
+        self.next_selector.value = option
         self.next_selector.disabled = not bool(options)
         previous = []
         for src, tgts in self._graph.items():
@@ -544,7 +543,7 @@ def init(self):
         """
         Initialize the Pipeline before first display.
         """
-        if self._linear or not self._graph:
+        if not self._graph:
             self.define_graph(self._graph)
         else:
             self._update_progress()

From bde9bf39b985d796ae2955009896eb61d4ca0a7c Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Fri, 25 Oct 2019 14:30:22 -0500
Subject: [PATCH 13/19] Fixed Pipeline tests

---
 panel/tests/test_pipeline.py | 62 ++++++++++++++++++++----------------
 1 file changed, 34 insertions(+), 28 deletions(-)

diff --git a/panel/tests/test_pipeline.py b/panel/tests/test_pipeline.py
index c6b16d8b60..d11071ba39 100644
--- a/panel/tests/test_pipeline.py
+++ b/panel/tests/test_pipeline.py
@@ -7,6 +7,7 @@
 from panel.pane import HoloViews
 from panel.param import Param, ParamMethod
 from panel.pipeline import Pipeline
+from panel.widgets import Button
 from panel.tests.util import hv_available
 
 if LooseVersion(param.__version__) < '1.8.2':
@@ -19,6 +20,8 @@ class Stage1(param.Parameterized):
 
     b = param.Number(default=5, bounds=(0, 10))
 
+    ready = param.Boolean(default=False)
+
     @param.output(c=param.Number)
     def output(self):
         return self.a * self.b
@@ -54,20 +57,21 @@ def test_pipeline_from_classes():
 
     assert isinstance(layout, Column)
     assert isinstance(layout[0], Row)
-    error, progress, prev_button, next_button = layout[0].objects
+    (title, error), progress, (prev_button, next_button) = layout[0].objects
 
-    assert isinstance(error, Spacer)
-    assert isinstance(prev_button, Param)
-    assert isinstance(next_button, Param)
+    assert isinstance(error, Row)
+    assert isinstance(prev_button, Button)
+    assert isinstance(next_button, Button)
     assert isinstance(progress, HoloViews)
 
     hv_obj = progress.object
-    points = hv_obj.get(1)
-    assert isinstance(points, hv.Points)
-    assert len(points) == 2
-    labels = hv_obj.get(2)
+    graph = hv_obj.get(0)
+    assert isinstance(graph, hv.Graph)
+    assert len(graph) == 1
+    labels = hv_obj.get(1)
     assert isinstance(labels, hv.Labels)
-    assert list(labels['text']) == ['Stage 1', 'Stage 2']
+    print(labels)
+    assert list(labels['Stage']) == ['Stage 1', 'Stage 2']
 
     stage = layout[1][0]
     assert isinstance(stage, Row)
@@ -97,20 +101,21 @@ def test_pipeline_from_instances():
 
     assert isinstance(layout, Column)
     assert isinstance(layout[0], Row)
-    error, progress, prev_button, next_button = layout[0].objects
+    (title, error), progress, (prev_button, next_button) = layout[0].objects
 
-    assert isinstance(error, Spacer)
-    assert isinstance(prev_button, Param)
-    assert isinstance(next_button, Param)
+    assert isinstance(error, Row)
+    assert isinstance(prev_button, Button)
+    assert isinstance(next_button, Button)
     assert isinstance(progress, HoloViews)
 
     hv_obj = progress.object
-    points = hv_obj.get(1)
-    assert isinstance(points, hv.Points)
-    assert len(points) == 2
-    labels = hv_obj.get(2)
+    graph = hv_obj.get(0)
+    assert isinstance(graph, hv.Graph)
+    assert len(graph) == 1
+    labels = hv_obj.get(1)
     assert isinstance(labels, hv.Labels)
-    assert list(labels['text']) == ['Stage 1', 'Stage 2']
+    print(labels)
+    assert list(labels['Stage']) == ['Stage 1', 'Stage 2']
 
     stage = layout[1][0]
     assert isinstance(stage, Row)
@@ -142,20 +147,21 @@ def test_pipeline_from_add_stages():
 
     assert isinstance(layout, Column)
     assert isinstance(layout[0], Row)
-    error, progress, prev_button, next_button = layout[0].objects
+    (title, error), progress, (prev_button, next_button) = layout[0].objects
 
-    assert isinstance(error, Spacer)
-    assert isinstance(prev_button, Param)
-    assert isinstance(next_button, Param)
+    assert isinstance(error, Row)
+    assert isinstance(prev_button, Button)
+    assert isinstance(next_button, Button)
     assert isinstance(progress, HoloViews)
 
     hv_obj = progress.object
-    points = hv_obj.get(1)
-    assert isinstance(points, hv.Points)
-    assert len(points) == 2
-    labels = hv_obj.get(2)
+    graph = hv_obj.get(0)
+    assert isinstance(graph, hv.Graph)
+    assert len(graph) == 1
+    labels = hv_obj.get(1)
     assert isinstance(labels, hv.Labels)
-    assert list(labels['text']) == ['Stage 1', 'Stage 2']
+    print(labels)
+    assert list(labels['Stage']) == ['Stage 1', 'Stage 2']
 
     stage = layout[1][0]
     assert isinstance(stage, Row)
@@ -194,7 +200,7 @@ def test_pipeline_add_stage_validate_add_twice():
 def test_pipeline_getitem():
     pipeline = Pipeline()
     pipeline.add_stage('Stage 1', Stage1)
-    assert pipeline[0] == Stage1
+    assert pipeline['Stage 1'] == Stage1
 
 
 @hv_available

From 267043a1c1216328e0518f04a0e2fc101bd0ef5c Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Fri, 25 Oct 2019 14:31:04 -0500
Subject: [PATCH 14/19] Removed stray print

---
 panel/pipeline.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/panel/pipeline.py b/panel/pipeline.py
index 8cd2dd61f2..ae6f251aa4 100644
--- a/panel/pipeline.py
+++ b/panel/pipeline.py
@@ -280,7 +280,6 @@ def _init_stage(self):
             self._state.param.watch(self._unblock, ready_param, onlychanged=False)
 
         next_param = stage_kwargs.get('next_parameter', self.next_parameter)
-        print(next_param)
         if next_param and next_param in stage.param:
             self._state.param.watch(self._select_next, next_param, onlychanged=False)
 

From 0cb4993efc5d67be2358d228a3a2a29d58ac2353 Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Fri, 25 Oct 2019 14:31:27 -0500
Subject: [PATCH 15/19] Removed flake

---
 panel/pipeline.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/panel/pipeline.py b/panel/pipeline.py
index ae6f251aa4..2732e3ae13 100644
--- a/panel/pipeline.py
+++ b/panel/pipeline.py
@@ -7,7 +7,6 @@
 from collections import OrderedDict, defaultdict
 
 import param
-import numpy as np
 
 from .layout import Row, Column, HSpacer, VSpacer, Spacer
 from .pane import HoloViews, Pane, Markdown

From 60e369ae7c6095ff316b7f73376362680cc3d26e Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Fri, 25 Oct 2019 15:29:40 -0500
Subject: [PATCH 16/19] Fixed error condition bug

---
 panel/pipeline.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/panel/pipeline.py b/panel/pipeline.py
index 2732e3ae13..35e972dc62 100644
--- a/panel/pipeline.py
+++ b/panel/pipeline.py
@@ -359,7 +359,7 @@ def _next(self):
                 raise e
             return e
         else:
-            self.error = [Spacer(width=100)]
+            self.error[:] = []
             self._error = None
             self._update_button()
             self._route.append(self._stage)
@@ -390,7 +390,7 @@ def _previous(self):
             if self.debug:
                 raise e
         else:
-            self.error[:] = [Spacer(width=100)]
+            self.error[:] = []
             self._error = None
             self._update_button()
             self._route.pop()

From 19e8ba31af15391ba5a9ef05b0c28e71d826a0d0 Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Fri, 25 Oct 2019 15:44:08 -0500
Subject: [PATCH 17/19] Added new tests

---
 panel/tests/test_pipeline.py | 224 ++++++++++++++++++++++++++++++++---
 1 file changed, 210 insertions(+), 14 deletions(-)

diff --git a/panel/tests/test_pipeline.py b/panel/tests/test_pipeline.py
index d11071ba39..e1f1b2e59d 100644
--- a/panel/tests/test_pipeline.py
+++ b/panel/tests/test_pipeline.py
@@ -7,12 +7,16 @@
 from panel.pane import HoloViews
 from panel.param import Param, ParamMethod
 from panel.pipeline import Pipeline
-from panel.widgets import Button
-from panel.tests.util import hv_available
+from panel.widgets import Button, Select
 
 if LooseVersion(param.__version__) < '1.8.2':
     pytest.skip("skipping if param version < 1.8.2", allow_module_level=True)
 
+try:
+    import holoviews as hv
+except:
+    pytest.skip('Pipeline requires HoloViews.')
+
 
 class Stage1(param.Parameterized):
 
@@ -22,6 +26,8 @@ class Stage1(param.Parameterized):
 
     ready = param.Boolean(default=False)
 
+    next = param.String(default=None)
+
     @param.output(c=param.Number)
     def output(self):
         return self.a * self.b
@@ -47,10 +53,28 @@ def view(self):
     def panel(self):
         return Row(self.param, self.view)
 
-@hv_available
-def test_pipeline_from_classes():
-    import holoviews as hv
 
+class Stage2b(param.Parameterized):
+
+    c = param.Number(default=5, precedence=-1, bounds=(0, None))
+
+    root = param.Parameter(default=0.1)
+
+    @param.depends('c', 'root')
+    def view(self):
+        return '%s^-%s=%.3f' % (self.c, self.root, self.c**(-self.root))
+
+    def panel(self):
+        return Row(self.param, self.view)
+
+
+class DummyStage(param.Parameterized):
+
+    def panel(self):
+        return 'foo'
+
+
+def test_pipeline_from_classes():
     pipeline = Pipeline([('Stage 1', Stage1), ('Stage 2', Stage2)])
 
     layout = pipeline.layout
@@ -91,10 +115,7 @@ def test_pipeline_from_classes():
     assert stage[1].object() == '5 * 5 = 25'
 
 
-@hv_available
 def test_pipeline_from_instances():
-    import holoviews as hv
-
     pipeline = Pipeline([('Stage 1', Stage1()), ('Stage 2', Stage2())])
 
     layout = pipeline.layout
@@ -135,9 +156,7 @@ def test_pipeline_from_instances():
     assert stage[1].object() == '5 * 5 = 25'
 
 
-@hv_available
 def test_pipeline_from_add_stages():
-    import holoviews as hv
 
     pipeline = Pipeline()
     pipeline.add_stage('Stage 1', Stage1)
@@ -181,14 +200,193 @@ def test_pipeline_from_add_stages():
     assert stage[1].object() == '5 * 5 = 25'
 
 
-@hv_available
+def test_pipeline_define_graph_missing_node():
+    pipeline = Pipeline()
+    pipeline.add_stage('Stage 1', Stage1)
+    pipeline.add_stage('Stage 2', Stage2)
+
+    with pytest.raises(ValueError):
+        pipeline.define_graph({'Stage 1': ('Stage 2', 'Stage 2b')})
+
+
+def test_pipeline_define_graph():
+    pipeline = Pipeline()
+    pipeline.add_stage('Stage 2', Stage2)
+    pipeline.add_stage('Stage 2b', Stage2b)
+    pipeline.add_stage('Stage 1', Stage1)
+
+    pipeline.define_graph({'Stage 1': ('Stage 2', 'Stage 2b')})
+
+    pipeline.init()
+
+    assert pipeline._stage == 'Stage 1'
+
+    assert isinstance(pipeline.buttons, Row)
+    (pselect, pbutton), (nselect, nbutton) = pipeline.buttons
+    assert isinstance(pselect, Select)
+    assert pselect.disabled
+    assert isinstance(pbutton, Button)
+    assert pbutton.disabled
+
+    assert isinstance(nselect, Select)
+    assert not nselect.disabled
+    assert nselect.options == ['Stage 2', 'Stage 2b']
+    assert nselect.value == 'Stage 2'
+    assert isinstance(nbutton, Button)
+    assert not nbutton.disabled
+
+    pipeline._next()
+
+    assert isinstance(pipeline._state, Stage2)
+
+    pipeline._previous()
+
+    nselect.value = 'Stage 2b'
+
+    pipeline._next()
+
+    assert isinstance(pipeline._state, Stage2b)
+
+
+def test_pipeline_define_next_parameter_respected():
+    pipeline = Pipeline()
+    pipeline.add_stage('Stage 2', Stage2)
+    pipeline.add_stage('Stage 2b', Stage2b)
+    pipeline.add_stage('Stage 1', Stage1(next='Stage 2b'), next_parameter='next')
+
+    pipeline.define_graph({'Stage 1': ('Stage 2', 'Stage 2b')})
+
+    pipeline.init()
+
+    assert pipeline.next_selector.value == 'Stage 2b'
+
+    pipeline._state.next = 'Stage 2'
+
+    assert pipeline.next_selector.value == 'Stage 2'
+
+
+def test_pipeline_error_condition():
+    pipeline = Pipeline()
+    stage2b = Stage2b(root='error')
+    pipeline.add_stage('Stage 2', Stage2)
+    pipeline.add_stage('Stage 2b', stage2b)
+    pipeline.add_stage('Stage 1', Stage1)
+
+    pipeline.define_graph({'Stage 1': ('Stage 2', 'Stage 2b')})
+
+    pipeline.init()
+
+    pipeline.next_selector.value = 'Stage 2b'
+    pipeline._next()
+
+    assert isinstance(pipeline.error[0], Button)
+    
+    stage2b.root = 2
+
+    pipeline._next()
+
+    assert len(pipeline.error) == 0
+
+    
+def test_pipeline_previous_follows_initial_path():
+    pipeline = Pipeline()
+    pipeline.add_stage('Stage 1', Stage1)
+    pipeline.add_stage('Stage 2', Stage2)
+    pipeline.add_stage('Stage 2b', Stage2b)
+    pipeline.add_stage('Stage 3', DummyStage)
+
+    pipeline.define_graph({
+        'Stage 1': ('Stage 2', 'Stage 2b'),
+        'Stage 2': 'Stage 3',
+        'Stage 2b': 'Stage 3'
+    })
+
+    pipeline.init()
+
+    assert pipeline._route == ['Stage 1']
+
+    pipeline.next_selector.value = 'Stage 2b'
+    pipeline._next()
+
+    assert pipeline._route == ['Stage 1', 'Stage 2b']
+    
+    pipeline._next()
+
+    assert pipeline._route == ['Stage 1', 'Stage 2b', 'Stage 3']
+    
+    pipeline._previous()
+
+    assert pipeline._stage == 'Stage 2b'
+    assert pipeline._route == ['Stage 1', 'Stage 2b']
+
+    
+def test_pipeline_ready_respected():
+    pipeline = Pipeline(ready_parameter='ready')
+    pipeline.add_stage('Stage 1', Stage1)
+    pipeline.add_stage('Stage 2', Stage2)
+
+    pipeline.init()
+
+    assert pipeline.next_button.disabled
+
+    pipeline._state.ready = True
+
+    assert not pipeline.next_button.disabled
+
+
+def test_pipeline_auto_advance_respected():
+    pipeline = Pipeline(ready_parameter='ready', auto_advance=True)
+    pipeline.add_stage('Stage 1', Stage1)
+    pipeline.add_stage('Stage 2', Stage2)
+
+    pipeline.init()
+
+    assert pipeline.next_button.disabled
+
+    pipeline._state.ready = True
+
+    assert isinstance(pipeline._state, Stage2)
+
+
+def test_pipeline_network_diagram_states():
+    pipeline = Pipeline(ready_parameter='ready', auto_advance=True)
+    pipeline.add_stage('Stage 1', Stage1)
+    pipeline.add_stage('Stage 2', Stage2)
+    pipeline.add_stage('Stage 2b', Stage2b)
+
+    pipeline.define_graph({'Stage 1': ('Stage 2', 'Stage 2b')})
+
+    pipeline.init()
+
+    [s1, s2, s2b] = pipeline.network.object.nodes['State']
+
+    assert s1 == 'active'
+    assert s2 == 'inactive'
+    assert s2b == 'next'
+
+    pipeline._next()
+
+    [s1, s2, s2b] = pipeline.network.object.nodes['State']
+
+    assert s1 == 'inactive'
+    assert s2 == 'inactive'
+    assert s2b == 'active'
+
+    pipeline._previous()
+
+    [s1, s2, s2b] = pipeline.network.object.nodes['State']
+
+    assert s1 == 'active'
+    assert s2 == 'inactive'
+    assert s2b == 'next'
+
+
 def test_pipeline_add_stage_validate_wrong_type():
     pipeline = Pipeline()
     with pytest.raises(ValueError):
         pipeline.add_stage('Stage 1', 1)
 
 
-@hv_available
 def test_pipeline_add_stage_validate_add_twice():
     pipeline = Pipeline()
     pipeline.add_stage('Stage 1', Stage1)
@@ -196,14 +394,12 @@ def test_pipeline_add_stage_validate_add_twice():
         pipeline.add_stage('Stage 1', Stage1)
 
 
-@hv_available
 def test_pipeline_getitem():
     pipeline = Pipeline()
     pipeline.add_stage('Stage 1', Stage1)
     assert pipeline['Stage 1'] == Stage1
 
 
-@hv_available
 def test_pipeline_repr():
     pipeline = Pipeline()
     pipeline.add_stage('Stage 1', Stage1)

From 56790ed5a1848b578a52a6228cf8ef01466670dd Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Fri, 25 Oct 2019 15:45:04 -0500
Subject: [PATCH 18/19] Fixed flakes

---
 panel/tests/test_pipeline.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/panel/tests/test_pipeline.py b/panel/tests/test_pipeline.py
index e1f1b2e59d..97c635b86c 100644
--- a/panel/tests/test_pipeline.py
+++ b/panel/tests/test_pipeline.py
@@ -3,9 +3,9 @@
 import pytest
 import param
 
-from panel.layout import Row, Column, Spacer
+from panel.layout import Row, Column
 from panel.pane import HoloViews
-from panel.param import Param, ParamMethod
+from panel.param import ParamMethod
 from panel.pipeline import Pipeline
 from panel.widgets import Button, Select
 

From 208686402fa5a4d6b1e7a06dd183701791a97bfd Mon Sep 17 00:00:00 2001
From: Philipp Rudiger <prudiger@anaconda.com>
Date: Fri, 25 Oct 2019 15:56:58 -0500
Subject: [PATCH 19/19] Removed flake

---
 panel/pipeline.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/panel/pipeline.py b/panel/pipeline.py
index 35e972dc62..ce8d2e8a44 100644
--- a/panel/pipeline.py
+++ b/panel/pipeline.py
@@ -8,7 +8,7 @@
 
 import param
 
-from .layout import Row, Column, HSpacer, VSpacer, Spacer
+from .layout import Row, Column, HSpacer, VSpacer
 from .pane import HoloViews, Pane, Markdown
 from .widgets import Button, Select
 from .param import Param