00_more_data_streaming.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# bokeh in the notebook: data streaming / live plots update \n",
    "\n",
    "## About this notebook\n",
    "\n",
    "This notebook belongs to a series of small projects which aim is to evaluate the [Jupyter](http://jupyter.org/) ecosystem for science experiments control. The main idea is to use the _Juypter notebook_ as a convergence platform in order to offer a fully featured environment to scientists. \n",
    "\n",
    "## About bokeh\n",
    "\n",
    "Experiments control requires both static and dynamic (i.e live) data visualization. Since Jupyter doesn't provide any 'official' data visualization solution, we need to select one. Among the available solutions, [bokeh](http://bokeh.pydata.org/en/latest) presents the highest potential for our application.\n",
    "\n",
    "Bokeh as been selected for its:\n",
    "1. [built-in notebook integration](http://bokeh.pydata.org/en/latest/docs/user_guide/notebook.html)\n",
    "2. built-in [data streaming](http://bokeh.pydata.org/en/latest/docs/reference/models/sources.html#bokeh.models.sources.ColumnDataSource.patch) [features](http://bokeh.pydata.org/en/latest/docs/reference/models/sources.html#bokeh.models.sources.ColumnDataSource.stream) for live plots update \n",
    "3. ability to add [custom or specialized behaviors](http://bokeh.pydata.org/en/latest/docs/user_guide/interaction/callbacks.html) in response to property changes and other events\n",
    "4. [graphics quality](http://bokeh.pydata.org/en/latest/docs/gallery.html#gallery)\n",
    "\n",
    "Have a look to this [quickstart](http://bokeh.pydata.org/en/latest/docs/user_guide/quickstart.html) for a bokeh overview.\n",
    "\n",
    "## Topic of the day\n",
    "\n",
    "Asynchronous (i.e. non blocking) plots update. We arre going to use our tiny Datastream pattern whose model is the following:\n",
    "    \n",
    "<img src=\"./tango_meeting_florence_0617/resources/images/DataStreamUML.png\" width=\"650\" height=\"500\">\n",
    "\n",
    "The idea is to implement a DataSource (see XDS, YDS, ...) and attach it to a Channel, which is itself attached to a DataStream, which is itself attached to a DataStreamer, which is itself attached to an optional DataStreamerController. Note that the ScalarChannel and SprectrumChannel are multi-DataSource channels whereas the ImageChannel is a single DataSource one."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import threading\n",
    "import datetime\n",
    "import time\n",
    "import math\n",
    "import random\n",
    "import logging\n",
    "import copy\n",
    "from collections import deque\n",
    "import numpy as np\n",
    "from bokeh.models.ranges import Range1d\n",
    "from bokeh.palettes import Plasma256, Viridis256, Inferno256, Greys256\n",
    "from bokeh.layouts import column\n",
    "from common.plots import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "logging.getLogger('tornado').setLevel(logging.CRITICAL)\n",
    "logging.getLogger('bokeh.server').setLevel(logging.CRITICAL)\n",
    "logging.getLogger('fs.client.jupyter').setLevel(logging.DEBUG)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# select the context in which we are running: NOTEBOOK or LAB\n",
    "import os\n",
    "os.environ[\"JUPYTER_CONTEXT\"] = \"LAB\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#--------------------------------------------------------------------------\n",
    "class XDS(DataSource):\n",
    "    \n",
    "    def __init__(self, name, num_points=128):\n",
    "        DataSource.__init__(self, name)\n",
    "        self._l = num_points\n",
    "    \n",
    "    def pull_data(self): \n",
    "        cd = ChannelData(self.name)\n",
    "        start = random.uniform(-math.pi/2, math.pi/2)\n",
    "        end = 2 * math.pi + start\n",
    "        cd.buffer = np.linspace(start, end, self._l)\n",
    "        return cd\n",
    "        \n",
    "    def cleanup(self):\n",
    "        super(XDS, self).cleanup()\n",
    "        \n",
    "#--------------------------------------------------------------------------\n",
    "class YDS(DataSource):\n",
    "    \n",
    "    def __init__(self, name, channels=None, num_points=128):\n",
    "        DataSource.__init__(self, name)\n",
    "        self._l = num_points\n",
    "    \n",
    "    def pull_data(self): \n",
    "        cd = ChannelData(self.name)\n",
    "        p = random.uniform(-math.pi/2, math.pi/2)\n",
    "        start = 0 + p\n",
    "        end = 2 * (math.pi + p)\n",
    "        x = np.linspace(start, end, self._l)\n",
    "        d = random.uniform(1.0, 4.0) * np.sin(x)\n",
    "        cd.buffer = random.uniform(1.0, 4.0) * np.sin(x)\n",
    "        return cd\n",
    "\n",
    "    def cleanup(self):\n",
    "        super(YDS, self).cleanup() \n",
    "        \n",
    "#--------------------------------------------------------------------------\n",
    "class SCTest(SpectrumChannel):\n",
    "    \n",
    "    def __init__(self, name, data_sources=None, model_properties=None):\n",
    "        SpectrumChannel.__init__(self, name, data_sources, model_properties)\n",
    "        self._lock = threading.Lock()\n",
    "        self._selection_range = self.__initial_selection_range()\n",
    "        \n",
    "    def __initial_selection_range(self):\n",
    "        return {'x0':0, 'x1':0, 'y0':0, 'y1':0, 'width':0, 'height':0}\n",
    "    \n",
    "    def on_box_selection_change(self, selection_range):\n",
    "        with self._lock:\n",
    "            self._selection_range = selection_range\n",
    "        \n",
    "    def on_box_selection_reset(self):\n",
    "        with self._lock:\n",
    "            self._selection_range = self.__initial_selection_range()\n",
    "            \n",
    "    def cleanup(self):\n",
    "        super(SCTest, self).cleanup()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#--------------------------------------------------------------------------\n",
    "class XYDS(DataSource):\n",
    "    \n",
    "    def __init__(self, name):\n",
    "        DataSource.__init__(self, name)\n",
    "        self._inc = 1\n",
    "        self._current_index = 0\n",
    "        self._iw, self._ih = 1000, 1000\n",
    "        x, y = np.linspace(0, 10, self._iw), np.linspace(0, 10, self._ih)\n",
    "        xx, yy = np.meshgrid(x, y)\n",
    "        self._full_image = np.sin(xx) * np.cos(yy)\n",
    "\n",
    "    def pull_data(self):\n",
    "        cd = ChannelData(self.name)    \n",
    "        i = self._current_index\n",
    "        cd.buffer = self._full_image[0:i+1, 0:i+1]\n",
    "        self._current_index += self._inc\n",
    "        if self._current_index > self._ih:\n",
    "            self._current_index = self._ih\n",
    "            self._inc *= -1 \n",
    "        elif self._current_index < 0:\n",
    "            self._current_index = 0\n",
    "            self._inc *= -1\n",
    "        return cd\n",
    "\n",
    "    def cleanup(self):\n",
    "        super(XYDS, self).cleanup()\n",
    "    \n",
    "#--------------------------------------------------------------------------\n",
    "class ICTest(ImageChannel):\n",
    "    \n",
    "    def __init__(self, name, data_source=None, model_properties=None):\n",
    "        ImageChannel.__init__(self, name, data_source=data_source, model_properties=model_properties)\n",
    "        self._iw, self._ih = 1000, 1000\n",
    "        self._selection_range = self.__initial_selection_range()\n",
    "        self._bsm = None\n",
    "        self._lock = threading.Lock()\n",
    "\n",
    "    def __initial_selection_range(self):\n",
    "        return {'x0':0, 'x1':self._iw, 'y0':0, 'y1':self._ih, 'width':self._iw, 'height':self._ih}\n",
    "    \n",
    "    def setup_model(self, **kwargs):\n",
    "        if 'selection_manager' not in kwargs:\n",
    "            scb = self.on_box_selection_change\n",
    "            rcb = self.on_box_selection_reset\n",
    "            self._bsm = BoxSelectionManager(selection_callback=scb, reset_callback=rcb)\n",
    "            kwargs['selection_manager'] = self._bsm\n",
    "        return super(ICTest, self).setup_model(**kwargs)\n",
    "    \n",
    "    def on_box_selection_change(self, selection_range):\n",
    "        with self._lock:\n",
    "            self._selection_range = selection_range\n",
    "        \n",
    "    def on_box_selection_reset(self):\n",
    "        with self._lock:\n",
    "            self._selection_range = self.__initial_selection_range()\n",
    "        \n",
    "    def cleanup(self):\n",
    "        super(ICTest, self).cleanup()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def on_box_selection_change(selection_range):\n",
    "    print(\"on_box_selection_change: {}\".format(selection_range))\n",
    "\n",
    "def on_box_selection_reset():\n",
    "    print(\"on_box_selection_reset\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def layout_model_properties():\n",
    "    lmp = dict()\n",
    "    lmp['width'] = 475\n",
    "    lmp['height'] = 250\n",
    "    lmp['layout'] = 'grid'\n",
    "    lmp['merge_tools'] = True\n",
    "    lmp['show_legend'] = True\n",
    "    lmp['palette'] = Viridis256\n",
    "    scb = on_box_selection_change\n",
    "    rcb = on_box_selection_reset\n",
    "    lmp['selection_manager'] = BoxSelectionManager(selection_callback=scb, reset_callback=rcb)  \n",
    "    lmp['refresh_mode'] = 'one-by-one'\n",
    "    return lmp\n",
    "\n",
    "def image_model_properties(s=0, e=1000):\n",
    "    xshsp = dict()\n",
    "    xshsp['start'] = s\n",
    "    xshsp['end'] = e\n",
    "    xshsp['num_points'] = abs(e - s)\n",
    "    xshsp['label'] = 'x-axis'\n",
    "    xshsp['unit'] = 'mm'\n",
    "    yshsp = dict()\n",
    "    yshsp['start'] = s\n",
    "    yshsp['end'] = e\n",
    "    yshsp['num_points'] = abs(e - s)\n",
    "    yshsp['label'] = 'y-axis'\n",
    "    yshsp['unit'] = 'mm'\n",
    "    imp = dict()\n",
    "    imp['x_scale'] = Scale(**xshsp)\n",
    "    imp['y_scale'] = Scale(**yshsp)\n",
    "    return imp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# x Scale # ------------------------------------------------\n",
    "shsp = dict()\n",
    "shsp['label'] = 'angle'\n",
    "shsp['unit'] = 'rad'\n",
    "shsp['channel'] = 'x_scale'\n",
    "shsp['start'] = -1\n",
    "shsp['end'] = 1\n",
    "x_scale = Scale(**shsp)\n",
    "# y Scale # ------------------------------------------------\n",
    "spsp = dict()\n",
    "spsp['label'] = 'amplitude'\n",
    "spsp['unit'] = 'a.u.'\n",
    "y_scale = Scale(**spsp)\n",
    "# SpectrumChannel 2 ----------------------------------------\n",
    "s1 = list()\n",
    "s1.append(XDS('x_scale'))\n",
    "s1.extend([YDS(n) for n in ['y1.1', 'y1.2', 'y1.3']]) \n",
    "c1 = SCTest('c1', data_sources=s1)\n",
    "# SpectrumChannel 1 ----------------------------------------\n",
    "s2 = list()\n",
    "s2.append(XDS('x_scale'))\n",
    "s2.extend([YDS(n) for n in ['y2.1', 'y2.2', 'y2.3']]) \n",
    "c2 = SCTest('c2', data_sources=s2)\n",
    "# layout parameters  ---------------------------------------\n",
    "lmp = dict()\n",
    "lmp['width'] = 475\n",
    "lmp['height'] = 250\n",
    "lmp['layout'] = 'grid'\n",
    "lmp['merge_tools'] = True\n",
    "lmp['show_legend'] = True\n",
    "lmp['x_scale'] = x_scale\n",
    "lmp['y_scale'] = y_scale\n",
    "# LayoutChannel --------------------------------------------\n",
    "l1 = LayoutChannel('l1', channels=[c1, c2], model_properties=lmp)\n",
    "# ImageChannel 1 -------------------------------------------\n",
    "ic0 = ICTest(\"i0\", data_source=XYDS(\"is0\"), model_properties=image_model_properties(0, 1000))\n",
    "ic1 = ICTest(\"i1\", data_source=XYDS(\"is1\"), model_properties=image_model_properties(0, 1000))\n",
    "# LayoutChannel --------------------------------------------\n",
    "l2 = LayoutChannel('l2', channels=[ic0, ic1], model_properties=layout_model_properties())\n",
    "# DataStream -----------------------------------------------\n",
    "s1 = DataStream('s1', channels=[l1, l2])\n",
    "# DataStreamer ---------------------------------------------\n",
    "m1 = DataStreamer('m1', data_streams=[s1], update_period=1.)\n",
    "# DataStreamerController -----------------------------------\n",
    "c1 = DataStreamerController('c1', m1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "jupyter": {
     "source_hidden": true
    }
   },
   "source": [
    "Let's try a GenericChannel : unknown channel type till the first call to pull_data returns a SCALAR, a SPECTRUM or an IMAGE."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class RingBuffer(np.ndarray):\n",
    "    \"\"\"\n",
    "    a multidimensional ring buffer\n",
    "    see https://gist.github.com/mtambos/aa435461084b5c0025d1\n",
    "    \"\"\"\n",
    "\n",
    "    def __new__(cls, input_array):\n",
    "        obj = np.asarray(input_array).view(cls)\n",
    "        return obj\n",
    "\n",
    "    def __array_finalize__(self, obj):\n",
    "        if obj is None: return\n",
    "\n",
    "    def __array_wrap__(self, out_arr, context=None):\n",
    "        return np.ndarray.__array_wrap__(self, out_arr, context)\n",
    "\n",
    "    def append(self, x):\n",
    "        \"\"\"adds element x to the ring buffer\"\"\"\n",
    "        self[:-1] = self[1:]\n",
    "        self[-1] = x"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class TSDS(DataSource):\n",
    "    \n",
    "    def __init__(self, name, num_points=128):\n",
    "        DataSource.__init__(self, name)\n",
    "        array = np.empty((num_points,))\n",
    "        array.fill(np.nan)\n",
    "        self._data_buffer = RingBuffer(array)\n",
    "        array = np.empty((num_points,), dtype=float)\n",
    "        array.fill(np.nan)\n",
    "        self._time_buffer = RingBuffer(array)\n",
    "        self._cnt = -1\n",
    "    \n",
    "    def pull_data(self): \n",
    "        try:\n",
    "            self._cnt += 1\n",
    "            y = self._cnt % 16\n",
    "            val = ChannelData()\n",
    "            self._time_buffer.append( time.time() * 1000. )\n",
    "            if 4 <= self._cnt <= 8:\n",
    "                self._data_buffer.append(math.nan)\n",
    "                val.set_error(err='Testing error handling...', exc=None)\n",
    "            else:\n",
    "                self._data_buffer.append(y)\n",
    "            val.set_data(self._data_buffer, self._time_buffer, ChannelData.Format.SCALAR)\n",
    "            self._cnt = y\n",
    "        except Exception as e:\n",
    "            print(e)\n",
    "        return val\n",
    "    \n",
    "    def cleanup(self):\n",
    "        super(TSDS, self).cleanup()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sch = GenericChannel('sch', data_source=TSDS('tsds'))\n",
    "scd = DataStream('scd', channels=[sch])\n",
    "scm = DataStreamer('scm', data_streams=[scd], update_period=1.)\n",
    "scr = DataStreamerController('scr', scm)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.9"
  },
  "widgets": {
   "state": {
    "18252868359644188c5d35181279de02": {
     "views": [
      {
       "cell_index": 7
      }
     ]
    },
    "1c757d72e00c446aad9eb921446ecfac": {
     "views": [
      {
       "cell_index": 8
      }
     ]
    },
    "efc06b4082064aa8b99bb9b7ddb8538b": {
     "views": [
      {
       "cell_index": 6
      }
     ]
    }
   },
   "version": "1.2.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}