NI-DAQ multichannel pulse generation.ipynb

{
 "metadata": {
  "name": "",
  "signature": "sha256:6dfe921f11eae17b7ce9ac7befb3d8a2365ab77acc26f9ed023c6cbb09ce3f4f"
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from __future__ import print_function\n",
      "import sys\n",
      "import PyDAQmx as daq\n",
      "from PyDAQmx import uInt32, int32, int16, byref\n",
      "from contextlib import contextmanager"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 1
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "# Generating a multi-channel pulse sequence with NI-DAQmx\n",
      "\n",
      "I am trying to use a NI PCI-6602 with the BNC-2121 accessory to make a multichannel pulse sequence. To make sure things are working, first try to get 4 synchronous square pulses out.\n",
      "\n",
      "My first (naiive) approach is to make one task, add 4 counter out channels to it, and start the task."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def configure_counter(duration=.1):\n",
      "    \n",
      "    pulses = daq.Task()\n",
      "    channels = (2,3,4,5)\n",
      "\n",
      "    # configure pulses\n",
      "    for ch in channels:\n",
      "        channel = \"Dev1/ctr%d\" % ch\n",
      "        pulses.CreateCOPulseChanTime(\n",
      "            channel, \"\",             # physical channel, name to assign\n",
      "            daq.DAQmx_Val_Seconds,   # units: seconds\n",
      "            daq.DAQmx_Val_Low,       # idle state: low\n",
      "            0., duration, duration,  # initial delay, low time, high time\n",
      "        )\n",
      "    \n",
      "    return pulses"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 2
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def start_pulse(pulse):\n",
      "    pulse.StartTask()\n",
      "\n",
      "def finish_pulse(pulse):\n",
      "    pulse.WaitUntilTaskDone(10.)\n",
      "    pulse.StopTask()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 3
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "@contextmanager\n",
      "def pulsing(pulse):\n",
      "    \"\"\"This stops the pulsing task if I interrupt the python kernel\"\"\"\n",
      "    try:\n",
      "        yield\n",
      "    except KeyboardInterrupt:\n",
      "        # stop the counters\n",
      "        try:\n",
      "            pulse.StopTask()\n",
      "            print(\"stopped pulses\", file=sys.stderr)\n",
      "        except DAQError:\n",
      "            print(\"no need to stop pulses\", file=sys.stderr)\n",
      "        raise"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 4
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "The following should make 2us pulses on the 4 channels:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# make sure to press 'single shot' on the scope, then:\n",
      "\n",
      "pp = configure_counter(duration=2e-6)\n",
      "with pulsing(pp):\n",
      "    start_pulse(pp)\n",
      "    finish_pulse(pp)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 14
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Using this approach I see that the different pulses do not fire synchronously, but rather they fire in the order they were configured, staggered by almost 7us.\n",
      "\n",
      "So, apparently adding a bunch of channels to a single task doesn't work. Try defining them in individual tasks.\n",
      "\n",
      "##Individual tasks, with chained trigger"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def configure_counter(duration=.1):\n",
      "    \n",
      "    pulses = []\n",
      "    channels = (2,3,4,5)\n",
      "\n",
      "    # configure a task for each pulse.\n",
      "    for ch in channels:\n",
      "        pulse = daq.Task()\n",
      "        channel = \"Dev1/ctr%d\" % ch\n",
      "        pulse.CreateCOPulseChanTime(\n",
      "            channel, \"\",             # physical channel, name to assign\n",
      "            daq.DAQmx_Val_Seconds,   # units: seconds\n",
      "            daq.DAQmx_Val_Low,       # idle state: low\n",
      "            0., duration, duration,  # initial delay, low time, high time\n",
      "        )\n",
      "        pulses.append(pulse)\n",
      "        \n",
      "    # configure the first pulse to trigger off the second pulse, and so on\n",
      "    for pulse,ch in zip(pulses[:-1],channels[1:]):\n",
      "        trigch = \"Ctr%dInternalOutput\" % ch\n",
      "        pulse.CfgDigEdgeStartTrig(trigch, daq.DAQmx_Val_Rising)\n",
      "    \n",
      "    return pulses"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 15
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def start_pulses(*pulses):\n",
      "    for pulse in pulses:\n",
      "        pulse.StartTask()\n",
      "\n",
      "def finish_pulses(*pulses):\n",
      "    # wait for pulse to be done\n",
      "    for pulse in pulses:\n",
      "        pulse.WaitUntilTaskDone(10.)\n",
      "        pulse.StopTask()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 16
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "@contextmanager\n",
      "def pulsing(*pulses):\n",
      "    try:\n",
      "        yield\n",
      "    except KeyboardInterrupt:\n",
      "        # stop the counters\n",
      "        for i,pulse in enumerate(pulses):\n",
      "            try:\n",
      "                pulse.StopTask()\n",
      "                print(\"stopped pulse %d\" % i, file=sys.stderr)\n",
      "            except DAQError:\n",
      "                print(\"no need to stop pulse %d\" % i, file=sys.stderr)\n",
      "        raise"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 17
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# press 'single shot', then:\n",
      "\n",
      "pp = configure_counter(duration=100e-9)\n",
      "with pulsing(*pp):\n",
      "    start_pulses(*pp)\n",
      "    finish_pulses(*pp)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 21
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "The pulses now fire in the reverse of the order configured, staggered by 40ns or so. Much better, but would be better still if we could trigger all from the same channel."
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "## Individual tasks all triggered from the same ('dummy') channel"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def configure_counter(duration=.1):\n",
      "    \n",
      "    pulses = []\n",
      "    channels = (2,3,4,5)\n",
      "\n",
      "    # configure a task for each pulse.\n",
      "    for ch in channels:\n",
      "        pulse = daq.Task()\n",
      "        channel = \"Dev1/ctr%d\" % ch\n",
      "        pulse.CreateCOPulseChanTime(\n",
      "            channel, \"\",             # physical channel, name to assign\n",
      "            daq.DAQmx_Val_Seconds,   # units: seconds\n",
      "            daq.DAQmx_Val_Low,       # idle state: low\n",
      "            0., duration, duration,  # initial delay, low time, high time\n",
      "        )\n",
      "        pulses.append(pulse)\n",
      "        \n",
      "    # configure the all but the last pulse to trigger from the last pulse\n",
      "    trigch = \"Ctr%dInternalOutput\" % channels[-1]\n",
      "    for pulse in pulses[:-1]:\n",
      "        pulse.CfgDigEdgeStartTrig(trigch, daq.DAQmx_Val_Rising)\n",
      "    \n",
      "    return pulses"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 27
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# press 'single shot', then:\n",
      "\n",
      "pp = configure_counter(duration=100e-9)\n",
      "with pulsing(*pp):\n",
      "    start_pulses(*pp)\n",
      "    finish_pulses(*pp)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 30
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "On the scope, the pulses from channels 2 and 3 are synchronous, but a little behind channel 4, which is of course a little behind channel 5, which is the trigger source. I think the discrepcancy between channels 2 and 3 from 4 is that 4 is a paired counter with channel 5.\n",
      "\n",
      "I can get 3 synchronous channels by making sure that none of them is paired with the one I'm using to trigger from. That is, by changing `channels` to `(2,3,1,5)` above."
     ]
    }
   ],
   "metadata": {}
  }
 ]
}