# -*- coding: utf-8 -*-
#
-# Copyright (C) 2008-2010 Alberto Gomez-Casado
+# Copyright (C) 2008-2011 Alberto Gomez-Casado
# Fabrizio Benedetti
# Massimo Sandal <devicerandom@gmail.com>
# W. Trevor King <wking@drexel.edu>
import numpy
from scipy.optimize import newton
-from ..command import Command, Argument, Failure
+from ..command import Command, Argument, Success, Failure
from ..config import Setting
from ..curve import Data
-from ..plugin import Plugin, argument_to_setting
from ..util.callback import is_iterable
from ..util.fit import PoorFit, ModelFitter
+from ..util.peak import Peak
from ..util.si import join_data_label, split_data_label
-from .curve import CurveArgument
-from .vclamp import scale
+from . import Plugin, argument_to_setting
+from .curve import ColumnAccessCommand, ColumnAddingCommand
kB = 1.3806503e-23 # Joules/Kelvin
... }
>>> model = FJC(d_data, info=info, rescale=True)
>>> outqueue = Queue()
- >>> Lp,a = model.fit(outqueue=outqueue)
+ >>> L,a = model.fit(outqueue=outqueue)
>>> fit_info = outqueue.get(block=False)
- >>> model.L(Lp) # doctest: +ELLIPSIS
- 3.500...e-08
- >>> a # doctest: +ELLIPSIS
- 2.499...e-10
+ >>> print L
+ 3.5e-08
+ >>> print a
+ 2.5e-10
Fit the example data with a one-parameter fit (`L`). We introduce
some error in our fixed Kuhn length for fun.
>>> info['Kuhn length (m)'] = 2*a
>>> model = FJC(d_data, info=info, rescale=True)
- >>> Lp = model.fit(outqueue=outqueue)
+ >>> L, = model.fit(outqueue=outqueue)
>>> fit_info = outqueue.get(block=False)
- >>> model.L(Lp) # doctest: +ELLIPSIS
+ >>> print L # doctest: +ELLIPSIS
3.199...e-08
"""
def Lp(self, L):
self._model_data[:] = FJC_fn(x_data, T, L, a)
return self._model_data
+ def fit(self, *args, **kwargs):
+ params = super(FJC, self).fit(*args, **kwargs)
+ params[0] = self.L(params[0]) # convert Lp -> L
+ if len(params) > 1:
+ params[1] = abs(params[1]) # take the absolute value of `a`
+ return params
+
def guess_initial_params(self, outqueue=None):
"""Guess initial fitting parameters.
return [Lp]
return [Lp, a]
- def guess_scale(self, params, outqueue=None):
- """Guess parameter scales.
-
- Returns
- -------
- Lp_scale : float
- A guess at the reparameterized contour length scale in meters.
- a_scale : float (optional)
- A guess at the Kuhn length in meters. If the length of
- `params` is less than 2, `a_scale` is not returned.
- """
- Lp_scale = 1.0
- if len(params) == 1:
- return [Lp_scale]
- return [Lp_scale] + [x/10.0 for x in params[1:]]
-
def inverse_FJC_PEG_fn(F_data, T=300, N=1, k=150., Lp=3.58e-10, Lh=2.8e-10, dG=3., a=7e-10):
"""Inverse poly(ethylene-glycol) adjusted extended FJC model.
... }
>>> model = FJC_PEG(d_data, info=info, rescale=True)
>>> outqueue = Queue()
- >>> Nr,a = model.fit(outqueue=outqueue)
+ >>> N,a = model.fit(outqueue=outqueue)
>>> fit_info = outqueue.get(block=False)
- >>> model.L(Nr) # doctest: +ELLIPSIS
- 122.999...
- >>> a # doctest: +ELLIPSIS
- 6.999...e-10
+ >>> print N
+ 123.0
+ >>> print a
+ 7e-10
Fit the example data with a one-parameter fit (`N`). We introduce
some error in our fixed Kuhn length for fun.
>>> info['Kuhn length (m)'] = 2*kwargs['a']
>>> model = FJC_PEG(d_data, info=info, rescale=True)
- >>> Nr = model.fit(outqueue=outqueue)
+ >>> N, = model.fit(outqueue=outqueue)
>>> fit_info = outqueue.get(block=False)
- >>> model.L(Nr) # doctest: +ELLIPSIS
+ >>> print N # doctest: +ELLIPSIS
96.931...
"""
def Lr(self, L):
self._model_data[:] = FJC_PEG_fn(x_data, N=N, a=a, **kwargs)
return self._model_data
+ def fit(self, *args, **kwargs):
+ params = super(FJC_PEG, self).fit(*args, **kwargs)
+ params[0] = self.L(params[0]) # convert Nr -> N
+ if len(params) > 1:
+ params[1] = abs(params[1]) # take the absolute value of `a`
+ return params
+
def guess_initial_params(self, outqueue=None):
"""Guess initial fitting parameters.
return [Nr]
return [Nr, a]
- def guess_scale(self, params, outqueue=None):
- """Guess parameter scales.
-
- Returns
- -------
- N_scale : float
- A guess at the section count scale in meters.
- a_scale : float (optional)
- A guess at the Kuhn length in meters. If the length of
- `params` is less than 2, `a_scale` is not returned.
- """
- return [x/10.0 for x in params]
-
def WLC_fn(x_data, T, L, p):
"""The worm like chain interpolation model.
... }
>>> model = WLC(d_data, info=info, rescale=True)
>>> outqueue = Queue()
- >>> Lp,p = model.fit(outqueue=outqueue)
+ >>> L,p = model.fit(outqueue=outqueue)
>>> fit_info = outqueue.get(block=False)
- >>> model.L(Lp) # doctest: +ELLIPSIS
- 3.500...e-08
- >>> p # doctest: +ELLIPSIS
- 2.500...e-10
+ >>> print L
+ 3.5e-08
+ >>> print p
+ 2.5e-10
Fit the example data with a one-parameter fit (`L`). We introduce
some error in our fixed persistence length for fun.
>>> info['persistence length (m)'] = 2*p
>>> model = WLC(d_data, info=info, rescale=True)
- >>> Lp = model.fit(outqueue=outqueue)
+ >>> L, = model.fit(outqueue=outqueue)
>>> fit_info = outqueue.get(block=False)
- >>> model.L(Lp) # doctest: +ELLIPSIS
+ >>> print L # doctest: +ELLIPSIS
3.318...e-08
"""
def Lp(self, L):
self._model_data[:] = WLC_fn(x_data, T, L, p)
return self._model_data
+ def fit(self, *args, **kwargs):
+ params = super(WLC, self).fit(*args, **kwargs)
+ params[0] = self.L(params[0]) # convert Lp -> L
+ if len(params) > 1:
+ params[1] = abs(params[1]) # take the absolute value of `p`
+ return params
+
def guess_initial_params(self, outqueue=None):
"""Guess initial fitting parameters.
return [Lp]
return [Lp, p]
- def guess_scale(self, params, outqueue=None):
- """Guess parameter scales.
-
- Returns
- -------
- Lp_scale : float
- A guess at the reparameterized contour length scale in meters.
- p_scale : float (optional)
- A guess at the persistence length in meters. If the
- length of `params` is less than 2, `p_scale` is not
- returned.
- """
- Lp_scale = 1.0
- if len(params) == 1:
- return [Lp_scale]
- return [Lp_scale] + [x/10.0 for x in params[1:]]
-
class PolymerFitPlugin (Plugin):
"""Polymer model (WLC, FJC, etc.) fitting.
self._settings.append(argument_to_setting(
self.setting_section, argument))
argument.default = None # if argument isn't given, use the config.
- self._arguments.extend([ # Non-configurable arguments
- Argument(name='input distance column', type='string',
- default='cantilever adjusted extension (m)',
- help="""
+ self._input_columns = [
+ ('distance column', 'cantilever adjusted extension (m)',
+ """
Name of the column to use as the surface position input.
""".strip()),
- Argument(name='input deflection column', type='string',
- default='deflection (N)',
- help="""
+ ('deflection column', 'deflection (N)',
+ """
Name of the column to use as the deflection input.
""".strip()),
- ])
+ ]
self._commands = [
PolymerFitCommand(self), PolymerFitPeaksCommand(self),
+ TranslateFlatPeaksCommand(self),
]
def dependencies(self):
return self._settings
-class PolymerFitCommand (Command):
+class PolymerFitCommand (ColumnAddingCommand):
"""Polymer model (WLC, FJC, etc.) fitting.
Fits an entropic elasticity function to a given chunk of the
def __init__(self, plugin):
super(PolymerFitCommand, self).__init__(
name='polymer fit',
+ columns=plugin._input_columns,
+ new_columns=[
+ ('output tension column', 'polymer tension',
+ """
+Name of the column (without units) to use as the polymer tension output.
+""".strip()),
+ ],
arguments=[
- CurveArgument,
- Argument(name='block', aliases=['set'], type='int', default=0,
+ Argument(name='fit parameters info name', type='string',
+ default='polymer fit',
help="""
-Data block for which the fit should be calculated. For an
-approach/retract force curve, `0` selects the approaching curve and
-`1` selects the retracting curve.
+Name (without units) for storing the fit parameters in the `.info` dictionary.
""".strip()),
Argument(name='bounds', type='point', optional=False, count=2,
help="""
Indicies of points bounding the selected data.
""".strip()),
- ] + plugin._arguments + [
- Argument(name='output tension column', type='string',
- default='polymer tension',
- help="""
-Name of the column (without units) to use as the polymer tension output.
-""".strip()),
- Argument(name='fit parameters info name', type='string',
- default='surface deflection offset',
- help="""
-Name (without units) for storing the fit parameters in the `.info` dictionary.
-""".strip()),
- ],
+ ] + plugin._arguments,
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
- print params['curve'], id(params['curve']), 'PFC'
- for key,value in params.items():
- if value == None: # Use configured default value.
- params[key] = self.plugin.config[key]
- scale(hooke, params['curve'], params['block']) # TODO: is autoscaling a good idea? (explicit is better than implicit)
- data = params['curve'].data[params['block']]
- # 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...
+ params = self._setup_params(hooke, params)
+ data = self._block(hooke, params)
model = params['polymer model']
- new = Data((data.shape[0], data.shape[1]+1), dtype=data.dtype)
- new.info = copy.deepcopy(data.info)
- new[:,:-1] = data
- new.info['columns'].append(
- join_data_label(params['output tension column'], 'N'))
- print 'add column', id(params['curve']), new.info['columns'][-1], 'to', new.info['columns'][:-1]
- z_data = data[:,data.info['columns'].index(
- params['input distance column'])]
- d_data = data[:,data.info['columns'].index(
- params['input deflection column'])]
+ dist_data = self._get_column(
+ hooke=hooke, params=params, column_name='distance column')
+ def_data = self._get_column(
+ hooke=hooke, params=params, column_name='deflection column')
start,stop = params['bounds']
tension_data,ps = self.fit_polymer_model(
- params, z_data, d_data, start, stop, outqueue)
- new.info[params['fit parameters info name']] = ps
- new.info[params['fit parameters info name']]['model'] = model
- print 'add info', params['fit parameters info name']
- new[:,-1] = tension_data
- params['curve'].data[params['block']] = new
-
- def fit_polymer_model(self, params, z_data, d_data, start, stop,
+ params, dist_data, def_data, start, stop, outqueue)
+ data.info[params['fit parameters info name']] = ps
+ data.info[params['fit parameters info name']]['model'] = model
+ self._set_column(hooke=hooke, params=params,
+ column_name='output tension column',
+ values=tension_data)
+
+ def _setup_params(self, hooke, params):
+ for key,value in params.items():
+ if value == None: # Use configured default value.
+ params[key] = self.plugin.config[key]
+ name,def_unit = split_data_label(params['deflection column'])
+ params['output tension column'] = join_data_label(
+ params['output tension column'], def_unit)
+ return params
+
+ def fit_polymer_model(self, params, dist_data, def_data, start, stop,
outqueue=None):
"""Railyard for the `fit_*_model` family.
"""
fn = getattr(self, 'fit_%s_model'
% params['polymer model'].replace('-','_'))
- return fn(params, z_data, d_data, start, stop, outqueue)
+ return fn(params, dist_data, def_data, start, stop, outqueue)
- def fit_FJC_model(self, params, z_data, d_data, start, stop,
+ def fit_FJC_model(self, params, dist_data, def_data, start, stop,
outqueue=None):
"""Fit the data with :class:`FJC`.
"""
info = {
'temperature (K)': self.plugin.config['temperature'],
- 'x data (m)': z_data[start:stop],
+ 'x data (m)': dist_data[start:stop],
}
if True: # TODO: optionally free persistence length
info['Kuhn length (m)'] = (
params['FJC Kuhn length'])
- model = FJC(d_data[start:stop], info=info, rescale=True)
+ model = FJC(def_data[start:stop], info=info, rescale=True)
queue = Queue()
params = model.fit(outqueue=queue)
if True: # TODO: if Kuhn length fixed
- params = [params]
a = info['Kuhn length (m)']
else:
a = params[1]
- Lp = params[0]
- L = model.L(Lp)
+ L = params[0]
T = info['temperature (K)']
fit_info = queue.get(block=False)
- mask = numpy.zeros(z_data.shape, dtype=numpy.bool)
- mask[start:stop] = True
- return [FJC_fn(z_data, T=T, L=L, a=a) * mask,
- fit_info]
+ f_data = numpy.ones(dist_data.shape, dtype=dist_data.dtype) * numpy.nan
+ f_data[start:stop] = FJC_fn(dist_data[start:stop], T=T, L=L, a=a)
+ return [f_data, fit_info]
- def fit_FJC_PEG_model(self, params, z_data, d_data, start, stop,
+ def fit_FJC_PEG_model(self, params, dist_data, def_data, start, stop,
outqueue=None):
"""Fit the data with :class:`FJC_PEG`.
"""
info = {
'temperature (K)': self.plugin.config['temperature'],
- 'x data (m)': z_data[start:stop],
+ 'x data (m)': dist_data[start:stop],
# TODO: more info
}
if True: # TODO: optionally free persistence length
info['Kuhn length (m)'] = (
params['FJC Kuhn length'])
- model = FJC(d_data[start:stop], info=info, rescale=True)
+ model = FJC_PEG(def_data[start:stop], info=info, rescale=True)
queue = Queue()
params = model.fit(outqueue=queue)
if True: # TODO: if Kuhn length fixed
- params = [params]
a = info['Kuhn length (m)']
else:
a = params[1]
- Nr = params[0]
- N = model.L(Nr)
+ N = params[0]
T = info['temperature (K)']
fit_info = queue.get(block=False)
- mask = numpy.zeros(z_data.shape, dtype=numpy.bool)
- mask[start:stop] = True
- return [FJC_PEG_fn(z_data, **kwargs) * mask,
- fit_info]
+ f_data = numpy.ones(dist_data.shape, dtype=dist_data.dtype) * numpy.nan
+ f_data[start:stop] = FJC_PEG_fn(dist_data[start:stop], **kwargs)
+ return [f_data, fit_info]
- def fit_WLC_model(self, params, z_data, d_data, start, stop,
+ def fit_WLC_model(self, params, dist_data, def_data, start, stop,
outqueue=None):
"""Fit the data with :class:`WLC`.
"""
info = {
'temperature (K)': self.plugin.config['temperature'],
- 'x data (m)': z_data[start:stop],
+ 'x data (m)': dist_data[start:stop],
}
if True: # TODO: optionally free persistence length
info['persistence length (m)'] = (
params['WLC persistence length'])
- model = WLC(d_data[start:stop], info=info, rescale=True)
+ model = WLC(def_data[start:stop], info=info, rescale=True)
queue = Queue()
params = model.fit(outqueue=queue)
if True: # TODO: if persistence length fixed
- params = [params]
p = info['persistence length (m)']
else:
p = params[1]
- Lp = params[0]
- L = model.L(Lp)
+ info['persistence length (m)'] = p
+ L = params[0]
+ info['contour length (m)'] = L
T = info['temperature (K)']
fit_info = queue.get(block=False)
- mask = numpy.zeros(z_data.shape, dtype=numpy.bool)
- mask[start:stop] = True
- return [WLC_fn(z_data, T=T, L=L, p=p) * mask,
- fit_info]
+ info['fit'] = fit_info
+ info.pop('x data (m)')
+ f_data = numpy.ones(dist_data.shape, dtype=dist_data.dtype) * numpy.nan
+ f_data[start:stop] = WLC_fn(dist_data[start:stop], T=T, L=L, p=p)
+ return [f_data, info]
-
-class PolymerFitPeaksCommand (Command):
+class PolymerFitPeaksCommand (ColumnAccessCommand):
"""Polymer model (WLC, FJC, etc.) fitting.
Use :class:`PolymerFitCommand` to fit the each peak in a list of
def __init__(self, plugin):
super(PolymerFitPeaksCommand, self).__init__(
name='polymer fit peaks',
+ columns=plugin._input_columns,
arguments=[
- CurveArgument,
- Argument(name='block', aliases=['set'], type='int', default=0,
- help="""
-Data block for which the fit should be calculated. For an
-approach/retract force curve, `0` selects the approaching curve and
-`1` selects the retracting curve.
-""".strip()),
Argument(name='peak info name', type='string',
- default='flat filter peaks',
+ default='polymer peaks',
help="""
-Name for storing the distance offset in the `.info` dictionary.
+Name for the list of peaks in the `.info` dictionary.
""".strip()),
Argument(name='peak index', type='int', count=-1, default=None,
help="""
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
- print params['curve'], id(params['curve']), 'PFPC'
- data = params['curve'].data[params['block']]
- fit_command = [c for c in hooke.commands
- if c.name=='polymer fit'][0]
+ data = self._block(hooke, params)
+ fit_command = hooke.command_by_name['polymer fit']
inq = Queue()
outq = Queue()
+ curve = params['curve']
+ params['curve'] = None
p = copy.deepcopy(params)
p['curve'] = params['curve']
del(p['peak info name'])
p['output tension column'] = peak.name
p['fit parameters info name'] = peak.name
fit_command.run(hooke, inq, outq, **p)
+ ret = outq.get()
+ if not isinstance(ret, Success):
+ raise ret
+
+
+class TranslateFlatPeaksCommand (ColumnAccessCommand):
+ """Translate flat filter peaks into polymer peaks for fitting.
+
+ Use :class:`~hooke.plugin.flatfilt.FlatPeaksCommand` creates a
+ list of peaks for regions with large derivatives. For velocity
+ clamp measurements, these regions are usually the rebound phase
+ after a protein domain unfolds, the cantilever detaches, etc.
+ Because these features occur after the polymer loading phase, we
+ need to shift the selected regions back to align them with the
+ polymer loading regions.
+ """
+ def __init__(self, plugin):
+ super(TranslateFlatPeaksCommand, self).__init__(
+ name='flat peaks to polymer peaks',
+ columns=plugin._input_columns,
+ arguments=[
+ Argument(name='input peak info name', type='string',
+ default='flat filter peaks',
+ help="""
+Name for the input peaks in the `.info` dictionary.
+""".strip()),
+ Argument(name='output peak info name', type='string',
+ default='polymer peaks',
+ help="""
+Name for the ouptput peaks in the `.info` dictionary.
+""".strip()),
+ Argument(name='end offset', type='int', default=-1,
+ help="""
+Number of points between the end of a new peak and the start of the old.
+""".strip()),
+ Argument(name='start fraction', type='float', default=0.2,
+ help="""
+Place the start of the new peak at `start_fraction` from the end of
+the previous old peak to the end of the new peak. Because the first
+new peak will have no previous old peak, it uses a distance of zero
+instead.
+""".strip()),
+ ] + plugin._arguments,
+ help=self.__doc__, plugin=plugin)
+
+ def _run(self, hooke, inqueue, outqueue, params):
+ data = self._block(hooke, params)
+ dist_data = self._get_column(
+ hooke=hooke, params=params, column_name='distance column')
+ def_data = self._get_column(
+ hooke=hooke, params=params, column_name='deflection column')
+ previous_old_stop = numpy.absolute(dist_data).argmin()
+ new = []
+ try:
+ peaks = data.info[params['input peak info name']]
+ except KeyError, e:
+ raise Failure('No value for %s in %s.info (%s): %s'
+ % (params['input peak info name'], data.info['name'],
+ sorted(data.info.keys()), e))
+ for i,peak in enumerate(peaks):
+ next_old_start = peak.index
+ stop = next_old_start + params['end offset']
+ dist_start = (
+ dist_data[previous_old_stop]
+ + params['start fraction']*(
+ dist_data[stop] - dist_data[previous_old_stop]))
+ start = numpy.absolute(dist_data - dist_start).argmin()
+ p = Peak('polymer peak %d' % i,
+ index=start,
+ values=def_data[start:stop])
+ new.append(p)
+ previous_old_stop = peak.post_index()
+ data.info[params['output peak info name']] = new
+
# TODO:
# def dist2fit(self):
projects / hooke.git / blobdiff