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
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
|
"""
Conversion routines for sequences.
"""
import ctypes
__all__ = ["CTypesView", "to_ctypes", "to_list", "to_tuple", "create_array",
"MemoryView"]
# Hack around an import error using relative import paths in Python 2.7
_ARRAY = __import__("array")
def to_tuple(dataseq):
"""Converts a ``ctypes`` array to a tuple."""
return tuple(dataseq)
def to_list(dataseq):
"""Converts a ``ctypes`` array to a list."""
return list(dataseq)
def to_ctypes(dataseq, dtype, mcount=0):
"""Converts an arbitrary sequence to a ``ctypes`` array of a given type.
Args:
dataseq: A sequence to convert to a ``ctypes`` array.
dtype (class): The ``ctypes`` data type to use for the array (e.g.
``ctypes.c_uint8``).
mcount (int, optional): The number of elements in ``dataseq``. If
not specified, this will be inferred automatically.
Returns:
tuple: A tuple in the form ``(valset, count)``, where ``valset`` is
the converted ``ctypes`` array and ``count`` is the number of
elements in the array.
Raises:
TypeError: if any elements in the passed sequence do not
match the specified type.
"""
if mcount > 0:
count = mcount
else:
count = len(dataseq)
if isinstance(dataseq, CTypesView):
itemsize = ctypes.sizeof(dtype)
if itemsize == 1:
dataseq = dataseq.to_bytes()
elif itemsize == 2:
dataseq = dataseq.to_uint16()
elif itemsize == 4:
dataseq = dataseq.to_uint32()
elif itemsize == 8:
dataseq = dataseq.to_uint64()
else:
raise TypeError("unsupported data type for the passed CTypesView")
valset = (count * dtype)(*dataseq)
return valset, count
def create_array(obj, itemsize):
"""Creates an :obj:`array.array` copy of a given object.
Args:
obj: The object from which the array will be created.
itemsize: The size (in bytes) of each item in the given object.
Returns:
:obj:`array.array`: The array created from the object.
"""
if itemsize == 1:
return _ARRAY.array("B", obj)
elif itemsize == 2:
return _ARRAY.array("H", obj)
elif itemsize == 4:
return _ARRAY.array("I", obj)
elif itemsize == 8:
return _ARRAY.array("d", obj)
else:
raise TypeError("unsupported data type")
class CTypesView(object):
"""A proxy for accessing byte-wise ``ctypes`` data types.
This class provides read-write access for arbitrary ``ctypes`` objects
that are iterable. In case the object does not provide a :func:`buffer`
interface for direct access, a ``CTypesView`` can copy the object's contents
into an internal buffer from which data can be retrieved.
Depending on the item type stored in the iterable object, you might
need to manually specify the object's item size (in bytes). Additionally,
you may need to manually specify the number of items in the iterable if
it does not properly return its length using the ``len`` function.
For certain types, such as the bytearray, the original object must not be
reassigned after being encapsuled and used in ctypes bindings if the
contents are not copied.
Args:
obj: An arbitrary iterable ``ctypes`` object to access.
itemsize (int, optional): The size (in bytes) of each item of the
iterable object. Defaults to ``1``.
docopy (bool, optional): If True, the view will be created for a copy
of the object and will not modify the original ``ctypes`` instance.
Defalts to False (original object will be modified).
objsize (int, optional): The number of items in the iterable object.
If not specified, the number of items will try to be inferred
automatically. Defaults to ``None`` (automatic inferrence).
"""
def __init__(self, obj, itemsize=1, docopy=False, objsize=None):
self._obj = obj
self._isshared = True
self._view = None
self._itemsize = itemsize
self._create_view(itemsize, bool(docopy), objsize)
def _create_view(self, itemsize, docopy, objsize):
"""Creates the view on the specified object."""
self._isshared = not docopy
bsize = 0
if objsize is None:
bsize = len(self._obj) * itemsize
else:
bsize = objsize * itemsize
if docopy:
self._obj = create_array(self._obj, itemsize)
try:
self._view = (ctypes.c_ubyte * bsize).from_buffer(self._obj)
except AttributeError:
# pypy ctypes arrays do not feature a from_buffer() method.
self._isshared = False
# in case we requested a copy earlier, we do not need to recreate
# the array, since we have it already. In any other case, create
# a byte array.
if not docopy:
# Try to determine the itemsize again for array
# instances, just in case the user assumed it to work.
if isinstance(self._obj, _ARRAY.array):
itemsize = self._obj.itemsize
bsize = len(self._obj) * itemsize
self._obj = create_array(self._obj, itemsize)
self._view = (ctypes.c_ubyte * bsize)(*bytearray(self._obj))
def __repr__(self):
dtype = type(self._obj).__name__
bsize = self.bytesize
return "CTypesView(type=%s, bytesize=%d, shared=%s)" % (dtype, bsize,
self.is_shared)
def __len__(self):
"""Returns the length of the underlying object in bytes."""
return self.bytesize
def to_bytes(self):
"""Casts the object to an array of bytes.
If the view was created with ``docopy = False`` (the default), the
returned object provides direct read-write access to the object data.
Otherwise, the returned object will only modify a copy of the data.
Returns:
:obj:`ctypes._Pointer`: A pointer to a :class:`ctypes.c_uint8`
array.
"""
castval = ctypes.POINTER(ctypes.c_uint8 * self.bytesize)
return ctypes.cast(self.view, castval).contents
def to_uint16(self):
"""Casts the object to an array of 16-bit unsigned ints.
If the view was created with ``docopy = False`` (the default), the
returned object provides direct read-write access to the object data.
Otherwise, the returned object will only modify a copy of the data.
Returns:
:obj:`ctypes._Pointer`: A pointer to a :class:`ctypes.c_uint16`
array.
"""
castval = ctypes.POINTER(ctypes.c_uint16 * (self.bytesize // 2))
return ctypes.cast(self.view, castval).contents
def to_uint32(self):
"""Casts the object to an array of 32-bit unsigned ints.
If the view was created with ``docopy = False`` (the default), the
returned object provides direct read-write access to the object data.
Otherwise, the returned object will only modify a copy of the data.
Returns:
:obj:`ctypes._Pointer`: A pointer to a :class:`ctypes.c_uint32`
array.
"""
castval = ctypes.POINTER(ctypes.c_uint32 * (self.bytesize // 4))
return ctypes.cast(self.view, castval).contents
def to_uint64(self):
"""Casts the object to an array of 64-bit unsigned ints.
If the view was created with ``docopy = False`` (the default), the
returned object provides direct read-write access to the object data.
Otherwise, the returned object will only modify a copy of the data.
Returns:
:obj:`ctypes._Pointer`: A pointer to a :class:`ctypes.c_uint64`
array.
"""
castval = ctypes.POINTER(ctypes.c_uint64 * (self.bytesize // 8))
return ctypes.cast(self.view, castval).contents
@property
def bytesize(self):
"""int: The size (in bytes) of the underlying object."""
return ctypes.sizeof(self.view)
@property
def view(self):
"""Provides a read/write-aware ``ctypes`` view of the encapsuled
object.
"""
return self._view
@property
def is_shared(self):
"""bool: Whether any modifications to the view will also modify the
underlying object.
"""
return self._isshared
@property
def object(self):
"""The underlying object."""
return self._obj
class MemoryView(object):
"""A class that provides read-write access to indexable ``ctypes`` objects.
.. note: ``MemoryView`` makes heavy use of recursion for multi-dimensional
access, making it slow for many use-cases. For better performance, the
``numpy`` library can be used for fast access to many array-like data
types, but it may support fewer types of arbitrary objects than the
``MemoryView`` class.
Args:
source: An arbitrary indexable ``ctypes`` object to access.
itemsize (int): The size (in bytes) of each item of the indexable
object.
strides (tuple): The length of each dimension in the object (e.g.
``(4, 3)`` for a 4 x 3 2D array).
getfunc (function, optional): Deprecated, do not use.
setfunc (function, optional): Deprecated, do not use.
srcsize (int, optional): The size (in bytes) of the input object.
If ``len(source)`` returns the size of the object in bytes this will
be inferred automatically, otherwise it must be manually specified.
"""
def __init__(self, source, itemsize, strides, getfunc=None, setfunc=None,
srcsize=None):
self._source = source
self._itemsize = itemsize
self._strides = strides
self._srcsize = srcsize or len(source)
self._offset = 0
self._getfunc = getfunc or self._getbytes
self._setfunc = setfunc or self._setbytes
tsum = 1
for v in strides:
tsum *= v
if tsum > self._srcsize:
raise ValueError("strides exceed the accesible source size")
#if itemsize > strides[-1]:
# raise ValueError("itemsize exceeds the accessible stride length")
def _getbytes(self, start, end):
"""Gets the bytes within the range of start:end."""
return self._source[start:end]
def _setbytes(self, start, end, value):
"""Gets the bytes within the range of start:end to the passed
value.
"""
self._source[start:end] = value
def __len__(self):
"""The length of the MemoryView over the current dimension
(amount of items for the current dimension).
"""
return self.strides[0]
def __repr__(self):
retval = "["
slen = self.strides[0]
for dim in range(slen - 1):
retval += "%s, " % self[dim]
retval += str(self[slen - 1])
retval += "]"
return retval
def __getitem__(self, index):
"""Returns the item at the specified index."""
if type(index) is slice:
raise IndexError("slicing is not supported")
else:
if index >= len(self):
raise IndexError("index '%d'is out of bounds for '%d'" %
(index, len(self)))
if self.ndim == 1:
offset = self._offset + index * self.itemsize
return self._getfunc(offset, offset + self.itemsize)
else:
advance = self.itemsize
for b in self.strides[1:]:
advance *= b
offset = self._offset + advance * index
view = MemoryView(self._source, self.itemsize,
self.strides[1:], self._getfunc,
self._setfunc, self._srcsize)
view._offset = offset
return view
def __setitem__(self, index, value):
"""Sets the item at index to the specified value."""
if type(index) is slice:
raise IndexError("slicing is not supported")
else:
if index >= len(self):
raise IndexError("index '%d'is out of bounds for '%d'" %
(index, len(self)))
offset = self._offset + index * self.itemsize
if self.ndim == 1:
self._setfunc(offset, offset + self.itemsize, value)
else:
advance = self.itemsize
for b in self.strides[1:]:
advance *= b
offset = self._offset + advance * index
view = MemoryView(self._source, self.itemsize,
self.strides[1:], self._getfunc,
self._setfunc, self._srcsize)
view._offset = offset
if len(value) != len(view):
raise ValueError("value does not match the view strides")
for x in range(len(view)):
view[x] = value[x]
@property
def size(self):
"""int: The size (in bytes) of the underlying source object."""
return self._srcsize
@property
def strides(self):
"""tuple: The length of each dimension the MemoryView."""
return self._strides
@property
def itemsize(self):
"""int: The size (in bytes) of a single item in the object."""
return self._itemsize
@property
def ndim(self):
"""int: The number of dimensions of the MemoryView."""
return len(self.strides)
@property
def source(self):
"""The underlying data source."""
return self._source
|
