hooke/plugin/curve.py

   1 # Copyright (C) 2008-2010 Alberto Gomez-Casado
   2 #                         Fabrizio Benedetti
   3 #                         Massimo Sandal <devicerandom@gmail.com>
   4 #                         W. Trevor King <wking@drexel.edu>
   5 #
   6 # This file is part of Hooke.
   7 #
   8 # Hooke is free software: you can redistribute it and/or modify it
   9 # under the terms of the GNU Lesser General Public License as
  10 # published by the Free Software Foundation, either version 3 of the
  11 # License, or (at your option) any later version.
  12 #
  13 # Hooke is distributed in the hope that it will be useful, but WITHOUT
  14 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
  15 # or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General
  16 # Public License for more details.
  17 #
  18 # You should have received a copy of the GNU Lesser General Public
  19 # License along with Hooke.  If not, see
  20 # <http://www.gnu.org/licenses/>.
  21
  22 """The ``curve`` module provides :class:`CurvePlugin` and several
  23 associated :class:`hooke.command.Command`\s for handling
  24 :mod:`hooke.curve` classes.
  25 """
  26
  27 import numpy
  28
  29 from ..command import Command, Argument, Failure
  30 from ..curve import Data
  31 from ..plugin import Builtin
  32 from ..plugin.playlist import current_playlist_callback
  33 from ..util.calculus import derivative
  34 from ..util.fft import unitary_avg_power_spectrum
  35 from ..util.si import ppSI, split_data_label
  36
  37
  38 class CurvePlugin (Builtin):
  39     def __init__(self):
  40         super(CurvePlugin, self).__init__(name='curve')
  41         self._commands = [
  42             GetCommand(self), InfoCommand(self), DeltaCommand(self),
  43             ExportCommand(self), DifferenceCommand(self),
  44             DerivativeCommand(self), PowerSpectrumCommand(self)]
  45
  46
  47 # Define common or complicated arguments
  48
  49 def current_curve_callback(hooke, command, argument, value):
  50     if value != None:
  51         return value
  52     playlist = current_playlist_callback(hooke, command, argument, value)
  53     curve = playlist.current()
  54     if curve == None:
  55         raise Failure('No curves in %s' % playlist)
  56     return curve
  57
  58 CurveArgument = Argument(
  59     name='curve', type='curve', callback=current_curve_callback,
  60     help="""
  61 :class:`hooke.curve.Curve` to act on.  Defaults to the current curve
  62 of the current playlist.
  63 """.strip())
  64
  65
  66 # Define commands
  67
  68 class GetCommand (Command):
  69     """Return a :class:`hooke.curve.Curve`.
  70     """
  71     def __init__(self, plugin):
  72         super(GetCommand, self).__init__(
  73             name='get curve', arguments=[CurveArgument],
  74             help=self.__doc__, plugin=plugin)
  75
  76     def _run(self, hooke, inqueue, outqueue, params):
  77         outqueue.put(params['curve'])
  78
  79 class InfoCommand (Command):
  80     """Get selected information about a :class:`hooke.curve.Curve`.
  81     """
  82     def __init__(self, plugin):
  83         args = [
  84             CurveArgument,
  85             Argument(name='all', type='bool', default=False, count=1,
  86                      help='Get all curve information.'),
  87             ]
  88         self.fields = ['name', 'path', 'experiment', 'driver', 'filetype', 'note',
  89                        'blocks', 'block sizes']
  90         for field in self.fields:
  91             args.append(Argument(
  92                     name=field, type='bool', default=False, count=1,
  93                     help='Get curve %s' % field))
  94         super(InfoCommand, self).__init__(
  95             name='curve info', arguments=args,
  96             help=self.__doc__, plugin=plugin)
  97
  98     def _run(self, hooke, inqueue, outqueue, params):
  99         fields = {}
 100         for key in self.fields:
 101             fields[key] = params[key]
 102         if reduce(lambda x,y: x and y, fields.values()) == False:
 103             params['all'] = True # No specific fields set, default to 'all'
 104         if params['all'] == True:
 105             for key in self.fields:
 106                 fields[key] = True
 107         lines = []
 108         for key in self.fields:
 109             if fields[key] == True:
 110                 get = getattr(self, '_get_%s' % key.replace(' ', '_'))
 111                 lines.append('%s: %s' % (key, get(params['curve'])))
 112         outqueue.put('\n'.join(lines))
 113
 114     def _get_name(self, curve):
 115         return curve.name
 116
 117     def _get_path(self, curve):
 118         return curve.path
 119
 120     def _get_experiment(self, curve):
 121         return curve.info.get('experiment', None)
 122
 123     def _get_driver(self, curve):
 124         return curve.driver
 125
 126     def _get_filetype(self, curve):
 127         return curve.info.get('filetype', None)
 128
 129     def _get_note(self, curve):
 130         return curve.info.get('note', None)
 131
 132     def _get_blocks(self, curve):
 133         return len(curve.data)
 134
 135     def _get_block_sizes(self, curve):
 136         return [block.shape for block in curve.data]
 137
 138
 139 class DeltaCommand (Command):
 140     """Get distance information between two points.
 141
 142     With two points A and B, the returned distances are A-B.
 143     """
 144     def __init__(self, plugin):
 145         super(DeltaCommand, self).__init__(
 146             name='delta',
 147             arguments=[
 148                 CurveArgument,
 149                 Argument(name='block', type='int', default=0,
 150                     help="""
 151 Data block that points are selected from.  For an approach/retract
 152 force curve, `0` selects the approaching curve and `1` selects the
 153 retracting curve.
 154 """.strip()),
 155                 Argument(name='point', type='point', optional=False, count=2,
 156                          help="""
 157 Indicies of points bounding the selected data.
 158 """.strip()),
 159                 Argument(name='SI', type='bool', default=False,
 160                          help="""
 161 Return distances in SI notation.
 162 """.strip())
 163                 ],
 164             help=self.__doc__, plugin=plugin)
 165
 166     def _run(self, hooke, inqueue, outqueue, params):
 167         data = params['curve'].data[params['block']]
 168         As = data[params['point'][0],:]
 169         Bs = data[params['point'][1],:]
 170         ds = [A-B for A,B in zip(As, Bs)]
 171         if params['SI'] == False:
 172             out = [(name, d) for name,d in zip(data.info['columns'], ds)]
 173         else:
 174             out = []
 175             for name,d in zip(data.info['columns'], ds):
 176                 n,units = split_data_label(name)
 177                 out.append(
 178                   (n, ppSI(value=d, unit=units, decimals=2)))
 179         outqueue.put(out)
 180
 181
 182 class ExportCommand (Command):
 183     """Export a :class:`hooke.curve.Curve` data block as TAB-delimeted
 184     ASCII text.
 185
 186     A "#" prefixed header will optionally appear at the beginning of
 187     the file naming the columns.
 188     """
 189     def __init__(self, plugin):
 190         super(ExportCommand, self).__init__(
 191             name='export block',
 192             arguments=[
 193                 CurveArgument,
 194                 Argument(name='block', aliases=['set'], type='int', default=0,
 195                          help="""
 196 Data block to save.  For an approach/retract force curve, `0` selects
 197 the approaching curve and `1` selects the retracting curve.
 198 """.strip()),
 199                 Argument(name='output', type='file', default='curve.dat',
 200                          help="""
 201 File name for the output data.  Defaults to 'curve.dat'
 202 """.strip()),
 203                 Argument(name='header', type='bool', default=True,
 204                          help="""
 205 True if you want the column-naming header line.
 206 """.strip()),
 207                 ],
 208             help=self.__doc__, plugin=plugin)
 209
 210     def _run(self, hooke, inqueue, outqueue, params):
 211         data = params['curve'].data[params['block']]
 212
 213         f = open(params['output'], 'w')
 214         if params['header'] == True:
 215             f.write('# %s \n' % ('\t'.join(data.info['columns'])))
 216         numpy.savetxt(f, data, delimiter='\t')
 217         f.close()
 218
 219 class DifferenceCommand (Command):
 220     """Calculate the difference between two blocks of data.
 221     """
 222     def __init__(self, plugin):
 223         super(DifferenceCommand, self).__init__(
 224             name='block difference',
 225             arguments=[
 226                 CurveArgument,
 227                 Argument(name='block one', aliases=['set one'], type='int',
 228                          default=1,
 229                          help="""
 230 Block A in A-B.  For an approach/retract force curve, `0` selects the
 231 approaching curve and `1` selects the retracting curve.
 232 """.strip()),
 233                 Argument(name='block two', aliases=['set two'], type='int',
 234                          default=0,
 235                          help='Block B in A-B.'),
 236                 Argument(name='x column', type='int', default=0,
 237                          help="""
 238 Column of data to return as x values.
 239 """.strip()),
 240                 Argument(name='y column', type='int', default=1,
 241                          help="""
 242 Column of data block to difference.
 243 """.strip()),
 244                 ],
 245             help=self.__doc__, plugin=plugin)
 246
 247     def _run(self, hooke, inqueue, outqueue, params):
 248         a = params['curve'].data[params['block one']]
 249         b = params['curve'].data[params['block two']]
 250         assert a[:,params['x column']] == b[:,params['x column']]
 251         out = Data((a.shape[0],2), dtype=a.dtype)
 252         out[:,0] = a[:,params['x column']]
 253         out[:,1] = a[:,params['y column']] - b[:,params['y column']]
 254         outqueue.put(out)
 255
 256 class DerivativeCommand (Command):
 257     """Calculate the derivative (actually, the discrete differentiation)
 258     of a curve data block.
 259
 260     See :func:`hooke.util.calculus.derivative` for implementation
 261     details.
 262     """
 263     def __init__(self, plugin):
 264         super(DerivativeCommand, self).__init__(
 265             name='block derivative',
 266             arguments=[
 267                 CurveArgument,
 268                 Argument(name='block', aliases=['set'], type='int', default=0,
 269                          help="""
 270 Data block to differentiate.  For an approach/retract force curve, `0`
 271 selects the approaching curve and `1` selects the retracting curve.
 272 """.strip()),
 273                 Argument(name='x column', type='int', default=0,
 274                          help="""
 275 Column of data block to differentiate with respect to.
 276 """.strip()),
 277                 Argument(name='f column', type='int', default=1,
 278                          help="""
 279 Column of data block to differentiate.
 280 """.strip()),
 281                 Argument(name='weights', type='dict', default={-1:-0.5, 1:0.5},
 282                          help="""
 283 Weighting scheme dictionary for finite differencing.  Defaults to
 284 central differencing.
 285 """.strip()),
 286                 ],
 287             help=self.__doc__, plugin=plugin)
 288
 289     def _run(self, hooke, inqueue, outqueue, params):
 290         data = params['curve'].data[params['block']]
 291         outqueue.put(derivative(
 292                 block, x_col=params['x column'], f_col=params['f column'],
 293                 weights=params['weights']))
 294
 295 class PowerSpectrumCommand (Command):
 296     """Calculate the power spectrum of a data block.
 297     """
 298     def __init__(self, plugin):
 299         super(PowerSpectrumCommand, self).__init__(
 300             name='block power spectrum',
 301             arguments=[
 302                 CurveArgument,
 303                 Argument(name='block', aliases=['set'], type='int', default=0,
 304                          help="""
 305 Data block to act on.  For an approach/retract force curve, `0`
 306 selects the approaching curve and `1` selects the retracting curve.
 307 """.strip()),
 308                 Argument(name='column', type='int', default=1,
 309                          help="""
 310 Column of data block containing time-series data.
 311 """.strip()),
 312                 Argument(name='freq', type='float', default=1.0,
 313                          help="""
 314 Sampling frequency.
 315 """.strip()),
 316                 Argument(name='chunk size', type='int', default=2048,
 317                          help="""
 318 Number of samples per chunk.  Use a power of two.
 319 """.strip()),
 320                 Argument(name='overlap', type='bool', default=False,
 321                          help="""
 322 If `True`, each chunk overlaps the previous chunk by half its length.
 323 Otherwise, the chunks are end-to-end, and not overlapping.
 324 """.strip()),
 325                 ],
 326             help=self.__doc__, plugin=plugin)
 327
 328     def _run(self, hooke, inqueue, outqueue, params):
 329         data = params['curve'].data[params['block']]
 330         outqueue.put(unitary_avg_power_spectrum(
 331                 data[:,params['column']], freq=params['freq'],
 332                 chunk_size=params['chunk size'],
 333                 overlap=params['overlap']))