1 # Copyright (C) 2008-2012 Massimo Sandal <devicerandom@gmail.com>
2 # W. Trevor King <wking@drexel.edu>
4 # This file is part of Hooke.
6 # Hooke is free software: you can redistribute it and/or modify it under the
7 # terms of the GNU Lesser General Public License as published by the Free
8 # Software Foundation, either version 3 of the License, or (at your option) any
11 # Hooke is distributed in the hope that it will be useful, but WITHOUT ANY
12 # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
13 # A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
16 # You should have received a copy of the GNU Lesser General Public License
17 # along with Hooke. If not, see <http://www.gnu.org/licenses/>.
19 """Driver for JPK ForceRobot's velocity clamp data format.
21 This driver is based on JPK's :file:`JPKForceSpec.txt` version 0.12.
22 The specs are freely available from JPK, just email support@jpk.com.
31 from .. import curve as curve
32 from ..util.util import Closing as Closing
33 from ..util.si import join_data_label, split_data_label
34 from . import Driver as Driver
37 def slash_join(*args):
38 r"""Join path components with forward slashes regardless of OS.
42 From the `PKZIP Application Note`_, section J (Explanation of fields):
46 ... All slashes should be forward slashes ``/`` as opposed to
47 backwards slashes ``\`` for compatibility with Amiga and UNIX
50 .. _PKZIP Application Note:
51 http://www.pkware.com/documents/casestudies/APPNOTE.TXT
58 >>> print slash_join('abc', 'def/ghi', 'jkl\\mno')
60 >>> os.path.sep = '\\'
61 >>> print slash_join('abc', 'def/ghi', 'jkl\\mno')
65 Note that when :const:`os.path.sep` is ``/`` (e.g. UNIX),
66 ``def/ghi`` is a compound segment, but when :const:`os.path.sep`
67 is ``\`` (e.g. Windows), ``def/ghi`` is a single segment.
72 return os.path.join(*args)
77 class JPKDriver (Driver):
78 """Handle JPK ForceRobot's data format.
81 super(JPKDriver, self).__init__(name='jpk')
83 def is_me(self, path):
84 if os.path.isdir(path):
86 if zipfile.is_zipfile(path): # JPK file versions since at least 0.5
87 with Closing(zipfile.ZipFile(path, 'r')) as f:
88 if 'header.properties' not in f.namelist():
90 with Closing(f.open('header.properties')) as h:
91 if 'jpk-data-file' in h.read():
94 with Closing(open(path, 'r')) as f:
97 headlines.append(f.readline())
98 if headlines[0].startswith('# xPosition') \
99 and headlines[1].startswith('# yPosition'):
103 def read(self, path, info=None):
106 if zipfile.is_zipfile(path): # JPK file versions since at least 0.5
107 return self._read_zip(path, info)
109 return self._read_old(path, info)
111 def _read_zip(self, path, info):
112 with Closing(zipfile.ZipFile(path, 'r')) as f:
114 zip_info = self._zip_info(f)
115 version = zip_info['file-format-version']
116 if zip_info['jpk-data-file'] == 'jpk-data1D-file':
117 return self._zip_read_1d(
118 f, path, info, zip_info, version)
119 elif zip_info['jpk-data-file'] != 'spm-forcefile':
120 raise ValueError('unrecognized JPK file type "%s"'
121 % zip_info['jpk-data-file'])
123 for i in range(len([p for p in f.namelist()
124 if p.endswith('segment-header.properties')])):
125 segments.append(self._zip_segment(
126 f, path, info, zip_info, i, version))
127 if version not in ['0.%d' % i for i in range(13)]:
128 raise NotImplementedError(
129 'JPK file version %s not supported (yet).' % version)
130 curve_info = self._zip_translate_params(
131 zip_info, segments[0].info['raw info'], version)
132 for segment in segments: # HACK, should use curve-level spring constant
133 for key in ['spring constant (N/m)',
134 'z piezo sensitivity (m/V)']:
135 if key in curve_info:
136 segment.info['spring constant (N/m)'] = \
137 curve_info['spring constant (N/m)']
138 names = [segment.info['name'] for segment in segments]
139 for name in set(names): # ensure unique names
140 count = names.count(name)
143 for j,n in enumerate(names):
145 segments[j].info['name'] += '-%d' % i
147 return (segments, curve_info)
149 def _zip_info(self, zipfile):
150 with Closing(zipfile.open('header.properties')) as f:
151 info = self._parse_params(f.readlines())
154 def _zip_segment(self, zipfile, path, info, zip_info, index, version):
155 with Closing(zipfile.open(slash_join(
156 'segments', str(index), 'segment-header.properties'))
158 prop = self._parse_params(f.readlines())
159 expected_shape = (int(prop['force-segment-header']['num-points']),)
161 if 'list' not in prop['channels']:
162 prop['channels'] = {'list': prop['channels'].split()}
163 for chan in prop['channels']['list']:
164 chan_info = prop['channel'][chan]
165 channels.append(self._zip_channel(
166 zipfile, index, chan, chan_info))
167 if channels[-1].shape != expected_shape:
168 raise NotImplementedError(
169 'channel %d:%s in %s has strange shape %s != %s'
170 % (index, chan, zipfile.path,
171 channels[-1].shape, expected_shape))
172 if len(channels) > 0:
173 shape = (len(channels[0]), len(channels))
174 dtype = channels[0].dtype
175 else: # no channels for this data block
177 dtype = numpy.float32
181 info=self._zip_translate_segment_params(prop))
182 for i,chan in enumerate(channels):
184 return self._zip_scale_segment(d, path, info, version)
186 def _zip_channel(self, zipfile, segment_index, channel_name, chan_info):
187 if chan_info['data']['type'] in ['constant-data', 'raster-data']:
188 return self._zip_calculate_channel(chan_info)
189 with Closing(zipfile.open(slash_join(
190 'segments', str(segment_index),
191 chan_info['data']['file']['name']), 'r')) as f:
192 assert chan_info['data']['file']['format'] == 'raw', \
193 'non-raw data format:\n%s' % pprint.pformat(chan_info)
194 dtype = self._zip_channel_dtype(chan_info)
195 data = numpy.frombuffer(
198 if dtype.kind in ['i', 'u']:
199 data = self._zip_channel_decode(data, chan_info)
202 def _zip_calculate_channel(self, chan_info):
203 type_ = chan_info['data']['type']
204 n = int(chan_info['data']['num-points'])
205 if type_ == 'constant-data':
206 return float(chan_info['data']['value'])*numpy.ones(
209 elif type_ == 'raster-data':
210 start = float(chan_info['data']['start'])
211 step = float(chan_info['data']['step'])
214 stop=start + step*(n-0.5),
218 raise ValueError('unrecognized data format "%s"' % type_)
220 def _zip_channel_dtype(self, chan_info):
221 type_ = chan_info['data']['type']
222 if type_ in ['float-data', 'float']:
223 dtype = numpy.dtype(numpy.float32)
224 elif type_ in ['integer-data', 'memory-integer-data']:
225 encoder = chan_info['data']['encoder']['type']
226 if encoder in ['signedinteger', 'signedinteger-limited']:
227 dtype = numpy.dtype(numpy.int32)
228 elif encoder in ['unsignedinteger', 'unsignedinteger-limited']:
229 dtype = numpy.dtype(numpy.uint32)
231 raise ValueError('unrecognized encoder type "%s" for "%s" data'
233 elif type_ in ['short-data', 'short', 'memory-short-data']:
234 encoder = chan_info['data']['encoder']['type']
235 if encoder in ['signedshort', 'signedshort-limited']:
236 dtype = numpy.dtype(numpy.int16)
237 elif encoder in ['unsignedshort', 'unsignedshort-limited']:
238 dtype = numpy.dtype(numpy.uint16)
240 raise ValueError('unrecognized encoder type "%s" for "%s" data'
243 raise ValueError('unrecognized data format "%s"' % type_)
245 # '>' (big endian) byte order.
246 # From version 0.3 of JPKForceSpec.txt in the "Binary data" section:
247 # All forms of raw data are stored in chronological order
248 # (the order in which they were collected), and the
249 # individual values are stored in network byte order
250 # (big-endian). The data type used to store the data is
251 # specified by the "channel.*.data.type" property, and is
252 # either short (2 bytes per value), integer (4 bytes), or
253 # float (4 bytes, IEEE format).
254 return dtype.newbyteorder(byte_order)
256 def _zip_channel_decode(self, data, encoder_info):
257 return self._zip_apply_channel_scaling(
258 data, encoder_info['data']['encoder'])
260 def _zip_translate_params(self, params, chan_info, version):
263 #'time':self._time_from_TODO(raw_info[]),
265 if len(chan_info['channels']['list']) == 0:
267 force_unit = self._zip_unit(
268 chan_info['channel']['vDeflection']['conversion-set']['conversion']['force']['scaling'],
270 assert force_unit == 'N', force_unit
271 force_base = chan_info['channel']['vDeflection']['conversion-set']['conversion']['force']['base-calibration-slot']
272 assert force_base == 'distance', force_base
273 dist_unit = self._zip_unit(
274 chan_info['channel']['vDeflection']['conversion-set']['conversion']['distance']['scaling'],
276 assert dist_unit == 'm', dist_unit
277 distance_base = chan_info['channel']['vDeflection']['conversion-set']['conversion']['distance']['base-calibration-slot']
278 assert distance_base == 'volts', distance_base
279 base_conversion = chan_info['channel']['vDeflection']['conversion-set']['conversions']['base']
280 assert base_conversion == distance_base, base_conversion
281 if 'encoder' in chan_info['channel']['vDeflection']['data']:
282 distance_base_unit = self._zip_unit(
283 chan_info['channel']['vDeflection']['data']['encoder']['scaling'],
286 distance_base_unit = self._zip_unit(
287 chan_info['channel']['vDeflection']['data'],
289 assert distance_base_unit == 'V', distance_base_unit
291 chan_info['channel']['vDeflection']['conversion-set']['conversion']['force']['scaling']['multiplier'])
293 chan_info['channel']['vDeflection']['conversion-set']['conversion']['distance']['scaling']['multiplier'])
294 info['spring constant (N/m)'] = force_mult
295 info['z piezo sensitivity (m/V)'] = sens_mult
298 def _zip_translate_segment_params(self, params):
301 'columns': list(params['channels']['list']),
302 'name': self._zip_segment_name(params),
306 def _zip_segment_name(self, params):
307 name = params['force-segment-header']['name']['name']
308 if name.endswith('-spm'):
309 name = name[:-len('-spm')]
312 elif name.startswith('pause-at-'):
316 def _zip_scale_segment(self, segment, path, info, version):
322 segment.info['raw data'] = data
325 channels = segment.info['raw info']['channels']['list']
326 for i,channel in enumerate(channels):
328 if channel == 'vDeflection':
329 conversion = 'distance'
330 segment = self._zip_scale_channel(
331 segment, channel, conversion=conversion,
332 path=path, info=info, version=version)
333 name,unit = split_data_label(segment.info['columns'][i])
334 if name == 'vDeflection':
335 assert unit == 'm', segment.info['columns'][i]
336 segment.info['columns'][i] = join_data_label('deflection', 'm')
337 # Invert because deflection voltage increases as the
338 # tip moves away from the surface, but it makes more
339 # sense to me to have it increase as it moves toward
340 # the surface (positive tension on the protein chain).
342 elif name == 'height':
343 assert unit == 'm', segment.info['columns'][i]
344 segment.info['columns'][i] = join_data_label('z piezo', 'm')
347 def _zip_scale_channel(self, segment, channel_name,
348 conversion=None, path=None, info={}, version=None):
349 channel = segment.info['raw info']['channels']['list'].index(
351 conversion_set = segment.info['raw info']['channel'][channel_name]['conversion-set']
352 if conversion == None:
353 conversion = conversion_set['conversions']['default']
354 if conversion == conversion_set['conversions']['base']:
355 segment.info['columns'][channel] = join_data_label(
358 segment.info['raw info']['channel'][channel_name]['data'],
361 conversion_info = conversion_set['conversion'][conversion]
362 if conversion_info['base-calibration-slot'] \
363 != conversion_set['conversions']['base']:
364 # Our conversion is stacked on a previous conversion. Do
365 # the previous conversion first.
366 segment = self._zip_scale_channel(
367 segment, channel_name,
368 conversion_info['base-calibration-slot'],
369 path=path, info=info, version=version)
370 if conversion_info['type'] == 'file':
371 # Michael Haggerty at JPK points out that the conversion
372 # information stored in the external file is reproduced in
373 # the force curve file. So there is no need to actually
374 # read `conversion_info['file']`. In fact, the data there
375 # may have changed with future calibrations, while the
376 # information stored directly in conversion_info retains
377 # the calibration information as it was when the experiment
379 pass # Fall through to 'simple' conversion processing.
381 assert conversion_info['type'] == 'simple', conversion_info['type']
382 segment[:,channel] = self._zip_apply_channel_scaling(
383 segment[:,channel], conversion_info)
384 unit = self._zip_unit(conversion_info['scaling'], version)
385 segment.info['columns'][channel] = join_data_label(channel_name, unit)
388 def _zip_apply_channel_scaling(self, channel_data, conversion_info):
389 assert conversion_info['scaling']['type'] == 'linear', \
390 conversion_info['scaling']['type']
391 assert conversion_info['scaling']['style'] == 'offsetmultiplier', \
392 conversion_info['scaling']['style']
393 multiplier = float(conversion_info['scaling']['multiplier'])
394 offset = float(conversion_info['scaling']['offset'])
395 return channel_data * multiplier + offset
397 def _zip_unit(self, conversion_info, version):
398 if version in ['0.%d' % i for i in range(3)]:
399 return conversion_info['unit']
401 return conversion_info['unit']['unit']
403 def _parse_params(self, lines):
407 if line.startswith('#'):
410 # e.g.: force-segment-header.type=xy-position-segment-header
411 fields = line.split('=', 1)
412 assert len(fields) == 2, line
413 setting = fields[0].split('.')
414 sub_info = info # drill down, e.g. info['force-s..']['type']
415 for s in setting[:-1]:
416 if s not in sub_info:
418 sub_info = sub_info[s]
419 if setting[-1] == 'list': # split a space-delimited list
421 sub_info[setting[-1]] = fields[1].split(' ')
423 sub_info[setting[-1]] = []
425 sub_info[setting[-1]] = fields[1]
428 def _zip_read_1d(self, zipfile, path, info, zip_info, version):
429 expected_shape = (int(zip_info['data']['num-points']),)
430 if zip_info['data']['type'] in ['constant-data', 'raster-data']:
431 return self._zip_calculate_channel(zip_info)
432 with Closing(zipfile.open(
433 zip_info['data']['file']['name'], 'r')) as f:
434 assert zip_info['data']['file']['format'] == 'raw', \
435 'non-raw data format:\n%s' % pprint.pformat(chan_info)
436 dtype = self._zip_channel_dtype(zip_info)
437 data = numpy.frombuffer(
440 if dtype.kind in ['i', 'u']:
441 data = self._zip_channel_decode(data, zip_info)
442 if data.shape != expected_shape:
443 raise NotImplementedError(
444 'channel %s has strange shape %s != %s'
445 % (path, data.shape, expected_shape))
453 def _read_old(self, path, info):
454 raise NotImplementedError(
455 "Early JPK files (pre-zip) are not supported by Hooke. Please "
456 "use JPK's `out2jpk-force` script to convert your old files "
457 "(%s) to a more recent format before loading them with Hooke."