10_Modules_pygrass_parallel.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# How to run pygrass `Modules` in parallel"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It is possible to run `Modules` object in parallel. There are more ways, in this documents there are two of them.\n",
    "\n",
    "* using `subprocess` module\n",
    "* using `ParallelModuleQueue` pygrass modules"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## subprocess"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`subprocess` is the Python module to spawn new processes, connect to their input/output/error pipes, and obtain their return codes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from multiprocessing import Queue, Process\n",
    "import grass.script as grass"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First it is required to create a function with the analysis needed, in this case create a buffer around the input vector map."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def calculate(inp, dist):\n",
    "    # function to create a buffer\n",
    "    out = \"{ip}_buffer_{di}\".format(ip=inp, di=dist)\n",
    "    grass.run_command(\"v.buffer\", input=inp, output=out, distance=dist, overwrite=True, quiet=True)\n",
    "    return out"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def spawn(func):\n",
    "    def funct(q_in, q_out):\n",
    "        while True:\n",
    "            vec, dis = q_in.get()\n",
    "            if vec is None:\n",
    "                break\n",
    "            q_out.put(func(vec, dis))\n",
    "    return funct"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now create the needed input"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# two queue object, one for input and output\n",
    "q_in = Queue(1)\n",
    "q_out = Queue()\n",
    "# number of process to create\n",
    "nproc = 3\n",
    "# input vector and distances\n",
    "invect = \"schools\"\n",
    "inbuffer = range(0,220,20)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Queue?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a list of process according to `nproc` variable, the target is the `spawn` function with `calculate` as variable for `spawn`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "procs = [Process(target=spawn(calculate), args=(q_in, q_out)) for _ in range(nproc)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "Process?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For each process set it as daemon and start the process"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for proc in procs:\n",
    "    proc.daemon = True\n",
    "    proc.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Set the inputs (vector map and distance)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ans = [q_in.put((invect, i)) for i in inbuffer if i > 0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Set `None` to each processes to terminate them"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "[q_in.put((None, None)) for proc in procs]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At this point check if the processes run correctly and if the maps are created"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "[proc.join() for proc in procs]\n",
    "processed = [q_out.get() for _ in ans]\n",
    "if len(processed) != len(inbuffer) - 1:\n",
    "    print(\"An error occurs\")\n",
    "print(grass.read_command('g.list', type=\"vector\", pattern=\"schools_buffer*\", mapset='.'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ParallelModuleQueue"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `ParallelModuleQueue` is designed to run an arbitrary number of pygrass Module or MultiModule processes in parallel. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from grass.pygrass.modules import Module, ParallelModuleQueue"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ParallelModuleQueue?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Initialize the `ParallelModuleQueue` object"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "queue = ParallelModuleQueue(nprocs=nproc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Put the `Modules` objects inside the `queue` and wait until the finish"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "val = 0\n",
    "while val < 200:\n",
    "    queue.put(Module(\"r.mapcalc\",\n",
    "                     expression=\"elevation_{va}=if(elevation@PERMANENT>{va} && elevation@PERMANENT<={va2}, elevation@PERMANENT, null())\".format(va=val, va2=val+50),\n",
    "                     overwrite=True, run_=False, finish_=False)\n",
    "             )\n",
    "    val += 50\n",
    "queue.wait()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Check how the processes terminate"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Number of running process \".format(queue.get_num_run_procs()))\n",
    "print(\"Number of max number of process \".format(queue.get_max_num_procs()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for mapcalc in queue.get_finished_modules():\n",
    "    print(mapcalc.popen.returncode)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Summary"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "We have seen\n",
    "\n",
    "* how to run GRASS modules using the standard library `subprocess` module\n",
    "* how to use `ParallelModuleQueue` pygrass module"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercise"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.16"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}