{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import panel as pn\n",
    "\n",
    "from streamz import Stream\n",
    "\n",
    "pn.extension('vega')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The ``Streamz`` pane renders [Streamz](https://streamz.readthedocs.io/en/latest/) Stream objects emitting arbitrary objects, unlike the ``DataFrame`` pane which specifically handles streamz DataFrame and Series objects and exposes various formatting objects.\n",
    "\n",
    "<div class=\"alert alert-warning\">\n",
    "\n",
    "If you are not already a user of the Streamz library, we recommend using functionality from the Param and Panel ecosystem, such as [reactive expressions](https://param.holoviz.org/user_guide/Reactive_Expressions.html), [generator functions](https://param.holoviz.org/user_guide/Generators.html) and/ or *periodic callbacks*. We find these features to be more robustly supported.\n",
    "\n",
    "</div>\n",
    "\n",
    "#### Parameters:\n",
    "\n",
    "For details on other options for customizing the component see the [layout](../../how_to/layout/index.md) and [styling](../../how_to/styling/index.md) how-to guides.\n",
    "\n",
    "* **``always_watch``** (boolean, default=False): Whether to watch the stream even when not displayed.\n",
    "* **``object``** (streamz.Stream): The streamz.Stream object being watched\n",
    "* **``rate_limit``** (float, default=0.1): The minimum interval between events in seconds.\n",
    "\n",
    "___"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `Streamz` pane uses the default panel resolution to figure out the appropriate way to render the object returned by a Stream. By default the pane will only watch the ``Stream`` if it is displayed, we can tell it to watch the stream as soon as it is created by setting `always_watch=True`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def increment(x):\n",
    "    return x + 1\n",
    "\n",
    "source = Stream()\n",
    "\n",
    "streamz_pane = pn.pane.Streamz(source.map(increment), always_watch=True)\n",
    "\n",
    "# Note: To ensure that a static render of the stream displays anything\n",
    "#       we set always_watch=True and emit an event before displaying\n",
    "source.emit(1)\n",
    "\n",
    "streamz_pane"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can now define a periodic callback which emits an increasing count on the `Stream`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "count = 1 \n",
    "\n",
    "def emit_count():\n",
    "    global count\n",
    "    count += 1\n",
    "    source.emit(count)\n",
    "\n",
    "pn.state.add_periodic_callback(emit_count, period=100, count=9);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets reset the pane"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "source.emit(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `Streamz` stream can be used to stream any kind of data, e.g. we can create a streamz DataFrame, accumulate the data into a sliding window and then map it to a altair `line_plot` function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import altair as alt\n",
    "import pandas as pd\n",
    "from datetime import datetime\n",
    "from streamz.dataframe import DataFrame as sDataFrame\n",
    "\n",
    "df = sDataFrame(example=pd.DataFrame({'y': []}, index=pd.DatetimeIndex([])))\n",
    "\n",
    "def line_plot(data):\n",
    "    return alt.Chart(pd.concat(data).reset_index()).mark_line().encode(\n",
    "        x='index',\n",
    "        y='y',\n",
    "    ).properties(width=\"container\")\n",
    "\n",
    "altair_stream = df.cumsum().stream.sliding_window(50).map(line_plot)\n",
    "\n",
    "altair_pane = pn.pane.Streamz(altair_stream, height=350, sizing_mode=\"stretch_width\", always_watch=True)\n",
    "\n",
    "for i in range(100):\n",
    "    df.emit(pd.DataFrame({'y': [np.random.randn()]}, index=pd.DatetimeIndex([datetime.now()])))\n",
    "\n",
    "altair_pane"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can emit additional data on the DataFrame and watch the plot update:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def emit():\n",
    "    df.emit(pd.DataFrame({'y': [np.random.randn()]}, index=pd.DatetimeIndex([datetime.now()])))\n",
    "\n",
    "pn.state.add_periodic_callback(emit, period=100, count=50);"
   ]
  }
 ],
 "metadata": {
  "language_info": {
   "name": "python",
   "pygments_lexer": "ipython3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}