def test_int_pointers(self):
from ctypes import c_short, c_uint, c_int, c_long, POINTER, pointer
LPINT = POINTER(c_int)
## p = pointer(c_int(42))
## x = LPINT.from_param(p)
x = LPINT.from_param(pointer(c_int(42)))
self.assertEqual(x.contents.value, 42)
self.assertEqual(LPINT(c_int(42)).contents.value, 42)
self.assertEqual(LPINT.from_param(None), None)
if c_int != c_long:
self.assertRaises(TypeError, LPINT.from_param, pointer(c_long(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_uint(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_short(42)))
python类c_short()的实例源码
def dataSources():
"""Return a list with [name, descrition]"""
dsn = create_buffer(1024)
desc = create_buffer(1024)
dsn_len = c_short()
desc_len = c_short()
dsn_list = {}
try:
lock.acquire()
if shared_env_h is None:
AllocateEnv()
finally:
lock.release()
while 1:
ret = ODBC_API.SQLDataSources(shared_env_h, SQL_FETCH_NEXT, \
dsn, len(dsn), ADDR(dsn_len), desc, len(desc), ADDR(desc_len))
if ret == SQL_NO_DATA_FOUND:
break
elif not ret in (SQL_SUCCESS, SQL_SUCCESS_WITH_INFO):
ctrl_err(SQL_HANDLE_ENV, shared_env_h, ret)
else:
dsn_list[dsn.value] = desc.value
return dsn_list
def dataSources():
"""Return a list with [name, descrition]"""
dsn = create_buffer(1024)
desc = create_buffer(1024)
dsn_len = c_short()
desc_len = c_short()
dsn_list = {}
try:
lock.acquire()
if shared_env_h is None:
AllocateEnv()
finally:
lock.release()
while 1:
ret = ODBC_API.SQLDataSources(shared_env_h, SQL_FETCH_NEXT, \
dsn, len(dsn), ADDR(dsn_len), desc, len(desc), ADDR(desc_len))
if ret == SQL_NO_DATA_FOUND:
break
elif not ret in (SQL_SUCCESS, SQL_SUCCESS_WITH_INFO):
ctrl_err(SQL_HANDLE_ENV, shared_env_h, ret)
else:
dsn_list[dsn.value] = desc.value
return dsn_list
def dataSources():
"""Return a list with [name, descrition]"""
dsn = create_buffer(1024)
desc = create_buffer(1024)
dsn_len = c_short()
desc_len = c_short()
dsn_list = {}
try:
lock.acquire()
if shared_env_h is None:
AllocateEnv()
finally:
lock.release()
while 1:
ret = ODBC_API.SQLDataSources(shared_env_h, SQL_FETCH_NEXT, \
dsn, len(dsn), ADDR(dsn_len), desc, len(desc), ADDR(desc_len))
if ret == SQL_NO_DATA_FOUND:
break
elif not ret in (SQL_SUCCESS, SQL_SUCCESS_WITH_INFO):
ctrl_err(SQL_HANDLE_ENV, shared_env_h, ret)
else:
dsn_list[dsn.value] = desc.value
return dsn_list
def test_int_pointers(self):
from ctypes import c_short, c_uint, c_int, c_long, POINTER, pointer
LPINT = POINTER(c_int)
## p = pointer(c_int(42))
## x = LPINT.from_param(p)
x = LPINT.from_param(pointer(c_int(42)))
self.assertEqual(x.contents.value, 42)
self.assertEqual(LPINT(c_int(42)).contents.value, 42)
self.assertEqual(LPINT.from_param(None), None)
if c_int != c_long:
self.assertRaises(TypeError, LPINT.from_param, pointer(c_long(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_uint(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_short(42)))
def test_array_pointers(self):
from ctypes import c_short, c_uint, c_int, c_long, POINTER
INTARRAY = c_int * 3
ia = INTARRAY()
self.assertEqual(len(ia), 3)
self.assertEqual([ia[i] for i in range(3)], [0, 0, 0])
# Pointers are only compatible with arrays containing items of
# the same type!
LPINT = POINTER(c_int)
LPINT.from_param((c_int*3)())
self.assertRaises(TypeError, LPINT.from_param, c_short*3)
self.assertRaises(TypeError, LPINT.from_param, c_long*3)
self.assertRaises(TypeError, LPINT.from_param, c_uint*3)
## def test_performance(self):
## check_perf()
def dataSources():
"""Return a list with [name, descrition]"""
dsn = create_buffer(1024)
desc = create_buffer(1024)
dsn_len = c_short()
desc_len = c_short()
dsn_list = {}
try:
lock.acquire()
if shared_env_h is None:
AllocateEnv()
finally:
lock.release()
while 1:
ret = ODBC_API.SQLDataSources(shared_env_h, SQL_FETCH_NEXT, \
dsn, len(dsn), ADDR(dsn_len), desc, len(desc), ADDR(desc_len))
if ret == SQL_NO_DATA_FOUND:
break
elif not ret in (SQL_SUCCESS, SQL_SUCCESS_WITH_INFO):
ctrl_err(SQL_HANDLE_ENV, shared_env_h, ret)
else:
dsn_list[dsn.value] = desc.value
return dsn_list
def test_int_pointers(self):
from ctypes import c_short, c_uint, c_int, c_long, POINTER, pointer
LPINT = POINTER(c_int)
## p = pointer(c_int(42))
## x = LPINT.from_param(p)
x = LPINT.from_param(pointer(c_int(42)))
self.assertEqual(x.contents.value, 42)
self.assertEqual(LPINT(c_int(42)).contents.value, 42)
self.assertEqual(LPINT.from_param(None), None)
if c_int != c_long:
self.assertRaises(TypeError, LPINT.from_param, pointer(c_long(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_uint(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_short(42)))
def test_int_pointers(self):
from ctypes import c_short, c_uint, c_int, c_long, POINTER, pointer
LPINT = POINTER(c_int)
## p = pointer(c_int(42))
## x = LPINT.from_param(p)
x = LPINT.from_param(pointer(c_int(42)))
self.assertEqual(x.contents.value, 42)
self.assertEqual(LPINT(c_int(42)).contents.value, 42)
self.assertEqual(LPINT.from_param(None), None)
if c_int != c_long:
self.assertRaises(TypeError, LPINT.from_param, pointer(c_long(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_uint(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_short(42)))
def test_int_pointers(self):
from ctypes import c_short, c_uint, c_int, c_long, POINTER, pointer
LPINT = POINTER(c_int)
## p = pointer(c_int(42))
## x = LPINT.from_param(p)
x = LPINT.from_param(pointer(c_int(42)))
self.assertEqual(x.contents.value, 42)
self.assertEqual(LPINT(c_int(42)).contents.value, 42)
self.assertEqual(LPINT.from_param(None), None)
if c_int != c_long:
self.assertRaises(TypeError, LPINT.from_param, pointer(c_long(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_uint(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_short(42)))
def dataSources():
"""Return a list with [name, descrition]"""
dsn = create_buffer(1024)
desc = create_buffer(1024)
dsn_len = c_short()
desc_len = c_short()
dsn_list = {}
try:
lock.acquire()
if shared_env_h is None:
AllocateEnv()
finally:
lock.release()
while 1:
ret = ODBC_API.SQLDataSources(shared_env_h, SQL_FETCH_NEXT, \
dsn, len(dsn), ADDR(dsn_len), desc, len(desc), ADDR(desc_len))
if ret == SQL_NO_DATA_FOUND:
break
elif not ret in (SQL_SUCCESS, SQL_SUCCESS_WITH_INFO):
ctrl_err(SQL_HANDLE_ENV, shared_env_h, ret)
else:
dsn_list[dsn.value] = desc.value
return dsn_list
def _generate_data(self, num_bytes, offset):
if self._bytes_per_sample == 1:
start = offset
samples = num_bytes
bias = 127
amplitude = 127
data = (ctypes.c_ubyte * samples)()
else:
start = offset >> 1
samples = num_bytes >> 1
bias = 0
amplitude = 32767
data = (ctypes.c_short * samples)()
step = self.frequency * (math.pi * 2) / self.audio_format.sample_rate
envelope = self._envelope_array
env_offset = offset // self._bytes_per_sample
for i in range(samples):
data[i] = int(math.sin(step * (i + start)) *
amplitude * envelope[i+env_offset] + bias)
return data
def _generate_data(self, num_bytes, offset):
# XXX TODO consider offset
if self._bytes_per_sample == 1:
samples = num_bytes
value = 127
maximum = 255
minimum = 0
data = (ctypes.c_ubyte * samples)()
else:
samples = num_bytes >> 1
value = 0
maximum = 32767
minimum = -32768
data = (ctypes.c_short * samples)()
step = (maximum - minimum) * self.frequency / self._sample_rate
envelope = self._envelope_array
env_offset = offset // self._bytes_per_sample
for i in range(samples):
value += step
if value > maximum:
value = minimum + (value % maximum)
data[i] = int(value * envelope[i+env_offset])
return data
def _generate_data(self, num_bytes, offset):
if self._bytes_per_sample == 1:
start = offset
samples = num_bytes
bias = 127
amplitude = 127
data = (ctypes.c_ubyte * samples)()
else:
start = offset >> 1
samples = num_bytes >> 1
bias = 0
amplitude = 32767
data = (ctypes.c_short * samples)()
self._advance(start)
ring_buffer = self.ring_buffer
decay = self.decay
for i in range(samples):
data[i] = int(ring_buffer[0] * amplitude + bias)
ring_buffer.append(decay * (ring_buffer[0] + ring_buffer[1]) / 2)
return data
def _generate_data(self, bytes, offset):
if self._bytes_per_sample == 1:
start = offset
samples = bytes
bias = 127
amplitude = 127
data = (ctypes.c_ubyte * samples)()
else:
start = offset >> 1
samples = bytes >> 1
bias = 0
amplitude = 32767
data = (ctypes.c_short * samples)()
step = self.frequency * (math.pi * 2) / self.audio_format.sample_rate
for i in range(samples):
data[i] = future_round(math.sin(step * (i + start)) * amplitude + bias)
return data
def _generate_data(self, bytes, offset):
# XXX TODO consider offset
if self._bytes_per_sample == 1:
samples = bytes
value = 127
max = 255
min = 0
data = (ctypes.c_ubyte * samples)()
else:
samples = bytes >> 1
value = 0
max = 32767
min = -32768
data = (ctypes.c_short * samples)()
step = (max - min) * 2 * self.frequency / self.audio_format.sample_rate
for i in range(samples):
value += step
if value > max:
value = max - (value - max)
step = -step
if value < min:
value = min - (value - min)
step = -step
data[i] = future_round(value)
return data
def _generate_data(self, bytes, offset):
# XXX TODO consider offset
if self._bytes_per_sample == 1:
samples = bytes
value = 0
amplitude = 255
data = (ctypes.c_ubyte * samples)()
else:
samples = bytes >> 1
value = -32768
amplitude = 65535
data = (ctypes.c_short * samples)()
period = self.audio_format.sample_rate / self.frequency / 2
count = 0
for i in range(samples):
count += 1
if count == period:
value = amplitude - value
count = 0
data[i] = future_round(value)
return data
def test_int_pointers(self):
from ctypes import c_short, c_uint, c_int, c_long, POINTER, pointer
LPINT = POINTER(c_int)
## p = pointer(c_int(42))
## x = LPINT.from_param(p)
x = LPINT.from_param(pointer(c_int(42)))
self.assertEqual(x.contents.value, 42)
self.assertEqual(LPINT(c_int(42)).contents.value, 42)
self.assertEqual(LPINT.from_param(None), None)
if c_int != c_long:
self.assertRaises(TypeError, LPINT.from_param, pointer(c_long(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_uint(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_short(42)))
def dataSources():
"""Return a list with [name, descrition]"""
dsn = create_buffer(1024)
desc = create_buffer(1024)
dsn_len = c_short()
desc_len = c_short()
dsn_list = {}
try:
lock.acquire()
if shared_env_h is None:
AllocateEnv()
finally:
lock.release()
while 1:
ret = ODBC_API.SQLDataSources(shared_env_h, SQL_FETCH_NEXT, \
dsn, len(dsn), ADDR(dsn_len), desc, len(desc), ADDR(desc_len))
if ret == SQL_NO_DATA_FOUND:
break
elif not ret in (SQL_SUCCESS, SQL_SUCCESS_WITH_INFO):
ctrl_err(SQL_HANDLE_ENV, shared_env_h, ret)
else:
dsn_list[dsn.value] = desc.value
return dsn_list
def test_int_pointers(self):
from ctypes import c_short, c_uint, c_int, c_long, POINTER, pointer
LPINT = POINTER(c_int)
## p = pointer(c_int(42))
## x = LPINT.from_param(p)
x = LPINT.from_param(pointer(c_int(42)))
self.assertEqual(x.contents.value, 42)
self.assertEqual(LPINT(c_int(42)).contents.value, 42)
self.assertEqual(LPINT.from_param(None), None)
if c_int != c_long:
self.assertRaises(TypeError, LPINT.from_param, pointer(c_long(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_uint(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_short(42)))
def test_array_pointers(self):
from ctypes import c_short, c_uint, c_int, c_long, POINTER
INTARRAY = c_int * 3
ia = INTARRAY()
self.assertEqual(len(ia), 3)
self.assertEqual([ia[i] for i in range(3)], [0, 0, 0])
# Pointers are only compatible with arrays containing items of
# the same type!
LPINT = POINTER(c_int)
LPINT.from_param((c_int*3)())
self.assertRaises(TypeError, LPINT.from_param, c_short*3)
self.assertRaises(TypeError, LPINT.from_param, c_long*3)
self.assertRaises(TypeError, LPINT.from_param, c_uint*3)
## def test_performance(self):
## check_perf()
def test_int_pointers(self):
from ctypes import c_short, c_uint, c_int, c_long, POINTER, pointer
LPINT = POINTER(c_int)
## p = pointer(c_int(42))
## x = LPINT.from_param(p)
x = LPINT.from_param(pointer(c_int(42)))
self.assertEqual(x.contents.value, 42)
self.assertEqual(LPINT(c_int(42)).contents.value, 42)
self.assertEqual(LPINT.from_param(None), None)
if c_int != c_long:
self.assertRaises(TypeError, LPINT.from_param, pointer(c_long(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_uint(42)))
self.assertRaises(TypeError, LPINT.from_param, pointer(c_short(42)))
def dataSources():
"""Return a list with [name, descrition]"""
dsn = create_buffer(1024)
desc = create_buffer(1024)
dsn_len = c_short()
desc_len = c_short()
dsn_list = {}
try:
lock.acquire()
if shared_env_h is None:
AllocateEnv()
finally:
lock.release()
while 1:
ret = ODBC_API.SQLDataSources(shared_env_h, SQL_FETCH_NEXT, \
dsn, len(dsn), ADDR(dsn_len), desc, len(desc), ADDR(desc_len))
if ret == SQL_NO_DATA_FOUND:
break
elif not ret in (SQL_SUCCESS, SQL_SUCCESS_WITH_INFO):
ctrl_err(SQL_HANDLE_ENV, shared_env_h, ret)
else:
dsn_list[dsn.value] = desc.value
return dsn_list
def getshort(data, index):
# return two bytes from data as a signed 16-bit value
return c_short((data[index + 1] << 8) + data[index]).value
def read_short(self, addr):
"""Read a ``SHORT`` at ``addr``"""
sizeof_short = sizeof(ctypes.c_short)
return struct.unpack("<H", self.read_memory(addr, sizeof_short))[0]
def test_byref_pointer(self):
# The from_param class method of POINTER(typ) classes accepts what is
# returned by byref(obj), it type(obj) == typ
from ctypes import c_short, c_uint, c_int, c_long, pointer, POINTER, byref
LPINT = POINTER(c_int)
LPINT.from_param(byref(c_int(42)))
self.assertRaises(TypeError, LPINT.from_param, byref(c_short(22)))
if c_int != c_long:
self.assertRaises(TypeError, LPINT.from_param, byref(c_long(22)))
self.assertRaises(TypeError, LPINT.from_param, byref(c_uint(22)))
def test_byref_pointerpointer(self):
# See above
from ctypes import c_short, c_uint, c_int, c_long, pointer, POINTER, byref
LPLPINT = POINTER(POINTER(c_int))
LPLPINT.from_param(byref(pointer(c_int(42))))
self.assertRaises(TypeError, LPLPINT.from_param, byref(pointer(c_short(22))))
if c_int != c_long:
self.assertRaises(TypeError, LPLPINT.from_param, byref(pointer(c_long(22))))
self.assertRaises(TypeError, LPLPINT.from_param, byref(pointer(c_uint(22))))
def test_array_pointers(self):
from ctypes import c_short, c_uint, c_int, c_long, POINTER
INTARRAY = c_int * 3
ia = INTARRAY()
self.assertEqual(len(ia), 3)
self.assertEqual([ia[i] for i in range(3)], [0, 0, 0])
# Pointers are only compatible with arrays containing items of
# the same type!
LPINT = POINTER(c_int)
LPINT.from_param((c_int*3)())
self.assertRaises(TypeError, LPINT.from_param, c_short*3)
self.assertRaises(TypeError, LPINT.from_param, c_long*3)
self.assertRaises(TypeError, LPINT.from_param, c_uint*3)
def _NumOfCols(self):
"""Get the number of cols"""
if not self.connection:
self.close()
NOC = c_short()
ret = SQLNumResultCols(self.stmt_h, ADDR(NOC))
if ret != SQL_SUCCESS:
check_success(self, ret)
return NOC.value
def _NumOfCols(self):
"""Get the number of cols"""
if not self.connection:
self.close()
NOC = c_short()
ret = SQLNumResultCols(self.stmt_h, ADDR(NOC))
if ret != SQL_SUCCESS:
check_success(self, ret)
return NOC.value
