30fabdf0c0817a162687fcc32e3489b5110f249b
[hooke.git] / hooke / driver / jpk.py
1 # Copyright (C) 2008-2010 Massimo Sandal <devicerandom@gmail.com>
2 #                         W. Trevor King <wking@drexel.edu>
3 #
4 # This file is part of Hooke.
5 #
6 # Hooke is free software: you can redistribute it and/or modify it
7 # under the terms of the GNU Lesser General Public License as
8 # published by the Free Software Foundation, either version 3 of the
9 # License, or (at your option) any later version.
10 #
11 # Hooke is distributed in the hope that it will be useful, but WITHOUT
12 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
13 # or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General
14 # Public License for more details.
15 #
16 # You should have received a copy of the GNU Lesser General Public
17 # License along with Hooke.  If not, see
18 # <http://www.gnu.org/licenses/>.
19
20 """Driver for JPK ForceRobot's velocity clamp data format.
21 """
22
23 import os.path
24 import pprint
25 import zipfile
26
27 import numpy
28
29 from .. import curve as curve
30 from .. import experiment as experiment
31 from ..util.util import Closing as Closing
32 from . import Driver as Driver
33
34
35 class JPKDriver (Driver):
36     """Handle JPK ForceRobot's data format.
37     """
38     def __init__(self):
39         super(JPKDriver, self).__init__(name='jpk')
40
41     def is_me(self, path):
42         if os.path.isdir(path):
43             return False
44         if zipfile.is_zipfile(path):  # JPK file versions since at least 0.5
45             with Closing(zipfile.ZipFile(path, 'r')) as f:
46                 if 'header.properties' not in f.namelist():
47                     return False
48                 with Closing(f.open('header.properties')) as h:
49                     if 'jpk-data-file' in h.read():
50                         return True
51         else:
52             with Closing(open(path, 'r')) as f:
53                 headlines = []
54                 for i in range(3):
55                     headlines.append(f.readline())
56             if headlines[0].startswith('# xPosition') \
57                     and headlines[1].startswith('# yPosition'):
58                 return True
59         return False
60
61     def read(self, path, info=None):
62         if info == None:
63             info = {}
64         if zipfile.is_zipfile(path):  # JPK file versions since at least 0.5
65             return self._read_zip(path, info)
66         else:
67             return self._read_old(path, info)
68
69     def _read_zip(self, path, info):
70         with Closing(zipfile.ZipFile(path, 'r')) as f:
71             f.path = path
72             zip_info = self._zip_info(f)
73             segments = []
74             for i in range(len([p for p in f.namelist()
75                                 if p.endswith('segment-header.properties')])):
76                 segments.append(self._zip_segment(f, path, info, zip_info, i))
77         if zip_info['file-format-version'] not in ['0.5']:
78             raise NotImplementedError(
79                 'JPK file version %s not supported (yet).'
80                 % zip_info['file-format-version'])
81         for name in ['approach', 'retract']:
82             if len([s for s in segments if s.info['name'] == name]) == 0:
83                 raise ValueError(
84                     'No segment for %s in %s, only %s'
85                     % (name, path, [s.info['name'] for s in segments]))
86         return (segments,
87                 self._zip_translate_params(zip_info,
88                                            segments[0].info['raw info']))
89
90     def _zip_info(self, zipfile):
91         with Closing(zipfile.open('header.properties')) as f:
92             info = self._parse_params(f.readlines())
93             return info
94
95     def _zip_segment(self, zipfile, path, info, zip_info, index):
96         prop_file = zipfile.open(os.path.join(
97                 'segments', str(index), 'segment-header.properties'))
98         prop = self._parse_params(prop_file.readlines())
99         prop_file.close()
100         expected_shape = (int(prop['force-segment-header']['num-points']),)
101         channels = []
102         for chan in prop['channels']['list']:
103             chan_info = prop['channel'][chan]
104             channels.append(self._zip_channel(zipfile, index, chan, chan_info))
105             if channels[-1].shape != expected_shape:
106                     raise NotImplementedError(
107                         'Channel %d:%s in %s has strange shape %s != %s'
108                         % (index, chan, zipfile.path,
109                            channels[-1].shape, expected_shape))
110         d = curve.Data(
111             shape=(len(channels[0]), len(channels)),
112             dtype=channels[0].dtype,
113             info=self._zip_translate_segment_params(prop))
114         for i,chan in enumerate(channels):
115             d[:,i] = chan
116         return self._zip_scale_segment(d, path, info)
117
118     def _zip_channel(self, zipfile, segment_index, channel_name, chan_info):
119         f = zipfile.open(os.path.join(
120                 'segments', str(segment_index),
121                 chan_info['data']['file']['name']), 'r')
122         assert chan_info['data']['file']['format'] == 'raw', \
123             'Non-raw data format:\n%s' % pprint.pformat(chan_info)
124         assert chan_info['data']['type'] == 'float-data', \
125             'Non-float data format:\n%s' % pprint.pformat(chan_info)
126         data = numpy.frombuffer(
127             buffer(f.read()),
128             dtype=numpy.dtype(numpy.float32).newbyteorder('>'))
129         # '>' (big endian) byte order.
130         # From version 0.3 of JPKForceSpec.txt in the "Binary data" section:
131         #    All forms of raw data are stored in chronological order
132         #    (the order in which they were collected), and the
133         #    individual values are stored in network byte order
134         #    (big-endian). The data type used to store the data is
135         #    specified by the "channel.*.data.type" property, and is
136         #    either short (2 bytes per value), integer (4 bytes), or
137         #    float (4 bytes, IEEE format).
138         f.close()
139         return data
140
141     def _zip_translate_params(self, params, chan_info):
142         info = {
143             'raw info':params,
144             #'time':self._time_from_TODO(raw_info[]),
145             }
146         force_unit = chan_info['channel']['vDeflection']['conversion-set']['conversion']['force']['scaling']['unit']['unit']
147         assert force_unit == 'N', force_unit
148         force_base = chan_info['channel']['vDeflection']['conversion-set']['conversion']['force']['base-calibration-slot']
149         assert force_base == 'distance', force_base
150         dist_unit = chan_info['channel']['vDeflection']['conversion-set']['conversion']['distance']['scaling']['unit']['unit']
151         assert dist_unit == 'm', dist_unit
152         force_mult = float(
153             chan_info['channel']['vDeflection']['conversion-set']['conversion']['force']['scaling']['multiplier'])
154         info['spring constant (N/m)'] = force_mult
155         return info
156
157     def _zip_translate_segment_params(self, params):
158         info = {
159             'raw info':params,
160             'columns':list(params['channels']['list']),
161             'name':params['force-segment-header']['name']['name'],
162             }
163         if info['name'] in ['extend-spm', 'retract-spm', 'pause-at-end-spm']:
164             info['name'] = info['name'][:-len('-spm')]
165             if info['name'] == 'extend':
166                 info['name'] = 'approach'
167         else:
168             raise NotImplementedError(
169                 'Unrecognized segment type %s' % info['name'])
170         return info
171
172     def _zip_scale_segment(self, segment, path, info):
173         data = curve.Data(
174             shape=segment.shape,
175             dtype=segment.dtype,
176             info={})
177         data[:,:] = segment
178         segment.info['raw data'] = data
179
180         # raw column indices
181         channels = segment.info['raw info']['channels']['list']
182         z_col = channels.index('height')
183         d_col = channels.index('vDeflection')
184         
185         segment = self._zip_scale_channel(
186             segment, z_col, 'calibrated', path, info)
187         segment = self._zip_scale_channel(
188             segment, d_col, 'distance', path, info)
189
190         assert segment.info['columns'][z_col] == 'height (m)', \
191             segment.info['columns'][z_col]
192         assert segment.info['columns'][d_col] == 'vDeflection (m)', \
193             segment.info['columns'][d_col]
194
195         # scaled column indices same as raw column indices,
196         # because columns is a copy of channels.list
197         segment.info['columns'][z_col] = 'z piezo (m)'
198         segment.info['columns'][d_col] = 'deflection (m)'
199         return segment
200
201     def _zip_scale_channel(self, segment, channel, conversion, path, info):
202         channel_name = segment.info['raw info']['channels']['list'][channel]
203         conversion_set = segment.info['raw info']['channel'][channel_name]['conversion-set']
204         conversion_info = conversion_set['conversion'][conversion]
205         if conversion_info['base-calibration-slot'] \
206                 != conversion_set['conversions']['base']:
207             # Our conversion is stacked on a previous conversion.  Do
208             # the previous conversion first.
209             segment = self._zip_scale_channel(
210                 segment, channel, conversion_info['base-calibration-slot'],
211                 path=path, info=info)
212         if conversion_info['type'] == 'file':
213             # Michael Haggerty at JPK points out that the conversion
214             # information stored in the external file is reproduced in
215             # the force curve file.  So there is no need to actually
216             # read `conversion_info['file']`.  In fact, the data there
217             # may have changed with future calibrations, while the
218             # information stored directly in conversion_info retains
219             # the calibration information as it was when the experiment
220             # was performed.
221             pass  # Fall through to 'simple' conversion processing.
222         else:
223             assert conversion_info['type'] == 'simple', conversion_info['type']
224         assert conversion_info['scaling']['type'] == 'linear', \
225             conversion_info['scaling']['type']
226         assert conversion_info['scaling']['style'] == 'offsetmultiplier', \
227             conversion_info['scaling']['style']
228         multiplier = float(conversion_info['scaling']['multiplier'])
229         offset = float(conversion_info['scaling']['offset'])
230         unit = conversion_info['scaling']['unit']['unit']
231         segment[:,channel] = segment[:,channel] * multiplier + offset
232         segment.info['columns'][channel] = '%s (%s)' % (channel_name, unit)
233         return segment
234
235     def _parse_params(self, lines):
236         info = {}
237         for line in lines:
238             line = line.strip()
239             if line.startswith('#'):
240                 continue
241             else:
242                 # e.g.: force-segment-header.type=xy-position-segment-header
243                 fields = line.split('=', 1)
244                 assert len(fields) == 2, line
245                 setting = fields[0].split('.')
246                 sub_info = info  # drill down, e.g. info['force-s..']['type']
247                 for s in setting[:-1]:
248                     if s not in sub_info:
249                         sub_info[s] = {}
250                     sub_info = sub_info[s]
251                 if setting[-1] == 'list':  # split a space-delimited list
252                     sub_info[setting[-1]] = fields[1].split(' ')
253                 else:
254                     sub_info[setting[-1]] = fields[1]
255         return info
256
257     def _read_old(self, path, info):
258         raise NotImplementedError('No old-style JPK files were available for testing, please send us yours: %s' % path)