Simple Broker Demo.ipynb

{
 "metadata": {
  "name": "",
  "signature": "sha256:2cc055adb0cc95c2e8107c72c02967d3446ac322eb2653f5a87e00650eba8111"
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "# Simple Data Broker Demo\n",
      "\n",
      "Here, we demonstrate that a simple implementation of the data broker can fetch and combine data from all three sources. It has only one interface, a `search` function.\n",
      "\n",
      "Note the example in the docstring below, illustrating the format of the returned data."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from databroker import broker"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 1
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "help(broker.search)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "Help on function search in module databroker.broker.simple_broker:\n",
        "\n",
        "search(beamline_id, start_time, end_time, ca_host, channels=None)\n",
        "    Get data from all events from a given beamline between two times.\n",
        "    \n",
        "    Parameters\n",
        "    ----------\n",
        "    beamline_id : string\n",
        "        e.g., 'srx'\n",
        "    start_time : string or datetime object\n",
        "        e.g., datetime.datetime(2015, 1, 1) or '2015-01-01' (ISO format)\n",
        "    end_time : string or datetime object\n",
        "        e.g., datetime.datetime(2015, 1, 1) or '2015-01-01' (ISO format)\n",
        "    ca_host : URL string\n",
        "        the URL of your archiver's ArchiveDataServer.cgi. For example,\n",
        "        'http://cr01arc01/cgi-bin/ArchiveDataServer.cgi'\n",
        "    channels : list, optional\n",
        "        All queries will return applicable data from the N most popular\n",
        "        channels. If data from additional channels is needed, their full\n",
        "        identifiers (not human-readable names) must be given here as a list\n",
        "        of strings.\n",
        "    \n",
        "    Returns\n",
        "    -------\n",
        "    data : list\n",
        "        See example below illustrating the format of the returned dataset.\n",
        "    \n",
        "    Example\n",
        "    -------\n",
        "    >>> search('srx', '2015-01-01', '2015-01-02')\n",
        "    [(<unix epoch time>, {'chan1': <value>, 'chan2': <value>},\n",
        "     (<unix epoch time>, {'temp': <value>)}\n",
        "    \n",
        "    That is, it results a list of tuples, where each tuple contains a time and\n",
        "    a dictionary of name/value pairs. Every value is guaranteed to be either a\n",
        "    scalar Python primitive (int, float, string) or a numpy ndarray.\n",
        "\n"
       ]
      }
     ],
     "prompt_number": 2
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "## Sidebar: \"Dummy\" Data Sources for Demos, Testing, and Development\n",
      "\n",
      "The databroker repo also contains a `sources` module that imitates the APIs of the three sources. It will be useful for testing and development, and it serves as a stopgap until the new MDS format is fully specified and implemented.\n",
      "\n",
      "You can switch between using \"live\" and \"dummy\" versions of any source at runtime."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from __future__ import print_function\n",
      "from datetime import datetime as dt\n",
      "from datetime import timedelta as tdelta\n",
      "import uuid\n",
      "\n",
      "from databroker import sources\n",
      "from databroker.tests.test_broker import generate_ca_data\n",
      "from fileStore.test.t_utils import SynHandlerMod, SynHandlerEcho\n",
      "from fileStore.retrieve import register_handler\n",
      "\n",
      "\n",
      "# Switch to dummy versions of all sources. The API is unchanged.\n",
      "sources.switch(metadatastore=False, filestore=True, channelarchiver=False)\n",
      "\n",
      "# aliases for brevity\n",
      "mds = sources.metadataStore\n",
      "fs = sources.fileStore\n",
      "ca = sources.channelarchiver"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 8
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Make sure everything is clean.\n",
      "import databroker.sources.dummy_sources._metadataStore.api._dummies\n",
      "databroker.sources.dummy_sources._metadataStore.api._dummies._events.clear()\n",
      "databroker.sources.dummy_sources._metadataStore.api._dummies._ev_desc.clear()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 9
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "### Register two file handlers, handling different file types\n",
      "\n",
      "Here, we use two different handles to show that FileStore can deal with data stored in different formats."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "register_handler('syn-mod', SynHandlerMod)\n",
      "register_handler('syn-echo', SynHandlerEcho)\n",
      "\n",
      "file_types = ['syn-mod', 'syn-echo']  # will go in File Store\n",
      "scalar_names = ['mod_base', 'echo_base']  # will go in Metadata Store"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 10
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "### Load scalar data into (dummy) Metadata Store and array data into (the actual) File Store.\n",
      "\n",
      "We're doing something cute here to make a nontrivial example.\n",
      "\n",
      "In this example, each event has a scalar measurement and an array measurement. The scalar data labeled \"echo_base\" is accompanied by an array that is simply filled with that same scalar value. The scalar data labeled \"mod_base\" is accompanied by an array that contains linear ramp mod {mod_base}. Below, we can see that everything is read and returned correctly."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "start, end = dt.now(), dt.now() + tdelta(minutes=5)\n",
      "\n",
      "N = 2\n",
      "M = 4\n",
      "shape = (3, 4)\n",
      "\n",
      "for n in range(N):\n",
      "    # Each loop creates an event descriptor.\n",
      "    ev_desc = mds.api.collection.save_event_descriptor(\n",
      "        data_keys={scalar_names[n%len(scalar_names)]: {'source': 'PV:foo',},\n",
      "                   'img': {'source': 'CCD:syn',\n",
      "                           'external': 'FILESTORE'}})\n",
      "    # Make file descriptor entry.\n",
      "    fb = fs.commands.save_file_base(file_types[n%len(file_types)], 'filepath',\n",
      "                           {'shape': shape})\n",
      "    for m in range(M):\n",
      "        # Each loop creates an event.\n",
      "        a_val = n * M + m + 1\n",
      "        r_id = str(uuid.uuid4())\n",
      "        \n",
      "        # Create file link document.\n",
      "        fs.commands.save_file_event_link(fb, r_id, {'n': a_val})\n",
      "        \n",
      "        # Simulate \"data collection\"\n",
      "        data_dict = {scalar_names[n%len(scalar_names)]: {'value': a_val, 'timestamp': dt.now()},\n",
      "                     'img': {'value': r_id, 'timestamp': dt.now()}}\n",
      "        \n",
      "        # Save Metadata Store record.\n",
      "        event = mds.api.collection.save_event(event_descriptor=ev_desc, data=data_dict)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 12
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "### View the event records we just created\n",
      "\n",
      "Here are the documents stored in the Metadata Store. The user would not interact directly with the Metadata Store like this. This is only for illustrative purposes."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "raw_events = mds.api.analysis.find()\n",
      "  \n",
      "for re in raw_events:\n",
      "    print(\"event_id: {}\\nevent_descid: {}\\ndata: {}\\n\".format(re.uid, re.ev_desc.uid, repr(re.data)))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "event_id: ad4daf0f-8b83-40b1-a2ee-d9bef182048a\n",
        "event_descid: ed4fa773-e6bf-424a-ab42-57644fd0f0be\n",
        "data: {'mod_base': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 593240), 'value': 1}, 'img': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 593251), 'value': '726e4f9e-fc8e-4cff-883c-2049d23a0cfa'}}\n",
        "\n",
        "event_id: 286d8e85-121d-4b10-86da-9cc18dcc078f\n",
        "event_descid: ed4fa773-e6bf-424a-ab42-57644fd0f0be\n",
        "data: {'mod_base': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 594781), 'value': 2}, 'img': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 594793), 'value': '52c08bda-716b-4bc0-8cfb-57502f834991'}}\n",
        "\n",
        "event_id: 5934e02f-b5f7-4489-b77a-8777934dc5d8\n",
        "event_descid: ed4fa773-e6bf-424a-ab42-57644fd0f0be\n",
        "data: {'mod_base': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 595938), 'value': 3}, 'img': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 595949), 'value': '9c1da657-4db5-41ca-9188-010676f5e753'}}\n",
        "\n",
        "event_id: d17da047-e342-4c96-93df-9eaa53c28f11\n",
        "event_descid: ed4fa773-e6bf-424a-ab42-57644fd0f0be\n",
        "data: {'mod_base': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 597004), 'value': 4}, 'img': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 597011), 'value': '9ca3882d-46ce-4f60-9484-6985b07b20c3'}}\n",
        "\n",
        "event_id: 5ebc4ddb-4a85-4b75-aa38-cd824ba2c6d9\n",
        "event_descid: bf5253a7-cdc8-4419-8c44-3a0b28ea8213\n",
        "data: {'echo_base': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 598903), 'value': 5}, 'img': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 598911), 'value': '3e7f4400-7e7c-4414-bbac-7ea3776a4796'}}\n",
        "\n",
        "event_id: 1001416b-1f3f-460c-921b-fd999ba59b27\n",
        "event_descid: bf5253a7-cdc8-4419-8c44-3a0b28ea8213\n",
        "data: {'echo_base': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 599756), 'value': 6}, 'img': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 599763), 'value': '6f955425-999d-4b72-a30d-2bdbd25fdf58'}}\n",
        "\n",
        "event_id: 64a2e2a0-0d3d-4612-b89a-046c8395a713\n",
        "event_descid: bf5253a7-cdc8-4419-8c44-3a0b28ea8213\n",
        "data: {'echo_base': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 600676), 'value': 7}, 'img': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 600682), 'value': 'bd3d5957-001f-448e-8af5-3b4e44a5709e'}}\n",
        "\n",
        "event_id: 478ccc01-e806-46ea-ad3d-7e69fa93be81\n",
        "event_descid: bf5253a7-cdc8-4419-8c44-3a0b28ea8213\n",
        "data: {'echo_base': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 601737), 'value': 8}, 'img': {'timestamp': datetime.datetime(2015, 1, 26, 13, 48, 26, 601745), 'value': '9b90fead-ca7a-4aab-b654-32dfe8f3f770'}}\n",
        "\n"
       ]
      }
     ],
     "prompt_number": 13
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "### Load sample data into two channels of the dummy Channel Archiver\n",
      "\n",
      "To simulate channel data, we sneak instructions in where the host URL would normally go."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "simulated_ca_data = generate_ca_data(['SR11BCM01:LIFETIME_MONITOR', 'SR11BCM01:CURRENT_MONITOR'], start, end)\n",
      "ca.insert_data(simulated_ca_data)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 14
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "## A Sample Query\n",
      "\n",
      "A basic query returns all the applicable data from the MDS and FS along with data from the most commonly used CA channels."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "broker.search('srx', start, end, ca_host='whatever')"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 15,
       "text": [
        "[(datetime.datetime(2015, 1, 26, 13, 48, 26, 593347),\n",
        "  {'img': array([[0, 0, 0, 0],\n",
        "          [0, 0, 0, 0],\n",
        "          [0, 0, 0, 0]]), 'mod_base': 1}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 48, 26, 594868),\n",
        "  {'img': array([[0, 1, 0, 1],\n",
        "          [0, 1, 0, 1],\n",
        "          [0, 1, 0, 1]]), 'mod_base': 2}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 48, 26, 596018),\n",
        "  {'img': array([[0, 1, 2, 0],\n",
        "          [1, 2, 0, 1],\n",
        "          [2, 0, 1, 2]]), 'mod_base': 3}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 48, 26, 597068),\n",
        "  {'img': array([[0, 1, 2, 3],\n",
        "          [0, 1, 2, 3],\n",
        "          [0, 1, 2, 3]]), 'mod_base': 4}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 48, 26, 598962),\n",
        "  {'echo_base': 5, 'img': array([[ 5.,  5.,  5.,  5.],\n",
        "          [ 5.,  5.,  5.,  5.],\n",
        "          [ 5.,  5.,  5.,  5.]])}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 48, 26, 599821),\n",
        "  {'echo_base': 6, 'img': array([[ 6.,  6.,  6.,  6.],\n",
        "          [ 6.,  6.,  6.,  6.],\n",
        "          [ 6.,  6.,  6.,  6.]])}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 48, 26, 600734),\n",
        "  {'echo_base': 7, 'img': array([[ 7.,  7.,  7.,  7.],\n",
        "          [ 7.,  7.,  7.,  7.],\n",
        "          [ 7.,  7.,  7.,  7.]])}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 48, 26, 601805),\n",
        "  {'echo_base': 8, 'img': array([[ 8.,  8.,  8.,  8.],\n",
        "          [ 8.,  8.,  8.,  8.],\n",
        "          [ 8.,  8.,  8.,  8.]])}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 48, 26, 578003),\n",
        "  {'SR11BCM01:LIFETIME_MONITOR': 0}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 49, 26, 578003),\n",
        "  {'SR11BCM01:LIFETIME_MONITOR': 1}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 50, 26, 578003),\n",
        "  {'SR11BCM01:LIFETIME_MONITOR': 2}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 51, 26, 578003),\n",
        "  {'SR11BCM01:LIFETIME_MONITOR': 3}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 52, 26, 578003),\n",
        "  {'SR11BCM01:LIFETIME_MONITOR': 4}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 53, 26, 578003),\n",
        "  {'SR11BCM01:LIFETIME_MONITOR': 5}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 48, 26, 578003),\n",
        "  {'SR11BCM01:CURRENT_MONITOR': 0}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 49, 26, 578003),\n",
        "  {'SR11BCM01:CURRENT_MONITOR': 1}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 50, 26, 578003),\n",
        "  {'SR11BCM01:CURRENT_MONITOR': 2}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 51, 26, 578003),\n",
        "  {'SR11BCM01:CURRENT_MONITOR': 3}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 52, 26, 578003),\n",
        "  {'SR11BCM01:CURRENT_MONITOR': 4}),\n",
        " (datetime.datetime(2015, 1, 26, 13, 53, 26, 578003),\n",
        "  {'SR11BCM01:CURRENT_MONITOR': 5})]"
       ]
      }
     ],
     "prompt_number": 15
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "## What about the muggler?\n",
      "\n",
      "All communication with the three data sources takes place through the broker. (Fuller implementations of the broker will manage various optimizations in data retrieval. All of that must be isolated from higher layers.) The muggler can make requests for additional data, such as adding a source from the Channel Archiver, but it will do so through the broker."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "!gist \"Simple Broker Demo.ipynb\""
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "https://gist.github.com/ca331f830dcab966b9f4\r\n"
       ]
      }
     ],
     "prompt_number": 16
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [],
     "language": "python",
     "metadata": {},
     "outputs": []
    }
   ],
   "metadata": {}
  }
 ]
}