<commit_before><commit_msg>Partitioned the multicore notebook into two because it was too big.<commit_after>{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "import time\n",
    "import threading\n",
    "import multiprocessing\n",
    "\n",
    "sys.path.append(\"../\")\n",
    "from IoTPy.core.stream import Stream, StreamArray, run\n",
    "from IoTPy.agent_types.op import map_element, map_list, map_window\n",
    "from IoTPy.helper_functions.recent_values import recent_values\n",
    "from IoTPy.helper_functions.print_stream import print_stream\n",
    "\n",
    "from IoTPy.concurrency.multicore import get_processes, get_processes_and_procs\n",
    "from IoTPy.concurrency.multicore import terminate_stream\n",
    "from IoTPy.concurrency.multicore import get_proc_that_inputs_source\n",
    "from IoTPy.concurrency.multicore import extend_stream\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Actuators\n",
    "A multiprocessing block may need to interact asynchronously with some external device. To do so, the block puts data into a queue and uses threads responsible for interfacing between the queue and the device. This simple example illustrates the simplest actuator: a printer. Indeed printing can be done synchronously by the multiprocessing block. Printing doesn't need a queue to interface between it and the block. We use the printer in this example to illustrate the idea.\n",
    "<br>\n",
    "<br>\n",
    "Function <i>g</i> of process <i>p1</i> has an agent called 'copy_stream_s_to_queue_q' which  copies stream <i>s</i> to queue <i>q</i>. A thread, <b>my_thread</b> in <i>p1</i> prints values from the queue; this thread represents the thread that interfaces with an external actuator device. This thread is in addition to any source threads that may exist.\n",
    "<br>\n",
    "<br>\n",
    "Queue <i>q</i> is specified as an <b>output queue</b>. An output queue gets a special message <b>'_finished'</b> when the multiprocess block terminates.\n",
    "<br>\n",
    "<br>\n",
    "Threads (apart from source threads) and output queues are specified in <i>multicore_specifications</i>. See\n",
    "<br>\n",
    "<br>\n",
    "{'name': 'p1', 'agent': g, 'inputs': ['y'], \n",
    "<br>\n",
    " 'args': [q], <b>'output_queues'</b>: [q], <b>'threads'</b>: [my_thread]}\n",
    "<br>\n",
    "<br>\n",
    "The thread, <i>my_thread</i>, terminates when it receives a '_finished' message. We want this thread to terminate so that process <i>p1</i> terminates, and then the entire multiprocessing block can terminate as well."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "q.get() =  200\n",
      "q.get() =  202\n",
      "q.get() =  204\n",
      "q.get() =  206\n",
      "q.get() =  208\n",
      "q.get() =  210\n"
     ]
    }
   ],
   "source": [
    "import threading\n",
    "from IoTPy.agent_types.sink import stream_to_queue\n",
    "\n",
    "def f(in_streams, out_streams):\n",
    "    map_element(lambda v: v+100, in_streams[0], out_streams[0])\n",
    "\n",
    "def source_thread_target(procs):\n",
    "    for i in range(3):\n",
    "        extend_stream(procs, data=list(range(i*2, (i+1)*2)), stream_name='x')\n",
    "        time.sleep(0.001)\n",
    "    terminate_stream(procs, stream_name='x')\n",
    "\n",
    "def example_output_thread_with_queue():\n",
    "    q = multiprocessing.Queue()\n",
    "\n",
    "    def g(in_streams, out_streams, q):\n",
    "        s = Stream('s')\n",
    "        map_element(lambda v: v*2, in_streams[0], s)\n",
    "        stream_to_queue(s, q, name='copy_stream_s_to_queue_q')\n",
    "\n",
    "    def get_data_from_output_queue(q):\n",
    "        while True:\n",
    "            v = q.get()\n",
    "            if v == '_finished': break\n",
    "            else: print ('q.get() = ', v)\n",
    "\n",
    "    multicore_specification = [\n",
    "        # Streams\n",
    "        [('x', 'i'), ('y', 'i')],\n",
    "        # Processes\n",
    "        [{'name': 'p0', 'agent': f, 'inputs':['x'], 'outputs': ['y'], 'sources': ['x']},\n",
    "         {'name': 'p1', 'agent': g, 'inputs': ['y'], \n",
    "          'args': [q], 'output_queues': [q]}]\n",
    "    ]\n",
    "\n",
    "    processes, procs = get_processes_and_procs(multicore_specification)\n",
    "    source_thread = threading.Thread(target=source_thread_target, args=(procs,))\n",
    "    output_thread = threading.Thread(target=get_data_from_output_queue, args=(q,))\n",
    "    procs['p0'].threads = [source_thread, output_thread]\n",
    "\n",
    "    for process in processes: process.start()\n",
    "    for process in processes: process.join()\n",
    "    for process in processes: process.terminate()\n",
    "\n",
    "example_output_thread_with_queue()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example of Process Structure with Feedback\n",
    "The example shows a process structure with feedback. This example creates an echo from a spoken sound. (You can write more efficient and succinct code to compute echoes. The code in this example is here merely because it illustrates a concept.)\n",
    "<br>\n",
    "### Streams\n",
    "<ol>\n",
    "    <li><b>sound_made</b>: This is the sound made by a speaker in a large spherical space.</li>\n",
    "    <li><b>attenuated</b>: This is the sound made multiplied by an attenuation factor.</li>\n",
    "    <li><b>echo</b>: This is the echo of the sound made heard at the center of the room. The echo is a delay followed by an attenuation of the sound heard. </li>\n",
    "    <li><b>sound_heard</b>: This is the sound that is heard by the speaker. The heard sound is the sound made by the speaker plus the echo.</li>\n",
    "</ol>\n",
    "The equations that define the streams are:\n",
    "<ol>\n",
    "    <li>\n",
    "    <b>attentuated[n] = sound_heard[n]*attenuation</b>\n",
    "    </li>\n",
    "    <li>\n",
    "    <b>echo[n] = attentuated[n-delay]</b> for n > delay.\n",
    "    </li>\n",
    "    <li>\n",
    "    <b>sound_heard[n] = sound_made[n] + echo[n]</b> for n > delay.\n",
    "    </li>\n",
    "</ol>\n",
    "\n",
    "### Process Structure\n",
    "Process <i>p0</i> has a source which feeds one of its input streams <i>sound_made</i> with a stream of measurements obtained from a microphone. In this example, the stream is generated with numbers so that we can see how streams are processed.\n",
    "<br>\n",
    "<br>\n",
    "Process <i>p1</i> contains a single input stream which is the sound heard and a single output stream which is an attenuation of the sound heard. \n",
    "\n",
    "### Process Functions\n",
    "The function <i>f</i> of <i>p0</i> computes <i>echo</i> from <i>sound_made</i>. The first 4 , i.e., <b>delay</b>, units of the echo are empty (i.e. 0). \n",
    "<br>\n",
    "<b>map_element(lambda v: v, attenuated, echo)</b>\n",
    "<br>\n",
    "copies the attenuated stream to the echo stream; however, since the first 4 (i.e. delay) values of the echo stream are 0, the echo stream will consist of 4 zeroes followed by the attenuated stream.\n",
    "<br>\n",
    "<i>out_streams[0]</i> of process <i>p0</i> is <i>sound_heard</i>. Function <i>f</i> makes <i>sound_heard</i> the sum of the echo and the sound made.\n",
    "<br>\n",
    "The function <i>g</i> of process <i>p1</i> <i>p0</i> puts elements of its input stream (i.e. <i>sound_heard</i> on queue <i>q</i> and returns the elements multiplied by <i>attenuation</i>."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "heard sound = spoken + echo:  0.0\n",
      "heard sound = spoken + echo:  1.0\n",
      "heard sound = spoken + echo:  2.0\n",
      "heard sound = spoken + echo:  3.0\n",
      "heard sound = spoken + echo:  4.0\n",
      "heard sound = spoken + echo:  5.5\n",
      "heard sound = spoken + echo:  7.0\n",
      "heard sound = spoken + echo:  8.5\n",
      "heard sound = spoken + echo:  10.0\n",
      "heard sound = spoken + echo:  11.75\n",
      "heard sound = spoken + echo:  3.5\n",
      "heard sound = spoken + echo:  4.25\n",
      "heard sound = spoken + echo:  5.0\n",
      "heard sound = spoken + echo:  5.875\n",
      "heard sound = spoken + echo:  1.75\n",
      "heard sound = spoken + echo:  2.125\n",
      "heard sound = spoken + echo:  2.5\n",
      "heard sound = spoken + echo:  2.9375\n",
      "heard sound = spoken + echo:  0.875\n",
      "heard sound = spoken + echo:  1.0625\n"
     ]
    }
   ],
   "source": [
    "from IoTPy.agent_types.basics import *\n",
    "\n",
    "def example_echo_two_cores():\n",
    "    # This is the delay from when the made sound hits a\n",
    "    # reflecting surface.\n",
    "    delay = 4\n",
    "\n",
    "    # This is the attenuation of the reflected wave.\n",
    "    attenuation = 0.5\n",
    "\n",
    "    # The results are put in this queue. A thread reads this\n",
    "    # queue and feeds a speaker or headphone.\n",
    "    q = multiprocessing.Queue()\n",
    "\n",
    "    # Agent function for process named 'p0'\n",
    "    # echo is a delay of zeroes followed by attenuated heard sound.\n",
    "    # out_streams[0], which is the same as sound_heard is\n",
    "    # echo + sound_made\n",
    "    def f_echo(in_streams, out_streams, delay):\n",
    "        sound_made, attenuated = in_streams\n",
    "        echo = StreamArray('echo', dtype='float')\n",
    "        echo.extend(np.zeros(delay, dtype='float'))\n",
    "        map_element(lambda v: v, attenuated, echo)\n",
    "        # The zip_map output is the sound heard which is\n",
    "        # the sound heard plus the echo.\n",
    "        zip_map(sum, [sound_made, echo], out_streams[0])\n",
    "\n",
    "    # Agent function for process named 'p1'\n",
    "    # This process puts the sound heard into the output queue\n",
    "    # and returns an attenuated version of the sound_heard as \n",
    "    # its output stream.\n",
    "    def g_echo(in_streams, out_streams, attenuation, q):\n",
    "        def gg(v):\n",
    "            # v is the sound heard\n",
    "            q.put(v)\n",
    "            # v*attenuation is the echo\n",
    "            return v*attenuation\n",
    "        map_element(gg, in_streams[0], out_streams[0])\n",
    "\n",
    "    def source_thread_target(procs):\n",
    "        data=np.arange(10, dtype='float')\n",
    "        extend_stream(procs, data, stream_name='sound_made')\n",
    "        time.sleep(0.0001)\n",
    "        extend_stream(procs, data=np.zeros(10, dtype='float'), stream_name='sound_made')\n",
    "        terminate_stream(procs, stream_name='sound_made')\n",
    "\n",
    "    # Thread that gets data from the output queue\n",
    "    # This thread is included in 'threads' in the specification.\n",
    "    # Thread target\n",
    "    def get_data_from_output_queue(q):\n",
    "        finished_getting_output = False\n",
    "        while not finished_getting_output:\n",
    "            v = q.get()\n",
    "            if v == '_finished': break\n",
    "            print ('heard sound = spoken + echo: ', v)\n",
    "\n",
    "    multicore_specification = [\n",
    "        # Streams\n",
    "        [('sound_made', 'f'), ('attenuated', 'f'), ('sound_heard', 'f')],\n",
    "        # Processes\n",
    "        [{'name': 'p0', 'agent': f_echo, 'inputs': ['sound_made', 'attenuated'], \n",
    "          'outputs': ['sound_heard'], 'keyword_args' : {'delay' : delay}, 'sources': ['sound_made']},\n",
    "         {'name': 'p1', 'agent': g_echo, 'inputs': ['sound_heard'], 'outputs': ['attenuated'],\n",
    "          'args': [attenuation, q], 'output_queues': [q] } ]]\n",
    "\n",
    "    processes, procs = get_processes_and_procs(multicore_specification)\n",
    "     \n",
    "    source_thread = threading.Thread(target=source_thread_target, args=(procs,))\n",
    "    output_thread = threading.Thread(target=get_data_from_output_queue, args=(q,))\n",
    "    procs['p0'].threads = [source_thread, output_thread]\n",
    "\n",
    "    for process in processes: process.start()\n",
    "    for process in processes: process.join()\n",
    "    for process in processes: process.terminate()\n",
    "\n",
    "example_echo_two_cores()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example source and actuator thread with single process\n",
    "This example is the same as the previous one except that the computation is carried out in a single process rather in two processes. The example illustrates an actuator thread and a source thread in the same process."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "heard sound = spoken + echo:  0.0\n",
      "heard sound = spoken + echo:  1.0\n",
      "heard sound = spoken + echo:  2.0\n",
      "heard sound = spoken + echo:  3.0\n",
      "heard sound = spoken + echo:  4.0\n",
      "heard sound = spoken + echo:  5.5\n",
      "heard sound = spoken + echo:  7.0\n",
      "heard sound = spoken + echo:  8.5\n",
      "heard sound = spoken + echo:  10.0\n",
      "heard sound = spoken + echo:  11.75\n",
      "heard sound = spoken + echo:  3.5\n",
      "heard sound = spoken + echo:  4.25\n",
      "heard sound = spoken + echo:  5.0\n",
      "heard sound = spoken + echo:  5.875\n",
      "heard sound = spoken + echo:  1.75\n",
      "heard sound = spoken + echo:  2.125\n",
      "heard sound = spoken + echo:  2.5\n",
      "heard sound = spoken + echo:  2.9375\n",
      "heard sound = spoken + echo:  0.875\n",
      "heard sound = spoken + echo:  1.0625\n"
     ]
    }
   ],
   "source": [
    "def example_echo_single_core():\n",
    "    # This is the delay from when the made sound hits a\n",
    "    # reflecting surface.\n",
    "    delay = 4\n",
    "\n",
    "    # This is the attenuation of the reflected wave.\n",
    "    attenuation = 0.5\n",
    "\n",
    "    # The results are put in this queue. A thread reads this\n",
    "    # queue and feeds a speaker or headphone.\n",
    "    q = multiprocessing.Queue()\n",
    "\n",
    "    # Agent function for process named 'p0'\n",
    "    # echo is a delay of zeroes followed by attenuated heard sound.\n",
    "    # out_streams[0], which is the same as sound_heard is\n",
    "    # echo + sound_made\n",
    "    def f_echo(in_streams, out_streams, delay, attenuation, q):\n",
    "        echo = StreamArray(\n",
    "            'echo', initial_value=np.array([0.0]*delay, dtype='float'), dtype='float')\n",
    "        #Note: sound_made = in_streams[0]\n",
    "        sound_heard = in_streams[0] + echo\n",
    "        map_element(lambda v: v*attenuation, sound_heard, echo)\n",
    "        stream_to_queue(sound_heard, q)\n",
    "\n",
    "    def source_thread_target(procs):\n",
    "        extend_stream(procs, data=np.arange(10, dtype='float'), stream_name='sound_made')\n",
    "        time.sleep(0.0001)\n",
    "        extend_stream(procs=procs, data=np.zeros(10, dtype='float'), stream_name='sound_made')\n",
    "        terminate_stream(procs, stream_name='sound_made')\n",
    "\n",
    "    # Thread that gets data from the output queue\n",
    "    # This thread is included in 'threads' in the specification.\n",
    "    # Thread target\n",
    "    def get_data_from_output_queue(q):\n",
    "        finished_getting_output = False\n",
    "        while not finished_getting_output:\n",
    "            v = q.get()\n",
    "            if v == '_finished': break\n",
    "            print ('heard sound = spoken + echo: ', v)\n",
    "\n",
    "    multicore_specification = [\n",
    "        # Streams\n",
    "        [('sound_made', 'f')],\n",
    "        # Processes\n",
    "        [{'name': 'p0', 'agent': f_echo, 'inputs': ['sound_made'],\n",
    "          'args' : [delay, attenuation, q], 'sources': ['sound_made'],'output_queues': [q]}]]\n",
    "\n",
    "    processes, procs = get_processes_and_procs(multicore_specification)\n",
    "     \n",
    "    source_thread = threading.Thread(target=source_thread_target, args=(procs,))\n",
    "    output_thread = threading.Thread(target=get_data_from_output_queue, args=(q,))\n",
    "    procs['p0'].threads = [source_thread, output_thread]\n",
    "\n",
    "    for process in processes: process.start()\n",
    "    for process in processes: process.join()\n",
    "    for process in processes: process.terminate()\n",
    "\n",
    "example_echo_single_core()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example of a grid computation\n",
    "Grid computations are used in science, for example in computing the temperature of a metal plate. The grid is partitioned into regions with a process assigned to simulate each region. On the n-th step, each process reads the values of relevant parts of the grid and updates its own value.\n",
    "<br>\n",
    "<br>\n",
    "This example uses two copies of the grid; the two copies are <b>even</b> and <b>odd</b>. On even steps (i.e., steps 0, 2, 4,..) each process reads the <i>even</i> grid and updates its portion of the <i>odd</i> grid. So, each portion of the grid is modified by only one process. And no process is reading values while the values are being modified.\n",
    "<br>\n",
    "By symmetry, on odd steps, each process reads the <i>odd</i> grid and updates its portion of the <i>even</i> grid.\n",
    "<br>\n",
    "### The example problem\n",
    "A linear metal bar of length <i>N</i> is partitioned into a grid of <i>N</i> continuous regions. Grid 0 is kept at a constant temperature of 0 degrees while grid <i>N-1</i> is kept at a constant temperature of <i>N-1</i> degrees. Initially, the temperature at intermediate grid points is arbitrary; in the code below, the temperature at grid point <i>i</i> exceeds <i>i</i> by <i>DELTA</i>.\n",
    "<br>\n",
    "Let <b>TEMP[i][k]</b> be the temperature of the <i>i</i>-th region on step <i>k</i>. Then:\n",
    "<ol>\n",
    "    <li>TEMP[0][k] = 0 </li>\n",
    "    <li>TEMP[N-1][k] = N-1 </li>\n",
    "    <li>TEMP[i][k] = (TEMP[i-1][k] + TEMP[i][k] + TEMP[i+1][k])/3 i in [1, ..,N-2] </li>\n",
    "</ol>\n",
    "\n",
    "\n",
    "### Processes\n",
    "The computation uses <i>N-2</i> processes. The <i>i</i>-th process is called 'grid_i' and is responsible for simulating the <i>i</i>-th region.\n",
    "<br>\n",
    "Each process takes the <i>k + 1</i>-th step after it has finished the <i>k</i>-th step and it has determined that its neighbors have also finished the <i>k</i>-th step.\n",
    "<br>\n",
    "\n",
    "\n",
    "### Streams\n",
    "The system has one stream, <b>s_i</b> for the <i>i</i>-th process. This stream contains the elements [0, 1, .. , k] after the <i>i</i>-th process has completed <i>k</i>-th steps.\n",
    "<br>\n",
    "Process <i>grid_i</i> outputs stream <i>s_i</i> and inputs streams from its neighboring processes which are <i>grid_(i-1)</i> if <i>i</i> exceeds 1 and <i>grid_(i+1)</i> if <i>i</i> is less than <i>N-1</i>.\n",
    "\n",
    "### Process Structure\n",
    "The process structure is linear with each process getting input streams from each of its neighbors and sending its output stream to all its neighbors.\n",
    "\n",
    "### Process Function\n",
    "The process begins by sending 0 on its output stream to indicate that it has finished its 0-th step.\n",
    "<br>\n",
    "<br>\n",
    "The <i>k</i>-th value of <i>in_streams[j]</i> is <i>k</i> when the <i>j</i>-th neighboring process has completed its <i>k</i>-th step.\n",
    "<br>\n",
    "<br>\n",
    "<b>synch_stream</b> is an internal stream of the process. The <i>k</i>-th element of this stream is <i>k</i> after all neighboring processes have completed their <i>k</i>-th step.\n",
    "<br>\n",
    "<br>\n",
    "The zip_map function <i>r</i> operates on a list with one element from each neighbor. All the elements of the list will be <i>k</i> on the <i>k</i>-th step. The zip_map function returns <i>k</i> which is any element of the list. In this example it returns the 0-th element.\n",
    "<br>\n",
    "<br>\n",
    "Thus the zip_map function acts as a synchronizer. It waits until all neighbors have completed the <i>k</i>-step and then it outputs <i>k</i>.\n",
    "<br>\n",
    "<br>\n",
    "Function <i>g</i> is called for the <i>k</i>-th time when this process and all its neighbors have completed <i>k - 1</i> steps. Function <i>g</i> does the grid computation. Function <i>r</i> and the zip_map agent are used merely for synchronizing.\n",
    "### run()\n",
    "Function <i>f</i> calls <b>run</b> after it has declared all its agents. Without calling run() the function will take no action.\n",
    "<br>\n",
    "<br>\n",
    "Note that when using external source threads, you should not call <i>run</i> because the source threads are responsible for starting and stopping the main computational thread. This example has no source threads so you must call <i>run</i> to start the system."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Grid after  5  steps is: \n",
      "[0.0, 1.002880573272705, 2.0040740966796875, 3.002880573272705, 4.0]\n"
     ]
    }
   ],
   "source": [
    "from IoTPy.core.stream import _no_value\n",
    "def test_grid():\n",
    "    # N is the size of the grid\n",
    "    N = 5\n",
    "    # M is the number of steps of execution.\n",
    "    M = 5\n",
    "    # DELTA is the deviation from the final solution.\n",
    "    DELTA = 0.01\n",
    "    # even, odd are the grids that will be returned\n",
    "    # by this computation\n",
    "    even = multiprocessing.Array('f', N)\n",
    "    odd = multiprocessing.Array('f', N)\n",
    "    # Set up initial values of the grid.\n",
    "    for i in range(1, N-1):\n",
    "        even[i] = i + DELTA\n",
    "    even[N-1] = N-1\n",
    "    odd[N-1] = N-1\n",
    "        \n",
    "    def f(in_streams, out_streams, index, even, odd):\n",
    "        def g(v):\n",
    "            if (0 < index) and (index < N-1):\n",
    "                if v%2 == 0:\n",
    "                    odd[index] = (even[index-1] + even[index] + even[index+1])/3.0\n",
    "                else:\n",
    "                    even[index] = (odd[index-1] + odd[index] + odd[index+1])/3.0\n",
    "            return v+1\n",
    "\n",
    "        def r(lst, state):\n",
    "            if state < M:\n",
    "                return lst[0], state+1\n",
    "            else:\n",
    "                return _no_value, state\n",
    "        for out_stream in out_streams: out_stream.extend([0])\n",
    "        synch_stream = Stream('synch_stream')\n",
    "        zip_map(r, in_streams, synch_stream, state=0, name='zip_map_'+str(index))\n",
    "        map_element(g, synch_stream, out_streams[0], name='grid'+str(index))\n",
    "        run()\n",
    "\n",
    "    multicore_specification = [\n",
    "        # Streams\n",
    "        [('s_'+str(index), 'i') for index in range(1, N-1)],\n",
    "        # Processes\n",
    "        [{'name': 'grid_'+str(index), 'agent': f, \n",
    "          'inputs':['s_'+str(index+1), 's_'+str(index-1)], \n",
    "          'outputs':['s_'+str(index)], \n",
    "          'args': [index, even, odd]} for index in range(2, N-2)] + \\\n",
    "        [{'name': 'grid_'+str(1), 'agent': f, \n",
    "          'inputs':['s_'+str(2)], 'outputs':['s_'+str(1)], \n",
    "          'args': [1, even, odd]}] + \\\n",
    "        [{'name': 'grid_'+str(N-2), 'agent': f, \n",
    "          'inputs':['s_'+str(N-3)], 'outputs':['s_'+str(N-2)], \n",
    "          'args': [N-2, even, odd]}]\n",
    "    ]\n",
    "\n",
    "    # Execute processes (after including your own non IoTPy processes)\n",
    "    processes = get_processes(multicore_specification)\n",
    "    for process in processes: process.start()\n",
    "    for process in processes: process.join()\n",
    "    for process in processes: process.terminate()\n",
    "\n",
    "    print ('Grid after ', M, ' steps is: ')\n",
    "    if M%2 == 0:\n",
    "        print (even[:])\n",
    "    else:\n",
    "        print (odd[:])\n",
    "\n",
    "test_grid()"
   ]
  }
 ],
 "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
}
{'fraction_non_alphanumeric': 0.17902823426919573, 'fraction_numerical': 0.013780540925830594, 'mean_word_length': 2.4934292433254943, 'pattern_counts': {'":': 68, '<': 266, '<?xml version=': 0, '>': 265, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4926782', 'n_tokens_mistral': 9174, 'n_tokens_neox': 8866, 'n_words': 2471}