from numpy import diff
from scipy.signal.signaltools import medfilt
-from ..command import Command, Argument, Success, Failure
+from ..command import Command, Argument, Success, Failure, UncaughtException
from ..config import Setting
from ..experiment import VelocityClamp
from ..plugin import Plugin, argument_to_setting
from ..plugin.curve import CurveArgument
from ..plugin.playlist import FilterCommand
from ..plugin.vclamp import scale
+from ..util.fit import PoorFit
from ..util.peak import find_peaks, find_peaks_arguments, Peak, _kwargs
help=self.__doc__, plugin=plugin)
def _run(self, hooke, inqueue, outqueue, params):
- z_data,d_data,params = self._setup(params)
+ z_data,d_data,params = self._setup(hooke, params)
start_index = 0
- while z_data[start_index] < params['blind window']:
+ while (start_index < len(z_data)
+ and z_data[start_index] < params['blind window']):
start_index += 1
median = medfilt(d_data[start_index:], params['median window'])
deriv = diff(median)
peak.index += start_index
outqueue.put(peaks)
- def _setup(self, params):
+ def _setup(self, hooke, params):
"""Setup `params` from config and return the z piezo and
deflection arrays.
"""
if curve.info['experiment'] != VelocityClamp:
raise Failure('%s operates on VelocityClamp experiments, not %s'
% (self.name, curve.info['experiment']))
- for col in ['surface z piezo (m)', 'deflection (N)']:
+ for col in ['surface distance (m)', 'deflection (N)']:
if col not in curve.data[0].info['columns']:
- scale(curve)
+ scale(hooke, curve)
data = None
- for block in curve.data:
+ for i,block in enumerate(curve.data):
if block.info['name'].startswith('retract'):
data = block
break
if data == None:
raise Failure('No retraction blocks in %s.' % curve)
- z_data = data[:,data.info['columns'].index('surface z piezo (m)')]
+ z_data = data[:,data.info['columns'].index('surface distance (m)')]
if 'flattened deflection (N)' in data.info['columns']:
params['deflection column name'] = 'flattened deflection (N)'
else:
self.arguments.extend(plugin._arguments)
def filter(self, curve, hooke, inqueue, outqueue, params):
+ params['curve'] = curve
inq = Queue()
outq = Queue()
filt_command = [c for c in hooke.commands
if c.name=='flat filter peaks'][0]
filt_command.run(hooke, inq, outq, **params)
peaks = outq.get()
+ if isinstance(peaks, UncaughtException) \
+ and isinstance(peaks.exception, PoorFit):
+ return False
if not (isinstance(peaks, list) and (len(peaks) == 0
or isinstance(peaks[0], Peak))):
raise Failure('Expected a list of Peaks, not %s' % peaks)
ret = outq.get()
if not isinstance(ret, Success):
raise ret
+ if params['min peaks'] == None: # Use configured default value.
+ params['min peaks'] = self.plugin.config['min peaks']
return len(peaks) >= int(params['min peaks'])