Python numpy 模块，complex128() 实例源码
我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用numpy.complex128()。
def setWFPrec(self):
'''
Set wavefunction coefficients precision:
TAG = 45200: single precision complex, np.complex64, or complex(qs)
TAG = 45210: double precision complex, np.complex128, or complex(q)
'''
if self._rtag == 45200:
return np.complex64
elif self._rtag == 45210:
return np.complex128
elif self._rtag == 53300:
raise ValueError("VASP5 WAVECAR format, not implemented yet")
elif self._rtag == 53310:
raise ValueError("VASP5 WAVECAR format with double precision "
+"coefficients, not implemented yet")
else:
raise ValueError("Invalid TAG values: {}".format(self._rtag))
def logscale_spec(spec, sr=44100, factor=20.):
timebins, freqbins = np.shape(spec)
scale = np.linspace(0, 1, freqbins) ** factor
scale *= (freqbins-1)/max(scale)
scale = np.unique(np.round(scale))
# create spectrogram with new freq bins
newspec = np.complex128(np.zeros([timebins, len(scale)]))
for i in range(0, len(scale)):
if i == len(scale)-1:
newspec[:,i] = np.sum(spec[:,scale[i]:], axis=1)
else:
newspec[:,i] = np.sum(spec[:,scale[i]:scale[i+1]], axis=1)
# list center freq of bins
allfreqs = np.abs(np.fft.fftfreq(freqbins*2, 1./sr)[:freqbins+1])
freqs = []
for i in range(0, len(scale)):
if i == len(scale)-1:
freqs += [np.mean(allfreqs[scale[i]:])]
else:
freqs += [np.mean(allfreqs[scale[i]:scale[i+1]])]
return newspec, freqs
def logscale_spec(spec, sr=44100, factor=20.):
timebins, freqbins = np.shape(spec)
scale = np.linspace(0, 1, freqbins) ** factor
scale *= (freqbins-1)/max(scale)
scale = np.unique(np.round(scale))
# create spectrogram with new freq bins
newspec = np.complex128(np.zeros([timebins, len(scale)]))
for i in range(0, len(scale)):
if i == len(scale)-1:
newspec[:,i] = np.sum(spec[:,scale[i]:], axis=1)
else:
newspec[:,i] = np.sum(spec[:,scale[i]:scale[i+1]], axis=1)
# list center freq of bins
allfreqs = np.abs(np.fft.fftfreq(freqbins*2, 1./sr)[:freqbins+1])
freqs = []
for i in range(0, len(scale)):
if i == len(scale)-1:
freqs += [np.mean(allfreqs[scale[i]:])]
else:
freqs += [np.mean(allfreqs[scale[i]:scale[i+1]])]
return newspec, freqs
def guarantee_array(variable):
''' Guarantees that a varaible is a numpy ndarray and supports -, *, +, and other operators
Args:
variable (`number` or `numpy.ndarray`): variable to coalesce
Returns:
(type). Which supports * / and other operations with arrays
'''
if type(variable) in [float, np.ndarray, np.int32, np.int64, np.float32, np.float64, np.complex64, np.complex128]:
return variable
elif type(variable) is int:
return float(variable)
elif type(variable) is list:
return np.asarray(variable)
else:
raise ValueError(f'variable is of invalid type {type(variable)}')
def test_prod(self):
ba = [1, 2, 10, 11, 6, 5, 4]
ba2 = [[1, 2, 3, 4], [5, 6, 7, 9], [10, 3, 4, 5]]
for ctype in [np.int16, np.uint16, np.int32, np.uint32,
np.float32, np.float64, np.complex64, np.complex128]:
a = np.array(ba, ctype)
a2 = np.array(ba2, ctype)
if ctype in ['1', 'b']:
self.assertRaises(ArithmeticError, a.prod)
self.assertRaises(ArithmeticError, a2.prod, axis=1)
else:
assert_equal(a.prod(axis=0), 26400)
assert_array_equal(a2.prod(axis=0),
np.array([50, 36, 84, 180], ctype))
assert_array_equal(a2.prod(axis=-1),
np.array([24, 1890, 600], ctype))
def test_basic(self):
dts = [np.bool, np.int16, np.int32, np.int64, np.double, np.complex128,
np.longdouble, np.clongdouble]
for dt in dts:
c = np.ones(53, dtype=np.bool)
assert_equal(np.where( c, dt(0), dt(1)), dt(0))
assert_equal(np.where(~c, dt(0), dt(1)), dt(1))
assert_equal(np.where(True, dt(0), dt(1)), dt(0))
assert_equal(np.where(False, dt(0), dt(1)), dt(1))
d = np.ones_like(c).astype(dt)
e = np.zeros_like(d)
r = d.astype(dt)
c[7] = False
r[7] = e[7]
assert_equal(np.where(c, e, e), e)
assert_equal(np.where(c, d, e), r)
assert_equal(np.where(c, d, e[0]), r)
assert_equal(np.where(c, d[0], e), r)
assert_equal(np.where(c[::2], d[::2], e[::2]), r[::2])
assert_equal(np.where(c[1::2], d[1::2], e[1::2]), r[1::2])
assert_equal(np.where(c[::3], d[::3], e[::3]), r[::3])
assert_equal(np.where(c[1::3], d[1::3], e[1::3]), r[1::3])
assert_equal(np.where(c[::-2], d[::-2], e[::-2]), r[::-2])
assert_equal(np.where(c[::-3], d[::-3], e[::-3]), r[::-3])
assert_equal(np.where(c[1::-3], d[1::-3], e[1::-3]), r[1::-3])
def test_sum_complex(self):
for dt in (np.complex64, np.complex128, np.clongdouble):
for v in (0, 1, 2, 7, 8, 9, 15, 16, 19, 127,
128, 1024, 1235):
tgt = dt(v * (v + 1) / 2) - dt((v * (v + 1) / 2) * 1j)
d = np.empty(v, dtype=dt)
d.real = np.arange(1, v + 1)
d.imag = -np.arange(1, v + 1)
assert_almost_equal(np.sum(d), tgt)
assert_almost_equal(np.sum(d[::-1]), tgt)
d = np.ones(500, dtype=dt) + 1j
assert_almost_equal(np.sum(d[::2]), 250. + 250j)
assert_almost_equal(np.sum(d[1::2]), 250. + 250j)
assert_almost_equal(np.sum(d[::3]), 167. + 167j)
assert_almost_equal(np.sum(d[1::3]), 167. + 167j)
assert_almost_equal(np.sum(d[::-2]), 250. + 250j)
assert_almost_equal(np.sum(d[-1::-2]), 250. + 250j)
assert_almost_equal(np.sum(d[::-3]), 167. + 167j)
assert_almost_equal(np.sum(d[-1::-3]), 167. + 167j)
# sum with first reduction entry != 0
d = np.ones((1,), dtype=dt) + 1j
d += d
assert_almost_equal(d, 2. + 2j)
def test_zero_division(self):
with np.errstate(all="ignore"):
for t in [np.complex64, np.complex128]:
a = t(0.0)
b = t(1.0)
assert_(np.isinf(b/a))
b = t(complex(np.inf, np.inf))
assert_(np.isinf(b/a))
b = t(complex(np.inf, np.nan))
assert_(np.isinf(b/a))
b = t(complex(np.nan, np.inf))
assert_(np.isinf(b/a))
b = t(complex(np.nan, np.nan))
assert_(np.isnan(b/a))
b = t(0.)
assert_(np.isnan(b/a))
def test_basic(self):
ba = [1, 2, 10, 11, 6, 5, 4]
ba2 = [[1, 2, 3, 4], [5, 6, 7, 9], [10, 3, 4, 5]]
for ctype in [np.int8, np.uint8, np.int16, np.uint16, np.int32,
np.uint32, np.float32, np.float64, np.complex64, np.complex128]:
a = np.array(ba, ctype)
a2 = np.array(ba2, ctype)
tgt = np.array([1, 3, 13, 24, 30, 35, 39], ctype)
assert_array_equal(np.cumsum(a, axis=0), tgt)
tgt = np.array(
[[1, 2, 3, 4], [6, 8, 10, 13], [16, 11, 14, 18]], ctype)
assert_array_equal(np.cumsum(a2, axis=0), tgt)
tgt = np.array(
[[1, 3, 6, 10], [5, 11, 18, 27], [10, 13, 17, 22]], ctype)
assert_array_equal(np.cumsum(a2, axis=1), tgt)
def test_basic(self):
ba = [1, 2, 10, 11, 6, 5, 4]
ba2 = [[1, 2, 3, 4], [5, 6, 7, 9], [10, 3, 4, 5]]
for ctype in [np.int16, np.uint16, np.int32, np.uint32,
np.float32, np.float64, np.complex64, np.complex128]:
a = np.array(ba, ctype)
a2 = np.array(ba2, ctype)
if ctype in ['1', 'b']:
self.assertRaises(ArithmeticError, np.prod, a)
self.assertRaises(ArithmeticError, np.prod, a2, 1)
else:
assert_equal(a.prod(axis=0), 26400)
assert_array_equal(a2.prod(axis=0),
np.array([50, 36, 84, 180], ctype))
assert_array_equal(a2.prod(axis=-1),
np.array([24, 1890, 600], ctype))
def test_basic(self):
ba = [1, 2, 10, 11, 6, 5, 4]
ba2 = [[1, 2, 3, 4], [5, 6, 7, 9], [10, 3, 4, 5]]
for ctype in [np.int16, np.uint16, np.int32, np.uint32,
np.float32, np.float64, np.complex64, np.complex128]:
a = np.array(ba, ctype)
a2 = np.array(ba2, ctype)
if ctype in ['1', 'b']:
self.assertRaises(ArithmeticError, np.cumprod, a)
self.assertRaises(ArithmeticError, np.cumprod, a2, 1)
self.assertRaises(ArithmeticError, np.cumprod, a)
else:
assert_array_equal(np.cumprod(a, axis=-1),
np.array([1, 2, 20, 220,
1320, 6600, 26400], ctype))
assert_array_equal(np.cumprod(a2, axis=0),
np.array([[1, 2, 3, 4],
[5, 12, 21, 36],
[50, 36, 84, 180]], ctype))
assert_array_equal(np.cumprod(a2, axis=-1),
np.array([[1, 2, 6, 24],
[5, 30, 210, 1890],
[10, 30, 120, 600]], ctype))
def test_basic(self):
A = np.arange(100).reshape(10, 10)
mA = matrix(A)
mB = mA.copy()
O = np.ones((10, 10), np.float64) * 0.1
mB = mB + O
assert_(mB.dtype.type == np.float64)
assert_(np.all(mA != mB))
assert_(np.all(mB == mA+0.1))
mC = mA.copy()
O = np.ones((10, 10), np.complex128)
mC = mC * O
assert_(mC.dtype.type == np.complex128)
assert_(np.all(mA != mB))
def for_all_dtypes_combination(names=('dtyes',),
no_float16=False, no_bool=False, full=None,
no_complex=False):
"""Decorator that checks the fixture with a product set of all dtypes.
Args:
names(list of str): Argument names to which dtypes are passed.
no_float16(bool): If ``True``, ``numpy.float16`` is
omitted from candidate dtypes.
no_bool(bool): If ``True``, ``numpy.bool_`` is
omitted from candidate dtypes.
full(bool): If ``True``, then all combinations of dtypes
will be tested.
Otherwise, the subset of combinations will be tested
(see description in :func:`cupy.testing.for_dtypes_combination`).
no_complex(bool): If, True, ``numpy.complex64`` and
``numpy.complex128`` are omitted from candidate dtypes.
.. seealso:: :func:`cupy.testing.for_dtypes_combination`
"""
types = _make_all_dtypes(no_float16, no_bool, no_complex)
return for_dtypes_combination(types, names, full)
def _resizeFilter(self,size):
'''
Resize the filter
'''
if size == self.filter.shape:
return self.filter
if not self.filter_cache.has_key(size):
filter = np.fft.ifft2(self.filter)
w,h = size
fw,fh = filter.shape
tmp = np.zeros((w,h), np.complex128) #TODO: check this
w = min(w,fw)
h = min(h,fh)
tmp[ :w/2, :h/2] = filter[ :w/2, :h/2]
tmp[ :w/2,-h/2:] = filter[ :w/2,-h/2:]
tmp[-w/2:,-h/2:] = filter[-w/2:,-h/2:]
tmp[-w/2:, :h/2] = filter[-w/2:, :h/2]
self.filter_cache[size] = np.fft.fft2(tmp)
return self.filter_cache[size]
def resizeFilter(filter,size):
filter = np.fft.ifft2(filter)
w,h = size
fw,fh = filter.shape
tmp = np.zeros((w,h), np.complex128) #TODO: check this
w = min(w,fw)
h = min(h,fh)
tmp[ :w/2, :h/2] = filter[ :w/2, :h/2]
tmp[ :w/2,-h/2:] = filter[ :w/2,-h/2:]
tmp[-w/2:,-h/2:] = filter[-w/2:,-h/2:]
tmp[-w/2:, :h/2] = filter[-w/2:, :h/2]
return np.fft.fft2(tmp)
def test_writehdf5_complex(self):
exp = SweptTestExperiment()
exp.is_complex = True
clear_test_data()
wr = WriteToHDF5("test_writehdf5_complex.h5")
edges = [(exp.voltage, wr.sink)]
exp.set_graph(edges)
exp.voltage.descriptor.dtype = np.complex128
exp.update_descriptors()
exp.add_sweep(exp.field, np.linspace(0,100.0,4))
exp.add_sweep(exp.freq, np.linspace(0,10.0,3))
exp.run_sweeps()
self.assertTrue(os.path.exists("test_writehdf5_complex-0000.h5"))
with h5py.File("test_writehdf5_complex-0000.h5", 'r') as f:
self.assertTrue(0.0 not in f['main/data/voltage'])
self.assertTrue(np.sum(f['main/data/field']) == 5*3*np.sum(np.linspace(0,100.0,4)) )
self.assertTrue(np.sum(f['main/data/freq']) == 5*4*np.sum(np.linspace(0,10.0,3)) )
self.assertTrue(np.sum(f['main/data/samples']) == 3*4*np.sum(np.linspace(0,4,5)) )
self.assertTrue(f['main/data'].attrs['time_val'] == 0)
self.assertTrue(f['main/data'].attrs['unit_freq'] == "Hz")
os.remove("test_writehdf5_complex-0000.h5")
def get_descriptor(self, source_instr_settings, channel_settings):
# Create a channel
channel = X6Channel(channel_settings)
descrip = DataStreamDescriptor()
# If it's an integrated stream, then the time axis has already been eliminated.
# Otherswise, add the time axis.
if channel_settings['stream_type'] == 'Raw':
samp_time = 4.0e-9
descrip.add_axis(DataAxis("time", samp_time*np.arange(source_instr_settings['record_length']//4)))
descrip.dtype = np.float64
elif channel_settings['stream_type'] == 'Demodulated':
samp_time = 32.0e-9
descrip.add_axis(DataAxis("time", samp_time*np.arange(source_instr_settings['record_length']//32)))
descrip.dtype = np.complex128
else: # Integrated
descrip.dtype = np.complex128
return channel, descrip
def update_descriptors(self):
logger.debug("Updating Plotter %s descriptors based on input descriptor %s", self.name, self.sink.descriptor)
self.stream = self.sink.input_streams[0]
self.descriptor = self.sink.descriptor
try:
self.time_pts = self.descriptor.axes[self.descriptor.axis_num("time")].points
self.record_length = len(self.time_pts)
except ValueError:
raise ValueError("Single shot filter sink does not appear to have a time axis!")
self.num_segments = len(self.sink.descriptor.axes[self.descriptor.axis_num("segment")].points)
self.ground_data = np.zeros((self.record_length, self.num_segments//2), dtype=np.complex)
self.excited_data = np.zeros((self.record_length, self.num_segments//2), dtype=np.complex)
output_descriptor = DataStreamDescriptor()
output_descriptor.axes = [_ for _ in self.descriptor.axes if type(_) is SweepAxis]
output_descriptor._exp_src = self.sink.descriptor._exp_src
output_descriptor.dtype = np.complex128
for os in self.fidelity.output_streams:
os.set_descriptor(output_descriptor)
os.end_connector.update_descriptors()
def logscale_spec(spec, sr=44100, factor=20.):
""" scale frequency axis logarithmically """
timebins, freqbins = np.shape(spec)
scale = np.linspace(0, 1, freqbins) ** factor
scale *= (freqbins - 1) / max(scale)
scale = np.unique(np.round(scale))
# create spectrogram with new freq bins
newspec = np.complex128(np.zeros([timebins, len(scale)]))
for i in range(0, len(scale)):
if i == len(scale) - 1:
newspec[:, i] = np.sum(spec[:, scale[i]:], axis=1)
else:
newspec[:, i] = np.sum(spec[:, scale[i]:scale[i + 1]], axis=1)
# list center freq of bins
allfreqs = np.abs(np.fft.fftfreq(freqbins * 2 - 1, 1. / sr)[:freqbins + 1])
freqs = []
for i in range(0, len(scale)):
if i == len(scale) - 1:
freqs += [np.mean(allfreqs[scale[i]:])]
else:
freqs += [np.mean(allfreqs[scale[i]:scale[i + 1]])]
return newspec, freqs
def _nadata_total(self): #only one read operation--> twice faster than _nadata
attempt = 0
a, b, c, d = self._reads(0x140, 4)
while not ((a >> 31 == 0) and (b >> 31 == 0)
and (c >> 31 == 0) and (d >> 31 == 0)):
a, b, c, d = self._reads(0x140, 4)
self._logger.warning('NA data not ready yet. Try again!')
attempt += 1
if attempt > 10:
raise Exception("Trying to recover NA data while averaging is not finished. Some setting is wrong. ")
sum = np.complex128(self._to_pyint(int(a) + (int(b) << 31), bitlength=62)) \
+ np.complex128(self._to_pyint(int(c) + (int(d) << 31), bitlength=62)) * 1j
return sum
# the implementation of network_analyzer is not identical to na_trace
# there are still many bugs in it, which is why we will keep this function
# in the gui
def cont2discrete(r, p, c, dt=8e-9):
"""
Transforms residue and pole from continuous to discrete time
Parameters
----------
r: residues
p: poles
dt: sampling time
Returns
-------
(r, p) with the transformation applied
"""
r = np.asarray(r, dtype=np.complex128) * dt
p = np.exp(np.asarray(p, dtype=np.complex128) * dt)
return r, p, c
def discrete2cont(r, p, c, dt=8e-9):
"""
Transforms residues and poles from discrete time to continuous
Parameters
----------
r: residues
p: poles
c: constant term to be carried along
dt: sampling time (s)
Returns
-------
r, p with the transformation applied
"""
r = np.array(r, dtype=np.complex128) / dt
p = np.log(np.array(p, dtype=np.complex128)) / dt
return r, p, c
def tf_partialfraction(self, frequencies=None):
"""
Returns the transfer function just before the partial fraction
expansion for frequencies.
Parameters
----------
sys: (poles, zeros, k)
dt: sampling time
continuous: if True, returns the transfer function in continuous
time domain, if False converts to discrete one
method: method for scipy.signal.cont2discrete
alpha: alpha for above method (see scipy documentation)
Returns
-------
np.array(..., dtype=np.complex)
"""
# this code is more or less a direct copy of get_coeff()
if frequencies is None:
frequencies = self.frequencies
frequencies = np.asarray(frequencies, dtype=np.complex128)
r, p, c = self.rp_continuous
h = freqs_rp(r, p, c, frequencies*2*np.pi)
return h * self.tf_inputfilter(frequencies=frequencies)
def main(self, input_ring):
"""Initiate the writing to filename
@param[in] input_rings First ring in this list will be used for
data
@param[out] output_rings This list of rings won't be used."""
span_generator = self.iterate_ring_read(input_ring)
data_accumulate = None
for span in span_generator:
if self.nbit < 8:
unpacked_data = unpack(span.data_view(self.dtype), self.nbit)
else:
if self.dtype == np.complex64:
unpacked_data = span.data_view(self.dtype).view(np.float32)
elif self.dtype == np.complex128:
unpacked_data = span.data_view(self.dtype).view(np.float64)
else:
unpacked_data = span.data_view(self.dtype)
if data_accumulate is not None:
data_accumulate = np.concatenate((data_accumulate, unpacked_data[0]))
else:
data_accumulate = unpacked_data[0]
text_file = open(self.filename, 'a')
np.savetxt(text_file, data_accumulate.reshape((1, -1)))
def numpy2bifrost(dtype):
if dtype == np.int8: return _bf.BF_DTYPE_I8
elif dtype == np.int16: return _bf.BF_DTYPE_I16
elif dtype == np.int32: return _bf.BF_DTYPE_I32
elif dtype == np.uint8: return _bf.BF_DTYPE_U8
elif dtype == np.uint16: return _bf.BF_DTYPE_U16
elif dtype == np.uint32: return _bf.BF_DTYPE_U32
elif dtype == np.float16: return _bf.BF_DTYPE_F16
elif dtype == np.float32: return _bf.BF_DTYPE_F32
elif dtype == np.float64: return _bf.BF_DTYPE_F64
elif dtype == np.float128: return _bf.BF_DTYPE_F128
elif dtype == ci8: return _bf.BF_DTYPE_CI8
elif dtype == ci16: return _bf.BF_DTYPE_CI16
elif dtype == ci32: return _bf.BF_DTYPE_CI32
elif dtype == cf16: return _bf.BF_DTYPE_CF16
elif dtype == np.complex64: return _bf.BF_DTYPE_CF32
elif dtype == np.complex128: return _bf.BF_DTYPE_CF64
elif dtype == np.complex256: return _bf.BF_DTYPE_CF128
else: raise ValueError("Unsupported dtype: " + str(dtype))
def numpy2string(dtype):
if dtype == np.int8: return 'i8'
elif dtype == np.int16: return 'i16'
elif dtype == np.int32: return 'i32'
elif dtype == np.int64: return 'i64'
elif dtype == np.uint8: return 'u8'
elif dtype == np.uint16: return 'u16'
elif dtype == np.uint32: return 'u32'
elif dtype == np.uint64: return 'u64'
elif dtype == np.float16: return 'f16'
elif dtype == np.float32: return 'f32'
elif dtype == np.float64: return 'f64'
elif dtype == np.float128: return 'f128'
elif dtype == np.complex64: return 'cf32'
elif dtype == np.complex128: return 'cf64'
elif dtype == np.complex256: return 'cf128'
else: raise TypeError("Unsupported dtype: " + str(dtype))
def test_header_output(self):
"""Output a header for a ring explicitly"""
def generate_array_and_header():
"""Output the desired header of an array"""
header = {'dtype': 'complex128', 'nbit': 128}
yield np.array([1, 2, 3, 4]), header
def assert_expectation(array):
"Assert that the array has a complex datatype"
np.testing.assert_almost_equal(array, [1, 2, 3, 4])
self.assertEqual(array.dtype, np.dtype('complex128'))
self.occurences += 1
blocks = []
blocks.append((
NumpySourceBlock(generate_array_and_header, grab_headers=True),
{'out_1': 0}))
blocks.append((NumpyBlock(assert_expectation, outputs=0), {'in_1': 0}))
Pipeline(blocks).main()
self.assertEqual(self.occurences, 1)
def test_autocov():
x = np.random.randn(4096).astype(np.complex128) # 2x extra speedup from casting correct type
M=5
tic = time()
C= compute_autocovariance(x,M)
tocpy = time()-tic
#%%
tic = time()
Cc = S['c'].covariance.autocov(x,M)
tocfortcmpl = time()-tic
tic = time()
Cr = S['r'].covariance.autocov(x.real,M)
tocfortreal = time() - tic
#print(f'autocovariance: python {tocpy:.6f} sec fortran {tocfortcmpl:.6f} sec')
print('autocovariance: Fortran faster than Python by factor:',tocpy/tocfortcmpl)
np.testing.assert_allclose(C,Cc,rtol=1)
np.testing.assert_allclose(C.real,Cr,rtol=1)
def default(self, obj):
# convert dates and numpy objects in a json serializable format
if isinstance(obj, datetime):
return obj.strftime('%Y-%m-%dT%H:%M:%SZ')
elif isinstance(obj, date):
return obj.strftime('%Y-%m-%d')
elif type(obj) in (np.int_, np.intc, np.intp, np.int8, np.int16,
np.int32, np.int64, np.uint8, np.uint16,
np.uint32, np.uint64):
return int(obj)
elif type(obj) in (np.bool_,):
return bool(obj)
elif type(obj) in (np.float_, np.float16, np.float32, np.float64,
np.complex_, np.complex64, np.complex128):
return float(obj)
# Let the base class default method raise the TypeError
return json.JSONEncoder.default(self, obj)
def test_prod(self):
ba = [1, 2, 10, 11, 6, 5, 4]
ba2 = [[1, 2, 3, 4], [5, 6, 7, 9], [10, 3, 4, 5]]
for ctype in [np.int16, np.uint16, np.int32, np.uint32,
np.float32, np.float64, np.complex64, np.complex128]:
a = np.array(ba, ctype)
a2 = np.array(ba2, ctype)
if ctype in ['1', 'b']:
self.assertRaises(ArithmeticError, a.prod)
self.assertRaises(ArithmeticError, a2.prod, axis=1)
else:
assert_equal(a.prod(axis=0), 26400)
assert_array_equal(a2.prod(axis=0),
np.array([50, 36, 84, 180], ctype))
assert_array_equal(a2.prod(axis=-1),
np.array([24, 1890, 600], ctype))
def test_basic(self):
dts = [np.bool, np.int16, np.int32, np.int64, np.double, np.complex128,
np.longdouble, np.clongdouble]
for dt in dts:
c = np.ones(53, dtype=np.bool)
assert_equal(np.where( c, dt(0), dt(1)), dt(0))
assert_equal(np.where(~c, dt(0), dt(1)), dt(1))
assert_equal(np.where(True, dt(0), dt(1)), dt(0))
assert_equal(np.where(False, dt(0), dt(1)), dt(1))
d = np.ones_like(c).astype(dt)
e = np.zeros_like(d)
r = d.astype(dt)
c[7] = False
r[7] = e[7]
assert_equal(np.where(c, e, e), e)
assert_equal(np.where(c, d, e), r)
assert_equal(np.where(c, d, e[0]), r)
assert_equal(np.where(c, d[0], e), r)
assert_equal(np.where(c[::2], d[::2], e[::2]), r[::2])
assert_equal(np.where(c[1::2], d[1::2], e[1::2]), r[1::2])
assert_equal(np.where(c[::3], d[::3], e[::3]), r[::3])
assert_equal(np.where(c[1::3], d[1::3], e[1::3]), r[1::3])
assert_equal(np.where(c[::-2], d[::-2], e[::-2]), r[::-2])
assert_equal(np.where(c[::-3], d[::-3], e[::-3]), r[::-3])
assert_equal(np.where(c[1::-3], d[1::-3], e[1::-3]), r[1::-3])
def test_sum_complex(self):
for dt in (np.complex64, np.complex128, np.clongdouble):
for v in (0, 1, 2, 7, 8, 9, 15, 16, 19, 127,
128, 1024, 1235):
tgt = dt(v * (v + 1) / 2) - dt((v * (v + 1) / 2) * 1j)
d = np.empty(v, dtype=dt)
d.real = np.arange(1, v + 1)
d.imag = -np.arange(1, v + 1)
assert_almost_equal(np.sum(d), tgt)
assert_almost_equal(np.sum(d[::-1]), tgt)
d = np.ones(500, dtype=dt) + 1j
assert_almost_equal(np.sum(d[::2]), 250. + 250j)
assert_almost_equal(np.sum(d[1::2]), 250. + 250j)
assert_almost_equal(np.sum(d[::3]), 167. + 167j)
assert_almost_equal(np.sum(d[1::3]), 167. + 167j)
assert_almost_equal(np.sum(d[::-2]), 250. + 250j)
assert_almost_equal(np.sum(d[-1::-2]), 250. + 250j)
assert_almost_equal(np.sum(d[::-3]), 167. + 167j)
assert_almost_equal(np.sum(d[-1::-3]), 167. + 167j)
# sum with first reduction entry != 0
d = np.ones((1,), dtype=dt) + 1j
d += d
assert_almost_equal(d, 2. + 2j)
def test_zero_division(self):
with np.errstate(all="ignore"):
for t in [np.complex64, np.complex128]:
a = t(0.0)
b = t(1.0)
assert_(np.isinf(b/a))
b = t(complex(np.inf, np.inf))
assert_(np.isinf(b/a))
b = t(complex(np.inf, np.nan))
assert_(np.isinf(b/a))
b = t(complex(np.nan, np.inf))
assert_(np.isinf(b/a))
b = t(complex(np.nan, np.nan))
assert_(np.isnan(b/a))
b = t(0.)
assert_(np.isnan(b/a))
def test_basic(self):
ba = [1, 2, 10, 11, 6, 5, 4]
ba2 = [[1, 2, 3, 4], [5, 6, 7, 9], [10, 3, 4, 5]]
for ctype in [np.int8, np.uint8, np.int16, np.uint16, np.int32,
np.uint32, np.float32, np.float64, np.complex64, np.complex128]:
a = np.array(ba, ctype)
a2 = np.array(ba2, ctype)
tgt = np.array([1, 3, 13, 24, 30, 35, 39], ctype)
assert_array_equal(np.cumsum(a, axis=0), tgt)
tgt = np.array(
[[1, 2, 3, 4], [6, 8, 10, 13], [16, 11, 14, 18]], ctype)
assert_array_equal(np.cumsum(a2, axis=0), tgt)
tgt = np.array(
[[1, 3, 6, 10], [5, 11, 18, 27], [10, 13, 17, 22]], ctype)
assert_array_equal(np.cumsum(a2, axis=1), tgt)
def test_basic(self):
ba = [1, 2, 10, 11, 6, 5, 4]
ba2 = [[1, 2, 3, 4], [5, 6, 7, 9], [10, 3, 4, 5]]
for ctype in [np.int16, np.uint16, np.int32, np.uint32,
np.float32, np.float64, np.complex64, np.complex128]:
a = np.array(ba, ctype)
a2 = np.array(ba2, ctype)
if ctype in ['1', 'b']:
self.assertRaises(ArithmeticError, np.prod, a)
self.assertRaises(ArithmeticError, np.prod, a2, 1)
else:
assert_equal(a.prod(axis=0), 26400)
assert_array_equal(a2.prod(axis=0),
np.array([50, 36, 84, 180], ctype))
assert_array_equal(a2.prod(axis=-1),
np.array([24, 1890, 600], ctype))
def test_basic(self):
ba = [1, 2, 10, 11, 6, 5, 4]
ba2 = [[1, 2, 3, 4], [5, 6, 7, 9], [10, 3, 4, 5]]
for ctype in [np.int16, np.uint16, np.int32, np.uint32,
np.float32, np.float64, np.complex64, np.complex128]:
a = np.array(ba, ctype)
a2 = np.array(ba2, ctype)
if ctype in ['1', 'b']:
self.assertRaises(ArithmeticError, np.cumprod, a)
self.assertRaises(ArithmeticError, np.cumprod, a2, 1)
self.assertRaises(ArithmeticError, np.cumprod, a)
else:
assert_array_equal(np.cumprod(a, axis=-1),
np.array([1, 2, 20, 220,
1320, 6600, 26400], ctype))
assert_array_equal(np.cumprod(a2, axis=0),
np.array([[1, 2, 3, 4],
[5, 12, 21, 36],
[50, 36, 84, 180]], ctype))
assert_array_equal(np.cumprod(a2, axis=-1),
np.array([[1, 2, 6, 24],
[5, 30, 210, 1890],
[10, 30, 120, 600]], ctype))
def test_basic(self):
A = np.arange(100).reshape(10, 10)
mA = matrix(A)
mB = mA.copy()
O = np.ones((10, 10), np.float64) * 0.1
mB = mB + O
assert_(mB.dtype.type == np.float64)
assert_(np.all(mA != mB))
assert_(np.all(mB == mA+0.1))
mC = mA.copy()
O = np.ones((10, 10), np.complex128)
mC = mC * O
assert_(mC.dtype.type == np.complex128)
assert_(np.all(mA != mB))
def logscale_spec(spec, sr=44100, factor=20.):
timebins, freqbins = np.shape(spec)
scale = np.linspace(0, 1, freqbins) ** factor
scale *= (freqbins-1)/max(scale)
scale = np.unique(np.round(scale))
# create spectrogram with new freq bins
newspec = np.complex128(np.zeros([timebins, len(scale)]))
for i in range(0, len(scale)):
if i == len(scale)-1:
newspec[:,i] = np.sum(spec[:,int(scale[i]):], axis=1)
else:
newspec[:,i] = np.sum(spec[:,int(scale[i]):int(scale[i+1])], axis=1)
# list center freq of bins
allfreqs = np.abs(np.fft.fftfreq(freqbins*2, 1./sr)[:freqbins+1])
freqs = []
for i in range(0, len(scale)):
if i == len(scale)-1:
freqs += [np.mean(allfreqs[int(scale[i]):])]
else:
freqs += [np.mean(allfreqs[int(scale[i]):int(scale[i+1])])]
return newspec, freqs
def one_particle(sp, chli, chlo, Es=None):
"""
Single particle
Parameters
----------
s : Setup
Scattering setup
chi : int
Input channel
cho : int
Output channel
Es : ndarray
Input energies
"""
Es, T1 = tmatrix.one_particle(sp, chli, chlo, Es)
S1 = np.zeros(T1.shape, dtype=np.complex128)
if chli == chlo:
S1 += 1
S1 -= 2 * 1j * np.pi * T1
return Es, S1
def compare_g2_fock_state_to_g2_coherent_state():
N, U, = 2, 0
gs = (.2, .1)
model = scattering.Model(
omegas=[0]*N,
links=[(0, 1, 1)],
U=[2*U]*N)
channels = []
channels.append(scattering.Channel(site=0, strength=gs[0]))
channels.append(scattering.Channel(site=N-1, strength=gs[1]))
setup = scattering.Setup(model, channels)
Es = np.linspace(-3, 12, 1024)
dE = 0
g2f = np.zeros(Es.shape, dtype=np.complex128)
g2c = np.zeros(Es.shape, dtype=np.complex128)
for i, E in enumerate(Es):
g2s[i], _, _ = g2.fock_state(setup, (0, 0), (1, 1), E, dE)
def test_noninteracting_dimer_eigenenergies():
"""Test"""
m = scat.Model([0] * 2, [[0, 1, 1.1]], [0] * 2)
channels = [
scat.Channel(site=0, strength=1),
scat.Channel(site=1, strength=1),
]
sp = scat.Setup(m, channels)
E1, _, _ = sp.eigenbasis(1)
E2, _, _ = sp.eigenbasis(2)
E12 = np.zeros((len(E2), ), dtype=np.complex128)
for i in xrange(len(E1)):
for j in xrange(len(E1)):
E12[i + j] = E1[i] + E1[j]
E12 = E12[np.argsort(np.real(E12))]
E2 = E2[np.argsort(np.real(E2))]
assert np.allcose(E2, E12), \
'Non-interacting dimer, single particle energies do not coincide with the two-particle energies.'
def create(i, state0, basis0, basis1):
"""
Create a boson on site <i>
Parameters
----------
i : int
Site index
state0 : ndarray
Initial state
basis0 : list
Initial basis
basis1 : list
Final basis
"""
mbasis = np.copy(basis0.vs)
mbasis[:, i] += 1
state1 = np.zeros((basis1.len,), dtype=np.complex128)
index1 = basis1.index(mbasis)
state1[index1] = state0 * np.sqrt(mbasis[:, i])
return state1
def uniform(size, low=None, high=None, dtype=numpy.float32):
if dtype is numpy.float32 or dtype is numpy.float64:
if low is None:
low = -numpy.sqrt(6. / sum(size))
if high is None:
high = numpy.sqrt(6. / sum(size))
return numpy.asarray(
numpy.random.uniform(
low=low,
high=high,
size=size
),
dtype=dtype)
elif dtype is numpy.complex64:
return (uniform(size, low=low, high=high, dtype=numpy.float32) +
1j * uniform(size, low=low, high=high,
dtype=numpy.float32)).astype(numpy.complex64)
elif dtype is numpy.complex128:
return (uniform(size, low=low, high=high, dtype=numpy.float64) +
1j * uniform(size, low=low, high=high,
dtype=numpy.float64)).astype(numpy.complex128)
else:
raise ValueError('Requested dtype not available.')
def _ensure_numeric(x):
if isinstance(x, np.ndarray):
if is_integer_dtype(x) or is_bool_dtype(x):
x = x.astype(np.float64)
elif is_object_dtype(x):
try:
x = x.astype(np.complex128)
except:
x = x.astype(np.float64)
else:
if not np.any(x.imag):
x = x.real
elif not (is_float(x) or is_integer(x) or is_complex(x)):
try:
x = float(x)
except Exception:
try:
x = complex(x)
except Exception:
raise TypeError('Could not convert %s to numeric' % str(x))
return x
# NA-friendly array comparisons
def test_complex_fixed(self):
df = DataFrame(np.random.rand(4, 5).astype(np.complex64),
index=list('abcd'),
columns=list('ABCDE'))
with ensure_clean_path(self.path) as path:
df.to_hdf(path, 'df')
reread = read_hdf(path, 'df')
assert_frame_equal(df, reread)
df = DataFrame(np.random.rand(4, 5).astype(np.complex128),
index=list('abcd'),
columns=list('ABCDE'))
with ensure_clean_path(self.path) as path:
df.to_hdf(path, 'df')
reread = read_hdf(path, 'df')
assert_frame_equal(df, reread)
def test_complex_table(self):
df = DataFrame(np.random.rand(4, 5).astype(np.complex64),
index=list('abcd'),
columns=list('ABCDE'))
with ensure_clean_path(self.path) as path:
df.to_hdf(path, 'df', format='table')
reread = read_hdf(path, 'df')
assert_frame_equal(df, reread)
df = DataFrame(np.random.rand(4, 5).astype(np.complex128),
index=list('abcd'),
columns=list('ABCDE'))
with ensure_clean_path(self.path) as path:
df.to_hdf(path, 'df', format='table', mode='w')
reread = read_hdf(path, 'df')
assert_frame_equal(df, reread)
def test_basic(self):
dts = [np.bool, np.int16, np.int32, np.int64, np.double, np.complex128,
np.longdouble, np.clongdouble]
for dt in dts:
c = np.ones(53, dtype=np.bool)
assert_equal(np.where( c, dt(0), dt(1)), dt(0))
assert_equal(np.where(~c, dt(0), dt(1)), dt(1))
assert_equal(np.where(True, dt(0), dt(1)), dt(0))
assert_equal(np.where(False, dt(0), dt(1)), dt(1))
d = np.ones_like(c).astype(dt)
e = np.zeros_like(d)
r = d.astype(dt)
c[7] = False
r[7] = e[7]
assert_equal(np.where(c, e, e), e)
assert_equal(np.where(c, d, e), r)
assert_equal(np.where(c, d, e[0]), r)
assert_equal(np.where(c, d[0], e), r)
assert_equal(np.where(c[::2], d[::2], e[::2]), r[::2])
assert_equal(np.where(c[1::2], d[1::2], e[1::2]), r[1::2])
assert_equal(np.where(c[::3], d[::3], e[::3]), r[::3])
assert_equal(np.where(c[1::3], d[1::3], e[1::3]), r[1::3])
assert_equal(np.where(c[::-2], d[::-2], e[::-2]), r[::-2])
assert_equal(np.where(c[::-3], d[::-3], e[::-3]), r[::-3])
assert_equal(np.where(c[1::-3], d[1::-3], e[1::-3]), r[1::-3])
def test_zero_division(self):
with np.errstate(all="ignore"):
for t in [np.complex64, np.complex128]:
a = t(0.0)
b = t(1.0)
assert_(np.isinf(b/a))
b = t(complex(np.inf, np.inf))
assert_(np.isinf(b/a))
b = t(complex(np.inf, np.nan))
assert_(np.isinf(b/a))
b = t(complex(np.nan, np.inf))
assert_(np.isinf(b/a))
b = t(complex(np.nan, np.nan))
assert_(np.isnan(b/a))
b = t(0.)
assert_(np.isnan(b/a))
def test_basic(self):
A = np.arange(100).reshape(10, 10)
mA = matrix(A)
mB = mA.copy()
O = np.ones((10, 10), np.float64) * 0.1
mB = mB + O
assert_(mB.dtype.type == np.float64)
assert_(np.all(mA != mB))
assert_(np.all(mB == mA+0.1))
mC = mA.copy()
O = np.ones((10, 10), np.complex128)
mC = mC * O
assert_(mC.dtype.type == np.complex128)
assert_(np.all(mA != mB))
def readBandCoeff(self, ispin=1, ikpt=1, iband=1, norm=False):
'''
Read the planewave coefficients of specified KS states.
'''
self.checkIndex(ispin, ikpt, iband)
rec = self.whereRec(ispin, ikpt, iband)
self._wfc.seek(rec * self._recl)
nplw = self._nplws[ikpt - 1]
dump = np.fromfile(self._wfc, dtype=self._WFPrec, count=nplw)
cg = np.asarray(dump, dtype=np.complex128)
if norm:
cg /= np.linalg.norm(cg)
return cg