Function plot

python/pathway/stdlib/viz/plotting.py:34–139

Allows for plotting contents of the table visually in e.g. jupyter. If the table depends only on the bounded data sources, the plot will be generated right away. Otherwise (in streaming scenario), the plot will be auto-updating after running pw.run() Args: self (pw.Table):

(
    self: pw.Table,
    plotting_function,
    sorting_col=None,
)

@check_arg_types
@trace_user_frame
def plot(
    self: pw.Table,
    plotting_function,
    sorting_col=None,
):
    """
    Allows for plotting contents of the table visually in e.g. jupyter. If the table
    depends only on the bounded data sources, the plot will be generated right away.
    Otherwise (in streaming scenario), the plot will be auto-updating after running pw.run()

    Args:
        self (pw.Table): a table serving as a source of data
        plotting_function (Callable[[ColumnDataSource], Plot]): function for creating plot
            from ColumnDataSource

    Returns:
        pn.Column: visualization which can be displayed immediately or passed as a dashboard widget

    Example:

    >>> import pathway as pw
    >>> from bokeh.plotting import figure
    >>> def func(source):
    ...     plot = figure(height=400, width=400, title="CPU usage over time")
    ...     plot.scatter('a', 'b', source=source, line_width=3, line_alpha=0.6)
    ...     return plot
    >>> viz = pw.debug.table_from_pandas(pd.DataFrame({"a":[1,2,3],"b":[3,1,2]})).plot(func)  # doctest: +SKIP
    >>> type(viz)  # doctest: +SKIP
    <class 'panel.layout.base.Column'>
    """
    import panel as pn
    from bokeh.models import ColumnDataSource

    col_names = self.schema.column_names()

    gr = GraphRunner(parse_graph.G, debug=False, monitoring_level=MonitoringLevel.NONE)
    bounded = gr.has_bounded_input(self)

    source = ColumnDataSource(data={colname: [] for colname in col_names})

    plot = plotting_function(source)
    viz = pn.Column(
        pn.Row(
            "Static preview" if bounded else "Streaming mode",
            pn.widgets.TooltipIcon(
                value=(
                    "Immediate table preview is possible as the table depends only on static inputs"
                    if bounded
                    else "Table depends on streaming inputs. Please run pw.run()"
                )
            ),
        ),
        plot,
    )

    if bounded:
        [captured] = gr.run_tables(self)
        output_data = api.squash_updates(captured)