:mod:`hooke.curve` classes.
"""
+import copy
+
import numpy
from ..command import Command, Argument, Failure
from ..plugin.playlist import current_playlist_callback
from ..util.calculus import derivative
from ..util.fft import unitary_avg_power_spectrum
+from ..util.si import ppSI, join_data_label, split_data_label
+
class CurvePlugin (Builtin):
def __init__(self):
super(CurvePlugin, self).__init__(name='curve')
self._commands = [
- GetCommand(self), InfoCommand(self), ExportCommand(self),
- DifferenceCommand(self), DerivativeCommand(self),
- PowerSpectrumCommand(self)]
+ GetCommand(self), InfoCommand(self), DeltaCommand(self),
+ ExportCommand(self), DifferenceCommand(self),
+ DerivativeCommand(self), PowerSpectrumCommand(self)]
# Define common or complicated arguments
def _get_block_sizes(self, curve):
return [block.shape for block in curve.data]
+
+class DeltaCommand (Command):
+ """Get distance information between two points.
+
+ With two points A and B, the returned distances are A-B.
+ """
+ def __init__(self, plugin):
+ super(DeltaCommand, self).__init__(
+ name='delta',
+ arguments=[
+ CurveArgument,
+ Argument(name='block', type='int', default=0,
+ help="""
+Data block that points are selected from. For an approach/retract
+force curve, `0` selects the approaching curve and `1` selects the
+retracting curve.
+""".strip()),
+ Argument(name='point', type='point', optional=False, count=2,
+ help="""
+Indicies of points bounding the selected data.
+""".strip()),
+ Argument(name='SI', type='bool', default=False,
+ help="""
+Return distances in SI notation.
+""".strip())
+ ],
+ help=self.__doc__, plugin=plugin)
+
+ def _run(self, hooke, inqueue, outqueue, params):
+ data = params['curve'].data[params['block']]
+ As = data[params['point'][0],:]
+ Bs = data[params['point'][1],:]
+ ds = [A-B for A,B in zip(As, Bs)]
+ if params['SI'] == False:
+ out = [(name, d) for name,d in zip(data.info['columns'], ds)]
+ else:
+ out = []
+ for name,d in zip(data.info['columns'], ds):
+ n,units = split_data_label(name)
+ out.append(
+ (n, ppSI(value=d, unit=units, decimals=2)))
+ outqueue.put(out)
+
+
class ExportCommand (Command):
"""Export a :class:`hooke.curve.Curve` data block as TAB-delimeted
ASCII text.
name='export block',
arguments=[
CurveArgument,
- Argument(name='block', aliases=['set'], type='int', default=0,
+ Argument(name='block', type='int', default=0,
help="""
Data block to save. For an approach/retract force curve, `0` selects
the approaching curve and `1` selects the retracting curve.
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
- data = params['curve'].data[int(params['block'])] # HACK, int() should be handled by ui
+ data = params['curve'].data[params['block']]
f = open(params['output'], 'w')
if params['header'] == True:
f.close()
class DifferenceCommand (Command):
- """Calculate the difference between two blocks of data.
+ """Calculate the difference between two columns of data.
+
+ The difference is added to block A as a new column.
+
+ Note that the command will fail if the columns have different
+ lengths, so be careful when differencing columns from different
+ blocks.
"""
def __init__(self, plugin):
super(DifferenceCommand, self).__init__(
- name='block difference',
+ name='difference',
arguments=[
CurveArgument,
- Argument(name='block one', aliases=['set one'], type='int',
- default=1,
+ Argument(name='block A', type='int',
help="""
Block A in A-B. For an approach/retract force curve, `0` selects the
-approaching curve and `1` selects the retracting curve.
+approaching curve and `1` selects the retracting curve. Defaults to
+the first block.
""".strip()),
- Argument(name='block two', aliases=['set two'], type='int',
- default=0,
- help='Block B in A-B.'),
- Argument(name='x column', type='int', default=0,
+ Argument(name='block B', type='int',
help="""
-Column of data block to differentiate with respect to.
+Block B in A-B. Defaults to matching `block A`.
""".strip()),
- Argument(name='f column', type='int', default=1,
+ Argument(name='column A', type='string',
help="""
-Column of data block to differentiate.
+Column of data from block A to difference. Defaults to the first column.
+""".strip()),
+ Argument(name='column B', type='string', default=1,
+ help="""
+Column of data from block B to difference. Defaults to matching `column A`.
+""".strip()),
+ Argument(name='output column name', type='string',
+ help="""
+Name of the new column for storing the difference (without units, defaults to
+`difference of <block A> <column A> and <block B> <column B>`).
""".strip()),
],
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
- a = params['curve'].data[params['block one']]
- b = params['curve'].data[params['block two']]
- assert a[:,params['x column']] == b[:,params['x column']]
- out = Data((a.shape[0],2), dtype=a.dtype)
- out[:,0] = a[:,params['x column']]
- out[:,1] = a[:,params['f column']] - b[:,params['f column']]
- outqueue.put(out)
+ data_A = params['curve'].data[params['block A']]
+ data_B = params['curve'].data[params['block B']]
+ # HACK? rely on params['curve'] being bound to the local hooke
+ # playlist (i.e. not a copy, as you would get by passing a
+ # curve through the queue). Ugh. Stupid queues. As an
+ # alternative, we could pass lookup information through the
+ # queue...
+ new = Data((data_A.shape[0], data_A.shape[1]+1), dtype=data_A.dtype)
+ new.info = copy.deepcopy(data.info)
+ new[:,:-1] = data_A
+
+ a_col = data_A.info['columns'].index(params['column A'])
+ b_col = data_A.info['columns'].index(params['column A'])
+ out = data_A[:,a_col] - data_B[:,b_col]
+
+ a_name,a_units = split_data_label(params['column A'])
+ b_name,b_units = split_data_label(params['column B'])
+ assert a_units == b_units, (
+ 'Unit missmatch: %s != %s' % (a_units, b_units))
+ if params['output column name'] == None:
+ params['output column name'] = (
+ 'difference of %s %s and %s %s' % (
+ block_A.info['name'], params['column A'],
+ block_B.info['name'], params['column B']))
+ new.info['columns'].append(
+ join_data_label(params['output distance column'], a_units))
+ new[:,-1] = out
+ params['curve'].data[params['block A']] = new
+
class DerivativeCommand (Command):
"""Calculate the derivative (actually, the discrete differentiation)
"""
def __init__(self, plugin):
super(DerivativeCommand, self).__init__(
- name='block derivative',
+ name='derivative',
arguments=[
CurveArgument,
- Argument(name='block', aliases=['set'], type='int', default=0,
+ Argument(name='block', type='int', default=0,
help="""
Data block to differentiate. For an approach/retract force curve, `0`
selects the approaching curve and `1` selects the retracting curve.
""".strip()),
- Argument(name='x column', type='int', default=0,
+ Argument(name='x column', type='string',
help="""
Column of data block to differentiate with respect to.
""".strip()),
- Argument(name='f column', type='int', default=1,
+ Argument(name='f column', type='string',
help="""
Column of data block to differentiate.
""".strip()),
help="""
Weighting scheme dictionary for finite differencing. Defaults to
central differencing.
+""".strip()),
+ Argument(name='output column name', type='string',
+ help="""
+Name of the new column for storing the derivative (without units, defaults to
+`derivative of <f column name> with respect to <x column name>`).
""".strip()),
],
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
data = params['curve'].data[params['block']]
- outqueue.put(derivative(
- block, x_col=params['x column'], f_col=params['f column'],
- weights=params['weights']))
+ # HACK? rely on params['curve'] being bound to the local hooke
+ # playlist (i.e. not a copy, as you would get by passing a
+ # curve through the queue). Ugh. Stupid queues. As an
+ # alternative, we could pass lookup information through the
+ # queue...
+ new = Data((data.shape[0], data.shape[1]+1), dtype=data.dtype)
+ new.info = copy.deepcopy(data.info)
+ new[:,:-1] = data
+
+ x_col = data.info['columns'].index(params['x column'])
+ f_col = data.info['columns'].index(params['f column'])
+ d = derivative(
+ block, x_col=x_col, f_col=f_col, weights=params['weights'])
+
+ x_name,x_units = split_data_label(params['x column'])
+ f_name,f_units = split_data_label(params['f column'])
+ if params['output column name'] == None:
+ params['output column name'] = (
+ 'derivative of %s with respect to %s' % (
+ params['f column'], params['x column']))
+
+ new.info['columns'].append(
+ join_data_label(params['output distance column'],
+ '%s/%s' % (f_units/x_units)))
+ new[:,-1] = d[:,1]
+ params['curve'].data[params['block']] = new
+
class PowerSpectrumCommand (Command):
"""Calculate the power spectrum of a data block.
"""
def __init__(self, plugin):
super(PowerSpectrumCommand, self).__init__(
- name='block power spectrum',
+ name='power spectrum',
arguments=[
CurveArgument,
- Argument(name='block', aliases=['set'], type='int', default=0,
+ Argument(name='block', type='int', default=0,
help="""
Data block to act on. For an approach/retract force curve, `0`
selects the approaching curve and `1` selects the retracting curve.
""".strip()),
- Argument(name='f column', type='int', default=1,
+ Argument(name='column', type='string', optional=False,
help="""
-Column of data block to differentiate with respect to.
+Name of the data block column containing to-be-transformed data.
+""".strip()),
+ Argument(name='bounds', type='point', optional=True, count=2,
+ help="""
+Indicies of points bounding the selected data.
""".strip()),
Argument(name='freq', type='float', default=1.0,
help="""
Sampling frequency.
+""".strip()),
+ Argument(name='freq units', type='string', default='Hz',
+ help="""
+Units for the sampling frequency.
""".strip()),
Argument(name='chunk size', type='int', default=2048,
help="""
help="""
If `True`, each chunk overlaps the previous chunk by half its length.
Otherwise, the chunks are end-to-end, and not overlapping.
+""".strip()),
+ Argument(name='output block name', type='string',
+ help="""
+Name of the new data block for storing the power spectrum (defaults to
+`power spectrum of <source block name> <source column name>`).
""".strip()),
],
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
data = params['curve'].data[params['block']]
- outqueue.put(unitary_avg_power_spectrum(
- data[:,params['f column']], freq=params['freq'],
- chunk_size=params['chunk size'],
- overlap=params['overlap']))
+ col = data.info['columns'].index(params['column'])
+ d = data[:,col]
+ if bounds != None:
+ d = d[params['bounds'][0]:params['bounds'][1]]
+ freq_axis,power = unitary_avg_power_spectrum(
+ d, freq=params['freq'],
+ chunk_size=params['chunk size'],
+ overlap=params['overlap'])
+
+ name,data_units = split_data_label(params['column'])
+ b = Data((len(freq_axis),2), dtype=data.dtype)
+ if params['output block name'] == None:
+ params['output block name'] = 'power spectrum of %s %s' % (
+ params['output block name'], data.info['name'], params['column'])
+ b.info['name'] = params['output block name']
+ b.info['columns'] = [
+ join_data_label('frequency axis', params['freq units']),
+ join_data_label('power density',
+ '%s^2/%s' % (data_units, params['freq units'])),
+ ]
+ b[:,0] = freq_axis
+ b[:,1] = power
+ params['curve'].data.append(b)
+ outqueue.put(b)
projects / hooke.git / blobdiff