DataNode.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {},
   "outputs": [],
   "source": [
    "import uuid\n",
    "import msgpack\n",
    "from abc import ABC, abstractmethod\n",
    "\n",
    "class NodeFactory():\n",
    "    def __init__(self, server):\n",
    "        self.server = server\n",
    "        self.rootid = None\n",
    "        \n",
    "    def get_root(self):\n",
    "        self.rootid = msgpack.unpackb(self.server.request({'request': 'send_root'}))\n",
    "        return self.rootid\n",
    "        \n",
    "    def get_header(self, uid):\n",
    "        return msgpack.unpackb(self.server.request({'request': 'send_header', 'uid': uid}))\n",
    "        \n",
    "    def get_data(self, uid, **kwargs):\n",
    "        return msgpack.unpackb(self.server.request({'request': 'send_data', 'uid': uid, 'kwargs': kwargs}))\n",
    "        \n",
    "    def get_edges(self, uid, **kwargs):\n",
    "        return msgpack.unpackb(self.server.request({'request': 'send_edges', 'uid': uid, 'kwargs': kwargs}))\n",
    "    \n",
    "    def __call__(self, uid):\n",
    "        # I'll make this dynamic later.\n",
    "        # Can use entrypoints or importlib to discover node classes.\n",
    "        node = BlueskyNode(uid=uid,\n",
    "                          edges=self.get_edges(uid),\n",
    "                          meta=self.get_header(uid),\n",
    "                          data=(self.get_data(uid, i) for i in range(10)))\n",
    "        return node\n",
    "\n",
    "\n",
    "class BaseNode(ABC):\n",
    "    \"\"\"\n",
    "    This is an abstract base class that defines the required methods and attributes for a node in the graph.\n",
    "    This class is deigned to access data locally as well as remotely.\n",
    "    \"\"\"\n",
    "    def __init__(self, uid=None, edges=None, meta=None, data=None, server=None):\n",
    "        self.uid = uid or str(uuid.uuid4())\n",
    "        self.edges = edges or {}\n",
    "        self.info = {'module': 'Base', 'class': 'BaseNode', 'version': '0.0.1'}\n",
    "        self.meta = meta\n",
    "        self.data = data\n",
    "        if server:\n",
    "            self.factory = NodeFactory(server)\n",
    "        else:\n",
    "            self.factory = None\n",
    "        \n",
    "    @abstractmethod\n",
    "    def read(self):\n",
    "        \"\"\"\n",
    "        Convert raw data in self.data to a python object.\n",
    "        \"\"\"\n",
    "        pass\n",
    "\n",
    "    @abstractmethod\n",
    "    def load_meta(self):\n",
    "        \"\"\"\n",
    "        Load header data from the datasource.\n",
    "        \"\"\"\n",
    "        pass\n",
    "    \n",
    "    @abstractmethod\n",
    "    def load_data(self):\n",
    "        \"\"\"\n",
    "        This method loads the data from the source to self.data\n",
    "        self.data is a list of partitions, each partition must be serializable.\n",
    "        \"\"\"\n",
    "        pass\n",
    "    \n",
    "    def send_meta(self, **kwargs):\n",
    "        \"\"\"\n",
    "        Sends header data to client.\n",
    "        \"\"\"\n",
    "        self.load_meta()\n",
    "        return self.meta\n",
    "    \n",
    "    def send_data(self, index, **kwargs):\n",
    "        \"\"\"\n",
    "        Sends the data partitions to the client.\n",
    "        \"\"\"\n",
    "        if self.data is None:\n",
    "            self.load_data()\n",
    "        return self.data[index]\n",
    "    \n",
    "    def send_edges(self, **kwargs):\n",
    "        \"\"\"\n",
    "        Send the list of edges to the client.\n",
    "        \"\"\"\n",
    "        return {**{key:value.uid for (key, value) in self.edges.items() if isinstance(value, BaseNode)},\n",
    "                **{key:value for (key, value) in self.edges.items() if not isinstance(value, BaseNode)}}\n",
    "        \n",
    "    def __getitem__(self, edge):\n",
    "        \"\"\"\n",
    "        Access adjacent nodes.\n",
    "        On the server side edges are the actual adjacent nodes.\n",
    "        On the client side edges are the uids of the adjacent nodes.\n",
    "        On the client side the factory access the server for the data to construct the node locally.\n",
    "        \"\"\"\n",
    "        if not isinstance(edge, BaseNode):\n",
    "            self.edges[edge] = self.factory(edge)\n",
    "        return self.edges[edge]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy\n",
    "\n",
    "class NumpyTestNode(BaseNode):  \n",
    "    def read(self):\n",
    "        # Convert data to a different format.\n",
    "        if self.data is None:\n",
    "            self.load_data()\n",
    "        flat_data = [item for sublist in self.data for item in sublist]\n",
    "        return numpy.array(flat_data)\n",
    "    \n",
    "    def load_meta(self):\n",
    "        # Load header data from a data source.\n",
    "        self.meta.update({'field1': 'testing'})\n",
    "    \n",
    "    def load_data(self):\n",
    "        self.data = [list(range(10)) for i in range(100)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class ServerSim:\n",
    "    def __init__(self, root_node):\n",
    "        self.root = root_node\n",
    "        self.nodes = {'root_node.id': root_node}\n",
    "        \n",
    "    def request(self, request):\n",
    "        if request['request'] == 'send_root':\n",
    "            return msgpack.packb(self.send_root())\n",
    "        else:\n",
    "            return msgpack.packb(getattr(self.nodes[request['uid']], request['request'])(**request['kwargs']))\n",
    "    \n",
    "    def send_root(self):\n",
    "        return self.root.uid"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Remote mode: start a server, and get the root on the client side"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Start a server with the graph equal root_node\n",
    "root_node = BlueskyNode()\n",
    "server = ServerSim(root_node)\n",
    "\n",
    "# Create a local copy of the node.\n",
    "local_copy = BaseNode(server)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Local mode: just build your graph locally, and access it locally"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1,\n",
       "       2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3,\n",
       "       4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5,\n",
       "       6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,\n",
       "       8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9,\n",
       "       0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1,\n",
       "       2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3,\n",
       "       4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5,\n",
       "       6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,\n",
       "       8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9,\n",
       "       0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1,\n",
       "       2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3,\n",
       "       4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5,\n",
       "       6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,\n",
       "       8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9,\n",
       "       0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1,\n",
       "       2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3,\n",
       "       4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5,\n",
       "       6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,\n",
       "       8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9,\n",
       "       0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1,\n",
       "       2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3,\n",
       "       4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5,\n",
       "       6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,\n",
       "       8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9,\n",
       "       0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1,\n",
       "       2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3,\n",
       "       4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5,\n",
       "       6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,\n",
       "       8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9,\n",
       "       0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1,\n",
       "       2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3,\n",
       "       4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5,\n",
       "       6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,\n",
       "       8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9,\n",
       "       0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1,\n",
       "       2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3,\n",
       "       4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5,\n",
       "       6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,\n",
       "       8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9,\n",
       "       0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1,\n",
       "       2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3,\n",
       "       4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5,\n",
       "       6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,\n",
       "       8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9,\n",
       "       0, 1, 2, 3, 4, 5, 6, 7, 8, 9])"
      ]
     },
     "execution_count": 90,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "root_node = NumpyTestNode()\n",
    "root_node.read()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}