Add slow_bend.py version 0.2.

[blog.git] / posts / slow_bend / slow_bend.py

diff --git a/posts/slow_bend/slow_bend.py b/posts/slow_bend/slow_bend.py
new file mode 100755 (executable)
index 0000000..bb0cd5d
--- /dev/null
+++ b/posts/slow_bend/slow_bend.py
@@ -0,0 +1,139 @@
+#!/usr/bin/python
+#
+# W. Trevor King, Jan. 29th, 2008
+#
+# Version 0.2, WT King, Feb. 20th, 2008
+# (added OptionParser code and stripchart debug parameters)
+
+"""
+Simple single analog channel polling software.
+Spends most of its time sleeping,
+waking up every dt seconds to record the current voltage.
+Designed for Liming's cellulase activity measurements.
+"""
+
+import comedi_single_aio
+import time
+import curses_check_for_keypress
+import stripchart
+import data_logger
+
+DEFAULT_STRIPVERBOSE = False
+TEXT_VERBOSE = False
+LOG_DIR = '~/data/slow_bend' # set LOG_DIR = None to disable logging
+
+class slow_bend :
+    """
+    Take measurements on a single channel every dt seconds.
+    Save '<time>\t<bit_val>\t<physical_val>\n' records.
+    """
+    def __init__(self, chan=0, dt=4, log_dir=LOG_DIR) :
+        if TEXT_VERBOSE :
+            print "Initializing slow_bend instance"
+            print " chan %d, dt %g, log_dir %s" % (chan, dt, log_dir)
+        self.ai = comedi_single_aio.ai_obj(chan=[chan])
+        self.dt = dt
+        self._data_logger = data_logger.data_log(log_dir, log_name="slow_bend")
+        self.savefilename,timestamp = self._data_logger.get_filename()
+        self._make_header()
+        self.stripchart = stripchart.stripchart('slow_bend_pipe',
+                                                title='slow bend data')
+        if DEFAULT_STRIPVERBOSE == True :
+            self.stripchart.open(debug=TEXT_VERBOSE)
+    def run(self) :
+        if TEXT_VERBOSE :
+            print "Running slow_bend"
+        self.start_time = time.time()
+        next_read_time = self.start_time
+        self.c = curses_check_for_keypress.check_for_keypress( \
+                    "Press any key to quit")
+        if TEXT_VERBOSE :
+            self.c.output("Starting loop\n")
+        while self.c.input() == None :
+            if TEXT_VERBOSE :
+                self.c.output("Taking measurement\n")
+            tm,bitval,physical = self._take_and_save_reading()
+            self.c.output("%s\t%g\n" % (tm-self.start_time, physical))
+            next_read_time += self.dt
+            self._wait_until(next_read_time)
+        del(self.c)
+    def _measure_voltage(self) :
+        # opening each time is slower than leaving it open
+        # but for a slow measurement that's not worth
+        # hogging the card.
+        self.ai.open()
+        bitval = self.ai.read()[0]
+        physical = self.ai.comedi_to_phys(0,bitval)
+        self.ai.close()
+        if TEXT_VERBOSE :
+            self.c.output("%d\t%g\n" % (bitval, physical))
+        if self.stripchart.status == "open" :
+            if bitval == 0 :
+                self.stripchart.add_point(-10)
+            elif bitval == 2**16-1 :
+                self.stripchart.add_point(10)
+            else :
+                self.stripchart.add_point(physical)
+        return (bitval, physical)
+    def _make_header(self) :
+        "create the save file header"
+        fid = file(self.savefilename, 'w')
+        fid.write('#%s\t%s\t%s\n' % ('time (s)',
+                                     'deflection (bits)',
+                                     'deflection (V)'))
+        fid.close()
+    def _save_reading(self, time, bitval, physical) :
+        fid = file(self.savefilename, 'a+')
+        fid.write("%g\t%d\t%g\n" % (time - self.start_time,
+                                    bitval, physical))
+        fid.close()
+    def _take_and_save_reading(self) :
+        tm = time.time()
+        bitval,physical = self._measure_voltage()
+        self._save_reading(tm, bitval, physical)
+        return tm, bitval, physical
+    def _wait_until(self, targ_time) :
+        dt = targ_time - time.time()
+        if dt > 0 :
+            time.sleep(dt)
+
+if __name__ == "__main__" :
+    from optparse import OptionParser
+    
+    usage_string = """%prog [options]
+
+2008, W. Trevor King.
+"""
+    parser = OptionParser(usage=usage_string, version="%prog 0.2")
+
+    parser.add_option('-t', '--timestep', dest="dt",
+                      help="Measure voltage evert DT seconds (default '%default')",
+                      type='float', metavar="DATABASE", default=4.0)
+    parser.add_option('-c', '--input-channel', dest="chan",
+                      help="Input voltage CHANNEL (default '%default')",
+                      type='int', metavar="CHANNEL", default=0)
+    parser.add_option('-d', '--log-directory', dest="log_dir",
+                      help="Write output to subdir of LOG_DIR (default %default)",
+                      type='string', metavar="LOG_DIR", default=LOG_DIR)
+    parser.add_option('-s', '--stripchart-off', dest="stripchart",
+                      action="store_false",
+                      help="Turn off the stripchart display (default)",
+                      default=False)
+    parser.add_option('-S', '--stripchart-on', dest="stripchart",
+                      action="store_true",
+                      help="Turn on the stripchart display")
+    parser.add_option('-v', '--verbose', dest="verbose", action="store_true",
+                      help="Print lots of debugging information",
+                      default=False)
+
+    (options, args) = parser.parse_args()
+    parser.destroy()
+
+    if options.verbose :
+        TEXT_VERBOSE = True
+
+    sb = slow_bend(chan=options.chan, dt=options.dt, log_dir=options.log_dir)
+    if options.stripchart :
+        sb.stripchart.open(debug=TEXT_VERBOSE)
+
+    sb.run()