/usr/bin/ns-convert is in python-neuroshare 0.9.2-1.
This file is owned by root:root, with mode 0o755.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 | #!/usr/bin/python
import os
import sys
import h5py
import neuroshare as ns
import numpy as np
import getopt
class ProgressIndicator(object):
def __init__(self, offset=0):
self._cur_value = offset
self._max_value = 0
def setup(self, max_value):
self._max_value = max_value
self.progress(self._max_value, 0)
def __add__(self, other):
self._cur_value += other
self.progress(self._max_value, self._cur_value)
return self
def progress(self, max_value, cur_value):
pass
class Converter(object):
def __init__(self, filepath, output=None, progress=None):
if not output:
(basefile, ext) = os.path.splitext(filepath)
output = "%s.hdf5" % basefile
nf = ns.File(filepath)
h5 = h5py.File(output, 'w')
self._nf = nf
self._h5 = h5
self._groups = {}
self.convert_map = {1: self.convert_event,
2: self.convert_analog,
3: self.convert_segment,
4: self.convert_neural}
if not progress:
progress = ProgressIndicator()
self._progress = progress
def get_group_for_type(self, entity_type):
name_map = {1: 'Event',
2: 'Analog',
3: 'Segment',
4: 'Neural'}
if entity_type not in self._groups:
name = name_map[entity_type]
group = self._h5.create_group(name)
self._groups[entity_type] = group
return self._groups[entity_type]
def convert(self):
progress = self._progress
progress.setup(len(self._nf.entities))
self.copy_metadata(self._h5, self._nf.metadata_raw)
for entity in self._nf.entities:
conv = self.convert_map[entity.entity_type]
conv(entity)
progress + 1
self._h5.close()
def convert_event(self, event):
dtype = self.dtype_by_event(event)
nitems = event.item_count
data = np.empty([nitems], dtype)
for n in xrange(0, event.item_count):
data[n] = event.get_data(n)
group = self.get_group_for_type(event.entity_type)
dset = group.create_dataset(event.label, data=data)
self.copy_metadata(dset, event.metadata_raw)
def convert_analog(self, analog):
(data, times, ic) = analog.get_data()
group = self.get_group_for_type(analog.entity_type)
d_t = np.vstack((times, data)).T
dset = group.create_dataset(analog.label, data=d_t)
self.copy_metadata(dset, analog.metadata_raw)
def convert_segment(self, segment):
if not segment.item_count:
return
group = self.get_group_for_type(segment.entity_type)
seg_group = group.create_group(segment.label)
self.copy_metadata(seg_group, segment.metadata_raw)
for index in xrange(0, segment.source_count):
source = segment.sources[index]
name = 'SourceInfo.%d.' % index
self.copy_metadata(seg_group, source.metadata_raw, prefix=name)
for index in xrange(0, segment.item_count):
(data, timestamp, samples, unit) = segment.get_data(index)
name = '%d - %f' % (index, timestamp)
dset = seg_group.create_dataset(name, data=data.T)
dset.attrs['Timestamp'] = timestamp
dset.attrs['Unit'] = unit
dset.attrs['Index'] = index
def convert_neural(self, neural):
data = neural.get_data()
group = self._groups[neural.entity_type]
name = "%d - %s" % (neural.id, neural.label)
dset = group.create_dataset(name, data=data)
self.copy_metadata(dset, neural.metadata_raw)
@classmethod
def copy_metadata(cls, target, metadata, prefix=None):
for (key, value) in metadata.iteritems():
if prefix is not None:
key = prefix + key
target.attrs[key] = value
@classmethod
def dtype_by_event(cls, event):
type_map = {ns.EventEntity.EVENT_TEXT : 'a',
ns.EventEntity.EVENT_CSV : 'a',
ns.EventEntity.EVENT_BYTE : 'b',
ns.EventEntity.EVENT_WORD : 'h',
ns.EventEntity.EVENT_DWORD : 'i'}
val_type = type_map[event.event_type]
if val_type == 'a':
val_type += str(event.max_data_length)
return np.dtype([('timestamp', 'd'), ('value', val_type)])
class ConsoleIndicator(ProgressIndicator):
def __init__(self):
super(ConsoleIndicator, self).__init__()
self._size = 60
self._last_msg = ""
def progress(self, max_value, cur_value):
size = self._size
prefix = "Converting"
x = int(size*cur_value/max_value)
msg = "%s [%s%s] %i/%i\r" % (prefix, "#"*x, "." * (size-x),
cur_value, max_value)
self._last_msg = msg
sys.stdout.write(msg)
sys.stdout.flush()
def cleanup(self):
sys.stdout.write('%s\r' % (' '*len(self._last_msg)))
sys.stdout.flush()
def main():
opts, rem = getopt.getopt(sys.argv[1:], 'o:', ['output=',
'version=',
])
output = None
for opt, arg in opts:
if opt in ("-o", "--output"):
output = arg
if len(rem) != 1:
print "Wrong number of arguments"
return -1
filename = rem[0]
ci = ConsoleIndicator()
converter = Converter(filename, output, progress=ci)
converter.convert()
ci.cleanup()
return 0
if __name__ == "__main__":
res = main()
sys.exit(res)
|